wirenboard / wb-mqtt-serial / 967

Committed 05 Oct 2026 09:41AM UTC coverage: 85.672% (+2.4%) from 83.268%

Build # 967

Build Type

push

github

Committed by

web-flow

Commit Message

Add config/Save RPC (#1273)

The RPC validates a configuration, writes it to the configuration file and
restarts the polling in-process. Port clients are stopped first, requests
arriving meanwhile get config-busy and queued ones get request-cancelled.

Coverage Stats

12445 of 14263 branches covered (87.25%)

1432 of 1623 new or added lines in 41 files covered. (88.23%)

8 existing lines in 4 files now uncovered.

21065 of 24588 relevant lines covered (85.67%)

1528.89 hits per line

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

70.12

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

wirenboard / wb-mqtt-serial / 967

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