OpossumLogger.h

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * ArmarX is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * ArmarX is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 *
 * @package    RobotAPI::ArmarXObjects::OpossumLogger
 * @author     Joana Plewnia ( joana dot plewnia at kit dot edu )
 * @date       2026
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
 
#pragma once
 
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <thread>
#include <vector>
 
#include <ArmarXCore/core/Component.h>
 
#include <RobotAPI/libraries/armem/client/Reader.h>
#include <RobotAPI/libraries/armem/client/plugins.h>
#include <RobotAPI/libraries/armem/client/util/SubscriptionHandle.h>
#include <RobotAPI/libraries/armem/core/MemoryID.h>
 
#include "ObjectSpokenNames.h"
#include "SkillEventRenderer.h"
 
namespace armarx
{
 
    /**
     * @defgroup Component-OpossumLogger OpossumLogger
     * @ingroup RobotAPI-Components
     *
     * Streams the robot's skill events to an OPOSSUM summary server.
     *
     * OPOSSUM (Observe, Process, Organize Scene-Level SUMmaries) is the shared
     * multi-agent log of the euROBIN demo: every robot posts scene-level log
     * lines over HTTP, the server sorts them by `robot_name` and summarises them.
     *
     * This component subscribes to the `SkillEvent` core segment of the Skill
     * memory and, for every skill that starts, succeeds, fails or is aborted,
     * renders one line such as
     * @code
     * skill:LookAtObject, object:dining table, called_by:ServeDrinks::ApproachAndAskForDrink, state:started, time_stamp:2026-07-08T16:45:40.341
     * @endcode
     * and POSTs it as `{"robot_name": ..., "message": ...}`.
     *
     * The line format matches `tools/gen_armar7_logs.py` in the `opossum_euROBIN`
     * repository, which produces the same lines offline from an LTM export.
     *
     * @class OpossumLogger
     * @ingroup Component-OpossumLogger
     */
    class OpossumLogger :
        virtual public armarx::Component,
        virtual public armarx::armem::ListeningClientPluginUser
    {
    public:
        ~OpossumLogger() override;
 
        std::string getDefaultName() const override;
        static std::string GetDefaultName();
 
    protected:
        armarx::PropertyDefinitionsPtr createPropertyDefinitions() override;
 
        void onInitComponent() override;
        void onConnectComponent() override;
        void onDisconnectComponent() override;
        void onExitComponent() override;
 
    private:
        /// Called by the memory listener for every batch of new `SkillEvent` snapshots.
        void onSkillEventUpdate(const armem::MemoryID& subscriptionID,
                                const std::vector<armem::MemoryID>& updatedSnapshotIDs);
 
        /// Whether this snapshot has not been handled before.
        bool isNew(const armem::MemoryID& snapshotID);
 
        /// Fetch the given snapshots' first instance each, skipping those the
        /// memory no longer holds.
        std::map<armem::MemoryID, armem::wm::EntityInstance>
        resolve(const std::vector<armem::MemoryID>& snapshotIDs);
 
        void enqueue(std::string line);
 
        /// Worker: posts queued lines and appends them to the output file.
        void sendLoop();
 
        void startSender();
        void stopSender();
 
        struct Properties
        {
            bool enabled = true;
 
            std::string robotName = "ARMAR-7 (KIT)";
            std::string host = "192.168.253.48";
            int port = 8001;
            std::string path = "message";
            // A POST to the summary server crosses from the robot network to
            // the venue network; 200 ms was enough only against localhost.
            int timeoutMs = 1000;
 
            int queueSize = 1000;
            // On by default: a failed POST discards its line for good, and the
            // summary server belongs to a project partner, so the local record
            // is the only one we control. Written before the POST is attempted.
            std::string outputFile = "/tmp/opossum_log.txt";
 
            std::string skillFilter;
            bool resolveObjectNames = true;
 
            std::string memoryName = "Skill";
            std::string coreSegmentName = "SkillEvent";
        };
 
        Properties p;
 
        armem::MemoryID skillEventID;
        armem::client::Reader skillMemoryReader;
 
        opossum::ObjectSpokenNames spokenNames;
        std::unique_ptr<opossum::SkillEventRenderer> renderer;
 
        std::thread sender;
        std::atomic_bool running{false};
 
        std::mutex queueMutex;
        std::condition_variable queueCondition;
        std::deque<std::string> queue;
        std::size_t droppedLines = 0;
 
        /// Snapshot IDs already handled, oldest first, bounded in size.
        std::mutex seenMutex;
        std::deque<std::string> seenOrder;
        std::set<std::string> seen;
 
        /// Declared last on purpose: members are destroyed in reverse
        /// declaration order, so this is released before everything
        /// `onSkillEventUpdate` touches (`renderer`, the queue, `seen`).
        /// Note that ~SubscriptionHandle does not release by itself.
        armem::client::util::SubscriptionHandle subscription;
    };
 
} // namespace armarx