Release v0.3.0

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.14)
 
-project(iotic VERSION 0.3.0 LANGUAGES C CXX)
+project(iotic VERSION 0.3.1 LANGUAGES C CXX)
 
 include(FetchContent)
 set(FETCHCONTENT_QUIET FALSE)

README.md

@@ -45,8 +45,8 @@ iotic implements several strategies to realize PC surplus charging of electric v
 You need at least Bookworm (not working with Bullseye).
 Installation
 ```
-wget https://github.com/mincequi/iotic/releases/download/v0.3.0/iotic_0.3.0_armhf.deb
-sudo apt install ./iotic_0.3.0_armhf.deb
+wget https://github.com/mincequi/iotic/releases/download/v0.3.1/iotic_0.3.1_armhf.deb
+sudo apt install ./iotic_0.3.1_armhf.deb
 ```
 Starting
 ```

src/http/HttpClient.cpp

@@ -39,8 +39,7 @@ void HttpClient::run(char const* host, char const* port, char const* target) {
 }
 
 void HttpClient::on_resolve(beast::error_code ec, tcp::resolver::results_type results) {
-    if (ec)
-        return fail(ec, "resolve");
+    if (ec) return fail(ec, "resolve");
 
     // Set a timeout on the operation
     _stream.expires_after(std::chrono::seconds(30));
@@ -52,8 +51,7 @@ void HttpClient::on_resolve(beast::error_code ec, tcp::resolver::results_type re
 }
 
 void HttpClient::on_connect(beast::error_code ec, tcp::resolver::results_type::endpoint_type) {
-    if (ec)
-        return fail(ec, "connect");
+    if (ec) return fail(ec, "connect");
 
     // Set a timeout on the operation
     _stream.expires_after(std::chrono::seconds(30));
@@ -63,16 +61,14 @@ void HttpClient::on_connect(beast::error_code ec, tcp::resolver::results_type::e
 }
 
 void HttpClient::on_write(beast::error_code ec, std::size_t /*bytes_transferred*/) {
-    if (ec)
-        return fail(ec, "write");
+    if (ec) return fail(ec, "write");
 
     // Receive the HTTP response
     http::async_read(_stream, _buffer, _response, beast::bind_front_handler(&HttpClient::on_read, shared_from_this()));
 }
 
 void HttpClient::on_read(beast::error_code ec, std::size_t /*bytes_transferred*/) {
-    if(ec)
-        return fail(ec, "read");
+    if (ec) return fail(ec, "read");
 
     // Write the message to standard out
     std::cout << _response << std::endl;

src/main.cpp

@@ -41,7 +41,7 @@ int main(int argc, char *argv[]) {
     QTimer timer;
     QObject::connect(&timer, &QTimer::timeout, []() {
         uvw::loop::get_default()->run(uvw::loop::run_mode::NOWAIT);
-        modbus::ModbusDiscovery::ioc.poll();
+        //modbus::ModbusDiscovery::ioc.poll();
     });
     timer.start(100ms);
 

src/modbus/CMakeLists.txt

@@ -1,5 +1,7 @@
 add_library(modbus STATIC
   ModbusDiscovery.cpp
+  ModbusThing.cpp
+  #ModbusThingDecorator.cpp
 )
 
 target_link_libraries(modbus

src/modbus/ModbusDiscovery.cpp

@@ -2,23 +2,18 @@
 
 #include <QHostAddress>
 #include <QNetworkInterface>
-#include <QTcpSocket>
 
 #include <common/Logger.h>
 #include <config/Config.h>
+#include <modbus/ModbusThing.h>
 #include <things/ThingsRepository.h>
-#include <things/sunspec/SunSpecManager.h>
-#include <things/sunspec/SunSpecThing.h>
 
 namespace modbus {
 
 using namespace std::placeholders;
 using namespace sunspec;
 
-asio::io_context ModbusDiscovery::ioc;
-
-ModbusDiscovery::ModbusDiscovery() :
-    _httpClient(std::make_shared<HttpClient>(ioc)) {
+ModbusDiscovery::ModbusDiscovery() {
     _discoveryTimer.callOnTimeout(this, &ModbusDiscovery::onStartDiscovering);
 }
 
@@ -48,7 +43,6 @@ void ModbusDiscovery::onStartDiscovering() {
     }
 
     // Scan subnets
-    QTcpSocket socket;
     LOG_S(INFO) << "discovering things in subnet: " << subnet << "0/24";
     for (uint8_t i = 1; i < 255; ++i) {
         const QString host = subnet + QString::number(i);
@@ -58,21 +52,22 @@ void ModbusDiscovery::onStartDiscovering() {
 
         //_httpClient->connect(host.toStdString(), 502);
         auto candidate = std::make_unique<SunSpecThing>(ThingInfo{ThingInfo::SunSpec, host.toStdString(), host.toStdString()});
-        auto sub = candidate->state().subscribe(std::bind(&ModbusDiscovery::onCandidateStateChanged, this, candidate.get(), _1));
-        candidate->connectDevice();
+        //auto candidate = std::make_unique<ModbusThing>(ThingInfo{ThingInfo::SunSpec, host.toStdString(), host.toStdString()});
+        auto sub = candidate->stateObservable().subscribe(std::bind(&ModbusDiscovery::onCandidateStateChanged, this, candidate.get(), _1));
+        candidate->connect();
         _candidates.push_back({std::move(candidate), sub});
     }
 }
 
-void ModbusDiscovery::onCandidateStateChanged(const SunSpecThing* candidate_, SunSpecThing::State state) {
+void ModbusDiscovery::onCandidateStateChanged(const SunSpecThing* candidate_, Thing::State state) {
     switch (state) {
-    case SunSpecThing::State::Failed:
+    case Thing::State::Failed:
         // TODO: do we actually need to unsubscribe?
         _candidates.remove_if([&](const auto& c) {
             return c.first.get() == candidate_;
         });
         break;
-    case SunSpecThing::State::Ready:
+    case Thing::State::Ready: {
         // Steal candidate from container
         auto it = std::find_if(_candidates.begin(), _candidates.end(), [&](const auto& c) {
             return c.first.get() == candidate_;
@@ -103,6 +98,9 @@ void ModbusDiscovery::onCandidateStateChanged(const SunSpecThing* candidate_, Su
         thingDiscoveredSubscriber().on_next(thing);
         break;
     }
+    default:
+        break;
+    }
 }
 
 } // namespace modbus

src/modbus/ModbusDiscovery.h

@@ -4,7 +4,6 @@
 
 #include <uvw/tcp.h>
 
-#include <http/HttpClient.h>
 #include <things/ThingsDiscovery.h>
 #include <things/sunspec/SunSpecThing.h>
 
@@ -20,16 +19,12 @@ class ModbusDiscovery : public QObject, public ThingsDiscovery {
     void start(int msec) override;
     void stop() override;
 
-    static asio::io_context ioc;
-
 private:
     void onStartDiscovering();
-    void onCandidateStateChanged(const sunspec::SunSpecThing* thing, sunspec::SunSpecThing::State state);
+    void onCandidateStateChanged(const sunspec::SunSpecThing* thing, Thing::State state);
 
     QTimer _discoveryTimer;
     std::list<std::pair<std::shared_ptr<sunspec::SunSpecThing>, rpp::composite_subscription>> _candidates;
-
-    std::shared_ptr<HttpClient> _httpClient;
 };
 
 } // namespace modbus

src/modbus/ModbusThing.cpp

@@ -0,0 +1,63 @@
+#include "ModbusThing.h"
+
+#include <QModbusTcpClient>
+
+#include <common/Logger.h>
+
+using namespace std::placeholders;
+
+ModbusThing::ModbusThing(const ThingInfo& info) :
+    Thing{info},
+    _modbusClient(new QModbusTcpClient) {
+
+    _modbusClient->connect(_modbusClient, &QModbusTcpClient::stateChanged, std::bind(&ModbusThing::onStateChanged, this, _1));
+    _modbusClient->connect(_modbusClient, &QModbusTcpClient::errorOccurred, std::bind(&ModbusThing::onErrorOccurred, this, _1));
+
+    _modbusClient->setConnectionParameter(QModbusDevice::NetworkPortParameter, 502);
+    _modbusClient->setConnectionParameter(QModbusDevice::NetworkAddressParameter, QString::fromStdString(info.host()));
+
+    _modbusClient->setTimeout(5000);
+    _modbusClient->setNumberOfRetries(2);
+}
+
+ModbusThing::~ModbusThing() {
+    _modbusClient->disconnect();
+    _modbusClient->deleteLater();
+}
+
+bool ModbusThing::connect() {
+    return _modbusClient->connectDevice();
+}
+
+void ModbusThing::disconnect() {
+    return _modbusClient->disconnectDevice();
+}
+
+void ModbusThing::doRead() {
+};
+
+void ModbusThing::onStateChanged(QModbusDevice::State state_) {
+    if (state_ == QModbusDevice::State::ConnectedState && state() == Thing::State::Uninitialized) {
+        LOG_S(INFO) << "host found: " << _modbusClient->connectionParameter(QModbusDevice::NetworkAddressParameter).toString();
+        stateSubscriber().on_next(State::Ready);
+        //pollNextUnitId();
+    } else if (state_ == QModbusDevice::State::UnconnectedState && state() != Thing::State::Uninitialized) {
+        // Elgris smart meters disconnects after 10s. So, we automatically reconnect.
+        LOG_S(WARNING) << id() << "> disconnected. Reconnecting...";
+        connect();
+    }
+}
+
+void ModbusThing::onErrorOccurred(QModbusDevice::Error error) {
+    if (error == QModbusDevice::Error::NoError) {
+        return;
+    }
+
+    if (state() == Thing::State::Uninitialized) {
+        stateSubscriber().on_next(State::Failed);
+        return;
+    }
+
+    // We filter for connection error, because remotes might hang up.
+    LOG_IF_S(WARNING, error != QModbusDevice::Error::ConnectionError) << host() << "> error occured: " << _modbusClient->errorString().toStdString();
+}

src/modbus/ModbusThing.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include <QModbusDevice>
+
+#include <things/Thing.h>
+
+class QModbusTcpClient;
+
+class ModbusThing : public Thing {
+public:
+    ModbusThing(const ThingInfo& info);
+    virtual ~ModbusThing();
+
+    bool connect();
+    void disconnect();
+
+private:
+    virtual void doRead() override;
+
+    void onStateChanged(QModbusDevice::State state);
+    void onErrorOccurred(QModbusDevice::Error error);
+
+    QModbusTcpClient* _modbusClient;
+};

src/things/Thing.cpp

@@ -70,9 +70,24 @@ dynamic_observable<std::map<Property, Value>> Thing::properties() const {
     return _propertiesObservable;
 }
 
-dynamic_observable<Thing::State> Thing::state() const {
+dynamic_observable<Thing::State> Thing::stateObservable() const {
     return _stateSubject.get_observable();
 }
 
 void Thing::onRead(const std::string&) {
 }
+
+void Thing::doSetProperty(MutableProperty, const Value&) {
+}
+
+Thing::State Thing::state() const {
+    return _stateSubject.get_value();
+}
+
+dynamic_subscriber<Thing::State> Thing::stateSubscriber() const {
+    return _stateSubject.get_subscriber().as_dynamic();
+}
+
+dynamic_subscriber<std::map<Property, Value>> Thing::propertiesSubscriber() const {
+    return _propertiesSubject.get_subscriber().as_dynamic();
+}

src/things/Thing.h

@@ -19,6 +19,7 @@ class Thing : public ThingInfo {
     };
 
     enum class State {
+        Uninitialized,
         Ready,
         Failed
     };
@@ -33,31 +34,32 @@ class Thing : public ThingInfo {
     // If thing does not fire updates itself, this can be called to trigger it.
     void read();
 
+    dynamic_observable<State> stateObservable() const;
+
     void setProperty(MutableProperty property, const Value& value);
     const std::map<MutableProperty, Value>& mutableProperties() const;
-
-    /**
-     * @brief properties
-     * @return
-     */
     dynamic_observable<std::map<Property, Value>> properties() const;
 
-    dynamic_observable<State> state() const;
-
 protected:
+    State state() const;
+    dynamic_subscriber<State> stateSubscriber() const;
+    dynamic_subscriber<std::map<Property, Value>> propertiesSubscriber() const;
+
+    Type _type = Type::Undefined;
+
+private:
     virtual void doRead() = 0;
     virtual void onRead(const std::string& response);
-    virtual void doSetProperty(MutableProperty property, const Value& value) = 0;
+    virtual void doSetProperty(MutableProperty property, const Value& value);
 
-    Type _type = Type::Undefined;
     uint16_t _materialIcon = 0;
 
     // TODO: do not store persistent properties here. Use Config as single source of truth.
     std::map<MutableProperty, Value> _mutableProperties;
     publish_subject<std::map<Property, Value>> _propertiesSubject;
     dynamic_observable<std::map<Property, Value>> _propertiesObservable;
 
-    publish_subject<State> _stateSubject;
+    behavior_subject<State> _stateSubject = State::Uninitialized;
 
     friend class ThingsRepository;
 };

src/things/ThingsRepository.cpp

@@ -32,7 +32,7 @@ void ThingsRepository::addThing(ThingPtr&& thing) {
     _things.push_back(std::move(thing));
     _thingAdded.get_subscriber().on_next(_things.back());
 
-    _things.back()->state().subscribe([this, id](auto state) {
+    _things.back()->stateObservable().subscribe([this, id](auto state) {
         if (state == Thing::State::Failed) {
             LOG_S(WARNING) << "thing completed: " << id;
             // We must not directly delete this thing because thing itself might still process something.

src/things/goe/GoeCharger.cpp

@@ -66,7 +66,7 @@ void GoeCharger::onRead(const std::string& response) {
 
     _status = goe::toStatus(doc["car"].toInt());
 
-    _propertiesSubject.get_subscriber().on_next({
+    propertiesSubscriber().on_next({
         { Property::status, static_cast<int>(_status) },
         { Property::power, nrg.at(11).toDouble() },
         { Property::voltage, voltage }

src/things/http/CMakeLists.txt

@@ -46,5 +46,7 @@ PRIVATE
 target_link_libraries(httpthing
 PUBLIC
   common
+  goe
   qmdnsengine
+  shelly
 )

src/things/shelly/Shelly.cpp

@@ -61,6 +61,6 @@ void Shelly::onRead(const std::string& response) {
     }
 
     if (!properties.empty()) {
-        _propertiesSubject.get_subscriber().on_next(properties);
+        propertiesSubscriber().on_next(properties);
     }
 }

src/things/sunspec/CMakeLists.txt

@@ -1,7 +1,7 @@
 add_library(sunspec STATIC
   SunSpecBlock.cpp
   SunSpecDataPoint.cpp
-  SunSpecFactory.cpp
+  SunSpecDiscovery.cpp
   SunSpecLogger.cpp
   SunSpecManager.cpp
   SunSpecMeasuredValue.cpp

src/things/sunspec/SunSpecDiscovery.cpp

@@ -0,0 +1,10 @@
+#include "SunSpecDiscovery.h"
+
+SunSpecDiscovery::SunSpecDiscovery() {
+}
+
+void SunSpecDiscovery::start(int msec) {
+}
+
+void SunSpecDiscovery::stop() {
+}