MemoryBase.h

Go to the documentation of this file.

#pragma once
 
#include <functional>
#include <list>
 
 
// BaseClass
#include "MemoryItem.h"
 
// ChildType
 
// ArmarX
#include <ArmarXCore/core/Component.h>
#include <ArmarXCore/core/logging/LoggingUtil.h>
#include <ArmarXCore/core/time.h>
#include <ArmarXCore/interface/core/UserException.h>
 
#include <RobotAPI/libraries/armem/core/MemoryID.h>
#include <RobotAPI/libraries/armem/core/operations.h>
#include <RobotAPI/libraries/armem/core/wm/aron_conversions.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
#include <RobotAPI/libraries/armem/server/test/ForgettingExperiments.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
namespace armarx::armem::server::ltm
{
    enum class RecordingMode
    {
        CONSOLIDATE_REMOVED, // only store whats removed from the WM
        CONSOLIDATE_ALL, // store whats in the WM (on commit)
        CONSOLIDATE_LATEST // store the latest snapshots at a fixed frequency (for now 1Hz)
    };
} // namespace armarx::armem::server::ltm
 
namespace armarx::armem::server::ltm::detail
{
    /// @brief Interface functions for the longterm memory classes
    template <class _CoreSegmentT>
    class MemoryBase : public MemoryItem
    {
    public:
        struct Statistics
        {
            armarx::core::time::DateTime lastEnabled = armarx::core::time::DateTime::Invalid();
            long recordedCoreSegments = 0;
            bool firstStart = true;
            bool firstStop = true;
            armarx::core::time::DateTime firstStarted = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime firstStopped = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime lastStarted = armarx::core::time::DateTime::Invalid();
            armarx::core::time::DateTime lastStopped = armarx::core::time::DateTime::Invalid();
        };
 
    public:
        using CoreSegmentT = _CoreSegmentT;
 
        MemoryBase(const std::string& exportName, const MemoryID& id) :
            MemoryItem(exportName, id), enabled(false)
        {
        }
 
        void
        setRecordingMode(const std::string& m)
        {
            if (m == "CONSOLIDATE_REMOVED")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_REMOVED;
            }
            else if (m == "CONSOLIDATE_ALL")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_ALL;
            }
            else if (m == "CONSOLIDATE_LATEST")
            {
                this->recordingMode = RecordingMode::CONSOLIDATE_LATEST;
            }
            else
            {
                ARMARX_WARNING << "Unknown recording mode: " << m;
            }
        }
 
        void
        setRecordingMode(const RecordingMode m)
        {
            this->recordingMode = m;
        }
 
        RecordingMode
        getRecordingMode() const
        {
            return recordingMode;
        }
 
        /// initialize config
        void
        configure()
        {
            bool en = p.enabled_on_startup;
            ARMARX_INFO << VAROUT(p.configuration_on_startup);
            ARMARX_INFO << VAROUT(p.export_name);
            ARMARX_INFO << VAROUT(p.export_path);
 
            this->setExportName(p.export_name);
            this->setRecordingMode(p.recordingMode);
 
            try
            {
                const auto json = nlohmann::json::parse(p.configuration_on_startup);
 
                // Processors are shared. So we only need to configure the root
                processors->configure(json);
 
                this->_configure(json);
            }
            catch (...)
            {
                ARMARX_WARNING << "Failed to parse `" << p.configuration_on_startup << "`";
                en = false;
            }
 
            if (en)
            {
                this->startRecording();
            }
        }
 
        /// enable this LTM
        void
        enable()
        {
            processors->resetFilterStatisticsForNewEpisode(); //make sure the statistics are
            //resetted before the recording starts
            auto now = armarx::core::time::DateTime::Now();
            ARMARX_INFO << "Enabling LTM " << id().str() << " at " << now.toDateTimeString();
            ARMARX_INFO << "Storing information at " << this->getExportName();
            enabled = true;
            if (statistics.firstStart)
            {
                statistics.firstStarted = now;
                statistics.firstStart = false;
            }
            statistics.lastStarted = now;
            this->_enable();
        }
 
        /// disable this LTM
        void
        disable()
        {
            auto now = armarx::core::time::DateTime::Now();
            ARMARX_INFO << "Disabling LTM " << id().str() << " at " << now.toDateTimeString();
            enabled = false;
            if (statistics.firstStop)
            {
                statistics.firstStopped = now;
                statistics.firstStop = false;
            }
            statistics.lastStopped = now;
            this->_disable();
            ARMARX_INFO << "Disabling of LTM finished";
        }
 
        /// return the full ltm as a wm::Memory with only references
        /// the ltm may be huge, use with caution
        armem::wm::Memory
        loadAllReferences()
        {
            armem::wm::Memory ret;
            loadAllReferences(ret);
            return ret;
        }
 
        void
        loadAllReferences(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_LoadAll);
            _loadAllReferences(memory);
            TIMING_END_STREAM(LTM_Memory_LoadAll, ARMARX_DEBUG);
        }
 
        //return the newest part of the ltm as a wm::Memory with only references
        //will return the newest n snapshots of each entity
        armem::wm::Memory
        loadLatestNReferences(int n)
        {
            armem::wm::Memory ret;
            loadLatestNReferences(n, ret);
            return ret;
        }
 
        //return the newest part of the ltm as a wm::Memory with only references
        //will return the newest n snapshots of each entity for only the CoreSegments matching one of the name sin the list
        void
        loadLatestNReferences(int n, armem::wm::Memory& memory, std::list<std::string> coreSegNames)
        {
            TIMING_START(LTM_Memory_loadNewestn_someCoreSegs);
            _loadLatestNReferences(n, memory, coreSegNames);
            TIMING_END_STREAM(LTM_Memory_loadNewestn_someCoreSegs, ARMARX_DEBUG);
        }
 
        void
        loadLatestNReferences(int n, armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_loadNewestn);
            _loadLatestNReferences(n, memory);
            TIMING_END_STREAM(LTM_Memory_loadNewestn, ARMARX_DEBUG);
        }
 
        /// return the full ltm as a wm::Memory and resolves the references
        /// the ltm may be huge, use with caution
        armem::wm::Memory
        loadAllAndResolve()
        {
            armem::wm::Memory ret;
            loadAllAndResolve(ret);
            return ret;
        }
 
        void
        loadAllAndResolve(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_LoadAllAndResolve);
            _loadAllReferences(memory);
            _resolve(memory);
            TIMING_END_STREAM(LTM_Memory_LoadAllAndResolve, ARMARX_DEBUG);
        }
 
        /// convert the references of the input into a wm::Memory
        void
        resolve(armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_Load);
            _resolve(memory);
            TIMING_END_STREAM(LTM_Memory_Load, ARMARX_DEBUG);
        }
 
        /// append a wm::Memory instance to the ltm
        void
        store(const armem::wm::Memory& memory)
        {
            TIMING_START(LTM_Memory_Append);
 
            for (auto& f : processors->memFilters)
            {
                if (!f->accept(memory))
                {
                    ARMARX_DEBUG
                        << deactivateSpam(60)
                        << "Ignoring to put a Memory into the LTM because it got filtered.";
                    return;
                }
            }
            _directlyStore(memory);
            TIMING_END_STREAM(LTM_Memory_Append, ARMARX_DEBUG);
        }
 
        /// append a wm::Memory instance to the ltm
        void
        store(const armem::server::wm::Memory& serverMemory)
        {
            armem::wm::Memory memory(serverMemory.name());
            memory.update(armem::toCommit(serverMemory), true, false);
            this->store(memory);
        }
 
        /// iterate over all core segments of this ltm
        virtual bool forEachCoreSegment(std::function<void(CoreSegmentT&)> func) const = 0;
 
        /// check if core segment exists
        virtual bool hasCoreSegment(const std::string&) const = 0;
 
        /// find core segment
        virtual std::shared_ptr<CoreSegmentT> findCoreSegment(const std::string&) const = 0;
 
        /// default parameters. Implementation should use the configuration to configure
        virtual void
        createPropertyDefinitions(PropertyDefinitionsPtr& defs, const std::string& prefix)
        {
            defs->optional(p.enabled_on_startup, prefix + "enabled");
            defs->optional(p.recordingMode, prefix + "recordingMode");
            defs->optional(p.configuration_on_startup, prefix + "configuration");
            defs->optional(p.export_name, prefix + "exportName");
            defs->optional(p.storeOnStop, prefix + "storeOnStop");
            defs->optional(p.export_path, prefix + "exportPath");
 
            defs->optional(p.importOnStartUp, prefix + "importOnStartUp");
            defs->optional(p.maxAmountOfSnapshotsLoaded, prefix + "maxAmountSnapshotsLoaded");
            defs->optional(p.coreSegmentsToLoad, prefix + "loadedCoreSegments");
 
            defs->optional(p.persistenceStrategyIdentifier,
                           prefix + "persistenceStrategyIdentifier");
            defs->optional(p.persistenceStrategies, prefix + "persistenceStrategies");
            defs->optional(p.restHost, prefix + "rest.host");
            defs->optional(p.restPort, prefix + "rest.port");
            defs->optional(p.restDisableIfNotAvailable, prefix + "rest.disableIfNotAvailable");
 
            defs->optional(p.minDiskSpace, prefix + "minDiskSpace");
        }
 
        /// enable/disable
        void
        startRecording()
        {
            statistics.lastEnabled = armarx::core::time::DateTime::Now();
 
            enable();
        }
 
        void
        stopRecording()
        {
 
            disable();
        }
 
        bool
        isRecording() const
        {
            return enabled;
        }
 
        /// statistics
        virtual void
        resetStatistics()
        {
            // enabled stays the same
            statistics.lastEnabled = armarx::core::time::DateTime::Invalid();
            statistics.recordedCoreSegments = 0;
        }
 
        Statistics
        getStatistics() const
        {
            return statistics;
        }
 
        std::map<std::string, processor::SnapshotFilter::FilterStatistics>
        getFilterStatistics()
        {
            try
            {
                ARMARX_INFO << "Trying to save statistics";
                auto stats = processors->getSnapshotFilterStatistics();
                return stats;
            }
            catch (InvalidArgumentException& e)
            { //no data was actually recorded
                ARMARX_INFO << e.what();
            }
            catch (...)
            {
                ARMARX_WARNING << "Saving statistics did not work";
            }
            std::map<std::string, processor::SnapshotFilter::FilterStatistics> emptyStatistics;
            return emptyStatistics;
        }
 
        /// get level name1
        static std::string
        getLevelName()
        {
            return "LT-Memory";
        }
 
        void
        loadOnStartup()
        {
            _loadOnStartup();
        }
 
 
    protected:
        /// configuration
        virtual void
        _configure(const nlohmann::json&)
        {
        }
 
        virtual void
        _enable()
        {
        }
 
        virtual void
        _disable()
        {
        }
 
        virtual void
        _setExportName(const std::string&)
        {
        }
 
        virtual void _loadAllReferences(armem::wm::Memory& memory) = 0;
        virtual void _resolve(armem::wm::Memory& memory) = 0;
        virtual void _store(const armem::wm::Memory& memory) = 0;
        virtual void _directlyStore(const armem::wm::Memory& memory,
                                    bool simulatedVersion = false) = 0;
        virtual void _loadOnStartup() = 0;
        virtual void _loadLatestNReferences(int n, armem::wm::Memory& memory) = 0;
        virtual void _loadLatestNReferences(int n,
                                            armem::wm::Memory& memory,
                                            std::list<std::string> coreSegNames) = 0;
 
    public:
        // stuff for scenario parameters
        struct Properties
        {
            // whether the LTM is enabled on startup
            bool enabled_on_startup = false;
 
            // How data is stored, if the LTM is enabled
            std::string recordingMode = "CONSOLIDATE_REMOVED";
 
            // Name and export path (TODO: Should this be part of the base class? Export path is a disk memory thing)
            std::string export_name = "MemoryExport";
            std::string export_path = "/tmp/ltm";
            bool storeOnStop = false;
            int minDiskSpace = 50; //GB
 
            // TODO: belong more to WM, but no good solution for that currently:
            bool importOnStartUp = false;
            int maxAmountOfSnapshotsLoaded = 1;
            std::string coreSegmentsToLoad = "";
 
            // Other configuration
            std::string configuration_on_startup =
                "{ \"SnapshotFrequencyFilter\": "
                "{\"WaitingTimeInMsForFilter\" : 50}}"; //record with 20 fps as standard
 
            std::string persistenceStrategies = "disk";
            std::string persistenceStrategyIdentifier = "localDisk";
            std::string restHost = "localhost";
            int restPort = 8080;
            bool restDisableIfNotAvailable = false;
 
            bool overwriteIdentifierOnLtmLoad = true;
 
        } p;
 
    protected:
        mutable std::recursive_mutex ltm_mutex;
 
        mutable Statistics statistics;
 
        std::atomic_bool enabled = true;
 
        RecordingMode recordingMode = RecordingMode::CONSOLIDATE_REMOVED;
    };
} // namespace armarx::armem::server::ltm::detail