Program Listing for File likelihoodsmerger.cpp

↰ Return to documentation for file (processors/likelihoodsmerger/likelihoodsmerger.cpp)

// ---------------------------------------------------------------------
// 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 <http://www.gnu.org/licenses/>.
// ---------------------------------------------------------------------

#include "likelihoodsmerger.hpp"

LikelihoodsMerger::LikelihoodsMerger() : IProcessor(){
    add_option("normalize", normalize_, "enable likelihood normalization");
}

void LikelihoodsMerger::CreatePorts(){
    data_in_port_ = create_input_port<ColumnsType<double>>("loglikelihood",
                                                           ColumnsType<double>::Capabilities({"logL"},
                                                                                             SampleRange(0, std::numeric_limits<uint32_t>::max()), true),
                                                            PortInPolicy(SlotRange(0, MAX_NCHANNELS)));

    data_out_port_ = create_output_port<ColumnsType<double>>("posterior", ColumnsType<double>::Parameters({"posterior likelihood"}, 0, true),
                                                             PortOutPolicy(SlotRange(1)));

    shared_decoder_= create_follower_state("decoder", (std::shared_ptr<Decoder>*)nullptr, Permission::NONE);

}

void LikelihoodsMerger::CompleteStreamInfo() {
    nslots_ = data_in_port_->number_of_slots();
    data_out_port_->streaminfo(0).set_stream_parameters(data_in_port_->streaminfo(0));

}

void LikelihoodsMerger::Prepare(GlobalContext &context) {
    if(shared_decoder_->get() == nullptr){
        throw ProcessingPrepareError("The decoder state is not connected "
                                     "to a Encoding processor (used for load/update the model).", name());
    }
    grid_size_ = shared_decoder_->get()->get()->grid_size();
    prior_ = shared_decoder_->get()->get()->prior();

}

void LikelihoodsMerger::Process(ProcessingContext &context){
    ColumnsType<double>::Data* data_in = nullptr;
    ColumnsType<double>::Data *data_out = nullptr;
    uint64_t hardware_ts;
    std::vector<double> logL(grid_size_);
    std::fill(logL.begin(), logL.end(), 0);
    bool should_break=false;

    while (!context.terminated()) {

        for (slot_ = 0; slot_ < nslots_; ++slot_) {

            if (!data_in_port_->slot(slot_)->RetrieveData(data_in)) {
                should_break = true;
                continue;
            }

            hardware_ts = data_in->hardware_timestamp();

            // Merge all logL sources by summation
            std::transform(logL.begin(), logL.end(), data_in->begin_column("logL"),
                           logL.begin(), std::plus<double>());

            data_in_port_->slot(slot_)->ReleaseData();
        }

        if(should_break){
            break;
        }

        compute_posterior(logL.data(), prior_, grid_size_, normalize_());
        data_out = data_out_port_->slot(0)->ClaimData(false);
        data_out->set_hardware_timestamp(hardware_ts);
        data_out->set_source_timestamp();
        data_out->set_data_column("posterior likelihood", logL);
        data_out_port_->slot(0)->PublishData();
        std::fill(logL.begin(), logL.end(), 0);

    }
}

REGISTERPROCESSOR(LikelihoodsMerger)