ranczo-io/libs/modbus.cpp

#include "config.hpp"

#include <ranczo-io/utils/modbus.hpp>

#include <boost/asio.hpp>
#include <boost/asio/associated_executor.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/smart_ptr/shared_ptr.hpp>

#include <modbus/modbus.h>

#include <cerrno>
#include <chrono>
#include <cstdint>
#include <memory>
#include <ratio>
#include <string>

namespace ranczo {

ModbusTcpContext::~ModbusTcpContext() {
    if(ctx_) {
        ::modbus_close(ctx_.get());
        ctx_.reset();
    }
    pool_.join();
}

awaitable_expected< void > ModbusTcpContext::async_connect() {
    co_return co_await async_call< void >([this](modbus_t *& out_ctx) -> expected< void > {
        _log.info("create ctx");
        modbus_t * raw = ::modbus_new_tcp(host_.c_str(), port_);
        if(!raw)
            return unexpected(errno_errc());
        std::unique_ptr< modbus_t, CtxDeleter > ctx(raw);

        if(::modbus_connect(ctx.get()) == -1) {
            _log.error("failed to conenct");
            return unexpected(errno_errc());
        }
        _connected = true;
        _log.info("connected");

        timeval tv{2, 0};
        ::modbus_set_response_timeout(ctx.get(), tv.tv_sec, tv.tv_usec);
        // ::modbus_set_byte_timeout(ctx.get(), tv.tv_sec, tv.tv_usec);

        std::scoped_lock lk(mx_);
        ctx_    = std::move(ctx);
        out_ctx = ctx_.get();
        return {};
    });
}

bool ModbusTcpContext::connected() const {
    return _connected;
}

awaitable_expected< void > ranczo::ModbusTcpContext::wait_between_commands() {
    namespace asio = boost::asio;
    using namespace std::chrono;

    const auto now    = steady_clock::now();
    const auto expiry = delay_timer_.expiry();

    // Jeśli timer ma expiry w przyszłości – czekamy
    if(expiry > now) {
        boost::system::error_code ec;
        _log.trace("Timer waits {:.2f}us to expire time between ",
          std::chrono::duration_cast< std::chrono::duration< double, std::micro > >(expiry - now).count());

        co_await delay_timer_.async_wait(asio::redirect_error(asio::use_awaitable, ec));
        if(ec && ec != asio::error::operation_aborted) {
            co_return unexpected(ec);
        }
    }

    co_return _void{};
}

awaitable_expected< void > ModbusTcpContext::async_close() {
    co_return co_await async_call< void >([this](modbus_t *&) -> expected< void > {
        std::scoped_lock lk(mx_);
        if(ctx_) {
            ::modbus_close(ctx_.get());
            ctx_.reset();
            _connected = false;
        }
        return {};
    });
}

boost::system::error_code ModbusTcpContext::errno_errc() {
    return {errno, boost::system::generic_category()};
}

std::shared_ptr< ModbusTcpContext >
ModbusTcpContext::create(boost::asio::io_context & io, std::string host, int port, std::size_t pool_size) {
    return std::make_shared< ModbusTcpContext >(io, std::move(host), port, pool_size);
}

struct on_exit {
    std::function< void() > _fn;

    ~on_exit() {
        if(_fn)
            _fn();
    }
};

template < typename T, typename F >
awaitable_expected< T > ModbusTcpContext::call_with_lock(F && op) {
    // Throttling – zapewnij kilka ms przerwy między komendami
    ASYNC_CHECK(wait_between_commands());

    co_return co_await async_call< T >([this, op = std::forward< F >(op)](modbus_t *& out) -> expected< T > {
        on_exit _{[&]() { // Ustaw nowe "okno" – od teraz za command_interval_ ms
            delay_timer_.expires_at(std::chrono::steady_clock::now() + command_interval_);
        }};
        std::scoped_lock lk(mx_);
        if(!ctx_)
            return unexpected(make_error_code(boost::system::errc::not_connected));
        out = ctx_.get();
        return op(out);
    });
    
    
}


template < typename T, typename Maker >
awaitable_expected< T > ModbusTcpContext::async_call(Maker && maker) {
    namespace asio = boost::asio;

    // Initiator wyprodukuje expected<T>
    auto initiator = [this, maker = std::forward< Maker >(maker)](auto && completion_handler) mutable {
        auto ex = asio::get_associated_executor(completion_handler, io_.get_executor());

        asio::post(pool_, [this, maker = std::move(maker), completion_handler = std::move(completion_handler), ex]() mutable {
            // maker może potrzebować zwrócić też wskaźnik na ctx (referencja wyjściowa)
            modbus_t * ctx_ptr   = nullptr;
            expected< T > result = maker(ctx_ptr);
            asio::post(ex, [completion_handler = std::move(completion_handler), result = std::move(result)]() mutable {
                completion_handler(std::move(result));
            });
        });
    };

    co_return co_await asio::async_initiate< decltype(asio::use_awaitable), void(expected< T >) >(
      std::move(initiator), asio::use_awaitable_t<>{});
}

awaitable_expected< uint16_t > ModbusDevice::async_read_holding_register(uint16_t address) {
    address -= 1;
    auto ctx = ctx_; // kopia shared_ptr dla bezpieczeństwa w tasku
    co_return co_await ctx->call_with_lock< std::uint16_t >([this, address](modbus_t * c) -> expected< std::uint16_t > {
        if(::modbus_set_slave(c, unit_id_) == -1) {
            _log.error("modbus_set_slave address {} failed with {}", address, errno);
            return unexpected(errno_errc());
        }
        
        uint16_t val = 0;
        int rc       = ::modbus_read_registers(c, static_cast< int >(address), 1, &val);
        if(rc == -1) {
            _log.error("modbus_read_registers address {} failed with {}", address, errno);
            return unexpected(errno_errc());
        }
        return val;
    });
}

awaitable_expected<std::pmr::vector<uint16_t> > ModbusDevice::async_read_holding_registers(uint16_t address, std::size_t number, std::pmr::memory_resource *mr)
{
    BOOST_ASSERT(mr);
    address -= 1;
    auto ctx = ctx_; // kopia shared_ptr dla bezpieczeństwa w tasku
    co_return co_await ctx->call_with_lock< std::pmr::vector< std::uint16_t > >(
      [this, address, mr, number](modbus_t * c) -> expected< std::pmr::vector< std::uint16_t > > {
          if(::modbus_set_slave(c, unit_id_) == -1) {
              _log.error("modbus_set_slave address {} failed with {}", address, errno);
              return unexpected(errno_errc());
          }

          std::pmr::vector< std::uint16_t > val{mr};
          val.resize(number);
          int rc = ::modbus_read_registers(c, static_cast< int >(address), number, val.data());
          if(rc == -1) {
              _log.error("modbus_read_registers address {} failed with {}", address, errno);
              return unexpected(errno_errc());
          }
          return val;
      });
}

awaitable_expected< void > ModbusDevice::async_write_coil(uint16_t address, bool value) {
    auto ctx = ctx_;
    co_return co_await ctx->call_with_lock< void >([this, address, value](modbus_t * c) -> expected< void > {
        if(::modbus_set_slave(c, unit_id_) == -1) {
            _log.error("Modbus modbus_set_slave address {} failed with {}", address, errno);
            return unexpected(errno_errc());
        }
        const int rc = ::modbus_write_bit(c, static_cast< int >(address), value ? 1 : 0);
        if(rc == -1) {
            _log.error("Modbus modbus_write_bit address {} failed with {}", address, errno);
            return unexpected(errno_errc());
        }
        return {};
    });
}

} // namespace ranczo