Entity.cpp

Go to the documentation of this file.

// Header
#include "Entity.h"
 
// SImox
#include <SimoxUtility/algorithm/string.h>
 
// ArmarX
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/core/time/TimeUtil.h>
 
#include <RobotAPI/libraries/armem/core/base/detail/negative_index_semantics.h>
#include <RobotAPI/libraries/armem/server/wm/memory_definitions.h>
 
#include "processors/filter/frequencyFilter/FrequencyFilter.h"
 
namespace armarx::armem::server::ltm
{
    namespace util
    {
 
 
    } // namespace util
 
    Entity::Entity(const detail::mixin::Path& p,
                   const detail::mixin::MongoDBSettings& s,
 
                   const std::string& exportName,
                   const armem::MemoryID& id /* UNESCAPED */,
                   const std::shared_ptr<Processors>& filters) :
        EntityBase(exportName, id, filters),
        DiskMemoryItemMixin(p, exportName, id),
        MongoDBStorageMixin(s, exportName, id)
    {
        //start();
    }
 
    bool
    Entity::forEachSnapshot(std::function<void(EntitySnapshot&)> func) const
    {
        std::lock_guard l(ltm_mutex);
 
        if (fullPathExists())
        {
 
            for (const auto& d : getAllDirectories()) // days
            {
                if (!util::fs::detail::isDateString(d.filename()))
                {
                    ARMARX_WARNING << "Found a non-date folder inside an entity '" << id().str()
                                   << "' with name '" << d.filename() << "'. "
                                   << "Ignoring this folder, however this is a bad situation.";
                    continue;
                }
 
 
                for (const auto& s : util::fs::getAllDirectories(d)) // seconds
                {
                    if (!util::fs::detail::isNumberString(s.filename()))
                    {
                        ARMARX_WARNING << "Found a non-timestamp folder inside an entity '"
                                       << id().str() << "' hours folder with name '" << s.filename()
                                       << "'. "
                                       << "Ignoring this folder, however this is a bad situation.";
                        continue;
                    }
 
                    for (const auto& us : util::fs::getAllDirectories(s)) // microseconds
                    {
                        if (!util::fs::detail::isNumberString(us.filename()))
                        {
                            ARMARX_WARNING
                                << "Found a non-timestamp folder inside an entity '" << id().str()
                                << "' seconds folder with name '" << us.filename() << "'. "
                                << "Ignoring this folder, however this is a bad situation.";
                            continue;
                        }
 
 
                        EntitySnapshot c(
                            getMemoryBasePath(),
                            getSettings(),
                            getExportName(),
                            id().withTimestamp(timeFromStringMicroSeconds(us.filename())),
                            processors);
                        func(c);
                    }
                }
            }
        }
 
        return true;
    }
 
    bool
    Entity::forEachSnapshotInIndexRange(long first,
                                        long last,
                                        std::function<void(EntitySnapshot&)> func) const
    {
        std::lock_guard l(ltm_mutex);
        //ARMARX_WARNING << "PLEASE NOTE THAT QUERYING THE LTM INDEX WISE MAY BE BUGGY BECAUSE THE FILESYSTEM ITERATOR IS UNSORTED!";
 
        if (first < 0 or last < 0)
        {
            // We need to know what the size of the memory is... May be slow
            unsigned long size = 0;
            auto f = [&](EntitySnapshot& e) { size++; };
            forEachSnapshot(std::move(f));
 
            first = armarx::armem::base::detail::negativeIndexSemantics(first, size);
            last = armarx::armem::base::detail::negativeIndexSemantics(last, size);
        }
 
        long checked = 0;
        auto f = [&](EntitySnapshot& e)
        {
            if (checked >= first && checked <= last)
            {
                func(e);
            }
            checked++;
        };
 
        return forEachSnapshot(std::move(f));
    }
 
    bool
    Entity::forEachSnapshotInTimeRange(const Time& min,
                                       const Time& max,
                                       std::function<void(EntitySnapshot&)> func) const
    {
        std::lock_guard l(ltm_mutex);
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts >= min && ts <= max)
            {
                func(e);
            }
        };
 
        return forEachSnapshot(std::move(f));
    }
 
    bool
    Entity::forEachSnapshotBeforeOrAt(const Time& time,
                                      std::function<void(EntitySnapshot&)> func) const
    {
        std::lock_guard l(ltm_mutex);
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts <= time)
            {
                func(e);
            }
        };
 
        return forEachSnapshot(std::move(f));
    }
 
    bool
    Entity::forEachSnapshotBefore(const Time& time, std::function<void(EntitySnapshot&)> func) const
    {
        std::lock_guard l(ltm_mutex);
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts < time)
            {
                func(e);
            }
        };
 
        return forEachSnapshot(std::move(f));
    }
 
    bool
    Entity::hasSnapshot(const Time& n) const
    {
        std::lock_guard l(ltm_mutex);
 
        if (fullPathExists())
        {
            auto c = EntitySnapshot(getMemoryBasePath(),
                                    getSettings(),
                                    getExportName(),
                                    id().withTimestamp(n),
                                    processors);
            return c.fullPathExists();
        }
        return false;
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findSnapshot(const Time& n) const
    {
        std::lock_guard l(ltm_mutex);
        if (!hasSnapshot(n))
        {
            return nullptr;
        }
        return std::make_shared<EntitySnapshot>(
            getMemoryBasePath(), getSettings(), getExportName(), id().withTimestamp(n), processors);
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findLatestSnapshot() const
    {
        std::lock_guard l(ltm_mutex);
        Time bestMatch = Time::Invalid();
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts > bestMatch)
            {
                bestMatch = ts;
            }
        };
 
        forEachSnapshot(std::move(f));
 
        if (bestMatch == Time::Invalid())
        {
            return nullptr;
        }
 
        return std::make_shared<EntitySnapshot>(getMemoryBasePath(),
                                                getSettings(),
                                                getExportName(),
                                                id().withTimestamp(bestMatch),
                                                processors);
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findLatestSnapshotBefore(const Time& time) const
    {
        std::lock_guard l(ltm_mutex);
        Time bestMatch = Time::Invalid();
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts < time && ts > bestMatch)
            {
                bestMatch = ts;
            }
        };
 
        forEachSnapshot(std::move(f));
 
        if (bestMatch == Time::Invalid())
        {
            return nullptr;
        }
 
        return std::make_shared<EntitySnapshot>(getMemoryBasePath(),
                                                getSettings(),
                                                getExportName(),
                                                id().withTimestamp(bestMatch),
                                                processors);
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findLatestSnapshotBeforeOrAt(const Time& time) const
    {
        std::lock_guard l(ltm_mutex);
        Time bestMatch = Time::Invalid();
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts <= time && ts > bestMatch)
            {
                bestMatch = ts;
            }
        };
 
        forEachSnapshot(std::move(f));
 
        if (bestMatch == Time::Invalid())
        {
            return nullptr;
        }
 
        return std::make_shared<EntitySnapshot>(getMemoryBasePath(),
                                                getSettings(),
                                                getExportName(),
                                                id().withTimestamp(bestMatch),
                                                processors);
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findFirstSnapshotAfter(const Time& time) const
    {
        std::lock_guard l(ltm_mutex);
        Time bestMatch{Duration::MicroSeconds(std::numeric_limits<long>::max())};
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts > time && ts < bestMatch)
            {
                bestMatch = ts;
            }
        };
 
        forEachSnapshot(std::move(f));
 
        if (bestMatch == Time(Duration::MicroSeconds(std::numeric_limits<long>::max())))
        {
            return nullptr;
        }
 
        return std::make_shared<EntitySnapshot>(getMemoryBasePath(),
                                                getSettings(),
                                                getExportName(),
                                                id().withTimestamp(bestMatch),
                                                processors);
    }
 
    std::shared_ptr<EntitySnapshot>
    Entity::findFirstSnapshotAfterOrAt(const Time& time) const
    {
        std::lock_guard l(ltm_mutex);
        Time bestMatch{Duration::MicroSeconds(std::numeric_limits<long>::max())};
        auto f = [&](EntitySnapshot& e)
        {
            auto ts = e.id().timestamp;
            if (ts >= time && ts < bestMatch)
            {
                bestMatch = ts;
            }
        };
 
        forEachSnapshot(std::move(f));
 
        if (bestMatch == Time(Duration::MicroSeconds(std::numeric_limits<long>::max())))
        {
            return nullptr;
        }
 
        return std::make_shared<EntitySnapshot>(getMemoryBasePath(),
                                                getSettings(),
                                                getExportName(),
                                                id().withTimestamp(bestMatch),
                                                processors);
    }
 
    void
    Entity::_loadAllReferences(armem::wm::Entity& e)
    {
        std::lock_guard l(ltm_mutex);
        e.id() = id().getEntityID().cleanID();
 
        forEachSnapshot(
            [&e](auto& x)
            {
                if (!e.hasSnapshot(
                        x.id()
                            .timestamp)) // we only load the references if the snapshot is not existant
                {
                    armem::wm::EntitySnapshot s;
                    x.loadAllReferences(s);
                    e.addSnapshot(s);
                }
            });
    }
 
    void
    Entity::_loadLatestNReferences(int n, armem::wm::Entity& e)
    {
        std::lock_guard l(ltm_mutex);
        e.id() = id().getEntityID().cleanID();
        int count = 0;
 
        //this is a little bit ugly, TODO: find an easier way to count the snapshots
        forEachSnapshot(
            [&e, &count](auto& x)
            {
                count++;
            }
        );
 
        int current = 0;
 
        forEachSnapshot(
            [&e, &n, &current, &count](auto& x)
            {
                if (!e.hasSnapshot(
                        x.id()
                            .timestamp)) // we only load the references if the snapshot is not existant
                {
                    if (current >= (count - n)){
                        ARMARX_INFO << e.id().coreSegmentName << ": count: " << count << ", n: " << n << ", current: " <<  current;
                        ARMARX_DEBUG << "Loading snapshot with timestamp " << x.id().timestamp << " into WM";
                        armem::wm::EntitySnapshot s;
                        x.loadAllReferences(s);
                        e.addSnapshot(s);
                    } else {
                        ARMARX_DEBUG << "Skipping snapshot with timestamp " << x.id().timestamp;
                    }
                    current++;
                }
            });
    }
 
    void
    Entity::_resolve(armem::wm::Entity& p)
    {
        std::lock_guard l(ltm_mutex);
 
        if (/*(connected() && collectionExists()) ||*/ fullPathExists())
        {
 
            p.forEachSnapshot(
                [&](auto& e)
                {
                    EntitySnapshot c(getMemoryBasePath(),
                                     getSettings(),
                                     getExportName(),
                                     id().withTimestamp(e.id().timestamp),
                                     processors);
                    c.resolve(e);
                });
        }
    }
 
    void
    Entity::_store(const armem::wm::Entity& c)
    {
        std::lock_guard l(ltm_mutex);
        if (id().entityName.empty())
        {
            ARMARX_WARNING
                << "During storage of segment '" << c.id().str()
                << "' I noticed that the corresponding LTM has no id set. "
                << "I set the id of the LTM to the same name, however this should not happen!";
            id().entityName = c.id().entityName;
        }
 
        /*if (!connected())
        {
            ARMARX_WARNING << "LTM ENTITY NOT CONNECTED ALTHOUGH ENABLED " << id().str();
            return;
        }*/
 
        //writeForeignKeyToPreviousDocument();
 
        c.forEachSnapshot(
            [&](const auto& snap)
            {
                EntitySnapshot c(getMemoryBasePath(),
                                 getSettings(),
                                 getExportName(),
                                 id().withTimestamp(snap.id().timestamp),
                                 processors);
 
                // check if snapshot already exists
 
                if (hasSnapshot(snap.id().timestamp))
                {
                    ARMARX_INFO  << "Ignoring to put an EntitiySnapshot into the LTM because "
                                   "the timestamp already existed (we assume snapshots are "
                                   "const and do not change outside the ltm).";
                    return;
                }
 
                for (auto& f : processors->snapFilters)
                {
 
                    bool accepted = f->accept(snap);
                    if (!accepted)
                    {
                        //ARMARX_INFO << "Ignoring to put an EntitiySnapshot into the LTM because it got filtered.";
                        return;
                    } else {
                        //ARMARX_INFO << "Storing EntitySnapshot";
                    }
                }
 
                c.store(snap);
                statistics.recordedSnapshots++;
            });
    }
} // namespace armarx::armem::server::ltm