Processors.cpp

Go to the documentation of this file.

#include "Processors.h"
 
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/interface/core/UserException.h>
 
#include "converter/data/image/exr/ExrConverter.h"
#include "converter/data/image/png/PngConverter.h"
#include "extractor/imageExtractor/DepthImageExtractor.h"
#include "extractor/imageExtractor/ImageExtractor.h"
#include "filter/equalityFilter/EqualityFilter.h"
#include "filter/frequencyFilter/FrequencyFilter.h"
#include "filter/importanceFilter/ImportanceFilter.h"
 
namespace armarx::armem::server::ltm
{
    std::vector<std::unique_ptr<processor::SnapshotFilter>>
    Processors::buildSnapshotFilters(const nlohmann::json& config)
    {
        std::vector<std::unique_ptr<processor::SnapshotFilter>> filters;
        if (config.contains(processor::filter::SnapshotFrequencyFilter::NAME))
        {
            ARMARX_IMPORTANT << "ADDING SNAPSHOT FREQUENCY FILTER";
            auto f = std::make_unique<processor::filter::SnapshotFrequencyFilter>();
            f->configure(config[processor::filter::SnapshotFrequencyFilter::NAME]);
            filters.push_back(std::move(f));
        }
        if (config.contains(processor::filter::SnapshotSimilarityFilter::NAME))
        {
            ARMARX_IMPORTANT << "ADDING SNAPSHOT SIMILARITY FILTER";
            auto f = std::make_unique<processor::filter::SnapshotSimilarityFilter>();
            f->configure(config[processor::filter::SnapshotSimilarityFilter::NAME]);
            filters.push_back(std::move(f));
        }
        if (config.contains(processor::filter::SnapshotImportanceFilter::NAME))
        {
            ARMARX_IMPORTANT << "ADDING SNAPSHOT IMPORTANCE FILTER";
            auto f = std::make_unique<processor::filter::SnapshotImportanceFilter>();
            f->configure(config[processor::filter::SnapshotImportanceFilter::NAME]);
            filters.push_back(std::move(f));
        }
        return filters;
    }
 
    void
    Processors::configure(const nlohmann::json& config)
    {
        // Global / default snapshot filters:
        snapFilters = buildSnapshotFilters(config);
 
        // Segment whitelist: which segments to store. Absent => store all segments.
        storeSegments.reset();
        if (config.contains("storeSegments"))
        {
            std::set<std::string> whitelist;
            for (const auto& entry : config["storeSegments"])
            {
                whitelist.insert(entry.get<std::string>());
            }
            ARMARX_IMPORTANT << "LTM storeSegments whitelist active with " << whitelist.size()
                             << " entries";
            storeSegments = std::move(whitelist);
        }
 
        // Per-segment snapshot filter overrides, keyed by "Core" or "Core/Provider".
        segmentSnapFilters.clear();
        if (config.contains("segments"))
        {
            for (const auto& [segmentPath, segmentConfig] : config["segments"].items())
            {
                ARMARX_IMPORTANT << "ADDING PER-SEGMENT FILTERS FOR " << segmentPath;
                segmentSnapFilters[segmentPath] = buildSnapshotFilters(segmentConfig);
            }
        }
 
        // Converters
        if (config.contains(processor::converter::data::image::PngConverter::NAME))
        {
            ARMARX_IMPORTANT << "ADDING IMG CONVERTER PNG";
            auto f = std::make_unique<processor::converter::data::image::PngConverter>();
            f->configure(config[processor::converter::data::image::PngConverter::NAME]);
            converters.push_back(std::move(f));
        }
        if (config.contains(processor::converter::data::image::ExrConverter::NAME))
        {
            ARMARX_IMPORTANT << "ADDING IMG CONVERTER EXR";
            auto f = std::make_unique<processor::converter::data::image::ExrConverter>();
            f->configure(config[processor::converter::data::image::ExrConverter::NAME]);
            converters.push_back(std::move(f));
        }
    }
 
    bool
    Processors::hasSegmentSelection() const
    {
        return storeSegments.has_value() || !segmentSnapFilters.empty();
    }
 
    bool
    Processors::isSegmentWhitelisted(const std::string& coreSegmentName,
                                     const std::string& providerSegmentName) const
    {
        if (!storeSegments.has_value())
        {
            // No whitelist configured => store all segments.
            return true;
        }
        const auto& whitelist = *storeSegments;
        // A core segment is admitted either by its core-segment name or by the
        // specific provider-segment path.
        if (whitelist.count(coreSegmentName) > 0)
        {
            return true;
        }
        if (!providerSegmentName.empty() &&
            whitelist.count(coreSegmentName + "/" + providerSegmentName) > 0)
        {
            return true;
        }
        return false;
    }
 
    bool
    Processors::acceptSnapshotForSegment(const std::string& coreSegmentName,
                                         const std::string& providerSegmentName,
                                         const armem::wm::EntitySnapshot& snapshot,
                                         bool simulatedVersion)
    {
        // 1. Whitelist: reject snapshots of segments that are not selected for storage.
        if (!isSegmentWhitelisted(coreSegmentName, providerSegmentName))
        {
            return false;
        }
 
        // 2. Pick the applicable filter set: provider-specific override, else core-segment
        //    override, else the global filters.
        const std::string providerPath = coreSegmentName + "/" + providerSegmentName;
        std::vector<std::unique_ptr<processor::SnapshotFilter>>* filters = &snapFilters;
        if (auto it = segmentSnapFilters.find(providerPath); it != segmentSnapFilters.end())
        {
            filters = &it->second;
        }
        else if (auto cit = segmentSnapFilters.find(coreSegmentName);
                 cit != segmentSnapFilters.end())
        {
            filters = &cit->second;
        }
 
        for (auto& filter : *filters)
        {
            if (!filter->accept(snapshot, simulatedVersion))
            {
                return false;
            }
        }
        return true;
    }
 
    std::map<std::string, processor::SnapshotFilter::FilterStatistics>
    Processors::getSnapshotFilterStatistics()
    {
        std::map<std::string, processor::SnapshotFilter::FilterStatistics> stats;
        bool recordedOverall = false;
        std::size_t numFilters = snapFilters.size();
 
        auto collect = [&](const std::string& keyPrefix,
                           const std::vector<std::unique_ptr<processor::SnapshotFilter>>& filters)
        {
            for (const auto& filter_ptr : filters)
            {
                auto statistics = filter_ptr->getFilterStatistics();
                auto recorded = statistics.accepted + statistics.rejected;
                stats[keyPrefix + filter_ptr->getName()] = statistics;
                if (recorded > 0)
                {
                    recordedOverall = true;
                }
            }
        };
 
        collect("", snapFilters);
        for (const auto& [segmentPath, filters] : segmentSnapFilters)
        {
            numFilters += filters.size();
            collect(segmentPath + ":", filters);
        }
 
        ARMARX_INFO << "Number of active filters: " << numFilters;
 
        // Only complain about empty statistics if there actually are filters configured.
        // With a pure segment whitelist (and no filters), recording nothing is expected.
        if (numFilters > 0 && !recordedOverall)
        {
            throw InvalidArgumentException(
                "NoFilters recorded any data being accepted or rejected, "
                "cant store empty statistics");
        }
        return stats;
    }
 
    void
    Processors::resetFilterStatisticsForNewEpisode()
    {
        ARMARX_DEBUG << "Resetting statistics for filters";
        for (const auto& filter_ptr : this->snapFilters)
        {
            filter_ptr->resetStatisticsForNewEpisode();
        }
        for (const auto& [segmentPath, filters] : this->segmentSnapFilters)
        {
            for (const auto& filter_ptr : filters)
            {
                filter_ptr->resetStatisticsForNewEpisode();
            }
        }
    }
} // namespace armarx::armem::server::ltm