FrequencyFilter.cpp

Go to the documentation of this file.

#include "FrequencyFilter.h"
 
#include <chrono>
 
namespace armarx::armem::server::ltm::processor::filter
{
 
    bool
    SnapshotFrequencyFilter::accept(const armem::wm::EntitySnapshot& e, bool simulatedVersion)
    {
        // Thread safety: lock the filter mutex for all mutable state access
        std::lock_guard<std::mutex> lock(filterMutex_);
 
        //accepting to many elements makes the filter slow and brings problems with the buffer with itself!
 
        // set default values for used variables:
        auto start = std::chrono::high_resolution_clock::now();
        bool instances_accepted = false;
        std::int64_t current = 0;
        // get entity id of this snapshot:
        auto memID = e.id().getEntityID();
 
        //find out if timestamps are corrupted:
        if (this->nonCorruptedType == TimestampType::NOT_SET)
        {
            ARMARX_DEBUG << "Setting timestamp type";
            this->setNonCorruptedTimestampType(e);
            ARMARX_DEBUG << VAROUT(this->nonCorruptedType);
        }
 
        if (this->nonCorruptedType == TimestampType::ALL_CORRUPTED)
        {
            return true;
        }
 
        if (this->lastTimesPerEntity.end() == this->lastTimesPerEntity.find(memID))
        {
            //this happens if the key is not in the map, which means this is the first time this
            //entity tries to save a snapshot
            ARMARX_DEBUG << "First time this entity is saved";
            auto firstIndex = e.getInstanceIndices()[0];
            auto firstInsance = e.getInstance(firstIndex);
            auto lastT = this->getNonCorruptedTimestamp(firstInsance, simulatedVersion);
            //for statistics sake:
            auto end = std::chrono::high_resolution_clock::now();
            stats.end_time = end;
            stats.additional_time += (end - start);
            stats.accepted += 1;
            //add this last time to the map:
            this->lastTimesPerEntity[memID] = lastT;
            return true; //the first one is always accepted
        }
        else
        {
            auto lastTime = this->lastTimesPerEntity.at(memID);
 
            // check if any one of the instances for this snapshot were sent in the last x
            // milliseconds since a snapshot was accepted last for this entity:
            e.forEachInstance(
                [this, &instances_accepted, &current, &lastTime, &simulatedVersion](
                    armem::wm::EntityInstance& i)
                {
                    auto t = this->getNonCorruptedTimestamp(i, simulatedVersion);
                    int difference = std::abs(t - lastTime);
                    if (difference > this->maxDifference)
                    { //at least one instance is older than the last saved instance
                        instances_accepted = true;
                        current = this->getNonCorruptedTimestamp(i, simulatedVersion);
                    }
                });
 
            if (instances_accepted)
            { //if one of the instances was accepted the time when an
                //instance was accepted last needs to be changed
                this->lastTimesPerEntity[memID] = current;
            }
        }
 
        //set stats:
        auto end = std::chrono::high_resolution_clock::now();
        stats.end_time = end;
        stats.additional_time += (end - start);
 
        if (instances_accepted)
        {
            this->stats.accepted += 1;
        }
        else
        {
            this->stats.rejected += 1;
        }
 
        return instances_accepted;
    }
 
    void
    SnapshotFrequencyFilter::setNonCorruptedTimestampType(const armem::wm::EntitySnapshot& e)
    {
        auto firstIndex = e.getInstanceIndices()[0];
        auto firstInsance = e.getInstance(firstIndex);
        auto sentTime = firstInsance.metadata().sentTime;
        auto arrivedTime = firstInsance.metadata().arrivedTime;
        auto referencedTime = firstInsance.metadata().referencedTime;
 
        ARMARX_DEBUG << VAROUT(sentTime.toMilliSecondsSinceEpoch());
        ARMARX_DEBUG << VAROUT(arrivedTime.toMilliSecondsSinceEpoch());
        ARMARX_DEBUG << VAROUT(referencedTime.toMilliSecondsSinceEpoch());
 
        //we want the referenced time, if not set correctly we take sent time, then arrvived time:
        if (referencedTime.toMilliSecondsSinceEpoch() < 946688400000)
        {
            //timestamp corrupted:
            if (sentTime.toMilliSecondsSinceEpoch() < 946688400000)
            {
                //sent also corrupted:
                if (arrivedTime.toMilliSecondsSinceEpoch() < 946688400000)
                {
                    //arrived also corrupted:
                    if (!corruptedWarningGiven)
                    {
                        ARMARX_WARNING << "LTM recording does not work correctly, as frequency "
                                          "filter is used, but "
                                       << "time sent, arrived and referenced are all corrupted. \n"
                                       << "Accepting all snapshots.";
                        this->nonCorruptedType = TimestampType::ALL_CORRUPTED;
                        corruptedWarningGiven = true;
                    }
                }
                else
                {
                    if (!corruptedWarningGiven)
                    { //only print this warning once
                        ARMARX_INFO << "Time referenced and sent for snapshot corrupted, using "
                                       "time arrived for filtering";
                        corruptedWarningGiven = true;
                    }
                    this->nonCorruptedType = TimestampType::ARRIVED;
                }
            }
            else
            {
                if (!corruptedWarningGiven)
                { //only print this warning once
                    ARMARX_INFO
                        << "Time referenced snapshot corrupted, using time sent for filtering";
                    corruptedWarningGiven = true;
                }
                this->nonCorruptedType = TimestampType::SENT;
            }
        }
        else
        {
            this->nonCorruptedType = TimestampType::REFERENCED;
        }
 
        /**
        if(sentTime.toMilliSecondsSinceEpoch() < 946688400000){
            // we assume the timestamp does not make sense if it is older than the year 2000
            if(arrivedTime.toMilliSecondsSinceEpoch() < 946688400000){
                if(referencedTime.toMilliSecondsSinceEpoch() < 946688400000){
                    ARMARX_WARNING << "LTM recording does not work correctly, as frequency filter is used, but "
                                   << "time sent, arrived and referenced are all corrupted. \n"
                                   << "Accepting all snapshots.";
                    this->nonCorruptedType = TimestampType::ALL_CORRUPTED;
                } else {
                    if(!corruptedWarningGiven){ //only print this warning once
                        ARMARX_INFO << "Time sent and arrived for snapshot corrupted, using time referenced for filtering";
                        corruptedWarningGiven = true;
                    }
                    //use time referenced from now on
                    this->nonCorruptedType = TimestampType::REFERENCED;
                }
            } else {
                if(!corruptedWarningGiven){ //only print this warning once
                        ARMARX_INFO << "Time sent for snapshot corrupted, using time arrived for filtering";
                        corruptedWarningGiven = true;
                }
                //use time arrived from now on
                this->nonCorruptedType = TimestampType::ARRIVED;
            }
        } else {
            this->nonCorruptedType = TimestampType::SENT;
        }
        */
    }
 
    int
    SnapshotFrequencyFilter::getNonCorruptedTimestamp(const armem::wm::EntityInstance& i,
                                                      bool simulatedVersion)
    {
        switch (this->nonCorruptedType)
        {
            case TimestampType::SENT:
                if (!simulatedVersion)
                {
                    return i.metadata().sentTime.toMilliSecondsSinceEpoch();
                }
                else
                {
                    [[fallthrough]];
                }
            case TimestampType::ARRIVED:
                return i.metadata().arrivedTime.toMilliSecondsSinceEpoch();
            case TimestampType::REFERENCED:
                return i.metadata().referencedTime.toMilliSecondsSinceEpoch();
            default:
                return -1;
        }
    }
 
    void
    SnapshotFrequencyFilter::configure(const nlohmann::json& json)
    {
        std::lock_guard<std::mutex> lock(filterMutex_);
        if (json.find(PARAM_WAITING_TIME) != json.end())
        {
            this->maxDifference = json.at(PARAM_WAITING_TIME);
            ARMARX_INFO << VAROUT(maxDifference);
            stats.additional_info = "Max Difference in ms: " + std::to_string(this->maxDifference);
        }
        stats.start_time = std::chrono::high_resolution_clock::now();
        stats.number_of_compared_objects = 1;
        stats.similarity_type =
            aron::similarity::NDArraySimilarity::Type::NONE; //information for statistics export
    }
 
    SnapshotFilter::FilterStatistics
    SnapshotFrequencyFilter::getFilterStatistics()
    {
        std::lock_guard<std::mutex> lock(filterMutex_);
        stats.end_time = std::chrono::high_resolution_clock::now();
        return this->stats;
    }
 
    std::string
    SnapshotFrequencyFilter::getName()
    {
        return this->NAME;
    }
} // namespace armarx::armem::server::ltm::processor::filter