SkillEventSegment.cpp

Go to the documentation of this file.

#include "SkillEventSegment.h"
 
#include <atomic>
 
#include <SimoxUtility/algorithm/string.h>
 
#include <RobotAPI/libraries/armem/core/operations.h>
#include <RobotAPI/libraries/armem/server/MemoryToIceAdapter.h>
#include <RobotAPI/libraries/armem_skills/aron/Skill.aron.generated.h>
#include <RobotAPI/libraries/armem_skills/aron_conversions.h>
#include <RobotAPI/libraries/aron/converter/json/NLohmannJSONConverter.h>
 
namespace armarx::skills::segment
{
 
    SkillEventCoreSegment::SkillEventCoreSegment(armem::server::MemoryToIceAdapter& iceMemory) :
        Base(iceMemory,
             CoreSegmentName,
             armarx::skills::arondto::SkillExecutionRequest::ToAronType())
    {
    }
 
    void
    SkillEventCoreSegment::defineProperties(PropertyDefinitionsPtr defs, const std::string& prefix)
    {
        this->setDefaultMaxHistorySize(300);
        Base::defineProperties(defs, prefix);
 
        defs->topic(p.skillEventTopic, "SkillEventListener", prefix + "tpc.pub.SkillEventListener");
    }
 
    void
    SkillEventCoreSegment::init()
    {
        Base::init();
    }
 
    void
    SkillEventCoreSegment::addSkillUpdateEvent(const skills::SkillStatusUpdate& update)
    {
        // create commit about new update
        armarx::skills::arondto::SkillStatusUpdate event;
        armem::toAron(event, update);
 
        armem::MemoryID commitId = id();
        commitId.providerSegmentName = event.skillId.providerId.providerName;
        commitId.entityName = event.skillId.skillName;
 
        auto aron = event.toAron();
 
        armem::Commit comm;
        auto& entityUpdate = comm.add();
        entityUpdate.confidence = 1.0;
        entityUpdate.referencedTime = armem::Time::Now();
        entityUpdate.instancesData = {aron};
        entityUpdate.entityID = commitId;
 
        // Publish to the skill event topic. Tolerate two failure modes that
        // would otherwise crash the SkillMemory component:
        //  1. The topic proxy is null (e.g. topic was never wired up).
        //  2. The Ice RPC throws (network error, marshalling, etc.).
        // In both cases we skip the publish, log a warning ONCE per failure
        // mode, and still commit the update to local memory below.
        if (p.skillEventTopic)
        {
            try
            {
                p.skillEventTopic->reportSkillEvent(
                    update.toProviderIce(), event.skillId.providerId.providerName);
            }
            catch (const std::exception& e)
            {
                static std::atomic_flag warned = ATOMIC_FLAG_INIT;
                if (!warned.test_and_set())
                {
                    ARMARX_WARNING << "Failed to publish skill event on topic '"
                                   << "SkillEventListener"
                                   << "': " << e.what()
                                   << ". Subsequent failures of this kind will be "
                                      "silently ignored. The local memory commit "
                                      "still proceeds.";
                }
            }
        }
        else
        {
            static std::atomic_flag warned = ATOMIC_FLAG_INIT;
            if (!warned.test_and_set())
            {
                ARMARX_WARNING << "SkillEventListener topic proxy is null; "
                                  "skill events will not be published. "
                                  "This warning is only logged once. "
                                  "The local memory commit still proceeds.";
            }
        }
 
        iceMemory.commitLocking(comm);
    }
 
    std::map<skills::SkillExecutionID, skills::SkillStatusUpdate>
    SkillEventCoreSegment::getSkillStatusUpdates()
    {
        std::map<skills::SkillExecutionID, skills::SkillStatusUpdate> ret;
        auto coreSegment = this->segmentPtr;
        ARMARX_CHECK(coreSegment);
 
        coreSegment->doLocked(
            [&]()
            {
                coreSegment->forEachInstance(
                    [&](const armem::wm::EntityInstance& i)
                    {
                        // dataAs<>() throws on type mismatch / corrupted aron
                        // data. Skip the bad entry instead of crashing the
                        // whole getSkillStatusUpdates() call.
                        skills::SkillStatusUpdate up;
                        try
                        {
                            auto event =
                                i.dataAs<armarx::skills::arondto::SkillStatusUpdate>();
                            armem::fromAron(event, up);
                        }
                        catch (const std::exception& e)
                        {
                            ARMARX_WARNING << "Skipping malformed SkillStatusUpdate "
                                              "instance in SkillEvent segment: "
                                           << e.what();
                            return;
                        }
 
                        if (auto it = ret.find(up.executionId); it != ret.end() && up < it->second)
                        {
                            return;
                        }
 
                        // set or replace
                        ret[up.executionId] = up;
                    });
            });
 
        //        for (const auto& [k, v] : ret)
        //        {
        //            ARMARX_IMPORTANT << "Skill " << k.skillId << " has stati: " << int(v.status);
        //        }
 
        return ret;
    }
 
    std::optional<skills::SkillStatusUpdate>
    SkillEventCoreSegment::getSkillStatusUpdate(const skills::SkillExecutionID& id)
    {
        std::optional<skills::SkillStatusUpdate> ret = std::nullopt;
        auto coreSegment = this->segmentPtr;
        ARMARX_CHECK(coreSegment);
 
        coreSegment->doLocked(
            [&]()
            {
                coreSegment->forEachInstance(
                    [&](const armem::wm::EntityInstance& i)
                    {
                        // dataAs<>() throws on type mismatch / corrupted aron
                        // data. Skip the bad entry instead of crashing the
                        // whole getSkillStatusUpdate() call.
                        skills::SkillStatusUpdate up;
                        try
                        {
                            auto event =
                                i.dataAs<armarx::skills::arondto::SkillStatusUpdate>();
                            armem::fromAron(event, up);
                        }
                        catch (const std::exception& e)
                        {
                            ARMARX_WARNING << "Skipping malformed SkillStatusUpdate "
                                              "instance in SkillEvent segment: "
                                           << e.what();
                            return;
                        }
 
                        if (up.executionId == id)
                        {
                            if (!ret || (ret.has_value() && *ret < up))
                            {
                                ret = up;
                            }
                        }
                    });
            });
        return ret;
    }
 
    std::size_t
    SkillEventCoreSegment::clearAndConsolidate(armem::server::ltm::Memory& ltm,
                                                const std::string& memoryName,
                                                bool& consolidated)
    {
        consolidated = false;
        std::size_t snapshotCount = 0;
        auto coreSegment = this->segmentPtr;
        ARMARX_CHECK(coreSegment);
 
        coreSegment->doLockedExclusive(
            [&]()
            {
                coreSegment->forEachSnapshot(
                    [&](const armem::wm::EntitySnapshot&) { ++snapshotCount; });
 
                if (ltm.isRecording() && snapshotCount > 0)
                {
                    // Build a WM slice holding only this core segment so that
                    // only the data we are about to clear is written to LTM.
                    armem::wm::Memory slice;
                    slice.setName(memoryName);
                    slice.update(armem::toCommit(*coreSegment), true);
                    ltm.directlyStore(slice);
                    consolidated = true;
                }
 
                coreSegment->clear();
            });
 
        return snapshotCount;
    }
} // namespace armarx::skills::segment