monitoring_manager.h

Go to the documentation of this file.

#pragma once
 
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <expected>
#include <functional>
#include <map>
#include <mutex>
#include <nlohmann/json_fwd.hpp>
#include <optional>
#include <string>
#include <thread>
#include <vector>
 
#include "node/device_manager.h"
#include "node/monitoring.h"
#include "node/parameter_refresher.h"
 
namespace mm::node {
 
class MonitoringManager {
 public:
  using PublishFn = std::function<void(std::string topic, std::string json)>;
 
  explicit MonitoringManager(DeviceManager& deviceManager);
 
  ~MonitoringManager();
 
  MonitoringManager(const MonitoringManager&) = delete;
  MonitoringManager& operator=(const MonitoringManager&) = delete;
 
  void setPublish(PublishFn publish);
 
  std::expected<Monitoring, std::string> create(Monitoring config);
 
  bool remove(const std::string& topic);
 
  std::optional<nlohmann::json> get(const std::string& topic) const;
 
  nlohmann::json list() const;
 
  void start();
 
  void stop();
 
  void sampleAll();
 
  std::size_t monitoringCount() const;
 
  std::size_t polledSdoCount() const;
 
 private:
  enum class Source { Pdo, Sdo };
 
  struct ParamPlan {
    uint16_t devicePosition = 0;
    uint16_t index = 0;
    uint8_t subindex = 0;
    Source source = Source::Pdo;
    std::optional<DeviceManager::PdoSampleSpec> pdoSpec;  // PDO only
  };
 
  struct Sample {
    int64_t timestampUs = 0;
    std::vector<std::optional<DeviceParameterValue>> values;
  };
 
  struct Entry {
    Monitoring config;
    std::vector<ParamPlan> plans;
    uint64_t cursor = 0;           // next recorder sequence number to deliver ([cursor, head))
    bool cursorPrimed = false;     // false until the first flush seeds cursor from recorderHead()
    uint64_t imageGeneration = 0;  // processImageGeneration the PDO specs were captured under
    std::chrono::steady_clock::time_point nextDue{};  // default (epoch) => due on the next wake
  };
 
  void flushEntry(Entry& entry);  // decode + publish [cursor, head); assumes mutex_ held
  void recaptureIfRemapped(
      Entry& entry);  // re-classify source + re-capture specs; assumes mutex_ held
  static nlohmann::json resourceJson(const Entry& entry);  // assumes mutex_ held
 
  void run();
 
  DeviceManager& deviceManager_;
  ParameterRefresher refresher_;
  mutable std::mutex mutex_;    
  std::condition_variable cv_;  
  std::map<std::string, Entry> entries_;
  PublishFn publish_;
  bool running_ = false;  
  std::thread thread_;
};
 
}  // namespace mm::node