mesytec-mnode/src/tools/mana_auto_replay.cc

// mana - mnode analysis / mini analysis
// usage: mana_auto_replay <zipfile>
//
// open the file
// read preamble
// create crateconfig from preamble data
//
// setup:
// for each event in the config:
//   for each module:
//     check for module type in meta data
//     find module type in vme_module_data_sources.json
//     for each filter in the modules data_sources:
//       calculate array size (use float as the storage type for now)
//       reserve buffer space for the array
//
// replay:
// for each event in the listfile:
//   for each module in the event:
//     locate the list of filters for the crate, event and module index triplet
//     for each filter:
//       match it against every word in the module data
//       if a word matches extract address and value
//       store value in the reserved array buffer
//
//    -> extracted u32 data is stored in the buffer space for this event
//    what now?
//      - histogram
//      - make arrays available to python and test numpy
//      - crrate root histograms and accumulate
//      - create root trees or the new rntuples(?)
//    -> want plugins. similar to the mvme listfile_reader but on analysis data

#include <cmrc/cmrc.hpp> // mnode::resources
#include <mesytec-mvlc/mesytec-mvlc.h>
#include <mesytec-mnode/mnode_cpp_types.h>
#include <mesytec-mnode/mnode_math.h>
#include <nlohmann/json.hpp>
#include <list>
#include <sstream>
#include "internal/mana_arena.h"
#include "internal/mana_lib.hpp"
#include "internal/mana_analysis.h"

CMRC_DECLARE(mnode::resources);

using namespace mesytec;
using namespace mesytec::mnode;

struct ListfileContext
{
    std::unique_ptr<mvlc::listfile::ZipReader> zipReader;
    mvlc::listfile::ReadHandle *readHandle;
    mvlc::listfile::ListfileReaderHelper readerHelper;
    mvlc::CrateConfig crateConfig;

    ListfileContext() = default;
    ListfileContext(const ListfileContext &) = delete;
    ListfileContext &operator=(const ListfileContext &) = delete;
    ListfileContext(ListfileContext &&) = default;
    ListfileContext &operator=(ListfileContext &&) = default;
};

std::optional<ListfileContext> make_listfile_context(const std::string &filename)
{
    try
    {
        ListfileContext ctx;
        ctx.zipReader = std::make_unique<mvlc::listfile::ZipReader>();
        ctx.zipReader->openArchive(filename);
        ctx.readHandle = ctx.zipReader->openEntry(ctx.zipReader->firstListfileEntryName());
        ctx.readerHelper = mvlc::listfile::make_listfile_reader_helper(ctx.readHandle);
        auto configData = ctx.readerHelper.preamble.findCrateConfig();
        if (!configData)
        {
            std::cerr << fmt::format("No MVLC crate config found in {}\n", filename);
            return {};
        }

        ctx.crateConfig = mvlc::crate_config_from_yaml(configData->contentsToString());
        std::cout << fmt::format("Found MVLC crate config in {}\n", filename);
        for (size_t i = 0; i < ctx.crateConfig.stacks.size(); ++i)
        {
            std::cout << fmt::format("event[{}] {}:\n", i, ctx.crateConfig.stacks[i].getName());
            size_t mi = 0;
            for (const auto &module_: ctx.crateConfig.stacks[i].getGroups())
            {
                auto meta = fmt::format("{}", fmt::join(module_.meta, ", "));
                std::cout << fmt::format("  module[{}]: name={}, meta={{{}}}", mi++, module_.name,
                                         meta);
                if (!mvlc::produces_output(module_))
                    std::cout << ", (no output)";
                std::cout << "\n";
            }
        }

        return ctx;
    }
    catch (const std::exception &e)
    {
        std::cerr << fmt::format("Error: {}\n", e.what());
        return {};
    }
}

struct ParserContext
{
    using Parser = mvlc::readout_parser::ReadoutParserState;
    using Counters = mvlc::readout_parser::ReadoutParserCounters;
    using Callbacks = mvlc::readout_parser::ReadoutParserCallbacks;

    Parser parser;
    Counters counters;
    Callbacks callbacks;
    mvlc::CrateConfig crateConfig;
};

std::optional<ParserContext> make_parser_context(const mvlc::CrateConfig &crateConfig,
                                                 ParserContext::Callbacks callbacks)
{
    try
    {
        ParserContext ctx{};
        ctx.parser = mvlc::readout_parser::make_readout_parser(crateConfig.stacks);
        ctx.crateConfig = crateConfig;
        ctx.callbacks = callbacks;
        return ctx;
    }
    catch (const std::exception &e)
    {
        std::cerr << fmt::format("Error: {}\n", e.what());
        return {};
    }
}

size_t process_one_buffer(size_t bufferNumber, ListfileContext &listfileContext,
                          ParserContext &parserContext, mana::ModuleDataStage &analysisContext)
{
    listfileContext.readerHelper.destBuf().clear();
    auto buffer = mvlc::listfile::read_next_buffer(listfileContext.readerHelper);

    if (!buffer->used())
        return 0;

    auto bufferView = buffer->viewU32();
    parserContext.parser.userContext = &analysisContext;

    mvlc::readout_parser::parse_readout_buffer(
        listfileContext.readerHelper.bufferFormat, parserContext.parser, parserContext.callbacks,
        parserContext.counters, bufferNumber, bufferView.data(), bufferView.size());

    return buffer->used();
}

int main(int argc, char *argv[])
{
    auto f = cmrc::mnode::resources::get_filesystem().open("data/vme_module_data_sources.json");
    const auto jModuleDataSources = nlohmann::json::parse(f.begin(), f.end());

    if (argc < 2)
    {
        std::cout << fmt::format("Usage: {} <listfile.zip>\n", argv[0]);
        return 1;
    }

    std::string filename = argv[1];
    auto listfileContext = make_listfile_context(filename);

    if (!listfileContext)
        return 1;

    mana::ManaCountingSink manaPlugin;
    auto mana = mana::make_module_data_stage(filename, mana::Arena(), listfileContext->crateConfig,
                                             jModuleDataSources, &manaPlugin, nullptr);

    auto event_data = [](void *ctx_, int crateIndex, int eventIndex,
                         const mvlc::readout_parser::ModuleData *moduleDataList,
                         unsigned moduleCount)
    {
        (void)crateIndex;
        auto ctx = reinterpret_cast<mana::ModuleDataStage *>(ctx_);
        mana::module_data_stage_process_module_data(*ctx, eventIndex, moduleDataList, moduleCount);
    };

    auto system_event = [](void *ctx_, int crateIndex, const u32 *header, u32 size)
    {
        (void)crateIndex;
        auto ctx = reinterpret_cast<mana::ModuleDataStage *>(ctx_);
        mana::module_data_stage_process_system_event(*ctx, header, size);
    };

    auto parserContext =
        make_parser_context(listfileContext->crateConfig, {event_data, system_event});

    if (!parserContext)
        return 1;

    mana.sinkContext = mana.sink->init();
    mana.sink->begin_run(mana.sinkContext, mana.runDescriptor.dump().c_str());

    size_t bufferNumber = 0;
    size_t totalBytesProcessed = 0;
    size_t bytesProcessed = 0;
    const std::chrono::milliseconds ReportInterval(500);
    mvlc::util::Stopwatch sw;

    auto report = [&]
    {
        [](const mvlc::util::Stopwatch sw, size_t bufferNumber, size_t totalBytesProcessed)
        {
            auto s = sw.get_elapsed().count() / (1000.0 * 1000.0);
            auto bytesPerSecond = totalBytesProcessed / s;
            auto MiBPerSecond = bytesPerSecond / (1u << 20);
            std::cout << fmt::format("Processed {} buffers, {} bytes. t={} s, rate={} MiB/s\n",
                                     bufferNumber, totalBytesProcessed, s, MiBPerSecond);
        }(sw, bufferNumber, totalBytesProcessed);
    };

    do
    {
        bytesProcessed = process_one_buffer(bufferNumber, *listfileContext, *parserContext, mana);
        totalBytesProcessed += bytesProcessed;
        ++bufferNumber;

        if (auto elapsed = sw.get_interval(); elapsed >= ReportInterval)
        {
            report();
            sw.interval();
        }
    }
    while (bytesProcessed > 0);

    report();

    mana.sink->end_run(mana.sinkContext, mana.runDescriptor.dump().c_str());
    mana.sink->shutdown(mana.sinkContext);

    spdlog::info(
        "ManaCountingSink: eventCounts=[{}], totalBytes={}, systemEvents={}, systemEventBytes={}",
        fmt::join(manaPlugin.eventCounts, ", "), manaPlugin.totalBytes, manaPlugin.systemEventCount,
        manaPlugin.systemEventBytes);

    for (size_t ei = 0; ei < manaPlugin.eventArrayHits.size(); ++ei)
    {
        spdlog::info("event[{}]: {} hits", ei, manaPlugin.eventCounts[ei]);
        for (size_t ai = 0; ai < manaPlugin.eventArrayHits[ei].size(); ++ai)
        {
            auto name = mana.runDescriptor["events"][ei]["outputs"][ai]["name"];
            spdlog::info("  {}: {}", name, manaPlugin.eventArrayHits[ei][ai]);
        }
    }

    return 0;
}