wirenboard / wb-mqtt-serial / 2

Committed 20 Oct 2026 05:26AM UTC coverage: 86.772% (+13.9%) from 72.836%

Build # 2

Build Type

Pull #1283

github

Specific Base e3f11f

Committed by

KraPete

Commit Message

Remove comment from service unit

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Pull Request Pull Request #1283: Order service after mosquitto and restart it without limit

Coverage Stats

31090 of 80727 branches covered (38.51%)

22401 of 25816 relevant lines covered (86.77%)

2589.67 hits per line

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

80.0

/src/rpc/rpc_device_handler.cpp

#include "rpc_device_handler.h"
#include "rpc_device_load_config_task.h"
#include "rpc_device_load_task.h"
#include "rpc_device_probe_task.h"
#include "rpc_device_set_task.h"
#include "rpc_helpers.h"

#define LOG(logger) ::logger.Log() << "[RPC] "

namespace
{
    bool IsSupportedByFw(const Json::Value& item, const std::string& fwVersion)
    {
        if (fwVersion.empty()) {
            return true;
        }
        std::string fw = item["fw"].asString();
        return fw.empty() || util::CompareVersionStrings(fw, fwVersion) <= 0;
    }

    TRPCRegister CreateRPCRegister(const TDeviceProtocolParams& protocolParams,
                                   PSerialDevice device,
                                   const std::string& id,
                                   const Json::Value& item,
                                   bool checkUnsupported)
    {
        auto config = LoadRegisterConfig(item,
                                         *protocolParams.protocol->GetRegTypes(),
                                         std::string(),
                                         *protocolParams.factory,
                                         protocolParams.factory->GetRegisterAddressFactory().GetBaseRegisterAddress(),
                                         0);
        TRPCRegister reg = {id,
                            item["condition"].asString(),
                            std::make_shared<TRegister>(device, config.RegisterConfig),
                            checkUnsupported};
        reg.Register->SetAvailable(TRegisterAvailability::AVAILABLE);

        // this code checks enums and ranges only for 16-bit register unsupported value 0xFFFE
        // it must be modified to check larger registers like 24, 32 or 64-bits
        if (reg.CheckUnsupported) {
            int unsupportedValue =
                config.RegisterConfig->Format == S16 ? static_cast<int16_t>(0xFFFE) : static_cast<uint16_t>(0xFFFE);
            if (item.isMember("enum")) {
                for (const auto& value: item["enum"]) {
                    try {
                        if (std::stoi(value.asString(), 0, 0) == unsupportedValue) {
                            reg.CheckUnsupported = false;
                            break;
                        }
                    } catch (const std::logic_error&) {
                    }
                }
            } else {
                if (item["min"].asInt() <= unsupportedValue && item["max"].asInt() >= unsupportedValue) {
                    reg.CheckUnsupported = false;
                }
            }
        }
        return reg;
    }
} // namespace

#ifndef __EMSCRIPTEN__
TRPCDeviceHelper::TRPCDeviceHelper(const Json::Value& request,
                                   const TSerialDeviceFactory& deviceFactory,
                                   PTemplateMap templates,
                                   PSerialDevice deviceFromConfig)
{
    if (deviceFromConfig == nullptr) {
        DeviceTemplate = templates->GetTemplate(request["device_type"].asString());
        auto protocolName = DeviceTemplate->GetProtocol();
        if (protocolName == "modbus" && request["modbus_mode"].asString() == "TCP") {
            protocolName += "-tcp";
        }
        ProtocolParams = deviceFactory.GetProtocolParams(protocolName);
        auto config = std::make_shared<TDeviceConfig>("RPC Device", request["slave_id"].asString(), protocolName);
        bool isWbDevice = !DeviceTemplate->GetHardware().empty() ||
                          DeviceTemplate->GetTemplate()["enable_wb_continuous_read"].asBool();
        LoadCommonDeviceParameters(*config, DeviceTemplate->GetTemplate(), isWbDevice);
        Device = ProtocolParams.factory->CreateDevice(DeviceTemplate->GetTemplate(), config, ProtocolParams.protocol);
        Device->SetWbDevice(isWbDevice);
    } else {
        Device = deviceFromConfig;
        DeviceTemplate = templates->GetTemplate(Device->DeviceConfig()->DeviceType);
        ProtocolParams = deviceFactory.GetProtocolParams(Device->Protocol()->GetName());
        DeviceFromConfig = true;
    }
    if (DeviceTemplate->WithSubdevices()) {
        throw TRPCException("Device \"" + DeviceTemplate->Type + "\" is not supported by this RPC",
                            TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }
}
#endif

TRPCDeviceRequest::TRPCDeviceRequest(const TDeviceProtocolParams& protocolParams,
                                     PSerialDevice device,
                                     PDeviceTemplate deviceTemplate,
                                     bool deviceFromConfig)
    : ProtocolParams(protocolParams),
      Device(device),
      DeviceTemplate(deviceTemplate),
      DeviceFromConfig(deviceFromConfig)
{
    Json::Value responseTimeout = DeviceTemplate->GetTemplate()["response_timeout_ms"];
    if (responseTimeout.isInt()) {
        ResponseTimeout = std::chrono::milliseconds(responseTimeout.asInt());
    }

    Json::Value frameTimeout = DeviceTemplate->GetTemplate()["frame_timeout_ms"];
    if (frameTimeout.isInt()) {
        FrameTimeout = std::chrono::milliseconds(frameTimeout.asInt());
    }
}

void TRPCDeviceRequest::ParseSettings(const Json::Value& request,
                                      WBMQTT::TMqttRpcServer::TResultCallback onResult,
                                      WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    SerialPortSettings = ParseRPCSerialPortSettings(request);
    WBMQTT::JSON::Get(request, "response_timeout", ResponseTimeout);
    WBMQTT::JSON::Get(request, "frame_timeout", FrameTimeout);
    WBMQTT::JSON::Get(request, "total_timeout", TotalTimeout);
    OnResult = onResult;
    OnError = onError;
}

#ifndef __EMSCRIPTEN__
TRPCDeviceHandler::TRPCDeviceHandler(const std::string& configFileName,
                                     const std::string& requestDeviceLoadConfigSchemaFilePath,
                                     const std::string& requestDeviceLoadSchemaFilePath,
                                     const std::string& requestDeviceSetSchemaFilePath,
                                     const std::string& requestDeviceProbeSchemaFilePath,
                                     const std::string& requestDeviceSetPollSchemaFilePath,
                                     const TSerialDeviceFactory& deviceFactory,
                                     PTemplateMap templates,
                                     TSerialClientTaskRunner& serialClientTaskRunner,
                                     WBMQTT::PMqttRpcServer rpcServer)
    : ConfigFileName(configFileName),
      DeviceFactory(deviceFactory),
      RequestDeviceLoadConfigSchema(LoadRPCRequestSchema(requestDeviceLoadConfigSchemaFilePath, "device/LoadConfig")),
      RequestDeviceLoadSchema(LoadRPCRequestSchema(requestDeviceLoadSchemaFilePath, "device/Load")),
      RequestDeviceSetSchema(LoadRPCRequestSchema(requestDeviceSetSchemaFilePath, "device/Set")),
      RequestDeviceProbeSchema(LoadRPCRequestSchema(requestDeviceProbeSchemaFilePath, "device/Probe")),
      RequestDeviceSetPollSchema(LoadRPCRequestSchema(requestDeviceSetPollSchemaFilePath, "device/SetPoll")),
      Templates(templates),
      SerialClientTaskRunner(serialClientTaskRunner)
{
    rpcServer->RegisterAsyncMethod("device",
                                   "LoadConfig",
                                   std::bind(&TRPCDeviceHandler::LoadConfig,
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));
    rpcServer->RegisterAsyncMethod("device",
                                   "Load",
                                   std::bind(&TRPCDeviceHandler::Load, //
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));
    rpcServer->RegisterAsyncMethod("device",
                                   "Set",
                                   std::bind(&TRPCDeviceHandler::Set, //
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));
    rpcServer->RegisterAsyncMethod("device",
                                   "Probe",
                                   std::bind(&TRPCDeviceHandler::Probe,
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));

    rpcServer->RegisterMethod("device", "SetPoll", std::bind(&TRPCDeviceHandler::SetPoll, this, std::placeholders::_1));
}

void TRPCDeviceHandler::LoadConfig(const Json::Value& request,
                                   WBMQTT::TMqttRpcServer::TResultCallback onResult,
                                   WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceLoadConfigSchema);
    try {
        SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache cache) {
            auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);
            // Without "hw" the model is not checked, the settings may be read from another device
            if (helper.DeviceTemplate->GetHardware().empty()) {
                throw TRPCException("Device \"" + helper.DeviceTemplate->Type +
                                        "\" without \"hw\" in the template is not supported by this RPC",
                                    TRPCResultCode::RPC_WRONG_PARAM_VALUE);
            }
            auto rpcRequest = ParseRPCDeviceLoadConfigRequest(request,
                                                              helper.ProtocolParams,
                                                              helper.Device,
                                                              helper.DeviceTemplate,
                                                              helper.DeviceFromConfig,
                                                              ConfigFileName,
                                                              cache,
                                                              onResult,
                                                              onError);
            return std::make_shared<TRPCDeviceLoadConfigSerialClientTask>(rpcRequest);
        });
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

void TRPCDeviceHandler::Load(const Json::Value& request,
                             WBMQTT::TMqttRpcServer::TResultCallback onResult,
                             WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceLoadSchema);
    try {
        SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache) {
            auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);
            auto rpcRequest = ParseRPCDeviceLoadRequest(request,
                                                        helper.ProtocolParams,
                                                        helper.Device,
                                                        helper.DeviceTemplate,
                                                        helper.DeviceFromConfig,
                                                        onResult,
                                                        onError);
            return std::make_shared<TRPCDeviceLoadSerialClientTask>(rpcRequest);
        });
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

void TRPCDeviceHandler::Set(const Json::Value& request,
                            WBMQTT::TMqttRpcServer::TResultCallback onResult,
                            WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceSetSchema);
    try {
        SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache) {
            auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);
            auto rpcRequest = ParseRPCDeviceSetRequest(request,
                                                       helper.ProtocolParams,
                                                       helper.Device,
                                                       helper.DeviceTemplate,
                                                       helper.DeviceFromConfig,
                                                       onResult,
                                                       onError);
            return std::make_shared<TRPCDeviceSetSerialClientTask>(rpcRequest);
        });
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

void TRPCDeviceHandler::Probe(const Json::Value& request,
                              WBMQTT::TMqttRpcServer::TResultCallback onResult,
                              WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceProbeSchema);
    try {
        SerialClientTaskRunner.RunTask(request, [&](PSerialDevice, PDeviceParametersCache) {
            return std::make_shared<TRPCDeviceProbeSerialClientTask>(request, onResult, onError);
        });
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

Json::Value TRPCDeviceHandler::SetPoll(const Json::Value& request)
{
    ValidateRPCRequest(request, RequestDeviceSetPollSchema);
    SerialClientTaskRunner.SetPoll(request, request["poll"].asBool());
    return Json::Value(Json::objectValue);
}
#endif

void PrepareSession(TPort& port, PSerialDevice device, int maxRetries)
{
    for (int i = 0; i <= maxRetries; i++) {
        try {
            device->Prepare(port, TDevicePrepareMode::WITHOUT_SETUP);
            break;
        } catch (const TSerialDeviceException& e) {
            if (i == maxRetries) {
                auto error = std::string("Failed to prepare session: ") + e.what();
                LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": " << error;
                throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);
            }
        }
    }
}

void EndSession(TPort& port, PSerialDevice device)
{
    try {
        device->EndSession(port);
    } catch (const TSerialDeviceException& e) {
        LOG(Warn) << port.GetDescription() << " " << device->ToString() << " unable to end session: " << e.what();
    }
}

TRPCRegisterList CreateChannelsRegisterList(const TDeviceProtocolParams& protocolParams,
                                            PSerialDevice device,
                                            const Json::Value& channels)
{
    // nullptr checks are needed for tests
    auto fwVersion = device ? device->GetWbFwVersion() : std::string();
    auto checkUnsupported = device && device->IsWbDevice();

    TRPCRegisterList registerList;
    for (const auto& item: channels) {
        if (item["address"].isNull() || !IsSupportedByFw(item, fwVersion)) {
            continue;
        }
        registerList.push_back(
            CreateRPCRegister(protocolParams, device, item["id"].asString(), item, checkUnsupported));
    }
    return registerList;
}

TRPCRegisterList CreateParametersRegisterList(const TDeviceProtocolParams& protocolParams,
                                              PSerialDevice device,
                                              const Json::Value& parameters,
                                              const Json::Value& knownValues,
                                              const std::set<std::string>& ids)
{
    // nullptr checks are needed for tests
    auto fwVersion = device ? device->GetWbFwVersion() : std::string();
    auto checkUnsupported = device && device->IsWbDevice();

    // The fw variants of a parameter (declarations with the same id and condition) define the same register,
    // the first variant supported by the device firmware gives the register of the group.
    // The enum matters only for the unsupported value marker check and no firmware adds 0xFFFE to an enum
    std::set<std::pair<std::string, std::string>> createdRegisters;
    TRPCRegisterList registerList;
    for (auto it = parameters.begin(); it != parameters.end(); ++it) {
        const auto& item = *it;
        auto id = parameters.isObject() ? it.key().asString() : item["id"].asString();
        if (item["address"].isNull() || !knownValues[id].isNull()) {
            continue;
        }
        if (!ids.empty() && !ids.count(id)) {
            continue;
        }
        if (!IsSupportedByFw(item, fwVersion)) {
            continue;
        }
        if (createdRegisters.emplace(id, item["condition"].asString()).second) {
            registerList.push_back(CreateRPCRegister(protocolParams, device, id, item, checkUnsupported));
        }
    }
    return registerList;
}

void ReadRegisterList(TPort& port, PSerialDevice device, TRPCRegisterList& registerList, int maxRetries)
{
    if (registerList.size() == 0) {
        return;
    }

    TRegisterComparePredicate compare;
    std::sort(registerList.begin(), registerList.end(), [compare](TRPCRegister& a, TRPCRegister& b) {
        return compare(b.Register, a.Register);
    });

    size_t index = 0;
    std::string error;
    while (index < registerList.size() && error.empty()) {
        auto first = registerList[index].Register;
        auto range = device->CreateRegisterRange();
        while (index < registerList.size() &&
               range->Add(port, registerList[index].Register, std::chrono::milliseconds::max()))
        {
            ++index;
        }
        for (int i = 0; i <= maxRetries; ++i) {
            try {
                device->ReadRegisterRange(port, range, true);
                break;
            } catch (const TSerialDevicePermanentRegisterException& e) {
                LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": "
                          << "Failed to read " << std::to_string(range->RegisterList().size())
                          << " registers starting from <" << first->GetConfig()->ToString() + ">: " + e.what();
                auto modbusDevice = dynamic_cast<TModbusDevice*>(device.get());
                if (modbusDevice != nullptr &&
                    modbusDevice->GetContinuousReadStatus() == TContinuousReadStatus::DISABLED)
                {
                    for (const auto& reg: range->RegisterList()) {
                        reg->SetSupported(false);
                    }
                }
                break;
            } catch (const TSerialDeviceException& e) {
                if (i == maxRetries) {
                    error = "Failed to read " + std::to_string(range->RegisterList().size()) +
                            " registers starting from <" + first->GetConfig()->ToString() + ">: " + e.what();
                }
            }
        }
    }

    if (!error.empty()) {
        LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": " << error;
        throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }
}

Json::Value RawValueToJSON(const TRegisterConfig& reg, TRegisterValue val)
{
    auto str = ConvertFromRawValue(reg, val);
    try {
        if (str.find('.') == std::string::npos) {
            if (str.at(0) == '-') {
                return static_cast<Json::Int64>(std::stoll(str.c_str(), 0));
            }
            auto value = std::stoull(str.c_str(), 0);
            if (value <= INT64_MAX) {
                // cast value to signed integer to match default Json::Value type for integers
                return static_cast<Json::Int64>(value);
            }
            return static_cast<Json::UInt64>(value);
        }
        return std::stod(str.c_str(), 0);
    } catch (const std::invalid_argument&) {
        return str;
    }
}

bool RegisterGotValue(const TRPCRegister& item)
{
    return item.Register->IsSupported() && !item.Register->GetErrorState().test(TRegister::TError::ReadError) &&
           item.Register->GetValue().GetType() != TRegisterValue::ValueType::Undefined;
}

void MergeRegisterListValues(const TRPCRegisterList& registerList, Json::Value& values)
{
    for (const auto& item: registerList) {
        if (RegisterGotValue(item)) {
            values[item.Id] = RawValueToJSON(*item.Register->GetConfig(), item.Register->GetValue());
        }
    }
}

1	#include "rpc_device_handler.h"
2	#include "rpc_device_load_config_task.h"
3	#include "rpc_device_load_task.h"
4	#include "rpc_device_probe_task.h"
5	#include "rpc_device_set_task.h"
6	#include "rpc_helpers.h"
7
8	#define LOG(logger) ::logger.Log() << "[RPC] "
9
10	namespace
11	{
12	bool IsSupportedByFw(const Json::Value& item, const std::string& fwVersion)	199 ✔
13	{
14	if (fwVersion.empty()) {	199 ✔
15	return true;	158 ✔
16	}
17	std::string fw = item["fw"].asString();	41 !
18	return fw.empty() \|\| util::CompareVersionStrings(fw, fwVersion) <= 0;	41 !
19	}	41 ✔
20
21	TRPCRegister CreateRPCRegister(const TDeviceProtocolParams& protocolParams,	178 ✔
22	PSerialDevice device,
23	const std::string& id,
24	const Json::Value& item,
25	bool checkUnsupported)
26	{
27	auto config = LoadRegisterConfig(item,
28	*protocolParams.protocol->GetRegTypes(),	178 !
29	std::string(),	356 ✔
30	*protocolParams.factory,	178 ✔
31	protocolParams.factory->GetRegisterAddressFactory().GetBaseRegisterAddress(),	178 !
32	0);	178 !
33	TRPCRegister reg = {id,
34	item["condition"].asString(),	356 ✔
35	std::make_shared<TRegister>(device, config.RegisterConfig),
36	checkUnsupported};	178 !
37	reg.Register->SetAvailable(TRegisterAvailability::AVAILABLE);	178 !
38
39	// this code checks enums and ranges only for 16-bit register unsupported value 0xFFFE
40	// it must be modified to check larger registers like 24, 32 or 64-bits
41	if (reg.CheckUnsupported) {	178 ✔
42	int unsupportedValue =
43	config.RegisterConfig->Format == S16 ? static_cast<int16_t>(0xFFFE) : static_cast<uint16_t>(0xFFFE);	85 ✔
44	if (item.isMember("enum")) {	85 !
45	for (const auto& value: item["enum"]) {	45 !
46	try {
47	if (std::stoi(value.asString(), 0, 0) == unsupportedValue) {	32 !
48	reg.CheckUnsupported = false;	3 ✔
49	break;	3 ✔
50	}
51	} catch (const std::logic_error&) {	×
52	}	×
53	}
54	} else {
55	if (item["min"].asInt() <= unsupportedValue && item["max"].asInt() >= unsupportedValue) {	69 !
56	reg.CheckUnsupported = false;	1 ✔
57	}
58	}
59	}
60	return reg;	178 ✔
61	}	178 ✔
62	} // namespace
63
64	#ifndef __EMSCRIPTEN__
65	TRPCDeviceHelper::TRPCDeviceHelper(const Json::Value& request,	3 ✔
66	const TSerialDeviceFactory& deviceFactory,
67	PTemplateMap templates,
68	PSerialDevice deviceFromConfig)	3 ✔
69	{
70	if (deviceFromConfig == nullptr) {	3 !
71	DeviceTemplate = templates->GetTemplate(request["device_type"].asString());	×
72	auto protocolName = DeviceTemplate->GetProtocol();	×
73	if (protocolName == "modbus" && request["modbus_mode"].asString() == "TCP") {	×
74	protocolName += "-tcp";	×
75	}
76	ProtocolParams = deviceFactory.GetProtocolParams(protocolName);	×
77	auto config = std::make_shared<TDeviceConfig>("RPC Device", request["slave_id"].asString(), protocolName);	×
78	bool isWbDevice = !DeviceTemplate->GetHardware().empty() \|\|	×
79	DeviceTemplate->GetTemplate()["enable_wb_continuous_read"].asBool();	×
80	LoadCommonDeviceParameters(*config, DeviceTemplate->GetTemplate(), isWbDevice);	×
81	Device = ProtocolParams.factory->CreateDevice(DeviceTemplate->GetTemplate(), config, ProtocolParams.protocol);	×
82	Device->SetWbDevice(isWbDevice);	×
83	} else {	×
84	Device = deviceFromConfig;	3 ✔
85	DeviceTemplate = templates->GetTemplate(Device->DeviceConfig()->DeviceType);	3 !
86	ProtocolParams = deviceFactory.GetProtocolParams(Device->Protocol()->GetName());	3 !
87	DeviceFromConfig = true;	3 ✔
88	}
89	if (DeviceTemplate->WithSubdevices()) {	3 !
90	throw TRPCException("Device \"" + DeviceTemplate->Type + "\" is not supported by this RPC",	×
91	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
92	}
93	}	3 ✔
94	#endif
95
96	TRPCDeviceRequest::TRPCDeviceRequest(const TDeviceProtocolParams& protocolParams,	103 ✔
97	PSerialDevice device,
98	PDeviceTemplate deviceTemplate,
99	bool deviceFromConfig)	103 ✔
100	: ProtocolParams(protocolParams),	103 ✔
101	Device(device),	103 ✔
102	DeviceTemplate(deviceTemplate),	103 ✔
103	DeviceFromConfig(deviceFromConfig)	103 !
104	{
105	Json::Value responseTimeout = DeviceTemplate->GetTemplate()["response_timeout_ms"];	103 !
106	if (responseTimeout.isInt()) {	103 !
107	ResponseTimeout = std::chrono::milliseconds(responseTimeout.asInt());	×
108	}
109
110	Json::Value frameTimeout = DeviceTemplate->GetTemplate()["frame_timeout_ms"];	103 !
111	if (frameTimeout.isInt()) {	103 !
112	FrameTimeout = std::chrono::milliseconds(frameTimeout.asInt());	×
113	}
114	}	103 ✔
115
116	void TRPCDeviceRequest::ParseSettings(const Json::Value& request,	2 ✔
117	WBMQTT::TMqttRpcServer::TResultCallback onResult,
118	WBMQTT::TMqttRpcServer::TErrorCallback onError)
119	{
120	SerialPortSettings = ParseRPCSerialPortSettings(request);	2 !
121	WBMQTT::JSON::Get(request, "response_timeout", ResponseTimeout);	4 !
122	WBMQTT::JSON::Get(request, "frame_timeout", FrameTimeout);	4 !
123	WBMQTT::JSON::Get(request, "total_timeout", TotalTimeout);	2 !
124	OnResult = onResult;	2 ✔
125	OnError = onError;	2 ✔
126	}	2 ✔
127
128	#ifndef __EMSCRIPTEN__
129	TRPCDeviceHandler::TRPCDeviceHandler(const std::string& configFileName,	53 ✔
130	const std::string& requestDeviceLoadConfigSchemaFilePath,
131	const std::string& requestDeviceLoadSchemaFilePath,
132	const std::string& requestDeviceSetSchemaFilePath,
133	const std::string& requestDeviceProbeSchemaFilePath,
134	const std::string& requestDeviceSetPollSchemaFilePath,
135	const TSerialDeviceFactory& deviceFactory,
136	PTemplateMap templates,
137	TSerialClientTaskRunner& serialClientTaskRunner,
138	WBMQTT::PMqttRpcServer rpcServer)	53 ✔
139	: ConfigFileName(configFileName),	53 ✔
140	DeviceFactory(deviceFactory),	53 ✔
141	RequestDeviceLoadConfigSchema(LoadRPCRequestSchema(requestDeviceLoadConfigSchemaFilePath, "device/LoadConfig")),	106 !
142	RequestDeviceLoadSchema(LoadRPCRequestSchema(requestDeviceLoadSchemaFilePath, "device/Load")),	106 !
143	RequestDeviceSetSchema(LoadRPCRequestSchema(requestDeviceSetSchemaFilePath, "device/Set")),	106 !
144	RequestDeviceProbeSchema(LoadRPCRequestSchema(requestDeviceProbeSchemaFilePath, "device/Probe")),	106 !
145	RequestDeviceSetPollSchema(LoadRPCRequestSchema(requestDeviceSetPollSchemaFilePath, "device/SetPoll")),	53 !
146	Templates(templates),	53 ✔
147	SerialClientTaskRunner(serialClientTaskRunner)	53 ✔
148	{
149	rpcServer->RegisterAsyncMethod("device",	265 !
150	"LoadConfig",
151	std::bind(&TRPCDeviceHandler::LoadConfig,	×
152	this,	106 !
153	std::placeholders::_1,
154	std::placeholders::_2,
155	std::placeholders::_3));
156	rpcServer->RegisterAsyncMethod("device",	265 !
157	"Load",
158	std::bind(&TRPCDeviceHandler::Load, //	×
159	this,	106 !
160	std::placeholders::_1,
161	std::placeholders::_2,
162	std::placeholders::_3));
163	rpcServer->RegisterAsyncMethod("device",	265 !
164	"Set",
165	std::bind(&TRPCDeviceHandler::Set, //	×
166	this,	106 !
167	std::placeholders::_1,
168	std::placeholders::_2,
169	std::placeholders::_3));
170	rpcServer->RegisterAsyncMethod("device",	265 !
171	"Probe",
172	std::bind(&TRPCDeviceHandler::Probe,	×
173	this,	106 !
174	std::placeholders::_1,
175	std::placeholders::_2,
176	std::placeholders::_3));
177
178	rpcServer->RegisterMethod("device", "SetPoll", std::bind(&TRPCDeviceHandler::SetPoll, this, std::placeholders::_1));	212 !
179	}	53 ✔
180
181	void TRPCDeviceHandler::LoadConfig(const Json::Value& request,	5 ✔
182	WBMQTT::TMqttRpcServer::TResultCallback onResult,
183	WBMQTT::TMqttRpcServer::TErrorCallback onError)
184	{
185	ValidateRPCRequest(request, RequestDeviceLoadConfigSchema);	5 ✔
186	try {
187	SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache cache) {	8 !
188	auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);	3 !
189	// Without "hw" the model is not checked, the settings may be read from another device
190	if (helper.DeviceTemplate->GetHardware().empty()) {	3 !
191	throw TRPCException("Device \"" + helper.DeviceTemplate->Type +	2 !
192	"\" without \"hw\" in the template is not supported by this RPC",	1 !
193	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	3 !
194	}
195	auto rpcRequest = ParseRPCDeviceLoadConfigRequest(request,	6 !
196	helper.ProtocolParams,
197	helper.Device,
198	helper.DeviceTemplate,
199	helper.DeviceFromConfig,
200	ConfigFileName,
201	cache,
202	onResult,
203	onError);	6 !
204	return std::make_shared<TRPCDeviceLoadConfigSerialClientTask>(rpcRequest);	4 !
205	});	3 ✔
206	} catch (const TRPCException& e) {	3 !
207	ProcessException(e, onError);	3 !
208	}	3 ✔
209	}	5 ✔
210
211	void TRPCDeviceHandler::Load(const Json::Value& request,	1 ✔
212	WBMQTT::TMqttRpcServer::TResultCallback onResult,
213	WBMQTT::TMqttRpcServer::TErrorCallback onError)
214	{
215	ValidateRPCRequest(request, RequestDeviceLoadSchema);	1 ✔
216	try {
217	SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache) {	2 !
218	auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);	×
219	auto rpcRequest = ParseRPCDeviceLoadRequest(request,	×
220	helper.ProtocolParams,
221	helper.Device,
222	helper.DeviceTemplate,
223	helper.DeviceFromConfig,
224	onResult,
225	onError);	×
226	return std::make_shared<TRPCDeviceLoadSerialClientTask>(rpcRequest);	×
227	});	×
228	} catch (const TRPCException& e) {	1 !
229	ProcessException(e, onError);	1 !
230	}	1 ✔
231	}	1 ✔
232
233	void TRPCDeviceHandler::Set(const Json::Value& request,	1 ✔
234	WBMQTT::TMqttRpcServer::TResultCallback onResult,
235	WBMQTT::TMqttRpcServer::TErrorCallback onError)
236	{
237	ValidateRPCRequest(request, RequestDeviceSetSchema);	1 ✔
238	try {
239	SerialClientTaskRunner.RunTask(request, [&](PSerialDevice deviceFromConfig, PDeviceParametersCache) {	2 !
240	auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, deviceFromConfig);	×
241	auto rpcRequest = ParseRPCDeviceSetRequest(request,	×
242	helper.ProtocolParams,
243	helper.Device,
244	helper.DeviceTemplate,
245	helper.DeviceFromConfig,
246	onResult,
247	onError);	×
248	return std::make_shared<TRPCDeviceSetSerialClientTask>(rpcRequest);	×
249	});	×
250	} catch (const TRPCException& e) {	1 !
251	ProcessException(e, onError);	1 !
252	}	1 ✔
253	}	1 ✔
254
255	void TRPCDeviceHandler::Probe(const Json::Value& request,	1 ✔
256	WBMQTT::TMqttRpcServer::TResultCallback onResult,
257	WBMQTT::TMqttRpcServer::TErrorCallback onError)
258	{
259	ValidateRPCRequest(request, RequestDeviceProbeSchema);	1 ✔
260	try {
261	SerialClientTaskRunner.RunTask(request, [&](PSerialDevice, PDeviceParametersCache) {	2 !
262	return std::make_shared<TRPCDeviceProbeSerialClientTask>(request, onResult, onError);	×
263	});
264	} catch (const TRPCException& e) {	1 !
265	ProcessException(e, onError);	1 !
266	}	1 ✔
267	}	1 ✔
268
269	Json::Value TRPCDeviceHandler::SetPoll(const Json::Value& request)	3 ✔
270	{
271	ValidateRPCRequest(request, RequestDeviceSetPollSchema);	3 ✔
272	SerialClientTaskRunner.SetPoll(request, request["poll"].asBool());	3 ✔
273	return Json::Value(Json::objectValue);	1 ✔
274	}
275	#endif
276
277	void PrepareSession(TPort& port, PSerialDevice device, int maxRetries)	52 ✔
278	{
279	for (int i = 0; i <= maxRetries; i++) {	52 !
280	try {
281	device->Prepare(port, TDevicePrepareMode::WITHOUT_SETUP);	52 !
282	break;	52 ✔
283	} catch (const TSerialDeviceException& e) {	×
284	if (i == maxRetries) {	×
285	auto error = std::string("Failed to prepare session: ") + e.what();	×
286	LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": " << error;	×
287	throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
288	}	×
289	}	×
290	}
291	}	52 ✔
292
293	void EndSession(TPort& port, PSerialDevice device)	52 ✔
294	{
295	try {
296	device->EndSession(port);	52 !
297	} catch (const TSerialDeviceException& e) {	×
298	LOG(Warn) << port.GetDescription() << " " << device->ToString() << " unable to end session: " << e.what();	×
299	}	×
300	}	52 ✔
301
302	TRPCRegisterList CreateChannelsRegisterList(const TDeviceProtocolParams& protocolParams,	9 ✔
303	PSerialDevice device,
304	const Json::Value& channels)
305	{
306	// nullptr checks are needed for tests
307	auto fwVersion = device ? device->GetWbFwVersion() : std::string();	9 !
308	auto checkUnsupported = device && device->IsWbDevice();	9 !
309
310	TRPCRegisterList registerList;	9 ✔
311	for (const auto& item: channels) {	67 !
312	if (item["address"].isNull() \|\| !IsSupportedByFw(item, fwVersion)) {	29 !
313	continue;	×
314	}
315	registerList.push_back(	29 !
316	CreateRPCRegister(protocolParams, device, item["id"].asString(), item, checkUnsupported));	58 !
317	}
318	return registerList;	9 ✔
319	}	9 ✔
320
321	TRPCRegisterList CreateParametersRegisterList(const TDeviceProtocolParams& protocolParams,	72 ✔
322	PSerialDevice device,
323	const Json::Value& parameters,
324	const Json::Value& knownValues,
325	const std::set<std::string>& ids)
326	{
327	// nullptr checks are needed for tests
328	auto fwVersion = device ? device->GetWbFwVersion() : std::string();	72 !
329	auto checkUnsupported = device && device->IsWbDevice();	72 !
330
331	// The fw variants of a parameter (declarations with the same id and condition) define the same register,
332	// the first variant supported by the device firmware gives the register of the group.
333	// The enum matters only for the unsupported value marker check and no firmware adds 0xFFFE to an enum
334	std::set<std::pair<std::string, std::string>> createdRegisters;	72 ✔
335	TRPCRegisterList registerList;	72 ✔
336	for (auto it = parameters.begin(); it != parameters.end(); ++it) {	548 !
337	const auto& item = *it;	238 !
338	auto id = parameters.isObject() ? it.key().asString() : item["id"].asString();	238 !
339	if (item["address"].isNull() \|\| !knownValues[id].isNull()) {	238 !
340	continue;	8 ✔
341	}
342	if (!ids.empty() && !ids.count(id)) {	230 !
343	continue;	60 ✔
344	}
345	if (!IsSupportedByFw(item, fwVersion)) {	170 !
346	continue;	11 ✔
347	}
348	if (createdRegisters.emplace(id, item["condition"].asString()).second) {	159 !
349	registerList.push_back(CreateRPCRegister(protocolParams, device, id, item, checkUnsupported));	149 !
350	}
351	}	238 ✔
352	return registerList;	72 ✔
353	}	72 ✔
354
355	void ReadRegisterList(TPort& port, PSerialDevice device, TRPCRegisterList& registerList, int maxRetries)	47 ✔
356	{
357	if (registerList.size() == 0) {	47 ✔
358	return;	6 ✔
359	}
360
361	TRegisterComparePredicate compare;
362	std::sort(registerList.begin(), registerList.end(), [compare](TRPCRegister& a, TRPCRegister& b) {	41 !
363	return compare(b.Register, a.Register);	88 ✔
364	});
365
366	size_t index = 0;	41 ✔
367	std::string error;	41 ✔
368	while (index < registerList.size() && error.empty()) {	123 !
369	auto first = registerList[index].Register;	82 ✔
370	auto range = device->CreateRegisterRange();	82 !
371	while (index < registerList.size() &&	293 ✔
372	range->Add(port, registerList[index].Register, std::chrono::milliseconds::max()))	293 !
373	{
374	++index;	85 ✔
375	}
376	for (int i = 0; i <= maxRetries; ++i) {	82 !
377	try {
378	device->ReadRegisterRange(port, range, true);	87 ✔
379	break;	77 ✔
380	} catch (const TSerialDevicePermanentRegisterException& e) {	5 !
381	LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": "	10 !
382	<< "Failed to read " << std::to_string(range->RegisterList().size())	10 !
383	<< " registers starting from <" << first->GetConfig()->ToString() + ">: " + e.what();	5 !
384	auto modbusDevice = dynamic_cast<TModbusDevice*>(device.get());	5 !
385	if (modbusDevice != nullptr &&	10 !
386	modbusDevice->GetContinuousReadStatus() == TContinuousReadStatus::DISABLED)	5 !
387	{
388	for (const auto& reg: range->RegisterList()) {	8 !
389	reg->SetSupported(false);	4 !
390	}
391	}
392	break;	5 ✔
393	} catch (const TSerialDeviceException& e) {	5 !
394	if (i == maxRetries) {	×
395	error = "Failed to read " + std::to_string(range->RegisterList().size()) +	×
396	" registers starting from <" + first->GetConfig()->ToString() + ">: " + e.what();	×
397	}
398	}	×
399	}
400	}	82 ✔
401
402	if (!error.empty()) {	41 !
403	LOG(Warn) << port.GetDescription() << " " << device->ToString() << ": " << error;	×
404	throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
405	}
406	}	41 ✔
407
408	Json::Value RawValueToJSON(const TRegisterConfig& reg, TRegisterValue val)	83 ✔
409	{
410	auto str = ConvertFromRawValue(reg, val);	83 !
411	try {
412	if (str.find('.') == std::string::npos) {	83 ✔
413	if (str.at(0) == '-') {	81 !
414	return static_cast<Json::Int64>(std::stoll(str.c_str(), 0));	3 !
415	}
416	auto value = std::stoull(str.c_str(), 0);	161 !
417	if (value <= INT64_MAX) {	79 !
418	// cast value to signed integer to match default Json::Value type for integers
419	return static_cast<Json::Int64>(value);	79 !
420	}
421	return static_cast<Json::UInt64>(value);	×
422	}
423	return std::stod(str.c_str(), 0);	6 !
424	} catch (const std::invalid_argument&) {	1 !
425	return str;	1 !
426	}	1 ✔
427	}	83 ✔
428
429	bool RegisterGotValue(const TRPCRegister& item)	13 ✔
430	{
431	return item.Register->IsSupported() && !item.Register->GetErrorState().test(TRegister::TError::ReadError) &&	20 !
432	item.Register->GetValue().GetType() != TRegisterValue::ValueType::Undefined;	20 !
433	}
434
435	void MergeRegisterListValues(const TRPCRegisterList& registerList, Json::Value& values)	6 ✔
436	{
437	for (const auto& item: registerList) {	17 ✔
438	if (RegisterGotValue(item)) {	11 !
439	values[item.Id] = RawValueToJSON(*item.Register->GetConfig(), item.Register->GetValue());	3 !
440	}
441	}
442	}	6 ✔

wirenboard / wb-mqtt-serial / 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