From b29e14e41391ea9dbfb48b15467df0c9681d118c Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 2 Sep 2021 13:41:55 +0200 Subject: [PATCH 01/22] 1st draft decoding feature processor --- processors/decodingfeatures/CMakeLists.txt | 3 + .../decodingfeatures/decodingfeatures.cpp | 78 +++++++++++++++++++ .../decodingfeatures/decodingfeatures.hpp | 49 ++++++++++++ processors/decodingfeatures/doc.yaml | 0 4 files changed, 130 insertions(+) create mode 100644 processors/decodingfeatures/CMakeLists.txt create mode 100644 processors/decodingfeatures/decodingfeatures.cpp create mode 100644 processors/decodingfeatures/decodingfeatures.hpp create mode 100644 processors/decodingfeatures/doc.yaml diff --git a/processors/decodingfeatures/CMakeLists.txt b/processors/decodingfeatures/CMakeLists.txt new file mode 100644 index 0000000..c29135a --- /dev/null +++ b/processors/decodingfeatures/CMakeLists.txt @@ -0,0 +1,3 @@ + +ADD_LIBRARY(decodingfeatures "decodingfeatures.cpp" ) +TARGET_LINK_LIBRARIES(decodingfeatures) diff --git a/processors/decodingfeatures/decodingfeatures.cpp b/processors/decodingfeatures/decodingfeatures.cpp new file mode 100644 index 0000000..48c936a --- /dev/null +++ b/processors/decodingfeatures/decodingfeatures.cpp @@ -0,0 +1,78 @@ +// --------------------------------------------------------------------- +// This file is part of falcon-core. +// +// Copyright (C) 2021-now Neuro-Electronics Research Flanders +// +// Falcon-server is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Falcon-server is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with falcon-core. If not, see . +// --------------------------------------------------------------------- + + +#include "decodingfeatures.hpp" + +DecodingFeatures::DecodingFeatures() : IProcessor(){ + add_option("channelmap", channelmap_def_, "Relation between channel number and depth.", true); +} + +void DecodingFeatures::Configure(const GlobalContext &context) { + + std::string f = context.resolve_path( + + channelmap_def_()["file"].as(), "filters"); + + //Load file in depths_ + channels_pos_ = new std::vector>(MAX_NCHANNELS); +} + +void DecodingFeatures::CreatePorts(){ + data_in_port_ = create_input_port( + "spikes", SpikeType::Capabilities(), + PortInPolicy(SlotRange(1))); + + + std::vector feature_labels({"hardware ts", "depthx", "depthy", "amplitude", "angle", "width"}); + data_out_port_ = create_output_port>( + "data", FeaturesType::Capabilities(feature_labels.size()), + FeaturesType::Parameters(feature_labels), PortOutPolicy(SlotRange(1))); +} + +void DecodingFeatures::Process(ProcessingContext &context) { + SpikeType::Data *data_in = nullptr; + FeaturesType::Data *data_out = nullptr; + + while (!context.terminated()) { + if (!data_in_port_->slot(0)->RetrieveData(data_in)) { + break; + } + + // claim output data buckets + data_out = data_out_port_->slot(0)->ClaimData(false); + + for(unsigned int i=0; in_detected_spikes(); i++){ + data_out->set_data_sample(i, "hardware ts", data_in->ts_detected_spikes(i)); + data_out->set_data_sample(i, "depthx", 0/* channels_pos_[data_in->channel_detected_spikes(i)][0]*/); + data_out->set_data_sample(i, "depthy", 0 /*channels_pos_[data_in->channel_detected_spikes(i)][1]*/); + data_out->set_data_sample(i, "amplitude", *(data_in->spike_amplitudes(i))); + data_out->set_data_sample(i, "angle", 0 /*TODO*/); + data_out->set_data_sample(i, "width", 0 /*TODO*/); + } + + // publish and release data + data_out_port_->slot(0)->PublishData(); + data_in_port_->slot(0)->ReleaseData(); + + } +} + + +REGISTERPROCESSOR(DecodingFeatures) diff --git a/processors/decodingfeatures/decodingfeatures.hpp b/processors/decodingfeatures/decodingfeatures.hpp new file mode 100644 index 0000000..f5eb572 --- /dev/null +++ b/processors/decodingfeatures/decodingfeatures.hpp @@ -0,0 +1,49 @@ +// --------------------------------------------------------------------- +// This file is part of falcon-core. +// +// Copyright (C) 2021-now Neuro-Electronics Research Flanders +// +// Falcon-server is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Falcon-server is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with falcon-core. If not, see . +// --------------------------------------------------------------------- + +#pragma once + +#include "iprocessor.hpp" +#include "featuresdata/featuresdata.hpp" +#include "spikedata/spikedata.hpp" + +class DecodingFeatures : public IProcessor { + // CONSTRUCTOR and OVERLOADED METHODS + public: + DecodingFeatures(); + void Configure(const GlobalContext &context); + void CreatePorts() override; + void Process(ProcessingContext &context) override; + + + // DATA PORTS + protected: + PortIn *data_in_port_; + PortOut> *data_out_port_; + const uint32_t MAX_NCHANNELS=384; + + +protected: + options::Value channelmap_def_{}; + + +private: + std::vector>* channels_pos_; + +}; diff --git a/processors/decodingfeatures/doc.yaml b/processors/decodingfeatures/doc.yaml new file mode 100644 index 0000000..e69de29 From dfe32f2056d599247b9056cd95da8e6b59fad642 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 7 Sep 2021 13:41:07 +0200 Subject: [PATCH 02/22] rename processor as spikefeatures --- processors/decodingfeatures/CMakeLists.txt | 3 --- processors/spikefeatures/CMakeLists.txt | 3 +++ .../{decodingfeatures => spikefeatures}/doc.yaml | 0 .../spikefeatures.cpp} | 12 ++++++------ .../spikefeatures.hpp} | 4 ++-- 5 files changed, 11 insertions(+), 11 deletions(-) delete mode 100644 processors/decodingfeatures/CMakeLists.txt create mode 100644 processors/spikefeatures/CMakeLists.txt rename processors/{decodingfeatures => spikefeatures}/doc.yaml (100%) rename processors/{decodingfeatures/decodingfeatures.cpp => spikefeatures/spikefeatures.cpp} (90%) rename processors/{decodingfeatures/decodingfeatures.hpp => spikefeatures/spikefeatures.hpp} (95%) diff --git a/processors/decodingfeatures/CMakeLists.txt b/processors/decodingfeatures/CMakeLists.txt deleted file mode 100644 index c29135a..0000000 --- a/processors/decodingfeatures/CMakeLists.txt +++ /dev/null @@ -1,3 +0,0 @@ - -ADD_LIBRARY(decodingfeatures "decodingfeatures.cpp" ) -TARGET_LINK_LIBRARIES(decodingfeatures) diff --git a/processors/spikefeatures/CMakeLists.txt b/processors/spikefeatures/CMakeLists.txt new file mode 100644 index 0000000..ca9483b --- /dev/null +++ b/processors/spikefeatures/CMakeLists.txt @@ -0,0 +1,3 @@ + +ADD_LIBRARY(spikefeatures "spikefeatures.cpp" ) +TARGET_LINK_LIBRARIES(spikefeatures) diff --git a/processors/decodingfeatures/doc.yaml b/processors/spikefeatures/doc.yaml similarity index 100% rename from processors/decodingfeatures/doc.yaml rename to processors/spikefeatures/doc.yaml diff --git a/processors/decodingfeatures/decodingfeatures.cpp b/processors/spikefeatures/spikefeatures.cpp similarity index 90% rename from processors/decodingfeatures/decodingfeatures.cpp rename to processors/spikefeatures/spikefeatures.cpp index 48c936a..2c436b4 100644 --- a/processors/decodingfeatures/decodingfeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -18,13 +18,13 @@ // --------------------------------------------------------------------- -#include "decodingfeatures.hpp" +#include "spikefeatures.hpp" -DecodingFeatures::DecodingFeatures() : IProcessor(){ +SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option("channelmap", channelmap_def_, "Relation between channel number and depth.", true); } -void DecodingFeatures::Configure(const GlobalContext &context) { +void SpikeFeatures::Configure(const GlobalContext &context) { std::string f = context.resolve_path( @@ -34,7 +34,7 @@ void DecodingFeatures::Configure(const GlobalContext &context) { channels_pos_ = new std::vector>(MAX_NCHANNELS); } -void DecodingFeatures::CreatePorts(){ +void SpikeFeatures::CreatePorts(){ data_in_port_ = create_input_port( "spikes", SpikeType::Capabilities(), PortInPolicy(SlotRange(1))); @@ -46,7 +46,7 @@ void DecodingFeatures::CreatePorts(){ FeaturesType::Parameters(feature_labels), PortOutPolicy(SlotRange(1))); } -void DecodingFeatures::Process(ProcessingContext &context) { +void SpikeFeatures::Process(ProcessingContext &context) { SpikeType::Data *data_in = nullptr; FeaturesType::Data *data_out = nullptr; @@ -75,4 +75,4 @@ void DecodingFeatures::Process(ProcessingContext &context) { } -REGISTERPROCESSOR(DecodingFeatures) +REGISTERPROCESSOR(SpikeFeatures) diff --git a/processors/decodingfeatures/decodingfeatures.hpp b/processors/spikefeatures/spikefeatures.hpp similarity index 95% rename from processors/decodingfeatures/decodingfeatures.hpp rename to processors/spikefeatures/spikefeatures.hpp index f5eb572..192b0a5 100644 --- a/processors/decodingfeatures/decodingfeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -23,10 +23,10 @@ #include "featuresdata/featuresdata.hpp" #include "spikedata/spikedata.hpp" -class DecodingFeatures : public IProcessor { +class SpikeFeatures : public IProcessor { // CONSTRUCTOR and OVERLOADED METHODS public: - DecodingFeatures(); + SpikeFeatures(); void Configure(const GlobalContext &context); void CreatePorts() override; void Process(ProcessingContext &context) override; From 37b1202789d2c308a3a81bd82f0056bed5704a55 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 7 Sep 2021 14:22:15 +0200 Subject: [PATCH 03/22] add spike waveform input and reading it --- processors/spikefeatures/spikefeatures.cpp | 33 +++++++++++++++++----- processors/spikefeatures/spikefeatures.hpp | 4 ++- 2 files changed, 29 insertions(+), 8 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 2c436b4..75cabce 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -35,10 +35,13 @@ void SpikeFeatures::Configure(const GlobalContext &context) { } void SpikeFeatures::CreatePorts(){ - data_in_port_ = create_input_port( - "spikes", SpikeType::Capabilities(), + data_in_port_ = create_input_port>( + "waveform", MultiChannelType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), PortInPolicy(SlotRange(1))); + data_spike_port_ = create_input_port( + "spikes", SpikeType::Capabilities(), + PortInPolicy(SlotRange(1))); std::vector feature_labels({"hardware ts", "depthx", "depthy", "amplitude", "angle", "width"}); data_out_port_ = create_output_port>( @@ -47,29 +50,45 @@ void SpikeFeatures::CreatePorts(){ } void SpikeFeatures::Process(ProcessingContext &context) { - SpikeType::Data *data_in = nullptr; + MultiChannelType::Data *data_waveform = nullptr; + SpikeType::Data *data_spikes = nullptr; + FeaturesType::Data *data_out = nullptr; while (!context.terminated()) { - if (!data_in_port_->slot(0)->RetrieveData(data_in)) { + if (!data_in_port_->slot(0)->RetrieveData(data_waveform) or !data_spike_port_->slot(0)->RetrieveData(data_spikes)) { // data should be both present break; } + // check if data are coherent + + if(data_waveform->nchannels() != data_spikes->n_detected_spikes()){ + throw ProcessingError("Received different numbers of waveform and spike detected.", name()); + } + // claim output data buckets data_out = data_out_port_->slot(0)->ClaimData(false); + auto spike_ts = data_waveform->sample_timestamps(); + for(unsigned int i=0; inchannels(); i++){ - for(unsigned int i=0; in_detected_spikes(); i++){ - data_out->set_data_sample(i, "hardware ts", data_in->ts_detected_spikes(i)); + data_out->set_data_sample(i, "hardware ts", data_spikes->ts_detected_spikes(i)); data_out->set_data_sample(i, "depthx", 0/* channels_pos_[data_in->channel_detected_spikes(i)][0]*/); data_out->set_data_sample(i, "depthy", 0 /*channels_pos_[data_in->channel_detected_spikes(i)][1]*/); - data_out->set_data_sample(i, "amplitude", *(data_in->spike_amplitudes(i))); + data_out->set_data_sample(i, "amplitude", *(data_spikes->spike_amplitudes(i))); + + + auto waveform = data_waveform->begin_channel(i); + // compute angle spike + // computer width spike data_out->set_data_sample(i, "angle", 0 /*TODO*/); data_out->set_data_sample(i, "width", 0 /*TODO*/); + } // publish and release data data_out_port_->slot(0)->PublishData(); data_in_port_->slot(0)->ReleaseData(); + data_spike_port_->slot(0)->ReleaseData(); } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 192b0a5..7942978 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -21,6 +21,7 @@ #include "iprocessor.hpp" #include "featuresdata/featuresdata.hpp" +#include "multichanneldata/multichanneldata.hpp" #include "spikedata/spikedata.hpp" class SpikeFeatures : public IProcessor { @@ -34,7 +35,8 @@ class SpikeFeatures : public IProcessor { // DATA PORTS protected: - PortIn *data_in_port_; + PortIn> *data_in_port_; + PortIn *data_spike_port_; PortOut> *data_out_port_; const uint32_t MAX_NCHANNELS=384; From f9a62146c2d9227dbb593da7ef0690a0401ea302 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 28 Oct 2021 11:54:15 +0200 Subject: [PATCH 04/22] WIP spike feature with spike detectin --- processors/spikefeatures/spikefeatures.cpp | 91 ++++++++++++---------- processors/spikefeatures/spikefeatures.hpp | 34 +++++--- 2 files changed, 73 insertions(+), 52 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 75cabce..d72f3f4 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -21,74 +21,83 @@ #include "spikefeatures.hpp" SpikeFeatures::SpikeFeatures() : IProcessor(){ - add_option("channelmap", channelmap_def_, "Relation between channel number and depth.", true); -} - -void SpikeFeatures::Configure(const GlobalContext &context) { - std::string f = context.resolve_path( + feature_labels_ = {"hardware ts", "depth","amplitude"}; - channelmap_def_()["file"].as(), "filters"); + add_option("channelmap", channel_pos_, "Relation between channel number and depth.", true); + add_option(THRESHOLD, initial_threshold_, + "Spike detection threshold in data units."); + add_option(PEAK_LIFETIME, initial_peak_lifetime_, + "Peak life time in samples"); - //Load file in depths_ - channels_pos_ = new std::vector>(MAX_NCHANNELS); } void SpikeFeatures::CreatePorts(){ - data_in_port_ = create_input_port>( - "waveform", MultiChannelType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), + data_in_port_ = create_input_port>( + "waveform", TimeSeriesType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), PortInPolicy(SlotRange(1))); - data_spike_port_ = create_input_port( - "spikes", SpikeType::Capabilities(), - PortInPolicy(SlotRange(1))); - std::vector feature_labels({"hardware ts", "depthx", "depthy", "amplitude", "angle", "width"}); - data_out_port_ = create_output_port>( - "data", FeaturesType::Capabilities(feature_labels.size()), - FeaturesType::Parameters(feature_labels), PortOutPolicy(SlotRange(1))); + data_out_port_ = create_output_port>( + "data", ColumnsType::Capabilities(feature_labels_.size()), + ColumnsType::Parameters(), PortOutPolicy(SlotRange(1))); + + threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, + Permission::WRITE); + + peak_lifetime_ = create_static_state(PEAK_LIFETIME, initial_peak_lifetime_(), + true, Permission::WRITE); } -void SpikeFeatures::Process(ProcessingContext &context) { - MultiChannelType::Data *data_waveform = nullptr; - SpikeType::Data *data_spikes = nullptr; +void SpikeFeatures::CompleteStreamInfo() { + data_out_port_->streaminfo(0).set_parameters(feature_labels_); +} - FeaturesType::Data *data_out = nullptr; +void SpikeFeatures::Prepare(GlobalContext &context) { + spike_detector_.reset(new dsp::algorithms::SpikeDetectorNeuropixels( + initial_threshold_(), initial_peak_lifetime_())); +} - while (!context.terminated()) { - if (!data_in_port_->slot(0)->RetrieveData(data_waveform) or !data_spike_port_->slot(0)->RetrieveData(data_spikes)) { // data should be both present - break; - } - // check if data are coherent +void SpikeFeatures::Process(ProcessingContext &context) { + TimeSeriesType::Data *data_in = nullptr; + ColumnsType::Data *data_out = nullptr; - if(data_waveform->nchannels() != data_spikes->n_detected_spikes()){ - throw ProcessingError("Received different numbers of waveform and spike detected.", name()); - } + while (!context.terminated()) { + spike_detector_->set_threshold(threshold_->get()); + spike_detector_->set_peak_life_time(peak_lifetime_->get()); + if (!data_in_port_->slot(0)->RetrieveData(data_in) ){ + break; + } // claim output data buckets data_out = data_out_port_->slot(0)->ClaimData(false); - auto spike_ts = data_waveform->sample_timestamps(); - for(unsigned int i=0; inchannels(); i++){ - data_out->set_data_sample(i, "hardware ts", data_spikes->ts_detected_spikes(i)); - data_out->set_data_sample(i, "depthx", 0/* channels_pos_[data_in->channel_detected_spikes(i)][0]*/); - data_out->set_data_sample(i, "depthy", 0 /*channels_pos_[data_in->channel_detected_spikes(i)][1]*/); - data_out->set_data_sample(i, "amplitude", *(data_spikes->spike_amplitudes(i))); + // if invert signal + // ?? --- Removed because not needed for now + + // detect spikes sample by sample and collect each detected spike + for (size_t channel = 0; channel < data_in->ncolumns(); ++channel) { + spike_detector_->find_spikes(channel, data_in->sample_timestamps(), data_in->begin_column_by_index(channel)); - auto waveform = data_waveform->begin_channel(i); - // compute angle spike - // computer width spike - data_out->set_data_sample(i, "angle", 0 /*TODO*/); - data_out->set_data_sample(i, "width", 0 /*TODO*/); + } + unsigned int i = 0; + for(unsigned int j=0; j< spike_detector_->nspikes(); j++){ + data_out->set_data_sample(i, "hardware ts", spike_detector_->timestamps_detected_spike()[j]); + auto channel = data_in->column_labels()[spike_detector_->channels_detected_spike()[j]]; + data_out->set_data_sample(i, "depth", channel_pos_().at(channel)); + data_out->set_data_sample(i, "amplitude", spike_detector_->amplitudes_detected_spike()[j]); + i++; } + data_out->set_source_timestamp(); + data_out->set_hardware_timestamp(data_in->hardware_timestamp()); + // publish and release data data_out_port_->slot(0)->PublishData(); data_in_port_->slot(0)->ReleaseData(); - data_spike_port_->slot(0)->ReleaseData(); } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 7942978..ee04d31 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -18,34 +18,46 @@ // --------------------------------------------------------------------- #pragma once - +#include "dsp/algorithms.hpp" #include "iprocessor.hpp" -#include "featuresdata/featuresdata.hpp" -#include "multichanneldata/multichanneldata.hpp" -#include "spikedata/spikedata.hpp" +#include "columnsdata/columnsdata.hpp" +#include "timeseriesdata/timeseriesdata.hpp" class SpikeFeatures : public IProcessor { // CONSTRUCTOR and OVERLOADED METHODS public: SpikeFeatures(); - void Configure(const GlobalContext &context); void CreatePorts() override; + void CompleteStreamInfo() override; + void Prepare(GlobalContext &context) override; void Process(ProcessingContext &context) override; // DATA PORTS protected: - PortIn> *data_in_port_; - PortIn *data_spike_port_; - PortOut> *data_out_port_; - const uint32_t MAX_NCHANNELS=384; + PortIn> *data_in_port_; + PortOut> *data_out_port_; + + +protected: + options::Value< std::map, false> channel_pos_{}; + std::unique_ptr spike_detector_; + // STATES protected: - options::Value channelmap_def_{}; + StaticState *threshold_; + StaticState *peak_lifetime_; + // CONSTANTS +public: + const uint32_t MAX_NCHANNELS=384; + const std::string PEAK_LIFETIME = "peak lifetime"; + const std::string THRESHOLD = "threshold"; + std::vector feature_labels_; private: - std::vector>* channels_pos_; + options::Double initial_threshold_{60.}; + options::Measurement initial_peak_lifetime_{8, "sample"}; }; From 67e3dd8b55551702ebfee8be9f9b33fd7fe774af Mon Sep 17 00:00:00 2001 From: marinechaput Date: Wed, 10 Nov 2021 16:28:49 +0100 Subject: [PATCH 05/22] use the spike detector algorithm on a group of channels --- processors/spikefeatures/spikefeatures.cpp | 48 ++++++++++++++-------- processors/spikefeatures/spikefeatures.hpp | 9 ++-- 2 files changed, 36 insertions(+), 21 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index d72f3f4..cb0f1d6 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -24,7 +24,7 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ feature_labels_ = {"hardware ts", "depth","amplitude"}; - add_option("channelmap", channel_pos_, "Relation between channel number and depth.", true); + add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); add_option(THRESHOLD, initial_threshold_, "Spike detection threshold in data units."); add_option(PEAK_LIFETIME, initial_peak_lifetime_, @@ -39,7 +39,7 @@ void SpikeFeatures::CreatePorts(){ data_out_port_ = create_output_port>( - "data", ColumnsType::Capabilities(feature_labels_.size()), + "data", ColumnsType::Capabilities(feature_labels_.size(), SampleRange(1, 1000)), ColumnsType::Parameters(), PortOutPolicy(SlotRange(1))); threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, @@ -50,12 +50,13 @@ void SpikeFeatures::CreatePorts(){ } void SpikeFeatures::CompleteStreamInfo() { - data_out_port_->streaminfo(0).set_parameters(feature_labels_); + data_out_port_->streaminfo(0).set_parameters(ColumnsType::Parameters(feature_labels_, 1000)); } void SpikeFeatures::Prepare(GlobalContext &context) { - spike_detector_.reset(new dsp::algorithms::SpikeDetectorNeuropixels( - initial_threshold_(), initial_peak_lifetime_())); + for(uint16_t i=0; i< data_in_port_->streaminfo(0).parameters().ncolumns/ngroup_; i++ ){ + spike_detectors_.push_back(std::make_unique(ngroup_, initial_threshold_(), initial_peak_lifetime_())); + } } @@ -64,8 +65,6 @@ void SpikeFeatures::Process(ProcessingContext &context) { ColumnsType::Data *data_out = nullptr; while (!context.terminated()) { - spike_detector_->set_threshold(threshold_->get()); - spike_detector_->set_peak_life_time(peak_lifetime_->get()); if (!data_in_port_->slot(0)->RetrieveData(data_in) ){ break; @@ -78,18 +77,33 @@ void SpikeFeatures::Process(ProcessingContext &context) { // ?? --- Removed because not needed for now // detect spikes sample by sample and collect each detected spike - for (size_t channel = 0; channel < data_in->ncolumns(); ++channel) { - spike_detector_->find_spikes(channel, data_in->sample_timestamps(), data_in->begin_column_by_index(channel)); + unsigned int spike_number = 0; + uint16_t d = 0; - } + for (size_t channel = 0; channel < data_in->ncolumns(); channel=channel+ngroup_) { + spike_detectors_[d]->set_threshold(threshold_->get()); + spike_detectors_[d]->set_peak_life_time(peak_lifetime_->get()); + + for (size_t sample = 0; sample < data_in->nsamples();++sample) { + + if (spike_detectors_[d]->is_spike( + data_in->sample_timestamp(sample), + data_in->begin_sample(sample)+channel)) { + + + auto amp = spike_detectors_[d]->amplitudes_detected_spike(); + //data_out->set_data_sample(i, "hardware ts", ); + auto max_index = std::max_element(amp.begin(),amp.end()) - amp.begin(); - unsigned int i = 0; - for(unsigned int j=0; j< spike_detector_->nspikes(); j++){ - data_out->set_data_sample(i, "hardware ts", spike_detector_->timestamps_detected_spike()[j]); - auto channel = data_in->column_labels()[spike_detector_->channels_detected_spike()[j]]; - data_out->set_data_sample(i, "depth", channel_pos_().at(channel)); - data_out->set_data_sample(i, "amplitude", spike_detector_->amplitudes_detected_spike()[j]); - i++; + //auto c_max = data_in->column_labels()[channel+max_index+1]; + //data_out->set_data_sample(i, "depth", std::stoi(c_max)); + std::vector f= {static_cast(spike_detectors_[d]->timestamp_detected_spike()),static_cast(channel+max_index+1), amp[max_index]}; + data_out->set_data_sample(spike_number, f); + LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index+1 << " - index: " << max_index; + spike_number++; + } + } + d++; } data_out->set_source_timestamp(); diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index ee04d31..497ef5b 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -41,7 +41,7 @@ class SpikeFeatures : public IProcessor { protected: options::Value< std::map, false> channel_pos_{}; - std::unique_ptr spike_detector_; + std::vector> spike_detectors_; // STATES @@ -51,13 +51,14 @@ class SpikeFeatures : public IProcessor { // CONSTANTS public: - const uint32_t MAX_NCHANNELS=384; + const uint32_t MAX_NCHANNELS=385; const std::string PEAK_LIFETIME = "peak lifetime"; const std::string THRESHOLD = "threshold"; std::vector feature_labels_; + const uint16_t ngroup_ = 1; private: - options::Double initial_threshold_{60.}; - options::Measurement initial_peak_lifetime_{8, "sample"}; + options::Double initial_threshold_{70.}; + options::Measurement initial_peak_lifetime_{1, "sample"}; }; From 36422391cb04b5768d5c530b10aee79b3a62b4ab Mon Sep 17 00:00:00 2001 From: marinechaput Date: Fri, 19 Nov 2021 18:56:44 +0100 Subject: [PATCH 06/22] add invert signal --- processors/spikefeatures/spikefeatures.cpp | 16 ++++++++++------ processors/spikefeatures/spikefeatures.hpp | 11 +++++++---- 2 files changed, 17 insertions(+), 10 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index cb0f1d6..475a2fa 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -30,6 +30,9 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option(PEAK_LIFETIME, initial_peak_lifetime_, "Peak life time in samples"); + add_option("invert_signal", invert_signal_, + "invert a signal to detect negative spikes"); + } void SpikeFeatures::CreatePorts(){ @@ -51,12 +54,17 @@ void SpikeFeatures::CreatePorts(){ void SpikeFeatures::CompleteStreamInfo() { data_out_port_->streaminfo(0).set_parameters(ColumnsType::Parameters(feature_labels_, 1000)); + + nsamples_= data_in_port_->slot(0)->streaminfo().parameters().nsamples; + ncolumns_ = data_in_port_->slot(0)->streaminfo().parameters().ncolumns; } void SpikeFeatures::Prepare(GlobalContext &context) { for(uint16_t i=0; i< data_in_port_->streaminfo(0).parameters().ncolumns/ngroup_; i++ ){ - spike_detectors_.push_back(std::make_unique(ngroup_, initial_threshold_(), initial_peak_lifetime_())); + spike_detectors_.push_back(std::make_unique(ngroup_, initial_threshold_(), + initial_peak_lifetime_(), invert_signal_())); } + } @@ -72,10 +80,6 @@ void SpikeFeatures::Process(ProcessingContext &context) { // claim output data buckets data_out = data_out_port_->slot(0)->ClaimData(false); - - // if invert signal - // ?? --- Removed because not needed for now - // detect spikes sample by sample and collect each detected spike unsigned int spike_number = 0; uint16_t d = 0; @@ -99,7 +103,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { //data_out->set_data_sample(i, "depth", std::stoi(c_max)); std::vector f= {static_cast(spike_detectors_[d]->timestamp_detected_spike()),static_cast(channel+max_index+1), amp[max_index]}; data_out->set_data_sample(spike_number, f); - LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index+1 << " - index: " << max_index; + LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index ; spike_number++; } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 497ef5b..88de9d0 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -38,11 +38,12 @@ class SpikeFeatures : public IProcessor { PortIn> *data_in_port_; PortOut> *data_out_port_; - protected: - options::Value< std::map, false> channel_pos_{}; std::vector> spike_detectors_; + size_t nsamples_; + size_t ncolumns_; + // STATES protected: @@ -51,7 +52,8 @@ class SpikeFeatures : public IProcessor { // CONSTANTS public: - const uint32_t MAX_NCHANNELS=385; + const uint32_t MAX_NCHANNELS=384; + const uint32_t MAX_NSPIKES=100; const std::string PEAK_LIFETIME = "peak lifetime"; const std::string THRESHOLD = "threshold"; std::vector feature_labels_; @@ -60,5 +62,6 @@ class SpikeFeatures : public IProcessor { private: options::Double initial_threshold_{70.}; options::Measurement initial_peak_lifetime_{1, "sample"}; - + options::Value< std::map, false> channel_pos_{}; + options::Bool invert_signal_{true}; }; From c7947e65a146a045638955606fadd6aa358647bf Mon Sep 17 00:00:00 2001 From: marinechaput Date: Fri, 3 Dec 2021 16:32:56 +0100 Subject: [PATCH 07/22] WIP try to add spike valley computation --- processors/spikefeatures/spikefeatures.cpp | 70 ++++++++++++++-------- 1 file changed, 45 insertions(+), 25 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 475a2fa..98fa141 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -22,7 +22,7 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ - feature_labels_ = {"hardware ts", "depth","amplitude"}; + feature_labels_ = {"hardware ts", "depth","amplitude", "slope"}; add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); add_option(THRESHOLD, initial_threshold_, @@ -32,7 +32,6 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option("invert_signal", invert_signal_, "invert a signal to detect negative spikes"); - } void SpikeFeatures::CreatePorts(){ @@ -42,8 +41,7 @@ void SpikeFeatures::CreatePorts(){ data_out_port_ = create_output_port>( - "data", ColumnsType::Capabilities(feature_labels_.size(), SampleRange(1, 1000)), - ColumnsType::Parameters(), PortOutPolicy(SlotRange(1))); + "data", ColumnsType::Parameters(feature_labels_, 1000), PortOutPolicy(SlotRange(1))); threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, Permission::WRITE); @@ -53,18 +51,25 @@ void SpikeFeatures::CreatePorts(){ } void SpikeFeatures::CompleteStreamInfo() { - data_out_port_->streaminfo(0).set_parameters(ColumnsType::Parameters(feature_labels_, 1000)); - - nsamples_= data_in_port_->slot(0)->streaminfo().parameters().nsamples; - ncolumns_ = data_in_port_->slot(0)->streaminfo().parameters().ncolumns; + nsamples_= data_in_port_->prototype(0).nsamples(); + ncolumns_ = data_in_port_->prototype(0).ncolumns(); } void SpikeFeatures::Prepare(GlobalContext &context) { - for(uint16_t i=0; i< data_in_port_->streaminfo(0).parameters().ncolumns/ngroup_; i++ ){ - spike_detectors_.push_back(std::make_unique(ngroup_, initial_threshold_(), - initial_peak_lifetime_(), invert_signal_())); + for(size_t i=0; i< ncolumns_/ngroup_; i++ ){ + + if(invert_signal_()){ + spike_detectors_.push_back( + std::make_unique(ngroup_, initial_threshold_(), + initial_peak_lifetime_(), + dsp::algorithms::SpikeDetectionSign::DOWN)); + }else{ + spike_detectors_.push_back( + std::make_unique(ngroup_, initial_threshold_(), + initial_peak_lifetime_(), + dsp::algorithms::SpikeDetectionSign::UP)); + } } - } @@ -82,32 +87,47 @@ void SpikeFeatures::Process(ProcessingContext &context) { // detect spikes sample by sample and collect each detected spike unsigned int spike_number = 0; - uint16_t d = 0; + size_t channel_detector = 0; for (size_t channel = 0; channel < data_in->ncolumns(); channel=channel+ngroup_) { - spike_detectors_[d]->set_threshold(threshold_->get()); - spike_detectors_[d]->set_peak_life_time(peak_lifetime_->get()); + // 1 major limitation - can have only 1 spike in the packet + // not much a limitation as we keep the epoch as small as possible (10 samples) + // but still a problem to raise if we could have multiple spikes in the same packet + spike_detectors_[channel_detector]->set_threshold(threshold_->get()); + spike_detectors_[channel_detector]->set_peak_life_time(peak_lifetime_->get()); for (size_t sample = 0; sample < data_in->nsamples();++sample) { - if (spike_detectors_[d]->is_spike( + if (spike_detectors_[channel_detector]->is_spike( data_in->sample_timestamp(sample), data_in->begin_sample(sample)+channel)) { + // if(spike_detectors_[d]->detection_mode_ == dsp::algorithms::SpikeDetectionMode::VALLEY){ + auto amp = spike_detectors_[channel_detector]->amplitudes_detected_spike(); + auto max_index =std::max_element(amp.begin(), amp.end()) - amp.begin(); + data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[channel_detector]->timestamp_detected_spike())); + data_out->set_data_sample(spike_number, "depth", static_cast(channel+max_index)); + data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); + data_out->set_data_sample(spike_number, "slope", spike_detectors_[channel_detector]->slopes_detected_spike()[channel+max_index]); + + + LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index ; + + /*} + else{ + // TODO + //auto amp = spike_detectors_[d]->amplitude_detected_valley(); + //auto max_index = spike_detectors_[d]->channel_max_peak_; + + data_out->set_data_sample(spike_number, "distance2peak", + data_in->sample_timestamp(sample)-data_out->data_sample(spike_number,"hardware ts")); - auto amp = spike_detectors_[d]->amplitudes_detected_spike(); - //data_out->set_data_sample(i, "hardware ts", ); - auto max_index = std::max_element(amp.begin(),amp.end()) - amp.begin(); + }*/ - //auto c_max = data_in->column_labels()[channel+max_index+1]; - //data_out->set_data_sample(i, "depth", std::stoi(c_max)); - std::vector f= {static_cast(spike_detectors_[d]->timestamp_detected_spike()),static_cast(channel+max_index+1), amp[max_index]}; - data_out->set_data_sample(spike_number, f); - LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index ; spike_number++; } } - d++; + channel_detector++; } data_out->set_source_timestamp(); From f4f44c6c33e526e3f627257d333673d537ff243c Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 11 Jan 2022 17:15:43 +0100 Subject: [PATCH 08/22] reorganize spike detection with datastream incoming --- processors/spikefeatures/spikefeatures.cpp | 130 ++++++++++----------- processors/spikefeatures/spikefeatures.hpp | 3 - 2 files changed, 64 insertions(+), 69 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 98fa141..b879349 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -1,4 +1,4 @@ -// --------------------------------------------------------------------- +// --------------------------------------------------------------------- // This file is part of falcon-core. // // Copyright (C) 2021-now Neuro-Electronics Research Flanders @@ -22,26 +22,29 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ - feature_labels_ = {"hardware ts", "depth","amplitude", "slope"}; + feature_labels_ = {"hardware ts", "depth","amplitude", "slope"}; + + //add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); + + add_option(THRESHOLD, initial_threshold_, + "Spike detection threshold in data units."); + add_option(PEAK_LIFETIME, initial_peak_lifetime_, + "Peak life time in samples"); - add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); - add_option(THRESHOLD, initial_threshold_, - "Spike detection threshold in data units."); - add_option(PEAK_LIFETIME, initial_peak_lifetime_, - "Peak life time in samples"); + add_option("invert_signal", invert_signal_, + "invert a signal to detect negative spikes"); - add_option("invert_signal", invert_signal_, - "invert a signal to detect negative spikes"); } void SpikeFeatures::CreatePorts(){ data_in_port_ = create_input_port>( - "waveform", TimeSeriesType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), - PortInPolicy(SlotRange(1))); + TimeSeriesType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), + PortInPolicy(SlotRange(1, MAX_NCHANNELS))); data_out_port_ = create_output_port>( - "data", ColumnsType::Parameters(feature_labels_, 1000), PortOutPolicy(SlotRange(1))); + ColumnsType::Parameters(feature_labels_, 1000), + PortOutPolicy(SlotRange(1, MAX_NCHANNELS))); threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, Permission::WRITE); @@ -51,21 +54,38 @@ void SpikeFeatures::CreatePorts(){ } void SpikeFeatures::CompleteStreamInfo() { - nsamples_= data_in_port_->prototype(0).nsamples(); - ncolumns_ = data_in_port_->prototype(0).ncolumns(); + // check if we have the same number of input and output slots + if (data_in_port_->number_of_slots() != data_out_port_->number_of_slots()) { + auto err_msg = "Number of output slots (" + + std::to_string(data_out_port_->number_of_slots()) + + ") on port '" + data_out_port_->name() + + "' does not match number of input slots (" + + std::to_string(data_in_port_->number_of_slots()) + + ") on port '" + data_in_port_->name() + "'."; + throw ProcessingStreamInfoError(err_msg, name()); + } + + for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { + data_out_port_->streaminfo(k).set_stream_rate( + data_in_port_->streaminfo(k).stream_rate()); + data_out_port_->streaminfo(k).set_parameters( + data_in_port_->prototype(k).parameters()); + } + } void SpikeFeatures::Prepare(GlobalContext &context) { - for(size_t i=0; i< ncolumns_/ngroup_; i++ ){ + // Create an independent spike detector by data stream input + for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { if(invert_signal_()){ spike_detectors_.push_back( - std::make_unique(ngroup_, initial_threshold_(), + std::make_unique(data_in_port_->prototype(k).ncolumns(), initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::DOWN)); }else{ spike_detectors_.push_back( - std::make_unique(ngroup_, initial_threshold_(), + std::make_unique(data_in_port_->prototype(k).ncolumns(), initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::UP)); } @@ -78,65 +98,43 @@ void SpikeFeatures::Process(ProcessingContext &context) { ColumnsType::Data *data_out = nullptr; while (!context.terminated()) { + for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { + if (!data_in_port_->slot(k)->RetrieveData(data_in) ){ + break; + } + // claim output data buckets + data_out = data_out_port_->slot(k)->ClaimData(false); + spike_detectors_[k]->set_threshold(threshold_->get()); + spike_detectors_[k]->set_peak_life_time(peak_lifetime_->get()); - if (!data_in_port_->slot(0)->RetrieveData(data_in) ){ - break; - } - // claim output data buckets - data_out = data_out_port_->slot(0)->ClaimData(false); - - // detect spikes sample by sample and collect each detected spike - unsigned int spike_number = 0; - size_t channel_detector = 0; - - for (size_t channel = 0; channel < data_in->ncolumns(); channel=channel+ngroup_) { - // 1 major limitation - can have only 1 spike in the packet - // not much a limitation as we keep the epoch as small as possible (10 samples) - // but still a problem to raise if we could have multiple spikes in the same packet - spike_detectors_[channel_detector]->set_threshold(threshold_->get()); - spike_detectors_[channel_detector]->set_peak_life_time(peak_lifetime_->get()); - + unsigned int spike_number = 0; for (size_t sample = 0; sample < data_in->nsamples();++sample) { - if (spike_detectors_[channel_detector]->is_spike( - data_in->sample_timestamp(sample), - data_in->begin_sample(sample)+channel)) { - - // if(spike_detectors_[d]->detection_mode_ == dsp::algorithms::SpikeDetectionMode::VALLEY){ - auto amp = spike_detectors_[channel_detector]->amplitudes_detected_spike(); - auto max_index =std::max_element(amp.begin(), amp.end()) - amp.begin(); - data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[channel_detector]->timestamp_detected_spike())); - data_out->set_data_sample(spike_number, "depth", static_cast(channel+max_index)); - data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); - data_out->set_data_sample(spike_number, "slope", spike_detectors_[channel_detector]->slopes_detected_spike()[channel+max_index]); - - - LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel+max_index ; + if (spike_detectors_[k]->is_spike( + data_in->sample_timestamp(sample), + data_in->begin_sample(sample))) { - /*} - else{ - // TODO - //auto amp = spike_detectors_[d]->amplitude_detected_valley(); - //auto max_index = spike_detectors_[d]->channel_max_peak_; + auto amp = spike_detectors_[k]->amplitudes_detected_spike(); + auto max_index =std::max_element(amp.begin(), amp.end()) - amp.begin(); + std::string channel_label = data_in->labels()[static_cast(max_index)]; + data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[k]->timestamp_detected_spike())); + data_out->set_data_sample(spike_number, "depth", std::stod(channel_label)); // transform the channel label from string to double + data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); + data_out->set_data_sample(spike_number, "slope", spike_detectors_[k]->slopes_detected_spike()[max_index]); - data_out->set_data_sample(spike_number, "distance2peak", - data_in->sample_timestamp(sample)-data_out->data_sample(spike_number,"hardware ts")); - }*/ - - spike_number++; + LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel_label ; + spike_number++; } } - channel_detector++; - } - data_out->set_source_timestamp(); - data_out->set_hardware_timestamp(data_in->hardware_timestamp()); - - // publish and release data - data_out_port_->slot(0)->PublishData(); - data_in_port_->slot(0)->ReleaseData(); + data_out->set_source_timestamp(); + data_out->set_hardware_timestamp(data_in->hardware_timestamp()); + // publish and release data + data_out_port_->slot(k)->PublishData(); + data_in_port_->slot(k)->ReleaseData(); + } } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 88de9d0..6c9dde5 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -41,9 +41,6 @@ class SpikeFeatures : public IProcessor { protected: std::vector> spike_detectors_; - size_t nsamples_; - size_t ncolumns_; - // STATES protected: From 625bb057ea98286fa45d95f6284f0c6396ed2d4c Mon Sep 17 00:00:00 2001 From: marinechaput Date: Fri, 14 Jan 2022 22:42:51 +0100 Subject: [PATCH 09/22] fix bug --- processors/spikefeatures/spikefeatures.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index b879349..cf2ced7 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -68,8 +68,9 @@ void SpikeFeatures::CompleteStreamInfo() { for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { data_out_port_->streaminfo(k).set_stream_rate( data_in_port_->streaminfo(k).stream_rate()); + data_out_port_->streaminfo(k).set_parameters( - data_in_port_->prototype(k).parameters()); + ColumnsType::Parameters(feature_labels_, 1000, data_in_port_->prototype(k).streamname())); } } From 12ec9621318c92a4462bc3d016828feed1c261c0 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Mon, 17 Jan 2022 19:16:47 +0100 Subject: [PATCH 10/22] remove timestamps from the feature --- processors/spikefeatures/spikefeatures.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index cf2ced7..be5e36b 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -22,7 +22,7 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ - feature_labels_ = {"hardware ts", "depth","amplitude", "slope"}; + feature_labels_ = {"depth","amplitude", "slope"}; //add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); @@ -118,7 +118,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { auto amp = spike_detectors_[k]->amplitudes_detected_spike(); auto max_index =std::max_element(amp.begin(), amp.end()) - amp.begin(); std::string channel_label = data_in->labels()[static_cast(max_index)]; - data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[k]->timestamp_detected_spike())); + //data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[k]->timestamp_detected_spike())); data_out->set_data_sample(spike_number, "depth", std::stod(channel_label)); // transform the channel label from string to double data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); data_out->set_data_sample(spike_number, "slope", spike_detectors_[k]->slopes_detected_spike()[max_index]); From ec26d4ee8d7ec9e5e47038e67ce4234ff23dc349 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 25 Jan 2022 13:10:12 +0100 Subject: [PATCH 11/22] add streamname --- processors/spikefeatures/spikefeatures.cpp | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index be5e36b..31f4597 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -66,11 +66,10 @@ void SpikeFeatures::CompleteStreamInfo() { } for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { - data_out_port_->streaminfo(k).set_stream_rate( - data_in_port_->streaminfo(k).stream_rate()); + data_out_port_->streaminfo(k).set_stream_parameters(data_in_port_->streaminfo(k)); data_out_port_->streaminfo(k).set_parameters( - ColumnsType::Parameters(feature_labels_, 1000, data_in_port_->prototype(k).streamname())); + ColumnsType::Parameters(feature_labels_, 1000)); } } From 4afff85aca14b703ebe8e0a92870cfc49d03fbe0 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 10 Feb 2022 10:51:16 +0100 Subject: [PATCH 12/22] fix bug spikefeatures --- processors/spikefeatures/spikefeatures.cpp | 108 +++++++++++++++------ processors/spikefeatures/spikefeatures.hpp | 10 +- 2 files changed, 87 insertions(+), 31 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 31f4597..d5fbf50 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -22,9 +22,9 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ - feature_labels_ = {"depth","amplitude", "slope"}; + add_option("features", features_, "Selection of features to compute.", true); - //add_option("channelmap", channel_pos_, "Relation between channel number and depth.", false); + add_option("channelmap", channel_pos_, "Relation between channel number and depth."); add_option(THRESHOLD, initial_threshold_, "Spike detection threshold in data units."); @@ -36,6 +36,33 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ } +void SpikeFeatures::Configure(const GlobalContext &context){ + features_labels_ = {}; + default_features_ = YAML::Load( "{ " + "amplitude: false," + "slope: false," + "channel index: false," + "depth: false" + "}"); + + + for(auto label: features_()){ + + if (!default_features_[label]){ + throw ProcessingConfigureError("The " + label + " is not implemented.", name()); + } + + default_features_[label] = true; + features_labels_.push_back(label); + } + + if(default_features_["depth"].as() and channel_pos_().size() == 0){ + throw ProcessingConfigureError("If the depth feature is selected, the channelmap option " + "needs to give the mapping between channel label and depth.", name()); + + } +} + void SpikeFeatures::CreatePorts(){ data_in_port_ = create_input_port>( TimeSeriesType::Capabilities(ChannelRange(1, MAX_NCHANNELS)), @@ -43,7 +70,7 @@ void SpikeFeatures::CreatePorts(){ data_out_port_ = create_output_port>( - ColumnsType::Parameters(feature_labels_, 1000), + ColumnsType::Parameters(features_labels_, MAX_NSPIKES), PortOutPolicy(SlotRange(1, MAX_NCHANNELS))); threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, @@ -65,27 +92,31 @@ void SpikeFeatures::CompleteStreamInfo() { throw ProcessingStreamInfoError(err_msg, name()); } - for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { - data_out_port_->streaminfo(k).set_stream_parameters(data_in_port_->streaminfo(k)); + for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { + data_out_port_->streaminfo(s_).set_stream_parameters(data_in_port_->streaminfo(s_)); - data_out_port_->streaminfo(k).set_parameters( - ColumnsType::Parameters(feature_labels_, 1000)); + data_out_port_->streaminfo(s_).set_parameters( + ColumnsType::Parameters(features_labels_, MAX_NSPIKES)); } } void SpikeFeatures::Prepare(GlobalContext &context) { // Create an independent spike detector by data stream input - for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { + spike_detectors_.clear(); + + for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { if(invert_signal_()){ spike_detectors_.push_back( - std::make_unique(data_in_port_->prototype(k).ncolumns(), initial_threshold_(), + std::make_unique(data_in_port_->prototype(s_).ncolumns(), + initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::DOWN)); }else{ spike_detectors_.push_back( - std::make_unique(data_in_port_->prototype(k).ncolumns(), initial_threshold_(), + std::make_unique(data_in_port_->prototype(s_).ncolumns(), + initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::UP)); } @@ -96,44 +127,67 @@ void SpikeFeatures::Prepare(GlobalContext &context) { void SpikeFeatures::Process(ProcessingContext &context) { TimeSeriesType::Data *data_in = nullptr; ColumnsType::Data *data_out = nullptr; + std::string channel_label; + std::vector amp; + double max_index; + std::string feature_str; + size_t sample = 0; while (!context.terminated()) { - for (int k = 0; k < data_in_port_->number_of_slots(); ++k) { - if (!data_in_port_->slot(k)->RetrieveData(data_in) ){ + for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { + + if (!data_in_port_->slot(s_)->RetrieveData(data_in) ){ break; } // claim output data buckets - data_out = data_out_port_->slot(k)->ClaimData(false); - spike_detectors_[k]->set_threshold(threshold_->get()); - spike_detectors_[k]->set_peak_life_time(peak_lifetime_->get()); + data_out = data_out_port_->slot(s_)->ClaimData(false); + spike_detectors_[s_]->set_threshold(threshold_->get()); + spike_detectors_[s_]->set_peak_life_time(peak_lifetime_->get()); unsigned int spike_number = 0; - for (size_t sample = 0; sample < data_in->nsamples();++sample) { + for (sample = 0; sample < data_in->nsamples();++sample) { - if (spike_detectors_[k]->is_spike( + if (spike_detectors_[s_]->is_spike( data_in->sample_timestamp(sample), data_in->begin_sample(sample))) { - auto amp = spike_detectors_[k]->amplitudes_detected_spike(); - auto max_index =std::max_element(amp.begin(), amp.end()) - amp.begin(); - std::string channel_label = data_in->labels()[static_cast(max_index)]; - //data_out->set_data_sample(spike_number, "hardware ts", static_cast(spike_detectors_[k]->timestamp_detected_spike())); - data_out->set_data_sample(spike_number, "depth", std::stod(channel_label)); // transform the channel label from string to double - data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); - data_out->set_data_sample(spike_number, "slope", spike_detectors_[k]->slopes_detected_spike()[max_index]); + amp = spike_detectors_[s_]->amplitudes_detected_spike(); + max_index = std::max_element(amp.begin(), amp.end()) - amp.begin(); + channel_label = data_in->labels()[static_cast(max_index)]; + + feature_str = "("; + if(default_features_["channel index"].as()){ + data_out->set_data_sample(spike_number, "channel index", std::stod(channel_label)); + feature_str += "channel index: " + channel_label; + } + + if(default_features_["amplitude"].as()){ + data_out->set_data_sample(spike_number, "amplitude", amp[max_index]); + feature_str += "amplitude: " + std::to_string(data_out->data_sample(spike_number, "amplitude")); + } + + if(default_features_["slope"].as()){ + data_out->set_data_sample(spike_number, "slope", spike_detectors_[s_]->slopes_detected_spike()[max_index]); + feature_str += "slope: " + std::to_string(data_out->data_sample(spike_number, "slope")); + } + if(default_features_["depth"].as()){ + data_out->set_data_sample(spike_number, "depth", channel_pos_().at(channel_label)); + feature_str += "depth: " + std::to_string(data_out->data_sample(spike_number, "depth")); + } - LOG(INFO)<< "detected 1 spike: " << amp[max_index] << " channel: "<< channel_label ; + LOG(DEBUG)<< "Spike detected : " + feature_str + ")"; spike_number++; } } + data_out->set_nreal_samples(spike_number); data_out->set_source_timestamp(); data_out->set_hardware_timestamp(data_in->hardware_timestamp()); // publish and release data - data_out_port_->slot(k)->PublishData(); - data_in_port_->slot(k)->ReleaseData(); + data_out_port_->slot(s_)->PublishData(); + data_in_port_->slot(s_)->ReleaseData(); } } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 6c9dde5..f790566 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -27,12 +27,12 @@ class SpikeFeatures : public IProcessor { // CONSTRUCTOR and OVERLOADED METHODS public: SpikeFeatures(); + void Configure(const GlobalContext &context) override; void CreatePorts() override; void CompleteStreamInfo() override; void Prepare(GlobalContext &context) override; void Process(ProcessingContext &context) override; - // DATA PORTS protected: PortIn> *data_in_port_; @@ -53,12 +53,14 @@ class SpikeFeatures : public IProcessor { const uint32_t MAX_NSPIKES=100; const std::string PEAK_LIFETIME = "peak lifetime"; const std::string THRESHOLD = "threshold"; - std::vector feature_labels_; - const uint16_t ngroup_ = 1; + std::vector features_labels_; + YAML::Node default_features_; + int s_ = 0; private: options::Double initial_threshold_{70.}; options::Measurement initial_peak_lifetime_{1, "sample"}; - options::Value< std::map, false> channel_pos_{}; + options::Value, false> features_{}; + options::Value, false> channel_pos_{}; options::Bool invert_signal_{true}; }; From d21306fa07815d51b79d91e0befd29b42c2b4979 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Mon, 14 Feb 2022 20:11:57 +0100 Subject: [PATCH 13/22] add time feature --- processors/spikefeatures/spikefeatures.cpp | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index d5fbf50..e418f1b 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -39,6 +39,7 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ void SpikeFeatures::Configure(const GlobalContext &context){ features_labels_ = {}; default_features_ = YAML::Load( "{ " + "time: false," "amplitude: false," "slope: false," "channel index: false," @@ -94,7 +95,6 @@ void SpikeFeatures::CompleteStreamInfo() { for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { data_out_port_->streaminfo(s_).set_stream_parameters(data_in_port_->streaminfo(s_)); - data_out_port_->streaminfo(s_).set_parameters( ColumnsType::Parameters(features_labels_, MAX_NSPIKES)); } @@ -156,6 +156,12 @@ void SpikeFeatures::Process(ProcessingContext &context) { channel_label = data_in->labels()[static_cast(max_index)]; feature_str = "("; + + if(default_features_["time"].as()){ + data_out->set_data_sample(spike_number, "time", sample+data_in->hardware_timestamp()); + feature_str += "time: " + std::to_string(sample+data_in->hardware_timestamp()); + } + if(default_features_["channel index"].as()){ data_out->set_data_sample(spike_number, "channel index", std::stod(channel_label)); feature_str += "channel index: " + channel_label; @@ -180,7 +186,6 @@ void SpikeFeatures::Process(ProcessingContext &context) { spike_number++; } } - data_out->set_nreal_samples(spike_number); data_out->set_source_timestamp(); data_out->set_hardware_timestamp(data_in->hardware_timestamp()); From 28f639446b9539e2d72f6bf4e1fdd9d78174e070 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 24 Mar 2022 12:30:20 +0100 Subject: [PATCH 14/22] add documentation --- processors/spikefeatures/doc.yaml | 48 +++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/processors/spikefeatures/doc.yaml b/processors/spikefeatures/doc.yaml index e69de29..e4c1759 100644 --- a/processors/spikefeatures/doc.yaml +++ b/processors/spikefeatures/doc.yaml @@ -0,0 +1,48 @@ +Description: Detect spikes in a TimeSeries data stream and emits a column data type packet containing all selected features for the decoding. + +Long description: possible features available are timestamp, amplitude, slope, channel index, channel depth. These features are disabled by default and need to be enabled in the graph options. + +Input ports: + - name: data + type: TimeSerieType + slots: 1-385 + description: + +Output port: + - name: data + type: ColumnData + slots: 1-385 + description: data packet containing spikes with the selected features (fixed size of 100 spikes - filled with 0 spikes) + +Options: + - &threshold + name: threshold + type: double + default: 70.0 + description: Spike detection threshold in data units. + - name: invert signal + type: bool + default: True + description: Invert signal before spike detection. + - &peaklifetime + name: peak lifetime + type: unsigned int + default: 1 samples + description: Peak life time in samples + - name: features + type : vector of string + default: empty, required option + description: list of features to use in output + - name: channelmap + type: map of string / double + default: empty, mandatory when the depth feature is activated + description: mapping between channel number and channel depth + +States: + Static: + - <<: *threshold + shared: true + external access: write + - <<: *peaklifetime + shared: true + external access: write From ac00b5bd08300b6ff1a6a7a5fd35bce48baa11b2 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Fri, 9 Sep 2022 15:27:56 +0200 Subject: [PATCH 15/22] follow modif in the columndata with automatic expander --- processors/spikefeatures/spikefeatures.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index e418f1b..616166b 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -24,7 +24,7 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option("features", features_, "Selection of features to compute.", true); - add_option("channelmap", channel_pos_, "Relation between channel number and depth."); + add_option("channeldepths", channel_pos_, "Relation between channel number and depth."); add_option(THRESHOLD, initial_threshold_, "Spike detection threshold in data units."); @@ -71,7 +71,7 @@ void SpikeFeatures::CreatePorts(){ data_out_port_ = create_output_port>( - ColumnsType::Parameters(features_labels_, MAX_NSPIKES), + ColumnsType::Parameters(features_labels_, 0, true), PortOutPolicy(SlotRange(1, MAX_NCHANNELS))); threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, @@ -96,7 +96,7 @@ void SpikeFeatures::CompleteStreamInfo() { for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { data_out_port_->streaminfo(s_).set_stream_parameters(data_in_port_->streaminfo(s_)); data_out_port_->streaminfo(s_).set_parameters( - ColumnsType::Parameters(features_labels_, MAX_NSPIKES)); + ColumnsType::Parameters(features_labels_, 0, true)); } } @@ -186,7 +186,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { spike_number++; } } - data_out->set_nreal_samples(spike_number); + data_out->set_source_timestamp(); data_out->set_hardware_timestamp(data_in->hardware_timestamp()); From 17be688e078322b9ef45bb57fc1f9bb1c26b86e9 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 29 Sep 2022 10:21:56 +0200 Subject: [PATCH 16/22] add check on the depth channelmap given in input --- processors/spikefeatures/spikefeatures.cpp | 48 ++++++++++++++-------- processors/spikefeatures/spikefeatures.hpp | 2 +- 2 files changed, 31 insertions(+), 19 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 616166b..b4bb9c6 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -57,7 +57,7 @@ void SpikeFeatures::Configure(const GlobalContext &context){ features_labels_.push_back(label); } - if(default_features_["depth"].as() and channel_pos_().size() == 0){ + if(default_features_["depth"].as() and channel_pos_().size() == 0 ){ throw ProcessingConfigureError("If the depth feature is selected, the channelmap option " "needs to give the mapping between channel label and depth.", name()); @@ -93,33 +93,45 @@ void SpikeFeatures::CompleteStreamInfo() { throw ProcessingStreamInfoError(err_msg, name()); } - for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { - data_out_port_->streaminfo(s_).set_stream_parameters(data_in_port_->streaminfo(s_)); - data_out_port_->streaminfo(s_).set_parameters( + for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { + data_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); + data_out_port_->streaminfo(slot_).set_parameters( ColumnsType::Parameters(features_labels_, 0, true)); - } + if(default_features_["depth"].as()){ + for(auto chan: data_in_port_->prototype(slot_).labels()){ + try { + channel_pos_().at(chan); + } catch (std::out_of_range) { + throw ProcessingConfigureError("The depth feature is selected but the channelmap given in input " + "does not contain the corresponding depth to the channel " + chan, name()); + } + } + } + } } void SpikeFeatures::Prepare(GlobalContext &context) { // Create an independent spike detector by data stream input spike_detectors_.clear(); - for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { + for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { if(invert_signal_()){ spike_detectors_.push_back( - std::make_unique(data_in_port_->prototype(s_).ncolumns(), + std::make_unique(data_in_port_->prototype(slot_).ncolumns(), initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::DOWN)); }else{ spike_detectors_.push_back( - std::make_unique(data_in_port_->prototype(s_).ncolumns(), + std::make_unique(data_in_port_->prototype(slot_).ncolumns(), initial_threshold_(), initial_peak_lifetime_(), dsp::algorithms::SpikeDetectionSign::UP)); } + + } } @@ -134,24 +146,24 @@ void SpikeFeatures::Process(ProcessingContext &context) { size_t sample = 0; while (!context.terminated()) { - for (s_ = 0; s_ < data_in_port_->number_of_slots(); ++s_) { + for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { - if (!data_in_port_->slot(s_)->RetrieveData(data_in) ){ + if (!data_in_port_->slot(slot_)->RetrieveData(data_in) ){ break; } // claim output data buckets - data_out = data_out_port_->slot(s_)->ClaimData(false); - spike_detectors_[s_]->set_threshold(threshold_->get()); - spike_detectors_[s_]->set_peak_life_time(peak_lifetime_->get()); + data_out = data_out_port_->slot(slot_)->ClaimData(false); + spike_detectors_[slot_]->set_threshold(threshold_->get()); + spike_detectors_[slot_]->set_peak_life_time(peak_lifetime_->get()); unsigned int spike_number = 0; for (sample = 0; sample < data_in->nsamples();++sample) { - if (spike_detectors_[s_]->is_spike( + if (spike_detectors_[slot_]->is_spike( data_in->sample_timestamp(sample), data_in->begin_sample(sample))) { - amp = spike_detectors_[s_]->amplitudes_detected_spike(); + amp = spike_detectors_[slot_]->amplitudes_detected_spike(); max_index = std::max_element(amp.begin(), amp.end()) - amp.begin(); channel_label = data_in->labels()[static_cast(max_index)]; @@ -173,7 +185,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { } if(default_features_["slope"].as()){ - data_out->set_data_sample(spike_number, "slope", spike_detectors_[s_]->slopes_detected_spike()[max_index]); + data_out->set_data_sample(spike_number, "slope", spike_detectors_[slot_]->slopes_detected_spike()[max_index]); feature_str += "slope: " + std::to_string(data_out->data_sample(spike_number, "slope")); } @@ -191,8 +203,8 @@ void SpikeFeatures::Process(ProcessingContext &context) { data_out->set_hardware_timestamp(data_in->hardware_timestamp()); // publish and release data - data_out_port_->slot(s_)->PublishData(); - data_in_port_->slot(s_)->ReleaseData(); + data_out_port_->slot(slot_)->PublishData(); + data_in_port_->slot(slot_)->ReleaseData(); } } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index f790566..3ed4a9f 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -55,7 +55,7 @@ class SpikeFeatures : public IProcessor { const std::string THRESHOLD = "threshold"; std::vector features_labels_; YAML::Node default_features_; - int s_ = 0; + int slot_ = 0; private: options::Double initial_threshold_{70.}; From 6f12f5a39ac8082d77839c51e852381de04dc124 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 29 Sep 2022 10:24:48 +0200 Subject: [PATCH 17/22] update doc --- processors/spikefeatures/doc.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/processors/spikefeatures/doc.yaml b/processors/spikefeatures/doc.yaml index e4c1759..a1b3eca 100644 --- a/processors/spikefeatures/doc.yaml +++ b/processors/spikefeatures/doc.yaml @@ -12,7 +12,7 @@ Output port: - name: data type: ColumnData slots: 1-385 - description: data packet containing spikes with the selected features (fixed size of 100 spikes - filled with 0 spikes) + description: data packet containing spikes with the selected features. The packet is expandable - aka nsamples is resized for the real number of spikes. Options: - &threshold From 009464aa23bc29cc8e3e8c033f38227739612c2b Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 29 Sep 2022 11:13:42 +0200 Subject: [PATCH 18/22] typo on the catching expression --- processors/spikefeatures/spikefeatures.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index b4bb9c6..8e3a490 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -102,7 +102,7 @@ void SpikeFeatures::CompleteStreamInfo() { for(auto chan: data_in_port_->prototype(slot_).labels()){ try { channel_pos_().at(chan); - } catch (std::out_of_range) { + } catch (const std::out_of_range& oor) { throw ProcessingConfigureError("The depth feature is selected but the channelmap given in input " "does not contain the corresponding depth to the channel " + chan, name()); } From 8ce0238292536531eb8bba6eb87e8762ece2ce05 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Thu, 20 Oct 2022 10:49:52 +0200 Subject: [PATCH 19/22] update processors for new columntype format --- processors/spikefeatures/spikefeatures.cpp | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 8e3a490..aa42c1d 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -157,16 +157,15 @@ void SpikeFeatures::Process(ProcessingContext &context) { spike_detectors_[slot_]->set_peak_life_time(peak_lifetime_->get()); unsigned int spike_number = 0; - for (sample = 0; sample < data_in->nsamples();++sample) { + for (sample = 0; sample < data_in->nsamples();sample++) { - if (spike_detectors_[slot_]->is_spike( + if (spike_detectors_[slot_]->is_spike::Data::sample_iterator>( data_in->sample_timestamp(sample), data_in->begin_sample(sample))) { amp = spike_detectors_[slot_]->amplitudes_detected_spike(); max_index = std::max_element(amp.begin(), amp.end()) - amp.begin(); - channel_label = data_in->labels()[static_cast(max_index)]; - + channel_label = data_in->labels()[max_index]; feature_str = "("; if(default_features_["time"].as()){ @@ -175,6 +174,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { } if(default_features_["channel index"].as()){ + data_out->set_data_sample(spike_number, "channel index", std::stod(channel_label)); feature_str += "channel index: " + channel_label; } @@ -190,11 +190,12 @@ void SpikeFeatures::Process(ProcessingContext &context) { } if(default_features_["depth"].as()){ + channel_label = data_in->labels()[max_index]; data_out->set_data_sample(spike_number, "depth", channel_pos_().at(channel_label)); feature_str += "depth: " + std::to_string(data_out->data_sample(spike_number, "depth")); } - LOG(DEBUG)<< "Spike detected : " + feature_str + ")"; + LOG(DEBUG) << name() << " Spike detected : " + feature_str + ") ts= " << sample << " channel =" << channel_label; spike_number++; } } From f58f1f552186afa608051a6f0fe51ff949f197b3 Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 25 Oct 2022 10:31:40 +0200 Subject: [PATCH 20/22] add encoding output and speed threshold --- processors/spikefeatures/spikefeatures.cpp | 51 +++++++++++++++++----- processors/spikefeatures/spikefeatures.hpp | 6 ++- 2 files changed, 45 insertions(+), 12 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index aa42c1d..5856969 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -34,6 +34,9 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option("invert_signal", invert_signal_, "invert a signal to detect negative spikes"); + add_option("do_encoding", do_encoding_, + "Stream the data on the encoding pipeline."); + } void SpikeFeatures::Configure(const GlobalContext &context){ @@ -70,33 +73,51 @@ void SpikeFeatures::CreatePorts(){ PortInPolicy(SlotRange(1, MAX_NCHANNELS))); - data_out_port_ = create_output_port>( + decoding_out_port_ = create_output_port>("decoding", ColumnsType::Parameters(features_labels_, 0, true), PortOutPolicy(SlotRange(1, MAX_NCHANNELS))); + encoding_out_port_ = create_output_port>("encoding", + ColumnsType::Parameters(features_labels_, 0, true), + PortOutPolicy(SlotRange(0, MAX_NCHANNELS))); + threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, Permission::WRITE); peak_lifetime_ = create_static_state(PEAK_LIFETIME, initial_peak_lifetime_(), true, Permission::WRITE); + + to_decode_ = create_follower_state("to decode", true, Permission::READ); } void SpikeFeatures::CompleteStreamInfo() { // check if we have the same number of input and output slots - if (data_in_port_->number_of_slots() != data_out_port_->number_of_slots()) { + if (encoding_out_port_->number_of_slots() != 0 + and data_in_port_->number_of_slots() != encoding_out_port_->number_of_slots()) { auto err_msg = "Number of output slots (" + - std::to_string(data_out_port_->number_of_slots()) + - ") on port '" + data_out_port_->name() + + std::to_string(encoding_out_port_->number_of_slots()) + + ") on port '" + encoding_out_port_->name() + "' does not match number of input slots (" + std::to_string(data_in_port_->number_of_slots()) + ") on port '" + data_in_port_->name() + "'."; throw ProcessingStreamInfoError(err_msg, name()); } + if (data_in_port_->number_of_slots() != decoding_out_port_->number_of_slots()) { + auto err_msg = "Number of output slots (" + + std::to_string(decoding_out_port_->number_of_slots()) + + ") on port '" + decoding_out_port_->name() + + "' does not match number of input slots (" + + std::to_string(data_in_port_->number_of_slots()) + + ") on port '" + data_in_port_->name() + "'."; + throw ProcessingStreamInfoError(err_msg, name()); + } for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { - data_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); - data_out_port_->streaminfo(slot_).set_parameters( - ColumnsType::Parameters(features_labels_, 0, true)); + decoding_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); + //encoding_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); BUG !!! + + decoding_out_port_->streaminfo(slot_).set_parameters(ColumnsType::Parameters(features_labels_, 0, true)); + encoding_out_port_->streaminfo(slot_).set_parameters(ColumnsType::Parameters(features_labels_, 0, true)); if(default_features_["depth"].as()){ for(auto chan: data_in_port_->prototype(slot_).labels()){ @@ -139,6 +160,7 @@ void SpikeFeatures::Prepare(GlobalContext &context) { void SpikeFeatures::Process(ProcessingContext &context) { TimeSeriesType::Data *data_in = nullptr; ColumnsType::Data *data_out = nullptr; + ColumnsType::Data *empty_data_out = nullptr; std::string channel_label; std::vector amp; double max_index; @@ -152,7 +174,14 @@ void SpikeFeatures::Process(ProcessingContext &context) { break; } // claim output data buckets - data_out = data_out_port_->slot(slot_)->ClaimData(false); + if(to_decode_->get()){ + data_out = encoding_out_port_->slot(slot_)->ClaimData(false); + empty_data_out = decoding_out_port_->slot(slot_)->ClaimData(false); + }else{ + data_out = decoding_out_port_->slot(slot_)->ClaimData(false); + empty_data_out = encoding_out_port_->slot(slot_)->ClaimData(false); + } + spike_detectors_[slot_]->set_threshold(threshold_->get()); spike_detectors_[slot_]->set_peak_life_time(peak_lifetime_->get()); @@ -202,9 +231,11 @@ void SpikeFeatures::Process(ProcessingContext &context) { data_out->set_source_timestamp(); data_out->set_hardware_timestamp(data_in->hardware_timestamp()); - + empty_data_out->set_source_timestamp(); + empty_data_out->set_hardware_timestamp(data_in->hardware_timestamp()); // publish and release data - data_out_port_->slot(slot_)->PublishData(); + encoding_out_port_->slot(slot_)->PublishData(); + decoding_out_port_->slot(slot_)->PublishData(); data_in_port_->slot(slot_)->ReleaseData(); } } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 3ed4a9f..9f0d27b 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -36,7 +36,8 @@ class SpikeFeatures : public IProcessor { // DATA PORTS protected: PortIn> *data_in_port_; - PortOut> *data_out_port_; + PortOut> *encoding_out_port_; + PortOut> *decoding_out_port_; protected: std::vector> spike_detectors_; @@ -46,7 +47,7 @@ class SpikeFeatures : public IProcessor { protected: StaticState *threshold_; StaticState *peak_lifetime_; - + FollowerState *to_decode_; // CONSTANTS public: const uint32_t MAX_NCHANNELS=384; @@ -63,4 +64,5 @@ class SpikeFeatures : public IProcessor { options::Value, false> features_{}; options::Value, false> channel_pos_{}; options::Bool invert_signal_{true}; + options::Bool do_encoding_{true}; }; From fdb0a5ac708d620a5218a77fbc01a30fe46faabc Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 25 Oct 2022 13:31:26 +0200 Subject: [PATCH 21/22] remove speed thresholding --- processors/spikefeatures/spikefeatures.cpp | 39 +++------------------- processors/spikefeatures/spikefeatures.hpp | 4 +-- 2 files changed, 5 insertions(+), 38 deletions(-) diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 5856969..953e71e 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -34,9 +34,6 @@ SpikeFeatures::SpikeFeatures() : IProcessor(){ add_option("invert_signal", invert_signal_, "invert a signal to detect negative spikes"); - add_option("do_encoding", do_encoding_, - "Stream the data on the encoding pipeline."); - } void SpikeFeatures::Configure(const GlobalContext &context){ @@ -73,35 +70,19 @@ void SpikeFeatures::CreatePorts(){ PortInPolicy(SlotRange(1, MAX_NCHANNELS))); - decoding_out_port_ = create_output_port>("decoding", + decoding_out_port_ = create_output_port>( ColumnsType::Parameters(features_labels_, 0, true), PortOutPolicy(SlotRange(1, MAX_NCHANNELS))); - encoding_out_port_ = create_output_port>("encoding", - ColumnsType::Parameters(features_labels_, 0, true), - PortOutPolicy(SlotRange(0, MAX_NCHANNELS))); - threshold_ = create_static_state(THRESHOLD, initial_threshold_(), true, Permission::WRITE); peak_lifetime_ = create_static_state(PEAK_LIFETIME, initial_peak_lifetime_(), true, Permission::WRITE); - - to_decode_ = create_follower_state("to decode", true, Permission::READ); } void SpikeFeatures::CompleteStreamInfo() { // check if we have the same number of input and output slots - if (encoding_out_port_->number_of_slots() != 0 - and data_in_port_->number_of_slots() != encoding_out_port_->number_of_slots()) { - auto err_msg = "Number of output slots (" + - std::to_string(encoding_out_port_->number_of_slots()) + - ") on port '" + encoding_out_port_->name() + - "' does not match number of input slots (" + - std::to_string(data_in_port_->number_of_slots()) + - ") on port '" + data_in_port_->name() + "'."; - throw ProcessingStreamInfoError(err_msg, name()); - } if (data_in_port_->number_of_slots() != decoding_out_port_->number_of_slots()) { auto err_msg = "Number of output slots (" + @@ -112,12 +93,10 @@ void SpikeFeatures::CompleteStreamInfo() { ") on port '" + data_in_port_->name() + "'."; throw ProcessingStreamInfoError(err_msg, name()); } + for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { decoding_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); - //encoding_out_port_->streaminfo(slot_).set_stream_parameters(data_in_port_->streaminfo(slot_)); BUG !!! - decoding_out_port_->streaminfo(slot_).set_parameters(ColumnsType::Parameters(features_labels_, 0, true)); - encoding_out_port_->streaminfo(slot_).set_parameters(ColumnsType::Parameters(features_labels_, 0, true)); if(default_features_["depth"].as()){ for(auto chan: data_in_port_->prototype(slot_).labels()){ @@ -160,7 +139,6 @@ void SpikeFeatures::Prepare(GlobalContext &context) { void SpikeFeatures::Process(ProcessingContext &context) { TimeSeriesType::Data *data_in = nullptr; ColumnsType::Data *data_out = nullptr; - ColumnsType::Data *empty_data_out = nullptr; std::string channel_label; std::vector amp; double max_index; @@ -174,14 +152,7 @@ void SpikeFeatures::Process(ProcessingContext &context) { break; } // claim output data buckets - if(to_decode_->get()){ - data_out = encoding_out_port_->slot(slot_)->ClaimData(false); - empty_data_out = decoding_out_port_->slot(slot_)->ClaimData(false); - }else{ - data_out = decoding_out_port_->slot(slot_)->ClaimData(false); - empty_data_out = encoding_out_port_->slot(slot_)->ClaimData(false); - } - + data_out = decoding_out_port_->slot(slot_)->ClaimData(false); spike_detectors_[slot_]->set_threshold(threshold_->get()); spike_detectors_[slot_]->set_peak_life_time(peak_lifetime_->get()); @@ -231,10 +202,8 @@ void SpikeFeatures::Process(ProcessingContext &context) { data_out->set_source_timestamp(); data_out->set_hardware_timestamp(data_in->hardware_timestamp()); - empty_data_out->set_source_timestamp(); - empty_data_out->set_hardware_timestamp(data_in->hardware_timestamp()); // publish and release data - encoding_out_port_->slot(slot_)->PublishData(); + decoding_out_port_->slot(slot_)->PublishData(); data_in_port_->slot(slot_)->ReleaseData(); } diff --git a/processors/spikefeatures/spikefeatures.hpp b/processors/spikefeatures/spikefeatures.hpp index 9f0d27b..280e556 100644 --- a/processors/spikefeatures/spikefeatures.hpp +++ b/processors/spikefeatures/spikefeatures.hpp @@ -36,7 +36,6 @@ class SpikeFeatures : public IProcessor { // DATA PORTS protected: PortIn> *data_in_port_; - PortOut> *encoding_out_port_; PortOut> *decoding_out_port_; protected: @@ -47,7 +46,7 @@ class SpikeFeatures : public IProcessor { protected: StaticState *threshold_; StaticState *peak_lifetime_; - FollowerState *to_decode_; + // CONSTANTS public: const uint32_t MAX_NCHANNELS=384; @@ -64,5 +63,4 @@ class SpikeFeatures : public IProcessor { options::Value, false> features_{}; options::Value, false> channel_pos_{}; options::Bool invert_signal_{true}; - options::Bool do_encoding_{true}; }; From b2aed6775f8b1bf8a6774c3b7686ef840e22ad6c Mon Sep 17 00:00:00 2001 From: marinechaput Date: Tue, 29 Nov 2022 12:10:56 +0100 Subject: [PATCH 22/22] parallelize spike feature processor --- processors/spikefeatures/CMakeLists.txt | 9 ++++++++- processors/spikefeatures/spikefeatures.cpp | 20 +++++++++++--------- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/processors/spikefeatures/CMakeLists.txt b/processors/spikefeatures/CMakeLists.txt index ca9483b..30f8902 100644 --- a/processors/spikefeatures/CMakeLists.txt +++ b/processors/spikefeatures/CMakeLists.txt @@ -1,3 +1,10 @@ ADD_LIBRARY(spikefeatures "spikefeatures.cpp" ) -TARGET_LINK_LIBRARIES(spikefeatures) + +find_package(OpenMP) +if(OpenMP_CXX_FOUND) + TARGET_LINK_LIBRARIES(spikefeatures PUBLIC OpenMP::OpenMP_CXX) + +else() + TARGET_LINK_LIBRARIES(spikefeatures) +endif() diff --git a/processors/spikefeatures/spikefeatures.cpp b/processors/spikefeatures/spikefeatures.cpp index 953e71e..35b6ce5 100644 --- a/processors/spikefeatures/spikefeatures.cpp +++ b/processors/spikefeatures/spikefeatures.cpp @@ -137,6 +137,8 @@ void SpikeFeatures::Prepare(GlobalContext &context) { void SpikeFeatures::Process(ProcessingContext &context) { +#pragma omp parallel +{ TimeSeriesType::Data *data_in = nullptr; ColumnsType::Data *data_out = nullptr; std::string channel_label; @@ -144,21 +146,22 @@ void SpikeFeatures::Process(ProcessingContext &context) { double max_index; std::string feature_str; size_t sample = 0; - - while (!context.terminated()) { + bool should_break=false; + while (!context.terminated() and !should_break) { + #pragma omp for nowait for (slot_ = 0; slot_ < data_in_port_->number_of_slots(); ++slot_) { if (!data_in_port_->slot(slot_)->RetrieveData(data_in) ){ - break; + should_break=true; + continue; } // claim output data buckets - data_out = decoding_out_port_->slot(slot_)->ClaimData(false); + data_out = decoding_out_port_->slot(slot_)->ClaimData(true); spike_detectors_[slot_]->set_threshold(threshold_->get()); spike_detectors_[slot_]->set_peak_life_time(peak_lifetime_->get()); unsigned int spike_number = 0; for (sample = 0; sample < data_in->nsamples();sample++) { - if (spike_detectors_[slot_]->is_spike::Data::sample_iterator>( data_in->sample_timestamp(sample), data_in->begin_sample(sample))) { @@ -198,16 +201,15 @@ void SpikeFeatures::Process(ProcessingContext &context) { LOG(DEBUG) << name() << " Spike detected : " + feature_str + ") ts= " << sample << " channel =" << channel_label; spike_number++; } - } - data_out->set_source_timestamp(); - data_out->set_hardware_timestamp(data_in->hardware_timestamp()); - // publish and release data + } + data_out->CloneTimestamps(*data_in); decoding_out_port_->slot(slot_)->PublishData(); data_in_port_->slot(slot_)->ReleaseData(); } } + } }