Merge pull request #82 from kivancsikert/valve/handle-events-and-over…

…ride

Publish events and override state from valve

src/main.cpp

@@ -1,8 +1,6 @@
 #include <atomic>
 #include <chrono>
 
-#include <Arduino.h>
-
 #include <ArduinoLog.h>
 
 #include <devices/Device.hpp>

src/peripherals/flow_control/FlowControl.hpp

@@ -29,7 +29,9 @@ class FlowControl : public Peripheral<FlowControlConfig> {
         PwmMotorDriver& controller, ValveControlStrategy& strategy,
         gpio_num_t pin, double qFactor, milliseconds measurementFrequency)
         : Peripheral<FlowControlConfig>(name, mqttRoot)
-        , valve(name, controller, strategy, mqttRoot)
+        , valve(name, controller, strategy, mqttRoot, [this]() {
+            publishTelemetry();
+        })
         , flowMeter(name, mqttRoot, pin, qFactor, measurementFrequency) {
     }
 

src/peripherals/valve/Valve.hpp

@@ -34,7 +34,9 @@ class Valve
 public:
     Valve(const String& name, PwmMotorDriver& controller, ValveControlStrategy& strategy, shared_ptr<MqttDriver::MqttRoot> mqttRoot)
         : Peripheral<ValveConfig>(name, mqttRoot)
-        , valve(name, controller, strategy, mqttRoot) {
+        , valve(name, controller, strategy, mqttRoot, [this]() {
+            publishTelemetry();
+        }) {
     }
 
     void configure(const ValveConfig& config) override {

src/peripherals/valve/ValveComponent.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <list>
 #include <memory>
@@ -15,6 +16,7 @@
 #include <kernel/Concurrent.hpp>
 #include <kernel/Service.hpp>
 #include <kernel/Task.hpp>
+#include <kernel/Time.hpp>
 #include <kernel/Telemetry.hpp>
 #include <kernel/drivers/MotorDriver.hpp>
 
@@ -152,10 +154,11 @@ class LatchingValveControlStrategy
 
 class ValveComponent : public Component {
 public:
-    ValveComponent(const String& name, PwmMotorDriver& controller, ValveControlStrategy& strategy, shared_ptr<MqttDriver::MqttRoot> mqttRoot)
+    ValveComponent(const String& name, PwmMotorDriver& controller, ValveControlStrategy& strategy, shared_ptr<MqttDriver::MqttRoot> mqttRoot, std::function<void()> publishTelemetry)
         : Component(name, mqttRoot)
         , controller(controller)
-        , strategy(strategy) {
+        , strategy(strategy)
+        , publishTelemetry(publishTelemetry) {
         Log.infoln("Creating valve '%s' with strategy %s",
             name.c_str(), strategy.describe().c_str());
 
@@ -179,14 +182,14 @@ class ValveComponent : public Component {
 
         Task::loop(name, 3072, [this, name](Task& task) {
             auto now = system_clock::now();
-            if (overrideState != ValveState::NONE && now > overrideUntil) {
+            if (overrideState != ValveState::NONE && now > overrideUntil.load()) {
                 Log.traceln("Valve '%s' override expired", name.c_str());
                 overrideUntil = time_point<system_clock>();
                 overrideState = ValveState::NONE;
             }
             ValveStateUpdate update;
             if (overrideState != ValveState::NONE) {
-                update = { overrideState, duration_cast<ticks>(overrideUntil - now) };
+                update = { overrideState, duration_cast<ticks>(overrideUntil.load() - now) };
                 Log.traceln("Valve '%s' override state is %d, will change after %F sec",
                     name.c_str(), static_cast<int>(update.state), update.transitionAfter.count() / 1000.0);
             } else {
@@ -197,15 +200,16 @@ class ValveComponent : public Component {
             setState(update.state);
             // TODO Account for time spent in setState()
             updateQueue.pollIn(update.transitionAfter, [this](const std::variant<OverrideSpec, ScheduleSpec>& change) {
-                std::visit([this](auto&& arg) {
-                    using T = std::decay_t<decltype(arg)>;
-                    if constexpr (std::is_same_v<T, OverrideSpec>) {
-                        overrideState = arg.state;
-                        overrideUntil = arg.until;
-                    } else if constexpr (std::is_same_v<T, ScheduleSpec>) {
-                        schedules = std::list(arg.schedules);
-                    }
-                },
+                std::visit(
+                    [this](auto&& arg) {
+                        using T = std::decay_t<decltype(arg)>;
+                        if constexpr (std::is_same_v<T, OverrideSpec>) {
+                            overrideState = arg.state;
+                            overrideUntil = arg.until;
+                        } else if constexpr (std::is_same_v<T, ScheduleSpec>) {
+                            schedules = std::list(arg.schedules);
+                        }
+                    },
                     change);
             });
         });
@@ -219,6 +223,15 @@ class ValveComponent : public Component {
 
     void populateTelemetry(JsonObject& telemetry) {
         telemetry["state"] = this->state;
+        auto overrideUntil = this->overrideUntil.load();
+        if (overrideUntil != time_point<system_clock>()) {
+            time_t rawtime = system_clock::to_time_t(overrideUntil);
+            auto timeinfo = gmtime(&rawtime);
+            char buffer[80];
+            strftime(buffer, 80, "%FT%TZ", timeinfo);
+            telemetry["overrideEnd"] = string(buffer);
+            telemetry["overrideState"] = this->overrideState.load();
+        }
     }
 
 private:
@@ -255,11 +268,15 @@ class ValveComponent : public Component {
                 // Ignore
                 break;
         }
-        // TODO Publish event
+        mqttRoot->publish("events/state", [=](JsonObject& json) {
+            json["state"] = state;
+        });
+        publishTelemetry();
     }
 
     PwmMotorDriver& controller;
     ValveControlStrategy& strategy;
+    std::function<void()> publishTelemetry;
 
     ValveState state = ValveState::NONE;
 
@@ -274,9 +291,9 @@ class ValveComponent : public Component {
         std::list<ValveSchedule> schedules;
     };
 
-    std::list<ValveSchedule> schedules;
-    ValveState overrideState = ValveState::NONE;
-    time_point<system_clock> overrideUntil;
+    std::list<ValveSchedule> schedules = {};
+    std::atomic<ValveState> overrideState = ValveState::NONE;
+    std::atomic<time_point<system_clock>> overrideUntil = time_point<system_clock>();
     Queue<std::variant<OverrideSpec, ScheduleSpec>> updateQueue { "eventQueue", 1 };
 };
 

src/peripherals/valve/ValveScheduler.hpp

@@ -39,9 +39,10 @@ class ValveSchedule {
         return duration;
     }
 
-    const time_point<system_clock> start;
-    const seconds period;
-    const seconds duration;
+private:
+    time_point<system_clock> start;
+    seconds period;
+    seconds duration;
 };
 
 struct ValveStateUpdate {

test/ValveSchedulerTest.cpp

@@ -31,9 +31,9 @@ class ValveSchedulerTest : public testing::Test {
 
 TEST_F(ValveSchedulerTest, can_create_schedule) {
     ValveSchedule schedule(base, hours { 1 }, minutes { 1 });
-    EXPECT_EQ(schedule.start, base);
-    EXPECT_EQ(schedule.period, hours { 1 });
-    EXPECT_EQ(schedule.duration, minutes { 1 });
+    EXPECT_EQ(schedule.getStart(), base);
+    EXPECT_EQ(schedule.getPeriod(), hours { 1 });
+    EXPECT_EQ(schedule.getDuration(), minutes { 1 });
 }
 
 // TEST_F(ValveSchedulerTest, can_create_schedule_from_json) {