tlbmc/sensors/virtual_sensor.cc

#include "tlbmc/sensors/virtual_sensor.h"

#include <chrono>  // NOLINT
#include <memory>
#include <optional>
#include <string>
#include <utility>

#include "absl/container/flat_hash_map.h"
#include "absl/functional/any_invocable.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/str_cat.h"
#include "virtual_sensor_config.pb.h"
#include "tlbmc/expression/expression.h"
#include "resource.pb.h"
#include "sensor.pb.h"
#include "tlbmc/sensors/sensor.h"
#include "tlbmc/time/time.h"

namespace milotic_tlbmc {

absl::StatusOr<std::shared_ptr<VirtualSensor>> VirtualSensor::Create(
    const VirtualSensorConfig& config,
    const std::shared_ptr<boost::asio::io_context>& io_context,
    const absl::flat_hash_map<std::string, std::shared_ptr<Sensor>>&
        sensors_map,
    std::unique_ptr<Expression> expression,
    std::optional<NotificationCb> on_batch_notify) {
  return std::make_shared<VirtualSensor>(
      Token(), config.name(), config.unit(), config.entity_common_config(),
      config.thresholds(), config.reading_ranges(), sensors_map,
      std::move(expression), io_context, on_batch_notify);
}

VirtualSensor::VirtualSensor(
    Token token, const std::string& sensor_name, SensorUnit sensor_unit,
    const EntityCommonConfig& entity_common_config,
    const ThresholdConfigs& threshold_configs,
    const ReadingRangeConfigs& reading_range_configs,
    const absl::flat_hash_map<std::string, std::shared_ptr<Sensor>>&
        sensors_map,
    std::unique_ptr<Expression> expression,
    const std::shared_ptr<boost::asio::io_context>& io_context,
    std::optional<NotificationCb> on_batch_notify)
    : Sensor(CreateStaticAttributes(sensor_name, sensor_unit,
                                    /*i2c_common_config=*/{},
                                    entity_common_config, reading_range_configs,
                                    /*reading_transform_config=*/{}),
             threshold_configs, on_batch_notify),
      io_context_(io_context),
      sensors_map_(sensors_map),
      expression_(std::move(expression)),
      sensor_name_(sensor_name) {}

void VirtualSensor::RefreshOnceAsync(
    absl::AnyInvocable<void(const std::shared_ptr<const SensorValue>&)>
        callback) {
  std::weak_ptr<VirtualSensor> self = shared_from_this();
  boost::asio::post(*io_context_, [self(std::move(self)),
                                   callback(std::move(callback))]() mutable {
    std::shared_ptr<VirtualSensor> sensor = self.lock();
    if (!sensor) {
      LOG(WARNING) << "Sensor is destroyed; cancel the refresh.";
      return;
    }
    std::chrono::steady_clock::time_point last_refresh_start_time =
        sensor->GetLastRefreshStartTime();
    if (last_refresh_start_time !=
        std::chrono::steady_clock::time_point::min()) {
      sensor->UpdateSoftwarePollingStartMetrics(
          std::chrono::steady_clock::now() - last_refresh_start_time);
    }
    sensor->SetLastRefreshStartTime(std::chrono::steady_clock::now());

    // Get the sensor value
    absl::StatusOr<double> sensor_value = sensor->GetSensorValue();

    if (!sensor_value.ok()) {
      State state;
      state.set_status(STATUS_STALE);
      state.set_status_message(
          absl::StrCat("Failed to read sensor value with error: ",
                       sensor_value.status().message()));
      sensor->UpdateState(std::move(state));
      if (callback) {
        callback(sensor->GetSensorData());
      }
      DLOG(INFO) << "VirtualSensor::RefreshOnceAsync failed: "
                 << sensor->sensor_name_ << " " << sensor_value.status();
    } else {
      SensorValue sensor_data;
      // milliseconds to timestamp
      *sensor_data.mutable_timestamp() = Now();
      sensor_data.set_reading(static_cast<double>(*sensor_value));
      DLOG(INFO) << "VirtualSensor::RefreshOnceAsync success: "
                 << sensor->sensor_name_ << " " << sensor_data.reading();
      sensor->StoreSensorData(
          std::make_shared<const SensorValue>(std::move(sensor_data)));
      State state;
      state.set_status(STATUS_READY);
      sensor->UpdateState(std::move(state));

      if (callback) {
        callback(sensor->GetSensorData());
      }
    }
    std::chrono::steady_clock::time_point last_refresh_end_time =
        sensor->GetLastRefreshEndTime();
    if (last_refresh_end_time != std::chrono::steady_clock::time_point::min()) {
      sensor->UpdateSoftwarePollingEndMetrics(std::chrono::steady_clock::now() -
                                              last_refresh_end_time);
    }
    sensor->SetLastRefreshEndTime(std::chrono::steady_clock::now());
  });
}

absl::StatusOr<double> VirtualSensor::GetSensorValue() {
  absl::flat_hash_map<std::string, double> sensor_readings_map;
  for (const auto& [sensor_name, sensor] : sensors_map_) {
    if (sensor == nullptr) {
      return absl::InvalidArgumentError(
          absl::StrCat("Sensor ", sensor_name, " is missing"));
    }
    const std::shared_ptr<const SensorValue> sensor_data =
        sensor->GetSensorData();
    if (sensor_data == nullptr) {
      return absl::InvalidArgumentError("Sensor data is null");
    }
    sensor_readings_map[sensor_name] = sensor_data->reading();
  }
  return expression_->Evaluate(sensor_readings_map);
}

}  // namespace milotic_tlbmc