wirenboard / wb-mqtt-smartweb / 2

Committed 31 Jul 2026 10:01AM UTC coverage: 33.897% (-0.4%) from 34.328%

Build # 2

Build Type

Pull #40

github

Specific Base 476708

Committed by

ekateluv

Commit Message

Lower coverage threshold to 33

Pull Request Pull Request #40: Use exported vars instead of MAKEFLAGS for coverage options

Coverage Stats

307 of 744 branches covered (41.26%)

501 of 1478 relevant lines covered (33.9%)

20.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

4.41

/src/MqttToSmartWebGateway.cpp

#include "MqttToSmartWebGateway.h"

#include <string.h>
#include <wblib/exceptions.h>

#include "exceptions.h"
#include "log.h"

using namespace WBMQTT;
using namespace CAN;
using namespace std;

namespace
{
    const auto KEEP_ALIVE_INTERVAL_S = TTimeIntervalS(10); // if nothing else to do - each 10 seconds send I_AM_HERE
    const auto CONNECTION_TIMEOUT_MIN =
        TTimeIntervalMin(10); // after 10 minutes without any messages connection is considered lost
    const auto SEND_MESSAGES_TIME_M = TTimeIntervalMin(10);        // send value during 10 minutes
    const auto SEND_MESSAGES_INTERVAL_MS = TTimeIntervalMs(30000); // interval between messages
    const auto READ_TIMEOUT_MS = TTimeIntervalMs(1000);            // 1 sec for messages waiting

    TTimePoint now()
    {
        return chrono::steady_clock::now();
    }
}

#define LOG(logger) ::logger.Log() << LOGGER_PREFIX

void print_frame(WBMQTT::TLogger& logger, const CAN::TFrame& frame, const std::string& prefix)
{
    if (logger.IsEnabled()) {
        SmartWeb::TCanHeader header;
        header.raw = frame.can_id;

        std::stringstream ss;
        for (int i = 0; i < frame.can_dlc; ++i) {
            ss << " " << std::hex << std::uppercase << std::setfill('0') << std::setw(2) << (int)frame.data[i];
        }
        logger.Log() << prefix << ": " << std::hex << std::uppercase << std::setfill('0') << std::setw(2)
                     << frame.can_id << " (pt: " << std::dec << (int)header.rec.program_type << ", pid: " << std::dec
                     << (int)header.rec.program_id << ", fid: " << std::dec << (int)header.rec.function_id
                     << ", mf: " << std::dec << (int)header.rec.message_format << ", mt: " << std::dec
                     << (int)header.rec.message_type << ")" << ss.str();
    }
}

void TMqttChannel::from_string(const string& deviceControl)
{
    auto delimiter_position = deviceControl.find('/');

    if (delimiter_position == string::npos) {
        throw TDriverError("unable to determine device id and control id from string '" + deviceControl + "'");
    }

    device = deviceControl.substr(0, delimiter_position);
    control = deviceControl.substr(delimiter_position + 1);

    if (device.empty() || control.empty()) {
        throw TDriverError("unable to determine device id or control id from string '" + deviceControl + "'");
    }
}

bool TMqttChannel::is_initialized() const
{
    return !device.empty() || !control.empty();
}

string TMqttChannel::to_string(const string& device, const string& control)
{
    return device + "/" + control;
}

string TMqttChannel::to_string() const
{
    return to_string(device, control);
}

void TChannelState::postpone_send()
{
    LastSendTimePoint = now();
    SendTimePoint = LastSendTimePoint + SEND_MESSAGES_INTERVAL_MS;
}

void TChannelState::postpone_send_end()
{
    SendEndTimePoint = now() + SEND_MESSAGES_TIME_M;
}

void TChannelState::schedule_to_send()
{
    SendTimePoint = now();
    postpone_send_end();
}

void TBroadcastChannel::schedule_to_send(const SmartWeb::TMappingPoint& mp)
{
    TChannelState::schedule_to_send();
    mapping_point = mp;
}

TMqttChannelTiming::TMqttChannelTiming(const TMqttChannelTiming& other)
    : LastUpdateTimePointMutex(),
      ValueTimeoutMin(other.ValueTimeoutMin)
{
    unique_lock<mutex> lock(other.LastUpdateTimePointMutex);
    LastUpdateTimePoint = other.LastUpdateTimePoint;
}

void TMqttChannelTiming::refresh_last_update_timepoint()
{
    unique_lock<mutex> lock(LastUpdateTimePointMutex);
    LastUpdateTimePoint = now();
}

bool TMqttChannelTiming::is_timed_out() const
{
    if (ValueTimeoutMin.count() < 0) {
        return false;
    }

    unique_lock<mutex> lock(LastUpdateTimePointMutex);
    return (now() - LastUpdateTimePoint) > ValueTimeoutMin;
}

TTimePoint TMqttChannelTiming::get_last_update_timepoint() const
{
    return LastUpdateTimePoint;
}

bool TMqttToSmartWebGateway::FilterIsSet = false;
std::mutex TMqttToSmartWebGateway::StartupMutex;

TMqttToSmartWebGateway::TMqttToSmartWebGateway(const TMqttToSmartWebConfig& config,
                                               std::shared_ptr<CAN::IPort> canPort,
                                               WBMQTT::PDeviceDriver driver)
    : DriverState(config),
      CanPort(canPort),
      Driver(driver)
{
    CONTROLLER_TYPE = 14; // External controller
    Enabled.store(true);
    CanPort->AddHandler(this);
    Thread = std::thread([this]() { TaskFn(); });
}

TMqttToSmartWebGateway::~TMqttToSmartWebGateway()
{
    CanPort->RemoveHandler(this);
    Enabled.store(false);
    if (Thread.joinable()) {
        Thread.join();
    }
}

bool TMqttToSmartWebGateway::Handle(const CAN::TFrame& frame)
{
    SmartWeb::TCanHeader header;
    header.raw = frame.can_id;
    if (!IsForMe(header, frame.data)) {
        return false;
    }
    std::unique_lock<std::mutex> waitLock(CanFramesMutex);
    CanFrames.push(frame);
    waitLock.unlock();
    CanFramesCv.notify_all();
    return true;
}

bool TMqttToSmartWebGateway::SelectTimeout(CAN::TFrame& frame)
{
    std::unique_lock<std::mutex> waitLock(CanFramesMutex);
    if (CanFrames.empty()) {
        if (std::cv_status::timeout == CanFramesCv.wait_for(waitLock, READ_TIMEOUT_MS)) {
            return false;
        }
    }
    frame = CanFrames.front();
    CanFrames.pop();
    return true;
}

bool TMqttToSmartWebGateway::IsForMe(const SmartWeb::TCanHeader& header, const TFrameData& data) const
{
    if (header.rec.program_id == DriverState.ProgramId) {
        return true;
    }
    if (header.rec.message_type == SmartWeb::MT_MSG_REQUEST && header.rec.program_type == SmartWeb::PT_CONTROLLER &&
        header.rec.function_id == SmartWeb::Controller::Function::GET_OUTPUT_VALUE)
    {
        SmartWeb::TMappingPoint mapping_point;
        memcpy(&mapping_point.raw, data, 2);
        return mapping_point.hostID ==
               DriverState.ProgramId; // NOLINT(clang-analyzer-core.UndefinedBinaryOperatorResult)
    }

    return false;
}

void TMqttToSmartWebGateway::TaskFn()
{
    WBMQTT::SetThreadName("MQTT to SW " + to_string(int(DriverState.ProgramId)));
    {
        std::unique_lock<std::mutex> lk(StartupMutex);
        if (!FilterIsSet) {
            Driver->SetFilter(GetAllDevicesFilter());
            Driver->WaitForReady();
            FilterIsSet = true;
        }
    }

    uint8_t program_id = DriverState.ProgramId;

    auto onHandler = Driver->On<TControlValueEvent>([&](const TControlValueEvent& event) {
        auto deviceControl = TMqttChannel::to_string(event.Control->GetDevice()->GetId(), event.Control->GetId());
        try {
            DriverState.MqttChannelsTiming.at(deviceControl).refresh_last_update_timepoint();
        } catch (out_of_range&) {
        }
    });

    auto get_response_frame = [&](SmartWeb::TCanHeader header) {
        if (header.rec.message_type != SmartWeb::MT_MSG_REQUEST) {
            throw TFrameError("Frame error: response to NOT request frame");
        }

        if (header.rec.program_id != program_id) {
            throw TFrameError("Frame error: response to request frame for different device (" +
                              to_string((int)header.rec.program_id) + ")");
        }

        header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;

        TFrame response{0};
        response.can_id = header.raw;

        return response;
    };

    auto postpone_i_am_here = [&] { SendIAmHereTime = now() + KEEP_ALIVE_INTERVAL_S; };
    auto postpone_connection_reset = [&] { ResetConnectionTime = now() + CONNECTION_TIMEOUT_MIN; };

    auto send_frame = [&](TFrame& frame, const std::string& prefix) {
        frame.can_id |= CAN_EFF_FLAG; // just in case
        try {
            CanPort->Send(frame);
            print_frame(DebugMqttToSw, frame, "[" + std::to_string(DriverState.ProgramId) + "] " + prefix);
        } catch (const std::exception& e) {
            print_frame(ErrorMqttToSw,
                        frame,
                        "[" + std::to_string(DriverState.ProgramId) + "] " + prefix + " " + e.what());
        }
    };

    auto read_mqtt_value = [&](const string& device_id, const string& control_id) {
        try {
            const auto& mqtt_channel_timing =
                DriverState.MqttChannelsTiming.at(TMqttChannel::to_string(device_id, control_id));
            if (mqtt_channel_timing.is_timed_out()) {
                WarnMqttToSw.Log() << "MQTT value of control " << control_id << " of device " << device_id
                                   << " timed out. Returning undefined value";
                return SmartWeb::SENSOR_UNDEFINED;
            }
        } catch (out_of_range&) {
            // Should never happen. Means that we did not add all mqtt channels to DriverState.MqttChannelsTiming at
            // startup as we should've.
            ErrorMqttToSw.Log() << "[error code 1] There is bug in code; Report error code to driver maintainer";
        }

        try {
            auto tx = Driver->BeginTx();
            if (auto device = tx->GetDevice(device_id)) {
                if (auto control = device->GetControl(control_id)) {
                    if (control->GetError().empty()) {
                        return SmartWeb::SensorData::FromDouble(control->GetValue().As<double>());
                    } else {
                        WarnMqttToSw.Log() << "Unable to read mqtt value because of error on control " << control_id
                                           << " of device " << device_id << ": " << control->GetError();
                        return SmartWeb::SENSOR_UNDEFINED;
                    }
                } else {
                    WarnMqttToSw.Log() << "Unable to read mqtt value because control " << control_id << " of device "
                                       << device_id << " does not exist";
                    return SmartWeb::SENSOR_UNDEFINED;
                }
            } else {
                WarnMqttToSw.Log() << "Unable to read mqtt value because device " << device_id << " does not exist";
                return SmartWeb::SENSOR_UNDEFINED;
            }
        } catch (const WBMQTT::TBaseException& e) {
            WarnMqttToSw.Log() << "Unable to read mqtt value: " << e.what();

            return SmartWeb::SENSOR_UNDEFINED;
        }
    };

    auto i_am_here = [&] {
        postpone_i_am_here();

        SmartWeb::TCanHeader header;

        header.rec.program_type = SmartWeb::PT_CONTROLLER;
        header.rec.program_id = program_id;
        header.rec.function_id = SmartWeb::Controller::Function::I_AM_HERE;
        header.rec.message_format = SmartWeb::MF_FORMAT_0;
        header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;

        TFrame frame{0};
        frame.can_id = header.raw | CAN_EFF_FLAG;
        frame.can_dlc = 1;
        frame.data[0] = CONTROLLER_TYPE;

        send_frame(frame, "send I_AM_HERE");
    };

    auto send_scheduled_i_am_here = [&] {
        if (SendIAmHereTime <= now()) {
            i_am_here();
        }
    };

    auto get_channel_number = [&](const SmartWeb::TCanHeader& header) {
        auto channel_number = max((size_t)DriverState.ParameterCount, DriverState.ParameterMapping.size());

        auto response = get_response_frame(header);
        response.can_dlc = 2;
        response.data[0] = 0xFF & channel_number;
        response.data[1] = 0xFF & channel_number >> 8;
        send_frame(response, "send channel number");
    };

    auto get_parameter_value = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {
        SmartWeb::TParameterData parameter_data{0};

        memcpy(&parameter_data.raw, data, 4);

        if (parameter_data.program_type != SmartWeb::PT_CONTROLLER) {
            throw TUnsupportedError("Unsupported program type " + to_string((int)parameter_data.program_type) +
                                    " for GET_PARAMETER_VALUE");
        }

        const auto& itDeviceChannel = DriverState.ParameterMapping.find(parameter_data.raw_info);

        int16_t value = SmartWeb::SENSOR_UNDEFINED;

        if (itDeviceChannel == DriverState.ParameterMapping.end()) {
            DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId
                                << "] unmapped parameter: type: " << (int)parameter_data.program_type
                                << ", id: " << (int)parameter_data.parameter_id
                                << ", index: " << (int)parameter_data.indexed_parameter.index;
        } else {
            DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId
                                << "] get parameter: type: " << (int)parameter_data.program_type
                                << ", id: " << (int)parameter_data.parameter_id
                                << ", index: " << (int)parameter_data.indexed_parameter.index << ", raw "
                                << (int)parameter_data.raw_info;
            value = read_mqtt_value(itDeviceChannel->second.device, itDeviceChannel->second.control);
        }

        auto response = get_response_frame(header);

        response.can_dlc = 5;

        memset(response.data, 0, sizeof response.data);

        memcpy(parameter_data.indexed_parameter.value, &value, sizeof value);

        memcpy(response.data, &parameter_data.raw, 5);

        send_frame(response, "get parameter response");

        DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId
                            << "] parameter {type: " << (int)parameter_data.program_type
                            << ", id: " << (int)parameter_data.parameter_id
                            << ", index: " << (int)parameter_data.indexed_parameter.index
                            << "} <== " << SmartWeb::SensorData::ToDouble(value);
    };

    auto get_output_value = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {
        SmartWeb::TMappingPoint mapping_point{};
        memcpy(mapping_point.rawID, data, 2);

        if (mapping_point.hostID != program_id) {
            throw TFrameError("hostID of mapping point does not match with driver program_id");
        }

        auto channel_id = mapping_point.channelID;
        if (channel_id >= CONTROLLER_OUTPUT_MAX) {
            throw TFrameError("channel_id of mapping point is out of bounds: " + to_string(channel_id));
        }

        auto& channel = DriverState.OutputMapping[channel_id];

        if (channel.is_initialized()) {
            channel.schedule_to_send(mapping_point);
            InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] scheduled output " << (int)channel_id;
        } else {
            WarnMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] unmapped output " << (int)channel_id;
        }
    };

    auto send_scheduled_outputs = [&] {
        SmartWeb::TCanHeader header;
        header.rec.program_type = SmartWeb::PT_CONTROLLER;
        header.rec.program_id = program_id;
        header.rec.function_id = SmartWeb::Controller::Function::GET_OUTPUT_VALUE;
        header.rec.message_format = SmartWeb::MF_FORMAT_0;
        header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;

        TFrame frame{0};

        frame.can_dlc = 4;

        for (uint8_t channel_id = 0; channel_id < CONTROLLER_OUTPUT_MAX; ++channel_id) {
            auto& channel = DriverState.OutputMapping[channel_id];

            if (channel.SendEndTimePoint < now()) {
                continue; // too late
            }

            if (channel.device.empty() || channel.control.empty()) {
                continue; // weird
            }

            auto lastUpdate =
                DriverState.MqttChannelsTiming.at(TMqttChannel::to_string(channel.device, channel.control))
                    .get_last_update_timepoint();
            if (lastUpdate <= channel.LastSendTimePoint) { // no channel updates
                if (channel.SendTimePoint > now()) {
                    continue; // too soon
                }
            }

            auto value = read_mqtt_value(channel.device, channel.control);

            frame.can_id = header.raw | CAN_EFF_FLAG;
            memcpy(frame.data, &channel.mapping_point.raw, sizeof channel.mapping_point.raw);
            frame.data[2] = 0xFF & value >> 8;
            frame.data[3] = 0xFF & value;

            send_frame(frame, "send output");

            DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] output {channel_id: " << (int)channel_id
                                << "} <== " << SmartWeb::SensorData::ToDouble(value);

            channel.postpone_send();
        }
    };

    auto get_controller_type = [&](const SmartWeb::TCanHeader& header) {
        auto response = get_response_frame(header);
        response.can_dlc = 1;
        response.data[0] = CONTROLLER_TYPE;
        send_frame(response, "send controller type");
    };

    auto handle_request = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {
        switch (header.rec.program_type) {
            case SmartWeb::PT_CONTROLLER:
                switch (header.rec.function_id) {
                    case SmartWeb::Controller::Function::GET_CHANNEL_NUMBER:
                        return get_channel_number(header);
                    case SmartWeb::Controller::Function::GET_CONTROLLER_TYPE:
                        return get_controller_type(header);
                    case SmartWeb::Controller::Function::GET_OUTPUT_VALUE:
                        return get_output_value(header, data);
                    case SmartWeb::Controller::Function::I_AM_HERE:
                        return i_am_here();
                    default:
                        throw TUnsupportedError("function id " + to_string((int)header.rec.function_id) +
                                                " is unsupported");
                }
            case SmartWeb::PT_REMOTE_CONTROL:
                switch (header.rec.function_id) {
                    case SmartWeb::RemoteControl::Function::GET_PARAMETER_VALUE:
                        return get_parameter_value(header, data);
                    default:
                        throw TUnsupportedError("function id " + to_string((int)header.rec.function_id) +
                                                " is unsupported");
                }
            default:
                throw TUnsupportedError("program_type " + to_string((int)header.rec.program_type) + " is unsupported");
        }
    };

    TFrame frame{0};

    SendIAmHereTime = now();

    SmartWeb::TCanHeader header{0};

    //   can0  0015AC0B   [8]  00 00 00 00 00 00 00 00  (CONTROLLER: JOURNAL (Get controller journal notes))
    //   can0  000AAC0B   [0]                           (CONTROLLER: GET_CHANNEL_NUMBER (Узнать количество
    //   входов/выходов)) can0  0003AC0B   [0]                           (CONTROLLER: GET_ACTIVE_PROGRAMS_LIST (Узнать
    //   список активных программ)) can0  0001AC16   [4]  0B 01 00 00              (REMOTE_CONTROL: GET_PARAMETER_VALUE)
    //   can0  0001AC16   [4]  0B 1E 00 00              (REMOTE_CONTROL: GET_PARAMETER_VALUE)
    //   can0  0001AC16   [4]  0B 02 00 00              (REMOTE_CONTROL: GET_PARAMETER_VALUE)
    //   can0  0008AC0B   [0]                           (CONTROLLER: GET_CONTROLLER_TYPE (Узнать тип контроллера))
    //   can0  0018AC0B   [1]  00                       (CONTROLLER: GET_RELAY_MAPPING (Get controller output binding))
    //   can0  0001AC16   [4]  0B 1C 00 00              (REMOTE_CONTROL: GET_PARAMETER_VALUE)
    //   can0  0001AC16   [4]  0B 1D 00 00              (REMOTE_CONTROL: GET_PARAMETER_VALUE)
    //   can0  0001AC16   [6]  0B 05 00 09 8A 2F        (REMOTE_CONTROL: GET_PARAMETER_VALUE)

    while (Enabled.load()) {
        memset(&frame, 0, sizeof(TFrame));

        bool frame_for_me = SelectTimeout(frame);

        if (!frame_for_me) {
            if (Status != DS_IDLE) {
                if (ResetConnectionTime <= now()) {
                    Status = DS_IDLE;
                    InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] CONNECTION LOST: TIMEOUT. IDLE";
                }
            }
        }

        if (frame_for_me) {
            print_frame(DebugMqttToSw, frame, "[" + std::to_string(DriverState.ProgramId) + "] got frame");
        }

        try {
            switch (Status) {
                case DS_IDLE:
                    if (frame_for_me) {
                        Status = DS_RUNNING;
                        InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] CONNECTION ESTABILISHED. RUNNING";
                    } else {
                        send_scheduled_i_am_here();
                        break;
                    }

                case DS_RUNNING:
                    if (frame_for_me) {
                        postpone_connection_reset();
                        header.raw = frame.can_id;
                        handle_request(header, frame.data);
                    }

                    send_scheduled_outputs();
                    send_scheduled_i_am_here();

                    break;
            }

        } catch (const TUnsupportedError& e) {
            DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();
        } catch (const TDriverError& e) {
            WarnMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();
        } catch (const std::exception& e) {
            ErrorMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();
            exit(1);
        }
    }
    Driver->RemoveEventHandler(onHandler);
}

1	#include "MqttToSmartWebGateway.h"
2
3	#include <string.h>
4	#include <wblib/exceptions.h>
5
6	#include "exceptions.h"
7	#include "log.h"
8
9	using namespace WBMQTT;
10	using namespace CAN;
11	using namespace std;
12
13	namespace
14	{
15	const auto KEEP_ALIVE_INTERVAL_S = TTimeIntervalS(10); // if nothing else to do - each 10 seconds send I_AM_HERE
16	const auto CONNECTION_TIMEOUT_MIN =
17	TTimeIntervalMin(10); // after 10 minutes without any messages connection is considered lost
18	const auto SEND_MESSAGES_TIME_M = TTimeIntervalMin(10); // send value during 10 minutes
19	const auto SEND_MESSAGES_INTERVAL_MS = TTimeIntervalMs(30000); // interval between messages
20	const auto READ_TIMEOUT_MS = TTimeIntervalMs(1000); // 1 sec for messages waiting
21
22	TTimePoint now()	4✔
23	{
24	return chrono::steady_clock::now();	4✔
25	}
26	}
27
28	#define LOG(logger) ::logger.Log() << LOGGER_PREFIX
29
30	void print_frame(WBMQTT::TLogger& logger, const CAN::TFrame& frame, const std::string& prefix)	×
31	{
32	if (logger.IsEnabled()) {	×
33	SmartWeb::TCanHeader header;
34	header.raw = frame.can_id;	×
35
36	std::stringstream ss;	×
37	for (int i = 0; i < frame.can_dlc; ++i) {	×
38	ss << " " << std::hex << std::uppercase << std::setfill('0') << std::setw(2) << (int)frame.data[i];	×
39	}
40	logger.Log() << prefix << ": " << std::hex << std::uppercase << std::setfill('0') << std::setw(2)	×
41	<< frame.can_id << " (pt: " << std::dec << (int)header.rec.program_type << ", pid: " << std::dec	×
42	<< (int)header.rec.program_id << ", fid: " << std::dec << (int)header.rec.function_id	×
43	<< ", mf: " << std::dec << (int)header.rec.message_format << ", mt: " << std::dec	×
44	<< (int)header.rec.message_type << ")" << ss.str();	×
45	}	×
46	}	×
47
48	void TMqttChannel::from_string(const string& deviceControl)	6✔
49	{
50	auto delimiter_position = deviceControl.find('/');	6✔
51
52	if (delimiter_position == string::npos) {	6✔
53	throw TDriverError("unable to determine device id and control id from string '" + deviceControl + "'");	×
54	}
55
56	device = deviceControl.substr(0, delimiter_position);	6✔
57	control = deviceControl.substr(delimiter_position + 1);	6✔
58
59	if (device.empty() \|\| control.empty()) {	6✔
60	throw TDriverError("unable to determine device id or control id from string '" + deviceControl + "'");	×
61	}
62	}	6✔
63
64	bool TMqttChannel::is_initialized() const	2✔
65	{
66	return !device.empty() \|\| !control.empty();	2✔
67	}
68
69	string TMqttChannel::to_string(const string& device, const string& control)	×
70	{
71	return device + "/" + control;	×
72	}
73
74	string TMqttChannel::to_string() const	×
75	{
76	return to_string(device, control);	×
77	}
78
79	void TChannelState::postpone_send()	×
80	{
81	LastSendTimePoint = now();	×
82	SendTimePoint = LastSendTimePoint + SEND_MESSAGES_INTERVAL_MS;	×
83	}	×
84
85	void TChannelState::postpone_send_end()	×
86	{
87	SendEndTimePoint = now() + SEND_MESSAGES_TIME_M;	×
88	}	×
89
90	void TChannelState::schedule_to_send()	×
91	{
92	SendTimePoint = now();	×
93	postpone_send_end();	×
94	}	×
95
96	void TBroadcastChannel::schedule_to_send(const SmartWeb::TMappingPoint& mp)	×
97	{
98	TChannelState::schedule_to_send();	×
99	mapping_point = mp;	×
100	}	×
101
102	TMqttChannelTiming::TMqttChannelTiming(const TMqttChannelTiming& other)	×
103	: LastUpdateTimePointMutex(),	×
104	ValueTimeoutMin(other.ValueTimeoutMin)	×
105	{
106	unique_lock<mutex> lock(other.LastUpdateTimePointMutex);	×
107	LastUpdateTimePoint = other.LastUpdateTimePoint;	×
108	}	×
109
110	void TMqttChannelTiming::refresh_last_update_timepoint()	4✔
111	{
112	unique_lock<mutex> lock(LastUpdateTimePointMutex);	4✔
113	LastUpdateTimePoint = now();	4✔
114	}	4✔
115
116	bool TMqttChannelTiming::is_timed_out() const	×
117	{
118	if (ValueTimeoutMin.count() < 0) {	×
119	return false;	×
120	}
121
122	unique_lock<mutex> lock(LastUpdateTimePointMutex);	×
123	return (now() - LastUpdateTimePoint) > ValueTimeoutMin;	×
124	}	×
125
126	TTimePoint TMqttChannelTiming::get_last_update_timepoint() const	×
127	{
128	return LastUpdateTimePoint;	×
129	}
130
131	bool TMqttToSmartWebGateway::FilterIsSet = false;
132	std::mutex TMqttToSmartWebGateway::StartupMutex;
133
134	TMqttToSmartWebGateway::TMqttToSmartWebGateway(const TMqttToSmartWebConfig& config,	×
135	std::shared_ptr<CAN::IPort> canPort,
136	WBMQTT::PDeviceDriver driver)	×
137	: DriverState(config),	×
138	CanPort(canPort),	×
139	Driver(driver)	×
140	{
141	CONTROLLER_TYPE = 14; // External controller	×
142	Enabled.store(true);	×
143	CanPort->AddHandler(this);	×
144	Thread = std::thread([this]() { TaskFn(); });	×
145	}	×
146
147	TMqttToSmartWebGateway::~TMqttToSmartWebGateway()	×
148	{
149	CanPort->RemoveHandler(this);	×
150	Enabled.store(false);	×
151	if (Thread.joinable()) {	×
152	Thread.join();	×
153	}
154	}	×
155
156	bool TMqttToSmartWebGateway::Handle(const CAN::TFrame& frame)	×
157	{
158	SmartWeb::TCanHeader header;
159	header.raw = frame.can_id;	×
160	if (!IsForMe(header, frame.data)) {	×
161	return false;	×
162	}
163	std::unique_lock<std::mutex> waitLock(CanFramesMutex);	×
164	CanFrames.push(frame);	×
165	waitLock.unlock();	×
166	CanFramesCv.notify_all();	×
167	return true;	×
168	}	×
169
170	bool TMqttToSmartWebGateway::SelectTimeout(CAN::TFrame& frame)	×
171	{
172	std::unique_lock<std::mutex> waitLock(CanFramesMutex);	×
173	if (CanFrames.empty()) {	×
174	if (std::cv_status::timeout == CanFramesCv.wait_for(waitLock, READ_TIMEOUT_MS)) {	×
175	return false;	×
176	}
177	}
178	frame = CanFrames.front();	×
179	CanFrames.pop();	×
180	return true;	×
181	}	×
182
183	bool TMqttToSmartWebGateway::IsForMe(const SmartWeb::TCanHeader& header, const TFrameData& data) const	×
184	{
185	if (header.rec.program_id == DriverState.ProgramId) {	×
186	return true;	×
187	}
188	if (header.rec.message_type == SmartWeb::MT_MSG_REQUEST && header.rec.program_type == SmartWeb::PT_CONTROLLER &&	×
189	header.rec.function_id == SmartWeb::Controller::Function::GET_OUTPUT_VALUE)	×
190	{
191	SmartWeb::TMappingPoint mapping_point;
192	memcpy(&mapping_point.raw, data, 2);	×
193	return mapping_point.hostID ==	×
194	DriverState.ProgramId; // NOLINT(clang-analyzer-core.UndefinedBinaryOperatorResult)	×
195	}
196
197	return false;	×
198	}
199
200	void TMqttToSmartWebGateway::TaskFn()	×
201	{
202	WBMQTT::SetThreadName("MQTT to SW " + to_string(int(DriverState.ProgramId)));	×
203	{
204	std::unique_lock<std::mutex> lk(StartupMutex);	×
205	if (!FilterIsSet) {	×
206	Driver->SetFilter(GetAllDevicesFilter());	×
207	Driver->WaitForReady();	×
208	FilterIsSet = true;	×
209	}
210	}	×
211
212	uint8_t program_id = DriverState.ProgramId;	×
213
214	auto onHandler = Driver->On<TControlValueEvent>([&](const TControlValueEvent& event) {	×
215	auto deviceControl = TMqttChannel::to_string(event.Control->GetDevice()->GetId(), event.Control->GetId());	×
216	try {
217	DriverState.MqttChannelsTiming.at(deviceControl).refresh_last_update_timepoint();	×
218	} catch (out_of_range&) {	×
219	}	×
220	});	×
221
222	auto get_response_frame = [&](SmartWeb::TCanHeader header) {	×
223	if (header.rec.message_type != SmartWeb::MT_MSG_REQUEST) {	×
224	throw TFrameError("Frame error: response to NOT request frame");	×
225	}
226
227	if (header.rec.program_id != program_id) {	×
228	throw TFrameError("Frame error: response to request frame for different device (" +	×
229	to_string((int)header.rec.program_id) + ")");	×
230	}
231
232	header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;	×
233
234	TFrame response{0};	×
235	response.can_id = header.raw;	×
236
237	return response;	×
238	};	×
239
240	auto postpone_i_am_here = [&] { SendIAmHereTime = now() + KEEP_ALIVE_INTERVAL_S; };	×
241	auto postpone_connection_reset = [&] { ResetConnectionTime = now() + CONNECTION_TIMEOUT_MIN; };	×
242
243	auto send_frame = [&](TFrame& frame, const std::string& prefix) {	×
244	frame.can_id \|= CAN_EFF_FLAG; // just in case	×
245	try {
246	CanPort->Send(frame);	×
247	print_frame(DebugMqttToSw, frame, "[" + std::to_string(DriverState.ProgramId) + "] " + prefix);	×
248	} catch (const std::exception& e) {	×
249	print_frame(ErrorMqttToSw,	×
250	frame,
251	"[" + std::to_string(DriverState.ProgramId) + "] " + prefix + " " + e.what());	×
252	}	×
253	};	×
254
255	auto read_mqtt_value = [&](const string& device_id, const string& control_id) {	×
256	try {
257	const auto& mqtt_channel_timing =
258	DriverState.MqttChannelsTiming.at(TMqttChannel::to_string(device_id, control_id));	×
259	if (mqtt_channel_timing.is_timed_out()) {	×
260	WarnMqttToSw.Log() << "MQTT value of control " << control_id << " of device " << device_id	×
261	<< " timed out. Returning undefined value";	×
262	return SmartWeb::SENSOR_UNDEFINED;	×
263	}
264	} catch (out_of_range&) {	×
265	// Should never happen. Means that we did not add all mqtt channels to DriverState.MqttChannelsTiming at
266	// startup as we should've.
267	ErrorMqttToSw.Log() << "[error code 1] There is bug in code; Report error code to driver maintainer";	×
268	}	×
269
270	try {
271	auto tx = Driver->BeginTx();	×
272	if (auto device = tx->GetDevice(device_id)) {	×
273	if (auto control = device->GetControl(control_id)) {	×
274	if (control->GetError().empty()) {	×
275	return SmartWeb::SensorData::FromDouble(control->GetValue().As<double>());	×
276	} else {
277	WarnMqttToSw.Log() << "Unable to read mqtt value because of error on control " << control_id	×
278	<< " of device " << device_id << ": " << control->GetError();	×
279	return SmartWeb::SENSOR_UNDEFINED;	×
280	}
281	} else {
282	WarnMqttToSw.Log() << "Unable to read mqtt value because control " << control_id << " of device "	×
283	<< device_id << " does not exist";	×
284	return SmartWeb::SENSOR_UNDEFINED;	×
285	}	×
286	} else {
287	WarnMqttToSw.Log() << "Unable to read mqtt value because device " << device_id << " does not exist";	×
288	return SmartWeb::SENSOR_UNDEFINED;	×
289	}	×
290	} catch (const WBMQTT::TBaseException& e) {	×
291	WarnMqttToSw.Log() << "Unable to read mqtt value: " << e.what();	×
292
293	return SmartWeb::SENSOR_UNDEFINED;	×
294	}	×
295	};	×
296
297	auto i_am_here = [&] {	×
298	postpone_i_am_here();	×
299
300	SmartWeb::TCanHeader header;
301
302	header.rec.program_type = SmartWeb::PT_CONTROLLER;	×
303	header.rec.program_id = program_id;	×
304	header.rec.function_id = SmartWeb::Controller::Function::I_AM_HERE;	×
305	header.rec.message_format = SmartWeb::MF_FORMAT_0;	×
306	header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;	×
307
308	TFrame frame{0};	×
309	frame.can_id = header.raw \| CAN_EFF_FLAG;	×
310	frame.can_dlc = 1;	×
311	frame.data[0] = CONTROLLER_TYPE;	×
312
313	send_frame(frame, "send I_AM_HERE");	×
314	};	×
315
316	auto send_scheduled_i_am_here = [&] {	×
317	if (SendIAmHereTime <= now()) {	×
318	i_am_here();	×
319	}
320	};	×
321
322	auto get_channel_number = [&](const SmartWeb::TCanHeader& header) {	×
323	auto channel_number = max((size_t)DriverState.ParameterCount, DriverState.ParameterMapping.size());	×
324
325	auto response = get_response_frame(header);	×
326	response.can_dlc = 2;	×
327	response.data[0] = 0xFF & channel_number;	×
328	response.data[1] = 0xFF & channel_number >> 8;	×
329	send_frame(response, "send channel number");	×
330	};	×
331
332	auto get_parameter_value = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {	×
333	SmartWeb::TParameterData parameter_data{0};	×
334
335	memcpy(&parameter_data.raw, data, 4);	×
336
337	if (parameter_data.program_type != SmartWeb::PT_CONTROLLER) {	×
338	throw TUnsupportedError("Unsupported program type " + to_string((int)parameter_data.program_type) +	×
339	" for GET_PARAMETER_VALUE");	×
340	}
341
342	const auto& itDeviceChannel = DriverState.ParameterMapping.find(parameter_data.raw_info);	×
343
344	int16_t value = SmartWeb::SENSOR_UNDEFINED;	×
345
346	if (itDeviceChannel == DriverState.ParameterMapping.end()) {	×
347	DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId	×
348	<< "] unmapped parameter: type: " << (int)parameter_data.program_type	×
349	<< ", id: " << (int)parameter_data.parameter_id	×
350	<< ", index: " << (int)parameter_data.indexed_parameter.index;	×
351	} else {
352	DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId	×
353	<< "] get parameter: type: " << (int)parameter_data.program_type	×
354	<< ", id: " << (int)parameter_data.parameter_id	×
355	<< ", index: " << (int)parameter_data.indexed_parameter.index << ", raw "	×
356	<< (int)parameter_data.raw_info;	×
357	value = read_mqtt_value(itDeviceChannel->second.device, itDeviceChannel->second.control);	×
358	}
359
360	auto response = get_response_frame(header);	×
361
362	response.can_dlc = 5;	×
363
364	memset(response.data, 0, sizeof response.data);	×
365
366	memcpy(parameter_data.indexed_parameter.value, &value, sizeof value);	×
367
368	memcpy(response.data, &parameter_data.raw, 5);	×
369
370	send_frame(response, "get parameter response");	×
371
372	DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId	×
373	<< "] parameter {type: " << (int)parameter_data.program_type	×
374	<< ", id: " << (int)parameter_data.parameter_id	×
375	<< ", index: " << (int)parameter_data.indexed_parameter.index	×
376	<< "} <== " << SmartWeb::SensorData::ToDouble(value);	×
377	};	×
378
379	auto get_output_value = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {	×
380	SmartWeb::TMappingPoint mapping_point{};	×
381	memcpy(mapping_point.rawID, data, 2);	×
382
383	if (mapping_point.hostID != program_id) {	×
384	throw TFrameError("hostID of mapping point does not match with driver program_id");	×
385	}
386
387	auto channel_id = mapping_point.channelID;	×
388	if (channel_id >= CONTROLLER_OUTPUT_MAX) {	×
389	throw TFrameError("channel_id of mapping point is out of bounds: " + to_string(channel_id));	×
390	}
391
392	auto& channel = DriverState.OutputMapping[channel_id];	×
393
394	if (channel.is_initialized()) {	×
395	channel.schedule_to_send(mapping_point);	×
396	InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] scheduled output " << (int)channel_id;	×
397	} else {
398	WarnMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] unmapped output " << (int)channel_id;	×
399	}
400	};	×
401
402	auto send_scheduled_outputs = [&] {	×
403	SmartWeb::TCanHeader header;
404	header.rec.program_type = SmartWeb::PT_CONTROLLER;	×
405	header.rec.program_id = program_id;	×
406	header.rec.function_id = SmartWeb::Controller::Function::GET_OUTPUT_VALUE;	×
407	header.rec.message_format = SmartWeb::MF_FORMAT_0;	×
408	header.rec.message_type = SmartWeb::MT_MSG_RESPONSE;	×
409
410	TFrame frame{0};	×
411
412	frame.can_dlc = 4;	×
413
414	for (uint8_t channel_id = 0; channel_id < CONTROLLER_OUTPUT_MAX; ++channel_id) {	×
415	auto& channel = DriverState.OutputMapping[channel_id];	×
416
417	if (channel.SendEndTimePoint < now()) {	×
418	continue; // too late	×
419	}
420
421	if (channel.device.empty() \|\| channel.control.empty()) {	×
422	continue; // weird	×
423	}
424
425	auto lastUpdate =
426	DriverState.MqttChannelsTiming.at(TMqttChannel::to_string(channel.device, channel.control))	×
427	.get_last_update_timepoint();	×
428	if (lastUpdate <= channel.LastSendTimePoint) { // no channel updates	×
429	if (channel.SendTimePoint > now()) {	×
430	continue; // too soon	×
431	}
432	}
433
434	auto value = read_mqtt_value(channel.device, channel.control);	×
435
436	frame.can_id = header.raw \| CAN_EFF_FLAG;	×
437	memcpy(frame.data, &channel.mapping_point.raw, sizeof channel.mapping_point.raw);	×
438	frame.data[2] = 0xFF & value >> 8;	×
439	frame.data[3] = 0xFF & value;	×
440
441	send_frame(frame, "send output");	×
442
443	DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] output {channel_id: " << (int)channel_id	×
444	<< "} <== " << SmartWeb::SensorData::ToDouble(value);	×
445
446	channel.postpone_send();	×
447	}
448	};	×
449
450	auto get_controller_type = [&](const SmartWeb::TCanHeader& header) {	×
451	auto response = get_response_frame(header);	×
452	response.can_dlc = 1;	×
453	response.data[0] = CONTROLLER_TYPE;	×
454	send_frame(response, "send controller type");	×
455	};	×
456
457	auto handle_request = [&](const SmartWeb::TCanHeader& header, const TFrameData& data) {	×
458	switch (header.rec.program_type) {	×
459	case SmartWeb::PT_CONTROLLER:	×
460	switch (header.rec.function_id) {	×
461	case SmartWeb::Controller::Function::GET_CHANNEL_NUMBER:	×
462	return get_channel_number(header);	×
463	case SmartWeb::Controller::Function::GET_CONTROLLER_TYPE:	×
464	return get_controller_type(header);	×
465	case SmartWeb::Controller::Function::GET_OUTPUT_VALUE:	×
466	return get_output_value(header, data);	×
467	case SmartWeb::Controller::Function::I_AM_HERE:	×
468	return i_am_here();	×
469	default:	×
470	throw TUnsupportedError("function id " + to_string((int)header.rec.function_id) +	×
471	" is unsupported");	×
472	}
473	case SmartWeb::PT_REMOTE_CONTROL:	×
474	switch (header.rec.function_id) {	×
475	case SmartWeb::RemoteControl::Function::GET_PARAMETER_VALUE:	×
476	return get_parameter_value(header, data);	×
477	default:	×
478	throw TUnsupportedError("function id " + to_string((int)header.rec.function_id) +	×
479	" is unsupported");	×
480	}
481	default:	×
482	throw TUnsupportedError("program_type " + to_string((int)header.rec.program_type) + " is unsupported");	×
483	}
484	};	×
485
486	TFrame frame{0};	×
487
488	SendIAmHereTime = now();	×
489
490	SmartWeb::TCanHeader header{0};	×
491
492	// can0 0015AC0B [8] 00 00 00 00 00 00 00 00 (CONTROLLER: JOURNAL (Get controller journal notes))
493	// can0 000AAC0B [0] (CONTROLLER: GET_CHANNEL_NUMBER (Узнать количество
494	// входов/выходов)) can0 0003AC0B [0] (CONTROLLER: GET_ACTIVE_PROGRAMS_LIST (Узнать
495	// список активных программ)) can0 0001AC16 [4] 0B 01 00 00 (REMOTE_CONTROL: GET_PARAMETER_VALUE)
496	// can0 0001AC16 [4] 0B 1E 00 00 (REMOTE_CONTROL: GET_PARAMETER_VALUE)
497	// can0 0001AC16 [4] 0B 02 00 00 (REMOTE_CONTROL: GET_PARAMETER_VALUE)
498	// can0 0008AC0B [0] (CONTROLLER: GET_CONTROLLER_TYPE (Узнать тип контроллера))
499	// can0 0018AC0B [1] 00 (CONTROLLER: GET_RELAY_MAPPING (Get controller output binding))
500	// can0 0001AC16 [4] 0B 1C 00 00 (REMOTE_CONTROL: GET_PARAMETER_VALUE)
501	// can0 0001AC16 [4] 0B 1D 00 00 (REMOTE_CONTROL: GET_PARAMETER_VALUE)
502	// can0 0001AC16 [6] 0B 05 00 09 8A 2F (REMOTE_CONTROL: GET_PARAMETER_VALUE)
503
504	while (Enabled.load()) {	×
505	memset(&frame, 0, sizeof(TFrame));	×
506
507	bool frame_for_me = SelectTimeout(frame);	×
508
509	if (!frame_for_me) {	×
510	if (Status != DS_IDLE) {	×
511	if (ResetConnectionTime <= now()) {	×
512	Status = DS_IDLE;	×
513	InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] CONNECTION LOST: TIMEOUT. IDLE";	×
514	}
515	}
516	}
517
518	if (frame_for_me) {	×
519	print_frame(DebugMqttToSw, frame, "[" + std::to_string(DriverState.ProgramId) + "] got frame");	×
520	}
521
522	try {
523	switch (Status) {	×
524	case DS_IDLE:	×
525	if (frame_for_me) {	×
526	Status = DS_RUNNING;	×
527	InfoMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] CONNECTION ESTABILISHED. RUNNING";	×
528	} else {
529	send_scheduled_i_am_here();	×
530	break;	×
531	}
532
533	case DS_RUNNING:
534	if (frame_for_me) {	×
535	postpone_connection_reset();	×
536	header.raw = frame.can_id;	×
537	handle_request(header, frame.data);	×
538	}
539
540	send_scheduled_outputs();	×
541	send_scheduled_i_am_here();	×
542
543	break;	×
544	}
545
546	} catch (const TUnsupportedError& e) {	×
547	DebugMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();	×
548	} catch (const TDriverError& e) {	×
549	WarnMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();	×
550	} catch (const std::exception& e) {	×
551	ErrorMqttToSw.Log() << "[" << (int)DriverState.ProgramId << "] " << e.what();	×
552	exit(1);	×
553	}	×
554	}
555	Driver->RemoveEventHandler(onHandler);	×
556	}	×

wirenboard / wb-mqtt-smartweb / 2

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous