ReadWritePluginUser.cpp

Go to the documentation of this file.

#include "ReadWritePluginUser.h"
 
#include <SimoxUtility/algorithm/string/string_tools.h> //for loadExportToLTM
 
#include <ArmarXCore/core/Component.h>
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
 
#include <RobotAPI/libraries/armem/core/Prediction.h>
#include <RobotAPI/libraries/armem/core/error.h>
#include <RobotAPI/libraries/armem/server/MemoryToIceAdapter.h>
#include <RobotAPI/libraries/armem/server/ltm/persistence/RedundantPersistenceStrategy.h>
#include <RobotAPI/libraries/armem/server/ltm/persistence/disk/DiskPersistence.h>
#include <RobotAPI/libraries/armem/server/ltm/persistence/rest/RestPersistence.h>
 
#include "Plugin.h"
 
namespace armarx::armem::server::plugins
{
 
    ReadWritePluginUser::ReadWritePluginUser()
    {
        addPlugin(plugin);
    }
 
    ReadWritePluginUser::~ReadWritePluginUser()
    {
    }
 
    void
    ReadWritePluginUser::setMemoryName(const std::string& memoryName)
    {
        plugin->setMemoryName(memoryName);
    }
 
    // WRITING
    data::AddSegmentsResult
    ReadWritePluginUser::addSegments(const data::AddSegmentsInput& input, const Ice::Current&)
    {
        ARMARX_TRACE;
        bool addCoreSegmentOnUsage = false;
        return addSegments(input, addCoreSegmentOnUsage);
    }
 
    data::AddSegmentsResult
    ReadWritePluginUser::addSegments(const data::AddSegmentsInput& input, bool addCoreSegments)
    {
        ARMARX_TRACE;
        data::AddSegmentsResult result = iceAdapter().addSegments(input, addCoreSegments);
        return result;
    }
 
    data::CommitResult
    ReadWritePluginUser::commit(const data::Commit& commitIce, const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().commitLocking(commitIce);
    }
 
    void
    ReadWritePluginUser::clearWorkingMemory(const Ice::Current&)
    {
        ARMARX_TRACE;
 
        ARMARX_DEBUG << "Clear working memory id="
                     << iceAdapter().workingMemory->id().cleanID().str();
 
        iceAdapter().workingMemory->forEachCoreSegment([](auto& coreSegment)
                                                       { coreSegment.clear(); });
    }
 
    // READING
    armem::query::data::Result
    ReadWritePluginUser::query(const armem::query::data::Input& input, const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().query(input);
    }
 
    armem::query::data::Result
    ReadWritePluginUser::queryLTM(const armem::query::data::Input& input,
                                  bool storeIntoWM,
                                  const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().queryLTM(input, storeIntoWM);
    }
 
    structure::data::GetServerStructureResult
    ReadWritePluginUser::getServerStructure(const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().getServerStructure();
    }
 
    // LTM STORING AND RECORDING
    dto::DirectlyStoreResult
    ReadWritePluginUser::directlyStore(const dto::DirectlyStoreInput& input, const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().directlyStore(input);
    }
 
    dto::StartRecordResult
    ReadWritePluginUser::startRecord(const dto::StartRecordInput& startRecordInput,
                                     const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().startRecord(startRecordInput);
    }
 
    dto::StopRecordResult
    ReadWritePluginUser::stopRecord(const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().stopRecord();
    }
 
    dto::RecordStatusResult
    ReadWritePluginUser::getRecordStatus(const Ice::Current&)
    {
        ARMARX_TRACE;
        return iceAdapter().getRecordStatus();
    }
 
    Plugin&
    ReadWritePluginUser::memoryServerPlugin()
    {
        return *plugin;
    }
 
    wm::Memory&
    ReadWritePluginUser::workingMemory()
    {
        return plugin->workingMemory;
    }
 
    MemoryToIceAdapter&
    ReadWritePluginUser::iceAdapter()
    {
        return plugin->iceAdapter;
    }
 
    ltm::Memory&
    ReadWritePluginUser::longtermMemory()
    {
        return plugin->longtermMemory;
    }
 
    // ACTIONS
    armem::actions::GetActionsOutputSeq
    ReadWritePluginUser::getActions(const armem::actions::GetActionsInputSeq& inputs,
                                    const ::Ice::Current& /*unused*/)
    {
        return getActions(inputs);
    }
 
    armem::actions::GetActionsOutputSeq
    ReadWritePluginUser::getActions(const armem::actions::GetActionsInputSeq& inputs)
    {
        (void)inputs;
        return {};
    }
 
    armem::actions::ExecuteActionOutputSeq
    ReadWritePluginUser::executeActions(const armem::actions::ExecuteActionInputSeq& inputs,
                                        const ::Ice::Current& /*unused*/)
    {
        return executeActions(inputs);
    }
 
    armem::actions::ExecuteActionOutputSeq
    ReadWritePluginUser::executeActions(const armem::actions::ExecuteActionInputSeq& inputs)
    {
        return {};
    }
 
    // PREDICTIONS
    armem::prediction::data::PredictionResultSeq
    ReadWritePluginUser::predict(const armem::prediction::data::PredictionRequestSeq& requests)
    {
        return iceAdapter().predict(requests);
    }
 
    armem::prediction::data::EngineSupportMap
    ReadWritePluginUser::getAvailableEngines()
    {
        return iceAdapter().getAvailableEngines();
    }
 
    armem::prediction::data::PredictionResultSeq
    ReadWritePluginUser::predict(const armem::prediction::data::PredictionRequestSeq& requests,
                                 const ::Ice::Current& /*unused*/)
    {
        return predict(requests);
    }
 
    armem::prediction::data::EngineSupportMap
    ReadWritePluginUser::getAvailableEngines(const ::Ice::Current& /*unused*/)
    {
        return getAvailableEngines();
    }
 
    armem::CommitResult
    ReadWritePluginUser::reloadFromLTM()
    {
        // Only take the core segments defined in the property 'loadedCoreSegments'
        // Why not full? Legacy code, previously also only retrieved not all core segments.
        iceAdapter().reloadFromLTMOnStartup();
        return armem::CommitResult();
    }
 
    void
    ReadWritePluginUser::useRestPersistenceStrategy(const std::string& identifier,
                                                    const std::string& exportName,
                                                    const std::string& host,
                                                    int port,
                                                    bool overwritePrevStrategy,
                                                    const ::Ice::Current&)
    {
        ARMARX_IMPORTANT << "Setting persistence strategy=rest via PROXY";
 
        std::shared_ptr<armem::server::ltm::persistence::RestPersistence> restPersistence =
            std::make_shared<armem::server::ltm::persistence::RestPersistence>(
                identifier, exportName, host, port);
 
        ARMARX_INFO << "MemoryID=" << iceAdapter().longtermMemory->getMemoryID().str();
        ARMARX_INFO << "Identifier=" << identifier;
        ARMARX_INFO << "Export name=" << exportName;
        ARMARX_INFO << "Host=" << host;
        ARMARX_INFO << "Port=" << port;
        ARMARX_INFO << "OverwritePrevStrategy=" << overwritePrevStrategy;
 
        if (overwritePrevStrategy)
        {
            iceAdapter().longtermMemory->getPersistenceStrategy()->clearStrategies();
            iceAdapter().longtermMemory->getPersistenceStrategy()->addStrategy(restPersistence);
        }
        else
        {
            iceAdapter().longtermMemory->getPersistenceStrategy()->addStrategy(restPersistence);
        }
    }
 
    void
    ReadWritePluginUser::useDiskPersistenceStrategy(const std::string& identifier,
                                                    const std::string& exportName,
                                                    const std::string& exportPath,
                                                    bool overwritePrevStrategy,
                                                    const ::Ice::Current&)
    {
        ARMARX_IMPORTANT << "Setting persistence strategy=disk via PROXY";
 
        std::shared_ptr<armem::server::ltm::persistence::DiskPersistence> diskPersistence =
            std::make_shared<armem::server::ltm::persistence::DiskPersistence>(
                identifier, exportName, std::filesystem::path(exportPath));
 
        ARMARX_INFO << "MemoryID=" << iceAdapter().longtermMemory->getMemoryID().str();
        ARMARX_INFO << "Identifier=" << identifier;
        ARMARX_INFO << "Export name=" << exportName;
        ARMARX_INFO << "Export path=" << exportPath;
        ARMARX_INFO << "OverwritePrevStrategy=" << overwritePrevStrategy;
 
        if (overwritePrevStrategy)
        {
            iceAdapter().longtermMemory->getPersistenceStrategy()->clearStrategies();
            iceAdapter().longtermMemory->getPersistenceStrategy()->addStrategy(diskPersistence);
        }
        else
        {
            iceAdapter().longtermMemory->getPersistenceStrategy()->addStrategy(diskPersistence);
        }
    }
 
    void
    ReadWritePluginUser::useMongoDbPersistenceStrategy(const std::string& identifier,
                                                       const std::string& exportName,
                                                       bool overwritePrevStrategy,
                                                       const ::Ice::Current&)
    {
        // TODO
    }
 
    void
    ReadWritePluginUser::loadLTMintoWM(bool complete, const ::Ice::Current&)
    {
        if (complete)
        {
            iceAdapter().reloadAllFromLTM();
        }
        else
        {
            // Only take the core segments defined in the property 'loadedCoreSegments'
            iceAdapter().reloadPropertyDefinedCoreSegmentsFromLTM();
        }
    }
 
    bool
    ReadWritePluginUser::loadExportIntoWM(const std::string& export_path,
                                          const std::string& memoryName,
                                          const std::vector<std::string>& coreSegmentNames,
                                          bool addNonExistingCoreSegments,
                                          int amountOfSnapshotsPerSegmentToLoad,
                                          const ::Ice::Current&)
    {
        ARMARX_INFO << "at ReadWritePluginUser";
        //we only build the directy and exportName from the export_path and then use the original function:
        auto splitAtLastSlash =
            [this, &export_path](const std::string& input) -> std::pair<std::string, std::string>
        {
            size_t lastSlash = input.rfind('/');
 
            if (lastSlash == std::string::npos)
            {
                ARMARX_ERROR
                    << "Could not load LTM into WM, as the given export_path does not contain '/', "
                       "which it needs to in order to split into directory_path and exportName. \n"
                    << "The path you have provided is: \n"
                    << export_path;
                return {"", ""};
            }
 
            std::string first = input.substr(0, lastSlash);
            std::string second = input.substr(lastSlash + 1);
 
            if (!second.empty())
            {
                return {first, second};
            }
 
            // If second part is empty, try finding an earlier slash
            size_t secondLastSlash = input.rfind('/', lastSlash - 1);
            if (secondLastSlash == std::string::npos)
            {
                ARMARX_ERROR << "The path you provided to load the LTM into the WM ends with a '/' "
                                "but does not contain any further '/'. "
                             << "Check if you provided the correct path. The path needs to consist "
                                "of a path to a directory, a '/' as a divider, and the name of the "
                                "export directory.\n"
                             << "You have given the following path: \n"
                             << export_path;
                return {"", ""};
            }
 
            return {input.substr(0, secondLastSlash),
                    input.substr(secondLastSlash + 1, lastSlash - secondLastSlash - 1)};
        };
 
        auto [directory, exportName] = splitAtLastSlash(export_path);
        if (directory == "" && exportName == "")
        {
            return false;
        }
        return this->loadExportIntoWM(directory,
                                      exportName,
                                      memoryName,
                                      coreSegmentNames,
                                      addNonExistingCoreSegments,
                                      amountOfSnapshotsPerSegmentToLoad);
    }
 
    bool
    ReadWritePluginUser::loadExportIntoWM(const std::string& directory,
                                          const std::string& exportName,
                                          const std::string& memoryName,
                                          const std::vector<std::string>& coreSegmentNames,
                                          bool addNonExistingCoreSegments,
                                          int amountOfSnapshotsPerSegmentToLoad)
    {
        ARMARX_IMPORTANT << "Starting to load data from disk into WM...";
        ARMARX_DEBUG << VAROUT(directory);
        ARMARX_DEBUG << VAROUT(exportName);
        ARMARX_DEBUG << VAROUT(memoryName);
        ARMARX_DEBUG << VAROUT(coreSegmentNames);
 
        ARMARX_DEBUG << VAROUT(addNonExistingCoreSegments);
        if (addNonExistingCoreSegments)
        {
            ARMARX_WARNING
                << "You tried to set addNonExistingCoreSegments to true, this is not supported yet";
        }
        ARMARX_DEBUG << VAROUT(amountOfSnapshotsPerSegmentToLoad);
 
        std::filesystem::path path(directory);
        path.append(exportName);
        path.append(memoryName);
 
        //check existance and depth and contains memory data
        if (not std::filesystem::is_directory(path))
        {
            ARMARX_WARNING << "You are trying to load data from a path which does not exist";
            ARMARX_WARNING << VAROUT(path);
            return false;
        }
 
        // Find out whether this is a single memory or a collection of memories by searching
        // for a data.aron.* or metadata.aron.* file at depth 5 (if 6 then it is collection of memories)
        bool isSingleMemory = false;
        for (auto i = std::filesystem::recursive_directory_iterator(path);
             i != std::filesystem::recursive_directory_iterator();
             ++i)
        {
            if (i.depth() >
                armem::server::ltm::persistence::MemoryPersistenceStrategy::DEPTH_TO_DATA_FILES + 2)
            {
                // In case we do not find any memory data in here
                ARMARX_WARNING
                    << "You are trying to import a directory which does not contain memory data";
                ARMARX_WARNING << VAROUT(path);
                return false;
            }
 
            auto& dir = *i;
 
            // if one matches it is enough to check
            if (std::filesystem::is_regular_file(dir.path()) &&
                simox::alg::starts_with(dir.path().filename(), "data.aron"))
            {
                isSingleMemory = (i.depth() == armem::server::ltm::persistence::
                                                   MemoryPersistenceStrategy::DEPTH_TO_DATA_FILES);
                if (!isSingleMemory)
                {
                    ARMARX_INFO << "Depth = " << i.depth();
                    ARMARX_INFO << "Depth to data files: "
                                << armem::server::ltm::persistence::MemoryPersistenceStrategy::
                                       DEPTH_TO_DATA_FILES;
                }
                break;
            }
        }
 
        if (!isSingleMemory)
        {
            ARMARX_WARNING << "Could not load data from LTM at " << directory
                           << " and write it to the current WM"
                           << "\n"
                           << "Reason: the path you have given is not a single memory but contains "
                              "more than one memory";
            return false;
        }
 
        ARMARX_DEBUG << "Now loading data from filesystem LTM...";
 
        armem::wm::Memory memoryData;
 
        if (std::filesystem::is_directory(directory))
        {
            std::string defaultIdent = "DefaultDisk";
            ARMARX_INFO << "Path of the folder trying to import: " << path;
            std::shared_ptr<armem::server::ltm::persistence::DiskPersistence> diskPersistence =
                std::make_shared<armem::server::ltm::persistence::DiskPersistence>(
                    defaultIdent, exportName, path.parent_path().parent_path());
 
            armem::server::ltm::Memory ltm(exportName, memoryName);
            if (ltm.getPersistenceStrategy())
            {
                ltm.getPersistenceStrategy()->addStrategy(diskPersistence);
            }
            else
            {
                std::shared_ptr<armem::server::ltm::persistence::RedundantPersistenceStrategy>
                    redundantPersistence = std::make_shared<
                        armem::server::ltm::persistence::RedundantPersistenceStrategy>();
 
                ltm.setPersistenceStrategy(redundantPersistence);
                ltm.getPersistenceStrategy()->addStrategy(diskPersistence);
            }
 
            armem::wm::Memory memory;
 
            if (amountOfSnapshotsPerSegmentToLoad != -1)
            {
                ARMARX_INFO << "You specified amountOfSnapshotsPerSegmentToLoad = "
                            << std::to_string(amountOfSnapshotsPerSegmentToLoad);
                if (coreSegmentNames.size() != 0)
                {
                    std::list<std::string> lst(coreSegmentNames.begin(), coreSegmentNames.end());
                    ltm.loadLatestNReferences(amountOfSnapshotsPerSegmentToLoad, memory, lst);
                }
                else
                {
                    memory = ltm.loadLatestNReferences(amountOfSnapshotsPerSegmentToLoad);
                }
                ltm.resolve(memory);
            }
            else
            {
                if (coreSegmentNames.size() != 0)
                {
                    std::list<std::string> lst(coreSegmentNames.begin(), coreSegmentNames.end());
                    ltm.loadAllAndResolve(memory, lst);
                }
                else
                {
                    memory = ltm.loadAllAndResolve(); // load list of references and load data
                }
            }
 
            memoryData = std::move(memory);
        }
 
        // we need to merge the current memory with the one from the disk.
        // Please note that this may ignores coresegments, if not already existent
        auto commit = armem::toCommit(memoryData);
        ARMARX_DEBUG << VAROUT(commit);
        data::Commit iceCommit;
        toIce(iceCommit, commit);
        this->commit(iceCommit);
 
        ARMARX_INFO << "Successfully loaded data from " << path
                    << " into the current working memory";
 
        return true;
    }
 
} // namespace armarx::armem::server::plugins