wirenboard / wb-mqtt-serial / 964

Committed 28 Sep 2026 06:54AM UTC coverage: 83.234% (+6.4%) from 76.817%

Build # 964

Build Type

push

github

Committed by

web-flow

Commit Message

Add disable_fast_modbus option to ports and Modbus devices (#1270)

Coverage Stats

11435 of 13496 branches covered (84.73%)

150 of 152 new or added lines in 11 files covered. (98.68%)

1282 existing lines in 91 files now uncovered.

19376 of 23279 relevant lines covered (83.23%)

1598.47 hits per line

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

43.71

/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

void TRPCDeviceParametersCache::RegisterCallbacks(PHandlerConfig handlerConfig)
{
    for (const auto& portConfig: handlerConfig->PortConfigs) {
        for (const auto& device: portConfig->Devices) {
            std::string id = GetId(*portConfig->Port, device->Device->DeviceConfig()->SlaveId);
            device->Device->AddOnConnectionStateChangedCallback([this, id](PSerialDevice device) {
                if (device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED) {
                    Remove(id);
                }
            });
        }
    }
}

std::string TRPCDeviceParametersCache::GetId(const TPort& port, const std::string& slaveId) const
{
    return port.GetDescription(false) + ":" + slaveId;
}

void TRPCDeviceParametersCache::Add(const std::string& id, const Json::Value& value)
{
    std::unique_lock lock(Mutex);
    DeviceParameters[id] = value;
}

void TRPCDeviceParametersCache::Remove(const std::string& id)
{
    std::unique_lock lock(Mutex);
    DeviceParameters.erase(id);
}

bool TRPCDeviceParametersCache::Contains(const std::string& id) const
{
    std::unique_lock lock(Mutex);
    return DeviceParameters.find(id) != DeviceParameters.end();
}

const Json::Value& TRPCDeviceParametersCache::Get(const std::string& id, const Json::Value& defaultValue) const
{
    std::unique_lock lock(Mutex);
    auto it = DeviceParameters.find(id);
    return it != DeviceParameters.end() ? it->second : defaultValue;
};

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

wirenboard / wb-mqtt-serial / 964

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