+ Completed minimal functionality to nbus revision 2

.idea/nbus_api.iml

@@ -4,7 +4,7 @@
     <content url="file://$MODULE_DIR$">
       <excludeFolder url="file://$MODULE_DIR$/im" />
     </content>
-    <orderEntry type="inheritedJdk" />
+    <orderEntry type="jdk" jdkName="Python 3.9 (nbus_api)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PackageRequirementsSettings">

main.py

@@ -1,9 +1,8 @@
 from nbus_hal.nbus_serial.serial_port import *
-from nbus_api.nbus_slave_module import NBusSlaveModule
-from nbus_api.nbus_slave_device import NBusSlaveDevice
+from nbus_api.nbus_module_slave import NBusSlaveModule
 from nbus_hal.nbus_serial.serial_config import *
-from nbus_types.nbus_data_fomat import NBusDataValue
-from nbus_types.nbus_parameter_type import NBusParameterID, NBusParameterValue
+from nbus_sensor_drivers.generic_sensor_driver import NBusGenericSensor
+from nbus_types.nbus_parameter_type import NBusParameterID
 
 # example config
 config = {
@@ -15,65 +14,76 @@ config = {
     "enable_log": False
 }
 
-
-class DummySlave(NBusSlaveDevice):
-    def data_parameters_loaded(self) -> bool:
-        return True
-
-    def map_parameter_get(self, param_id: NBusParameterID, param_value: int) -> NBusParameterValue:
-        return param_value
-
-    def map_parameter_set(self, param_id: NBusParameterID, param_value: NBusParameterValue) -> int:
-        return param_value
-
-    def map_data_get(self, values: list[int]) -> list[NBusDataValue]:
-        return values
-
-    def map_data_set(self, values: list[NBusDataValue]) -> list[int]:
-        return values
-
-
 if __name__ == "__main__":
-
-    try:
-
-        port = NBusSerialPort(NBusSerialConfig(**config))
-
-        module = NBusSlaveModule(port, 5)
-
-        module.add_device(DummySlave(1))
-        module.add_device(DummySlave(2))
-        module.add_device(DummySlave(3))
-        module.add_device(DummySlave(4))
-        module.add_device(DummySlave(5))
-        module.add_device(DummySlave(129))
-        module.add_device(DummySlave(130))
-
-        # test module get
-        print("CMD GET ECHO: ", module.cmd_get_echo(bytearray("Hello world!", "ascii")))
-        print("CMD GET PARAM SAMPLERATE: ", module.cmd_get_param(NBusParameterID.PARAM_SAMPLERATE))
-        print("CMD GET ALL PARAMS: ", module.cmd_get_all_params())
-        print("CMD GET SENSOR COUNT: ", module.cmd_get_sensor_cnt())
-        print("CMD GET SENSOR TYPE: ", module.cmd_get_sensor_type())
-        print("CMD GET INFO: ", module.cmd_get_info())
-        print("CMD GET SENSOR FORMAT: ", module.cmd_get_format())
-        print("CMD GET SENSOR DATA: ", module.cmd_get_data())
-
-
-       # print("ALL PARAMS: ", sensing_element.cmd_get_all_params())
-        #print("PARAM GAIN: ", sensing_element.cmd_get_param(NBusParameterID.PARAM_GAIN))
-       # print("SENSOR TYPE: ", sensing_element.cmd_get_sensor_type())
-       # print("SENSOR FORMAT: ", sensing_element.cmd_get_info(NBusInfoParam.INFO_FORMAT))
-        #print("DATA: ", sensing_element.cmd_get_data())
-
-        #print(module.cmd_get_format())
-       # print(sensing_element.cmd_get_data())
-        #print(sensing_element.cmd_set_data([129, 1]))
-        #print(sensing_element.cmd_get_data())
-
+    # create port
+    port = NBusSerialPort(NBusSerialConfig(**config))
+
+    # create module
+    module = NBusSlaveModule(port, 5)
+
+    # assemble module
+    module.add_sensor(NBusGenericSensor(1))
+    module.add_sensor(NBusGenericSensor(2))
+    module.add_sensor(NBusGenericSensor(3))
+    module.add_sensor(NBusGenericSensor(4))
+    module.add_sensor(NBusGenericSensor(5))
+    module.add_sensor(NBusGenericSensor(129))
+    module.add_sensor(NBusGenericSensor(130))
+
+    # get sensors
+    accelerometer = module.get_sensor(1)
+    led = module.get_sensor(129)
+
+    # test module get
+    print("<<TEST MODULE GET>>")
+    print("CMD GET ECHO: ", module.cmd_get_echo(bytearray("Hello world!", "ascii")))
+    print("CMD GET PARAM SAMPLERATE: ", module.cmd_get_param(NBusParameterID.PARAM_SAMPLERATE))
+    print("CMD GET ALL PARAMS: ", module.cmd_get_all_params())
+    print("CMD GET SENSOR COUNT: ", module.cmd_get_sensor_cnt())
+    print("CMD GET SENSOR TYPE: ", module.cmd_get_sensor_type())
+    print("CMD GET INFO: ", module.cmd_get_info())
+    print("CMD GET SENSOR FORMAT: ", module.cmd_get_format())
+    print("CMD GET SENSOR DATA: ", module.cmd_get_data())
+
+    # test module set
+    print("\n<<TEST MODULE SET>>")
+    print("CMD SET STOP: ", module.cmd_set_module_stop())
+    print("CMD SET START: ", module.cmd_set_module_start())
+
+    print("CMD SET PARAM SAMPLERATE: ", module.cmd_set_param(NBusParameterID.PARAM_SAMPLERATE, 12345))
+    print("CMD GET PARAM SAMPLERATE: ", module.cmd_get_param(NBusParameterID.PARAM_SAMPLERATE))
+
+    params = {NBusParameterID.PARAM_RANGE: 12, NBusParameterID.PARAM_RANGE0: 4234}
+    print("CMD SET MULTI PARAMS: ", module.cmd_set_multi_params(params))
+    print("CMD GET ALL PARAMS: ", module.cmd_get_all_params())
+
+    print("CMD SET CALIBRATE: ", module.cmd_set_calibrate())
+
+    data = {129: [1], 130: [10, -32]}
+    print("CMD SET DATA: ", module.cmd_set_data(data))
+    print("CMD GET SENSOR DATA: ", module.cmd_get_data())
+
+    # test sensor get
+    print("\n<<TEST SENSOR GET>>")
+    print("CMD GET PARAM SAMPLERATE: ", accelerometer.cmd_get_param(NBusParameterID.PARAM_SAMPLERATE))
+    print("CMD GET ALL PARAMS: ", accelerometer.cmd_get_all_params())
+    print("CMD GET SENSOR TYPE: ", accelerometer.cmd_get_sensor_type())
+    print("CMD GET SENSOR FORMAT: ", accelerometer.cmd_get_format())
+    print("CMD GET SENSOR DATA: ", accelerometer.cmd_get_data())
+
+    # test sensor set
+    print("\n<<TEST SENSOR SET>>")
+    print("CMD SET FIND: ", led.cmd_set_find(True))
+    print("CMD SET PARAM SAMPLERATE: ", led.cmd_set_param(NBusParameterID.PARAM_SAMPLERATE, 23456))
+    print("CMD GET PARAM SAMPLERATE: ", led.cmd_get_param(NBusParameterID.PARAM_SAMPLERATE))
+
+    params = {NBusParameterID.PARAM_RANGE: 12, NBusParameterID.PARAM_RANGE0: 4234}
+    print("CMD SET MULTI PARAMS: ", led.cmd_set_multi_params(params))
+    print("CMD GET ALL PARAMS: ", led.cmd_get_all_params())
+
+    print("CMD SET CALIBRATE: ", led.cmd_set_calibrate())
+
+    print("CMD SET DATA: ", led.cmd_set_data([1]))
+    print("CMD GET SENSOR DATA: ", led.cmd_get_data())
 
 
-    except Exception as Ex:
-        print("Error")
-        print(str(Ex))
-        print(Ex.args)

nbus_api/nbus_common_parser.py

@@ -1,13 +1,12 @@
 import struct
-from typing import Tuple, Any
+from typing import Tuple
 from nbus_types.nbus_data_fomat import NBusDataFormat, NBusDataValue
-from nbus_types.nbus_exceptions.nbus_api_exception import NBusErrorAPI, NBusErrorAPIType
 from nbus_types.nbus_parameter_type import NBusParameterID, NBusParameterValue
 
 
 class NbusCommonParser:
     """
-    Class for common parsers for both device and module.
+    Static class for common parsers for both device and module.
     These methods should be called from the inside of API modules.
     """
 
@@ -39,59 +38,97 @@ class NbusCommonParser:
                               byte_length=byte_len, samples=samples)
 
     @staticmethod
-    def data_from_response(nbus_device: Any, response: list[int]) -> Tuple[list[NBusDataValue], int]:
+    def parameters_from_response(resp_len: int, response: list[int]) \
+            -> list[Tuple[NBusParameterID, NBusParameterValue]]:
+        """
+        Parse multiple parameters from response.
+        :param resp_len: response length in bytes
+        :param response: response from nBus
+        :return: list of (parameter id, parameter value)
+        """
+        offset = 0  # response offset
+        params = []
+        while offset < resp_len:  # parse all params
+            new_offset = offset + 5
+
+            # get param id and type
+            param_id, *param_value = response[offset:new_offset]
+            param_type = NBusParameterID(param_id)
+
+            param_raw_value = struct.unpack("<I", bytearray(param_value))[0]  # get single raw value from bytes
+            params.append((param_type, param_raw_value))
+            offset = new_offset
+
+        return params
+
+    @staticmethod
+    def data_from_response(data_format: NBusDataFormat, response: list[int]) -> Tuple[list[NBusDataValue], int]:
         """
         Parse data from response.
-        :param nbus_device: device to parse for
+        :param data_format: format of data
         :param response: response from NBus
         :return: parsed data and length of parsed data in bytes
         """
-        if nbus_device.data_format is None:  # check for format and params
-            raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED, nbus_device)
-        if not nbus_device.data_parameters_loaded():
-            raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED, nbus_device)
-
         # variables for 2's complement
-        max_uint = 1 << (nbus_device.data_format.byte_length << 3)
+        max_uint = 1 << (data_format.byte_length << 3)
         half_max_uint = max_uint >> 1
 
         values = []
 
-        for s in range(nbus_device.data_format.samples):
+        for s in range(data_format.samples):
             value = 0
-            for i in range(nbus_device.data_format.byte_length):
-                value |= response[(i + 1) + (nbus_device.data_format.byte_length * s)] << (i << 3)
+            for i in range(data_format.byte_length):
+                value |= response[(i + 1) + (data_format.byte_length * s)] << (i << 3)
                 # compose value little endian
 
-            if nbus_device.data_format.sign == 1 and value >= half_max_uint:  # convert 2's complement
+            if data_format.sign == 1 and value >= half_max_uint:  # convert 2's complement
                 value -= max_uint
 
-            value *= 10 ** nbus_device.data_format.value_multiplier  # scale number
+            value *= 10 ** data_format.value_multiplier  # scale number
 
             values.append(value)  # append number to list
 
-        return nbus_device.map_data_get(values), nbus_device.data_format.samples * nbus_device.data_format.byte_length
+        return values, data_format.samples * data_format.byte_length
 
     @staticmethod
-    def parameters_from_response(resp_len: int, response: list[int]) \
-            -> list[Tuple[NBusParameterID, NBusParameterValue]]:
+    def parameters_to_request(params: dict[NBusParameterID, NBusParameterValue]) -> bytearray:
         """
-        Parse multiple parameters from response.
-        :param resp_len: response length in bytes
-        :param response: response from nBus
-        :return: list of (parameter id, parameter value)
+        Create request from parameters dictionary.
+        :param params: parameter dict
+        :return: parameter bytearray
         """
-        offset = 0  # response offset
-        params = []
-        while offset < resp_len:  # parse all params
-            new_offset = offset + 5
+        param_bytes = bytearray()
 
-            # get param id and type
-            param_id, *param_value = response[offset:new_offset]
-            param_type = NBusParameterID(param_id)
+        for p_id in params.keys():
+            param_id_raw = struct.pack("B", p_id.value)
+            param_val_raw = struct.pack("<I", params[p_id])
+            param_bytes += bytearray(param_id_raw) + bytearray(param_val_raw)
 
-            param_raw_value = struct.unpack("<I", bytearray(param_value))[0]  # get single raw value from bytes
-            params.append((param_type, param_raw_value))
-            offset = new_offset
+        return param_bytes
 
-        return params
+    @staticmethod
+    def data_to_request(data_format: NBusDataFormat, data: list[NBusDataValue]) -> list[int]:
+        """
+        Parse data to request format.
+        :param data_format: format of data
+        :param data: data values
+        :return: request bytes
+        """
+        # variables for 2's complement
+        max_uint = 1 << (data_format.byte_length << 3)
+
+        request = []
+
+        for value in data:
+            # Reverse scaling
+            value = int(value / (10 ** data_format.value_multiplier))
+
+            # Handle 2's complement for signed values
+            if data_format.sign == 1 and value < 0:
+                value += max_uint
+
+            # Break the value into bytes (little-endian)
+            for i in range(data_format.byte_length):
+                request.append((value >> (i << 3)) & 0xFF)
+
+        return request

nbus_api/nbus_module_slave.py

@@ -0,0 +1,358 @@
+import struct
+from nbus_api.nbus_sensor import NBusSensor
+from nbus_api.nbus_common_parser import NbusCommonParser
+from nbus_hal.nbus_serial.serial_port import *
+from nbus_types.nbus_address_type import NBusModuleAddress
+from nbus_types.nbus_data_fomat import NBusDataValue, NBusDataFormat
+from nbus_types.nbus_exceptions.nbus_api_exception import NBusErrorAPI, NBusErrorAPIType
+from nbus_types.nbus_parameter_type import NBusParameterID, NBusParameterValue
+from nbus_types.nbus_status_type import NBusStatusType
+from nbus_types.nbus_sensor_count_type import NBusSensorCount
+from nbus_types.nbus_info_type import NBusInfo
+from nbus_types.nbus_sensor_type import NBusSensorType
+
+
+@beartype
+class NBusSlaveModule:
+    """
+    Class representing nBus slave module.
+    """
+
+    def __init__(self, serial_port: NBusSerialPort, module_address: NBusModuleAddress):
+        """
+        Constructor.
+
+        :param serial_port: serial port
+        :param module_address: address of module
+        """
+        self.__port = serial_port
+        self.__module_addr = module_address
+        self.__params = {}
+        self.__devices = {}
+        self._map_param_get = lambda t, v: v    # dummy implementation
+        self._map_param_set = lambda t, v: v    # dummy implementation
+
+    """
+    ================================================================================================================
+                                                Module General Methods
+    ================================================================================================================
+    """
+
+    def set_module_parameter_mappers(self, map_param_get_cb: Callable[[NBusParameterID, NBusParameterValue], int],
+                                     map_param_set_cb: Callable[[NBusParameterID, int], NBusParameterValue]) -> None:
+        """
+        Set parameter mappers for module.
+
+        :param map_param_get_cb: callback for map param get
+        :param map_param_set_cb: callback for map param set
+        """
+        self._map_param_get = map_param_get_cb
+        self._map_param_set = map_param_set_cb
+
+    def add_sensor(self, sensor: NBusSensor) -> None:
+        """
+        Add sensor to container.
+
+        :param sensor: nbus sensor
+        """
+        sensor.set_parent_module_address(self.__module_addr)
+        sensor.set_device_port(self.__port)
+        self.__devices[sensor.address] = sensor
+
+    def get_sensor(self, sensor_address: NBusSensorAddress) -> NBusSensor:
+        """
+        Get module sensor.
+
+        :param sensor_address: address of sensor
+
+        :return: sensor
+        """
+        return self.__devices[sensor_address]
+
+    """
+    ================================================================================================================
+                                                Module Get Commands
+    ================================================================================================================
+    """
+    def cmd_get_echo(self, message: bytearray) -> bool:
+        """
+        Get echo from module.
+
+        :param message: message to send
+        :return: status (True = echo, False = no echo)
+        """
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_ECHO, message)
+        return response == list(message)
+
+    def cmd_get_param(self, parameter: NBusParameterID) -> NBusParameterValue:
+        """
+        Get single module parameter.
+
+        :param parameter: parameter id
+        :return: parameter value
+        """
+        # get response
+        resp_len, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_PARAM,
+                                                         bytearray([parameter.value]))
+
+        # parse parameter
+        param_id, param_val_raw = NbusCommonParser.parameters_from_response(resp_len, response)[0]
+        param_val = self._map_param_get(param_id, param_val_raw)
+
+        # store parameter value
+        self.__params[param_id] = param_val
+
+        return param_val
+
+    def cmd_get_all_params(self) -> dict[NBusParameterID, NBusParameterValue]:
+        """
+        Get all module parameters.
+
+        :return: dict of parameter id and parameter value
+        """
+
+        # get response
+        resp_len, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_PARAM, bytearray([]))
+
+        # parse parameters
+        params_raw = NbusCommonParser.parameters_from_response(resp_len, response)
+
+        for param_id, param_val_raw in params_raw:
+            param_val = self._map_param_get(param_id, param_val_raw)
+
+            # store parameters
+            self.__params[param_id] = param_val
+
+        return self.__params.copy()
+
+    def cmd_get_sensor_cnt(self) -> NBusSensorCount:
+        """
+        Get sensor count.
+
+        :return: count of read-only and read-write sensors
+        """
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_CNT, bytearray([]))
+        return NBusSensorCount(*response)
+
+    def cmd_get_data(self) -> dict[NBusSensorAddress, list[NBusDataValue]]:
+        """
+        Get data from all module sensors.
+
+        :return: dict of device addresses and data values
+        """
+
+        # get data
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_DATA, bytearray([]))
+
+        # parse data
+        begin_idx = 0
+        data = {}
+
+        while begin_idx < resp_length:
+            device_id = response[begin_idx]
+
+            # handle errors
+            if self.__devices[device_id].data_format is None:  # check for format and params
+                raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED)
+            if not self.__devices[device_id].data_parameters_loaded():
+                raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED)
+
+            values, offset = NbusCommonParser.data_from_response(self.__devices[device_id].data_format,
+                                                                 response[begin_idx:])
+            data[device_id] = self.__devices[device_id].map_data_get(values)
+            begin_idx += offset + 1
+
+        return data
+
+    def cmd_get_info(self) -> NBusInfo:
+        """
+        Get module info.
+
+        :return: module info
+        """
+
+        response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_INFO, bytearray([]))
+
+        name = str(response[1:9], "ascii")
+        typ = str(response[9:12], "ascii")
+        uuid = struct.unpack("<I", bytearray(response[12:16]))[0]
+        hw = str(response[16:19], "ascii")
+        fw = str(response[19:22], "ascii")
+        mem_id = struct.unpack("<Q", bytearray(response[22:30]))[0]
+        ro_count = int(response[30])
+        rw_count = int(response[31])
+
+        return NBusInfo(module_name=name, module_type=typ, uuid=uuid, hw=hw, fw=fw, memory_id=mem_id,
+                        read_only_sensors=ro_count, read_write_sensors=rw_count)
+
+    def cmd_get_format(self) -> dict[NBusSensorAddress, NBusDataFormat]:
+        """
+        Get format of all on-board sensors.
+
+        :return: dict of sensor addresses and sensor formats
+        """
+
+        # get response
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_FORMAT,
+                                                            bytearray([]))
+        begin_idx = 0
+        formats = {}
+
+        # parse format
+        while begin_idx < resp_length:
+            device_id = response[begin_idx]
+            device_format = NbusCommonParser.format_from_response(response[begin_idx:begin_idx + 4])
+
+            self.__devices[device_id].data_format = device_format
+            formats[device_id] = device_format
+            begin_idx += 4
+
+        return formats
+
+    def cmd_get_sensor_type(self) -> dict[NBusSensorAddress, NBusSensorType]:
+        """
+        Get type of all on-board sensors.
+
+        :return: dict of sensor addresses and sensor types
+        """
+        # get response
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_TYPE,
+                                                            bytearray([]))
+        # parse response
+        types = {}
+        i = 0
+
+        while i < resp_length - 1:
+            types[NBusSensorAddress(response[i])] = NBusSensorType(response[i + 1])
+            i += 2
+
+        return types
+
+    """
+    ================================================================================================================
+                                                Module Set Commands
+    ================================================================================================================
+    """
+    def cmd_set_module_stop(self) -> NBusStatusType:
+        """
+        Stop automatic measuring.
+
+        :return: status
+        """
+
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_STOP, bytearray([]))
+        return NBusStatusType(response[0])
+
+    def cmd_set_module_start(self) -> NBusStatusType:
+        """
+        Start automatic measuring.
+
+        :return: status
+        """
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_START, bytearray([]))
+        return NBusStatusType(response[0])
+
+    def cmd_set_param(self, param: NBusParameterID, value: NBusParameterValue) -> NBusStatusType:
+        """
+        Set module parameter.
+
+        :param param: parameter ID
+        :param value: parameter value
+        :return: status
+        """
+
+        # create request packet
+        param_id_raw = struct.pack("B", param.value)
+        param_val_raw = struct.pack("<I", self._map_param_set(param, value))
+        param_bytes = bytearray(param_id_raw) + bytearray(param_val_raw)
+
+        # proceed request
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_PARAM, param_bytes)
+
+        # if response is valid, store parameter
+        if response[0] == param.value and response[1] == NBusStatusType.STATUS_SUCCESS:
+            self.__params[param] = value
+
+        return NBusStatusType(response[1])
+
+    def cmd_set_multi_params(self, params: dict[NBusParameterID, NBusParameterValue]) \
+            -> dict[NBusParameterID, NBusStatusType]:
+        """
+        Set multiple module parameters.
+
+        :param params: parameters
+        :return: dict od statuses
+        """
+
+        # scale parameters
+        for p_id in params.keys():
+            params[p_id] = self._map_param_set(p_id, params[p_id])
+
+        # create request packet
+        param_bytes = NbusCommonParser.parameters_to_request(params)
+
+        # send request
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_PARAM,
+                                                            bytearray(param_bytes))
+        # parse statuses
+        statuses = {}
+
+        for i in range(0, resp_length - 1, 2):
+            p_id = NBusParameterID(response[i])
+            if response[i+1] == NBusStatusType.STATUS_SUCCESS:
+                self.__params[p_id] = params[p_id]  # if success, store param
+
+            statuses[p_id] = NBusStatusType(response[i + 1])
+
+        return statuses
+
+    def cmd_set_calibrate(self) -> NBusStatusType:
+        """
+        Send calibration command.
+
+        :return: calibration status
+        """
+
+        resp_length, *response = self.__port.request_module(self.__module_addr,
+                                                            NBusCommand.CMD_SET_CALIBRATE, bytearray([]))
+        print(response)
+        return NBusStatusType(response[0])
+
+    def cmd_set_data(self, data: dict[NBusSensorAddress, list[NBusDataValue]]) \
+            -> dict[NBusSensorAddress, NBusStatusType]:
+        """
+        Set data to read-write sensors.
+
+        :param data:
+        :return: operation statuses
+        """
+
+        # create request packet
+        request = []
+
+        # transform data
+        for addr in data.keys():
+            # handle errors
+            if self.__devices[addr].data_format is None:  # check for format and params
+                raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED)
+            if not self.__devices[addr].data_parameters_loaded():
+                raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED)
+
+            raw_data = self.__devices[addr].map_data_set(data[addr])
+            request.append(addr)
+            request.extend(NbusCommonParser.data_to_request(self.__devices[addr].data_format, raw_data))
+
+        # send request
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_DATA,
+                                                            bytearray(request))
+
+        # return response statuses
+        statuses = {}
+
+        for i in range(0, resp_length - 1, 2):
+            statuses[NBusSensorAddress(response[i])] = NBusStatusType(response[i + 1])
+
+        return statuses
+
+
+

nbus_api/nbus_sensor.py

@@ -0,0 +1,359 @@
+import struct
+from abc import abstractmethod, ABCMeta
+from nbus_hal.nbus_generic_port import NBusPort
+from nbus_types.nbus_command_type import NBusCommand
+from nbus_types.nbus_data_fomat import *
+from nbus_types.nbus_exceptions.nbus_api_exception import NBusErrorAPI, NBusErrorAPIType
+from nbus_types.nbus_parameter_type import *
+from nbus_types.nbus_address_type import NBusSensorAddress, NBusModuleAddress
+from nbus_types.nbus_sensor_type import NBusSensorType
+from nbus_api.nbus_common_parser import NbusCommonParser
+from nbus_types.nbus_status_type import NBusStatusType
+
+
+@beartype
+class NBusSensor(metaclass=ABCMeta):
+    """
+    Class representing nBus sensor type.
+    """
+
+    def __init__(self, address: NBusSensorAddress):
+        """
+        Constructor.
+
+        :param address: device address
+        """
+        self.__address = address
+        self.__module_address = None
+        self.__port = None
+        self.__data_format = None
+        self.__params = {}
+
+    """
+    ================================================================================================================
+                                                    Public Fields
+    ================================================================================================================
+    """
+
+    @property
+    def address(self):
+        """
+        Get sensor address.
+
+        :return: sensor address
+        """
+        return self.__address
+
+    @property
+    def data_format(self):
+        """
+        Get data format.
+
+        :return: data format
+        """
+        return self.__data_format
+
+    @data_format.setter
+    def data_format(self, data_format: NBusDataFormat):
+        """
+        Set data format.
+
+        :param data_format: format of data
+        """
+        self.__data_format = data_format
+
+    @property
+    def parameters(self) -> dict[NBusParameterID, NBusParameterValue]:
+        """
+        Get parameters.
+
+        :return: parameters
+        """
+        return self.__params
+
+    @parameters.setter
+    def parameters(self, values: dict[NBusParameterID, NBusParameterValue]):
+        """
+        Set parameters.
+
+        :param values: values to set
+        """
+        self.__params = values
+
+    """
+    ================================================================================================================
+                                                    Module-only Methods
+    ================================================================================================================
+    """
+
+    def set_parent_module_address(self, address: NBusModuleAddress) -> None:
+        """
+        Set address of parent module.
+
+        :param address: module address
+        """
+        self.__module_address = address
+
+    def set_device_port(self, port: NBusPort) -> None:
+        """
+        Set device communication port.
+        :param port: communicaiton port
+        """
+        self.__port = port
+
+    """
+    ================================================================================================================
+                                                    Abstract Methods
+    ================================================================================================================
+    """
+
+    @abstractmethod
+    def data_parameters_loaded(self) -> bool:
+        """
+        Verify that all necessary parameters are loaded
+        before performing data get/set conversion.
+
+        :return: true if ready for conversion, otherwise False
+        """
+        pass
+
+    @abstractmethod
+    def map_parameter_get(self, param_id: NBusParameterID, param_value: int) -> NBusParameterValue:
+        """
+        Convert a parameter from cmd_get_param() to its engineering range.
+
+        :param param_id: the ID of the parameter
+        :param param_value: the value of the parameter in binary format
+        :return: the converted parameter value in engineering units
+        """
+        pass
+
+    @abstractmethod
+    def map_parameter_set(self, param_id: NBusParameterID, param_value: NBusParameterValue) -> int:
+        """
+        Convert a parameter to its binary range for cmd_set_data().
+
+        :param param_id: the ID of the parameter
+        :param param_value: the value of the parameter in engineering units
+        :return: the converted parameter value in binary format
+        """
+        pass
+
+    @abstractmethod
+    def map_data_get(self, values: list[int]) -> list[NBusDataValue]:
+        """
+        Convert data from cmd_get_data() to its engineering range.
+
+        :param values: a list of values in binary format to be converted
+        :return: a list of converted values in engineering units
+        """
+        pass
+
+    @abstractmethod
+    def map_data_set(self, values: list[NBusDataValue]) -> list[int]:
+        """
+        Convert data to its binary range for cmd_set_data().
+
+        :param values: a list of values in engineering range to be converted
+        :return: a list of converted values in binary format
+        """
+        pass
+
+    """
+    ================================================================================================================
+                                                    Device Get Commands
+    ================================================================================================================
+    """
+
+    def cmd_get_param(self, parameter: NBusParameterID) -> NBusParameterValue:
+        """
+        Get single sensor parameter.
+
+        :param parameter: parameter id
+        :return: parameter value
+        """
+        # get response
+        resp_len, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                         NBusCommand.CMD_GET_PARAM, bytearray([parameter.value]))
+
+        # parse parameter
+        param_id, param_val_raw = NbusCommonParser.parameters_from_response(resp_len, response)[0]
+        param_val = self.map_parameter_get(param_id, param_val_raw)
+
+        # store parameter value
+        self.__params[param_id] = param_val
+
+        return param_val
+
+    def cmd_get_all_params(self) -> dict[NBusParameterID, NBusParameterValue]:
+        """
+        Get all sensor parameters.
+
+        :return: dict of parameter id and parameter value
+        """
+
+        # get response
+        resp_len, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                         NBusCommand.CMD_GET_PARAM, bytearray([]))
+
+        # parse parameters
+        params_raw = NbusCommonParser.parameters_from_response(resp_len, response)
+
+        for param_id, param_val_raw in params_raw:
+            param_val = self.map_parameter_get(param_id, param_val_raw)
+
+            # store parameters
+            self.__params[param_id] = param_val
+
+        return self.__params.copy()
+
+    def cmd_get_data(self) -> list[NBusDataValue]:
+        """
+       Get data from sensor.
+
+       :raises
+
+       :return: dict of device addresses and data values
+       """
+
+        if self.__data_format is None:  # check for format and params
+            raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED)
+        if not self.data_parameters_loaded():
+            raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED)
+
+        _, *resp = self.__port.request_sensor(self.__module_address, self.__address, NBusCommand.CMD_GET_DATA,
+                                              bytearray([]))
+        values, _ = NbusCommonParser.data_from_response(self.data_format, resp)
+        return self.map_data_get(values)
+
+    def cmd_get_sensor_type(self) -> NBusSensorType:
+        """
+        Get sensor type.
+
+        :return: sensor type
+        """
+        _, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                  NBusCommand.CMD_GET_SENSOR_TYPE, bytearray([]))
+        return NBusSensorType(response[0])
+
+    def cmd_get_format(self):
+        """
+        Get sensor format.
+
+        :return: sensor format
+        """
+        _, *response = self.__port.request_sensor(self.__module_address, self.__address, NBusCommand.CMD_GET_FORMAT,
+                                                  bytearray([]))
+
+        self.data_format = NbusCommonParser.format_from_response(response)
+        return self.data_format
+
+    """
+    ================================================================================================================
+                                                  Device Set Commands
+    ================================================================================================================
+    """
+    def cmd_set_find(self, enable: bool) -> NBusStatusType:
+        """
+        Turn on/off physical indicator of device.
+
+        :param enable: to start (True) / stop (False) finding
+        :return: status
+        """
+        _, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                  NBusCommand.CMD_SET_FIND, bytearray([enable]))
+        return NBusStatusType(response[0])
+
+    def cmd_set_param(self, param: NBusParameterID, value: NBusParameterValue) -> NBusStatusType:
+        """
+        Set sensor parameter.
+
+        :param param: parameter ID
+        :param value: parameter value
+        :return: status
+        """
+
+        # create request packet
+        param_id_raw = struct.pack("B", param.value)
+        param_val_raw = struct.pack("<I", self.map_parameter_set(param, value))
+        param_bytes = bytearray(param_id_raw) + bytearray(param_val_raw)
+
+        # proceed request
+        _, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                  NBusCommand.CMD_SET_PARAM, param_bytes)
+
+        # if response is valid, store parameter
+        if response[0] == param.value and response[1] == NBusStatusType.STATUS_SUCCESS:
+            self.__params[param] = value
+
+        return NBusStatusType(response[1])
+
+    def cmd_set_multi_params(self, params: dict[NBusParameterID, NBusParameterValue]) \
+            -> dict[NBusParameterID, NBusStatusType]:
+        """
+        Set multiple sensor parameters.
+
+        :param params: parameters
+        :return: dict od statuses
+        """
+
+        # scale parameters
+        for p_id in params.keys():
+            params[p_id] = self.map_parameter_set(p_id, params[p_id])
+
+        # create request packet
+        param_bytes = NbusCommonParser.parameters_to_request(params)
+
+        # send request
+        resp_length, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                            NBusCommand.CMD_SET_PARAM, bytearray(param_bytes))
+        # parse statuses
+        statuses = {}
+
+        for i in range(0, resp_length - 1, 2):
+            p_id = NBusParameterID(response[i])
+            if response[i + 1] == NBusStatusType.STATUS_SUCCESS:
+                self.__params[p_id] = params[p_id]  # if success, store param
+
+            statuses[p_id] = NBusStatusType(response[i + 1])
+
+        return statuses
+
+    def cmd_set_calibrate(self) -> NBusStatusType:
+        """
+        Send calibration command.
+
+        :return: calibration status
+        """
+
+        resp_length, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                            NBusCommand.CMD_SET_CALIBRATE, bytearray([]))
+        return NBusStatusType(response[0])
+
+    def cmd_set_data(self, data: list[NBusDataValue]) -> NBusStatusType:
+        """
+        Set data to read-write sensor.
+
+        :param data: data to set
+        :raises: NBusErrorAPIType
+        :return: operation status
+        """
+
+        if self.__data_format is None:  # check for format and params
+            raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED)
+        if not self.data_parameters_loaded():
+            raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED)
+
+        # create request packet
+        request = []
+
+        # transform data
+        raw_data = self.map_data_set(data)
+        request.append(self.__address)
+        request.extend(NbusCommonParser.data_to_request(self.data_format, raw_data))
+
+        # send request
+        resp_length, *response = self.__port.request_sensor(self.__module_address, self.__address,
+                                                            NBusCommand.CMD_SET_DATA, bytearray(request))
+        # return response status
+        return NBusStatusType(response[1])

nbus_api/nbus_slave_device.py

@@ -1,183 +0,0 @@
-import struct
-from abc import abstractmethod, ABCMeta
-from typing import Tuple
-from nbus_hal.nbus_generic_port import NBusPort
-from nbus_types.nbus_command_type import NBusCommand
-from nbus_types.nbus_data_fomat import *
-from nbus_types.nbus_exceptions.nbus_api_exception import NBusErrorAPI, NBusErrorAPIType
-from nbus_types.nbus_parameter_type import *
-from nbus_types.nbus_address_type import NBusDeviceAddress, NBusModuleAddress
-from nbus_types.nbus_sensor_type import NBusSensorType
-from nbus_api.nbus_common_parser import NbusCommonParser
-
-
-@beartype
-class NBusSlaveDevice(metaclass=ABCMeta):
-
-    def __init__(self, address: NBusDeviceAddress):
-        self.__address = address
-        self.__module_address = None
-        self.__port = None
-        self.__data_format = None
-        self.__parameters = {}
-
-    @property
-    def address(self):
-        return self.__address
-
-    @property
-    def data_format(self):
-        return self.__data_format
-
-    @data_format.setter
-    def data_format(self, data_format: NBusDataFormat) -> None:
-        self.__data_format = data_format
-
-    @property
-    def parameters(self):
-        return self.__parameters
-
-    @parameters.setter
-    def parameters(self, values):
-        self.__parameters = values
-
-    def set_parent_module_address(self, address: NBusModuleAddress):
-        self.__module_address = address
-
-    def set_device_port(self, port: NBusPort):
-        self.__port = port
-
-    """
-    ================================================================================================================
-                                                    Abstract Methods
-    ================================================================================================================
-    """
-
-    @abstractmethod
-    def data_parameters_loaded(self) -> bool:
-        """
-        Verify that all necessary parameters are loaded
-        before performing data get/set conversion.
-
-        :return: true if ready for conversion, otherwise False
-        """
-        pass
-
-    @abstractmethod
-    def map_parameter_get(self, param_id: NBusParameterID, param_value: int) -> NBusParameterValue:
-        """
-        Convert a parameter from cmd_get_param() to its engineering range.
-
-        :param param_id: the ID of the parameter
-        :param param_value: the value of the parameter in binary format
-        :return: the converted parameter value in engineering units
-        """
-        pass
-
-    @abstractmethod
-    def map_parameter_set(self, param_id: NBusParameterID, param_value: NBusParameterValue) -> int:
-        """
-        Convert a parameter to its binary range for cmd_set_data().
-
-        :param param_id: the ID of the parameter
-        :param param_value: the value of the parameter in engineering units
-        :return: the converted parameter value in binary format
-        """
-        pass
-
-    @abstractmethod
-    def map_data_get(self, values: list[int]) -> list[NBusDataValue]:
-        """
-        Convert data from cmd_get_data() to its engineering range.
-
-        :param values: a list of values in binary format to be converted
-        :return: a list of converted values in engineering units
-        """
-        pass
-
-    @abstractmethod
-    def map_data_set(self, values: list[NBusDataValue]) -> list[int]:
-        """
-        Convert data to its binary range for cmd_set_data().
-
-        :param values: a list of values in engineering range to be converted
-        :return: a list of converted values in binary format
-        """
-        pass
-
-    """
-    ================================================================================================================
-                                                    Device Get Commands
-    ================================================================================================================
-    """
-
-    def cmd_get_param(self, parameter: NBusParameterID) -> Tuple[NBusParameterID, NBusParameterValue]:
-        # get response
-        resp_len, *response = self.__port.request_device(self.__module_address, self.__address,
-                                                         NBusCommand.CMD_GET_PARAM, bytearray([parameter.value]))
-
-        param_id, param_val_raw = NbusCommonParser.parameters_from_response(resp_len, response)[0]
-        param_val = self.map_parameter_get(param_id, param_val_raw)
-
-        self.__parameters[param_id] = param_val
-
-        return param_id, param_val
-
-    def cmd_get_all_params(self) -> list[Tuple[NBusParameterID, NBusParameterValue]]:
-        resp_len, *response = self.__port.request_device(self.__module_address, self.__address,
-                                                         NBusCommand.CMD_GET_PARAM, bytearray([]))
-
-        params_raw = NbusCommonParser.parameters_from_response(resp_len, response)
-
-        params = []
-
-        for param_id, param_val_raw in params_raw:
-            param_val = self.map_parameter_get(param_id, param_val_raw)
-
-            params.append((param_id, param_val))
-            self.__parameters[param_id] = param_val
-
-        return params
-
-    def cmd_get_data(self):
-        _, *resp = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_GET_DATA, bytearray([]))
-        values, _ = NbusCommonParser.data_from_response(self, resp)
-        return values
-
-    def cmd_get_sensor_type(self):
-        _, *response = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_GET_SENSOR_TYPE,
-                                                  bytearray([]))
-        return NBusSensorType(response[0])
-
-    def cmd_get_format(self):
-        _, *response = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_GET_FORMAT,
-                                                  bytearray([]))
-
-        self.data_format = NbusCommonParser.format_from_response(response)
-        return self.data_format
-
-    """
-    ================================================================================================================
-                                                  Device Set Commands
-    ================================================================================================================
-    """
-
-    def cmd_set_param(self, param: NBusParameterID, value: NBusParameterValue) -> None:
-
-        raw_value = self.map_parameter_set(param, value)                        # get raw value
-        value_bytes = struct.pack('<I', raw_value & 0xFFFFFFFF)             # create bytes
-        self.__port.request_device(self.__module_address, self.__address,     # proceed request
-                                  NBusCommand.CMD_SET_PARAM, bytearray([param.value, *value_bytes]))
-        self.__parameters[param] = value     # store new value of parameter
-
-    def cmd_set_data(self, data: list[NBusDataValue]):
-
-        if self.__data_format is None:  # check for format and params
-            raise NBusErrorAPI(NBusErrorAPIType.FORMAT_NOT_LOADED)
-        if not self.data_parameters_loaded():
-            raise NBusErrorAPI(NBusErrorAPIType.PARAMS_NOT_LOADED)
-
-        binary_data = self.map_data_set(data)
-        _, *resp = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_SET_DATA, bytearray(binary_data))
-
-        return resp

nbus_api/nbus_slave_module.py

@@ -1,163 +0,0 @@
-import struct
-from abc import abstractmethod
-from typing import Tuple, Annotated
-
-from beartype.vale import Is
-
-from nbus_api.nbus_slave_device import NBusSlaveDevice
-from nbus_api.nbus_common_parser import NbusCommonParser
-from nbus_hal.nbus_serial.serial_port import *
-from nbus_types.nbus_address_type import NBusModuleAddress
-from nbus_types.nbus_parameter_type import NBusParameterID, NBusParameterValue
-from nbus_types.nbus_status_type import NBusStatusType
-from nbus_types.nbus_sensor_count_type import NBusSensorCount
-from nbus_types.nbus_info_type import NBusInfo
-from nbus_types.nbus_sensor_type import NBusSensorType
-
-
-@beartype
-class NBusSlaveModule:
-
-    def __init__(self, serial_port: NBusSerialPort, module_address: NBusModuleAddress):
-        self.__port = serial_port
-        self.__module_addr = module_address
-        self.__params = {}
-        self.__devices = {}
-        self._map_param_get = lambda t, v: v
-        self._map_param_set = lambda t, v: v
-
-    def set_module_parameter_mappers(self, map_param_get_cb: Callable[[NBusParameterID, NBusParameterValue], int],
-                                     map_param_set_cb: Callable[[NBusParameterID, int], NBusParameterValue]):
-        self._map_param_get = map_param_get_cb
-        self._map_param_set = map_param_set_cb
-
-    def add_device(self, device: NBusSlaveDevice) -> None:
-        device.set_parent_module_address(self.__module_addr)
-        device.set_device_port(self.__port)
-        self.__devices[device.address] = device
-
-    def get_device(self, device_address: NBusDeviceAddress) -> NBusSlaveDevice:
-        return self.__devices[device_address]
-
-
-
-    """
-    ================================================================================================================
-                                                Module Get Commands
-    ================================================================================================================
-    """
-    def cmd_get_echo(self, message: bytearray) -> bool:
-        """
-        Send Echo Command.
-        :param message: message to send
-        :return: status (True = echo, False = no echo)
-        """
-        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_ECHO, message)
-        return response == list(message)
-
-    def cmd_get_param(self, parameter: NBusParameterID) -> Tuple[NBusParameterID, NBusParameterValue]:
-        # get response
-        resp_len, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_PARAM,
-                                                         bytearray([parameter.value]))
-
-        param_id, param_val_raw = NbusCommonParser.parameters_from_response(resp_len, response)[0]
-        param_val = self._map_param_get(param_id, param_val_raw)
-
-        self.__params[param_id] = param_val
-
-        return param_id, param_val
-
-    def cmd_get_all_params(self) -> dict[NBusParameterID, NBusParameterValue]:
-        resp_len, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_PARAM, bytearray([]))
-        params_raw = NbusCommonParser.parameters_from_response(resp_len, response)
-
-        for param_id, param_val_raw in params_raw:
-            param_val = self._map_param_get(param_id, param_val_raw)
-
-            self.__params[param_id] = param_val
-
-        return self.__params.copy()
-
-    def cmd_get_sensor_cnt(self) -> NBusSensorCount:
-        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_CNT, bytearray([]))
-        return NBusSensorCount(*response)
-
-    def cmd_get_data(self):
-        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_DATA, bytearray([]))
-        begin_idx = 0
-        data = {}
-
-        while begin_idx < resp_length:
-            device_id = response[begin_idx]
-
-            values, offset = NbusCommonParser.data_from_response(self.__devices[device_id], response[begin_idx:])
-            data[device_id] = values
-            begin_idx += offset + 1
-
-        return data
-
-    def cmd_get_info(self):
-        response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_INFO, bytearray([]))
-
-        name = str(response[1:9], "ascii")
-        typ = str(response[9:12], "ascii")
-        uuid = struct.unpack("<I", bytearray(response[12:16]))[0]
-        hw = str(response[16:19], "ascii")
-        fw = str(response[19:22], "ascii")
-        mem_id = struct.unpack("<Q", bytearray(response[22:30]))[0]
-
-        return NBusInfo(module_name=name, module_type=typ, uuid=uuid, hw=hw, fw=fw, memory_id=mem_id)
-
-
-    def cmd_get_format(self):
-        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_FORMAT,
-                                                            bytearray([]))
-        begin_idx = 0
-        formats = {}
-
-        while begin_idx < resp_length:
-            device_id = response[begin_idx]
-            device_format = NbusCommonParser.format_from_response(response[begin_idx:begin_idx + 4])
-
-            self.__devices[device_id].data_format = device_format
-            formats[device_id] = device_format
-            begin_idx += 4
-
-        return formats
-
-
-    def cmd_get_sensor_type(self):
-        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_TYPE,
-                                                            bytearray([]))
-        types = {}
-        i = 0
-
-        while i < resp_length - 1:
-            types[NBusDeviceAddress(response[i])] = NBusSensorType(response[i+1])
-            i += 2
-
-        return types
-
-
-    """
-    ================================================================================================================
-                                                Module Set Commands
-    ================================================================================================================
-    """
-
-    def cmd_set_find(self, enable: bool) -> NBusStatusType:
-        """
-        Send Find Command.
-        :param enable: to start (True) / stop (False) finding
-        :return: status
-        """
-        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_FIND, bytearray([enable]))
-        return NBusStatusType(response[0])
-
-    def cmd_set_module_stop(self):
-        self._send_request_module(self.__module_addr, NBusCommand.CMD_SET_STOP, bytearray([]))
-
-    def cmd_set_module_start(self):
-        self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_START, bytearray([]))
-
-

nbus_hal/nbus_generic_port.py

@@ -2,7 +2,7 @@ from abc import ABCMeta, abstractmethod
 from typing import Annotated
 from beartype import beartype
 from beartype.vale import Is
-from nbus_types.nbus_address_type import NBusModuleAddress, NBusDeviceAddress
+from nbus_types.nbus_address_type import NBusModuleAddress, NBusSensorAddress
 from nbus_types.nbus_command_type import NBusCommand
 
 
@@ -63,12 +63,12 @@ class NBusPort(metaclass=ABCMeta):
         pass
 
     @abstractmethod
-    def request_device(self, module_addr: NBusModuleAddress, device_address: NBusDeviceAddress, command: NBusCommand,
+    def request_sensor(self, module_addr: NBusModuleAddress, sensor_address: NBusSensorAddress, command: NBusCommand,
                        data: bytearray, long_answer: NBusDelay = 0.0) -> bytearray:
         """
-        Make device request to nbus network.
+        Make sensor request to nbus network.
         :param module_addr: address of module
-        :param device_address: address of device
+        :param sensor_address: address of sensor
         :param command: command id
         :param data: command data to send
         :param long_answer: delay in s for longer answer

nbus_hal/nbus_serial/serial_port.py

@@ -9,7 +9,7 @@ from nbus_types.nbus_exceptions.nbus_network_exception import *
 from nbus_types.nbus_exceptions.nbus_node_exception import *
 from nbus_hal.nbus_serial.serial_config import NBusSerialConfig
 from nbus_hal.nbus_generic_port import NBusPort, NBusDelay
-from nbus_types.nbus_address_type import NBusModuleAddress, NBusDeviceAddress
+from nbus_types.nbus_address_type import NBusModuleAddress, NBusSensorAddress
 from nbus_hal.crc8 import crc8
 
 
@@ -99,18 +99,18 @@ class NBusSerialPort(NBusPort):
         """
         return self._request_response(module_addr, 0, command.value, data, long_answer)
 
-    def request_device(self, module_addr: NBusModuleAddress, device_address: NBusDeviceAddress,
+    def request_sensor(self, module_addr: NBusModuleAddress, sensor_address: NBusSensorAddress,
                        command: NBusCommand, data: bytearray, long_answer: NBusDelay = 0.0) -> bytearray:
         """
         Make device request to nbus network.
         :param module_addr: address of module
-        :param device_address: address of device
+        :param sensor_address: address of sensor
         :param command: command id
         :param data: command data to send
         :param long_answer: delay in s for longer answer
         :return: | payload length | payload |
         """
-        return self._request_response(module_addr, device_address, command.value, data, long_answer)
+        return self._request_response(module_addr, sensor_address, command.value, data, long_answer)
 
     """
     ================================================================================================================

nbus_sensor_drivers/generic_sensor_driver.py

@@ -0,0 +1,24 @@
+from nbus_api.nbus_sensor import NBusSensor
+from nbus_types.nbus_data_fomat import NBusDataValue
+from nbus_types.nbus_parameter_type import NBusParameterID, NBusParameterValue
+
+
+class NBusGenericSensor(NBusSensor):
+    """
+    Class for generic NBus sensor (no data transformation)
+    """
+
+    def data_parameters_loaded(self) -> bool:
+        return True
+
+    def map_parameter_get(self, param_id: NBusParameterID, param_value: int) -> NBusParameterValue:
+        return param_value
+
+    def map_parameter_set(self, param_id: NBusParameterID, param_value: NBusParameterValue) -> int:
+        return param_value
+
+    def map_data_get(self, values: list[int]) -> list[NBusDataValue]:
+        return values
+
+    def map_data_set(self, values: list[NBusDataValue]) -> list[int]:
+        return values

nbus_types/nbus_address_type.py

@@ -7,6 +7,6 @@ Typedef for NBus module address.
 NBusModuleAddress = Annotated[int, Is[lambda x: 1 <= x <= 127]]
 
 """
-Typedef for NBus device address.
+Typedef for NBus sensor address.
 """
-NBusDeviceAddress = Annotated[int, Is[lambda value: (1 <= value <= 31) or (129 <= value <= 159)]]
+NBusSensorAddress = Annotated[int, Is[lambda value: (1 <= value <= 31) or (129 <= value <= 159)]]

nbus_types/nbus_info_type.py

@@ -17,6 +17,8 @@ class NBusInfo:
     :ivar hw: firmware version. MAJOR.MINOR
     :ivar fw: memory ID number
     :ivar memory_id: sensor count
+    :ivar read_only_sensors: count of read-only sensors
+    :ivar read_write_sensors: count of read-write sensors
     """
     module_name: Annotated[str, Is[lambda value: len(value) == 8]]
     module_type: Annotated[str, Is[lambda value: len(value) == 3]]
@@ -24,6 +26,8 @@ class NBusInfo:
     hw: Annotated[str, Is[lambda value: len(value) == 3]]
     fw: Annotated[str, Is[lambda value: len(value) == 3]]
     memory_id: Annotated[int, Is[lambda value: 0 <= value <= np.iinfo(np.uint64).max]]
+    read_only_sensors: Annotated[int, Is[lambda value: 0 <= value <= 31]]
+    read_write_sensors: Annotated[int, Is[lambda value: 0 <= value <= 31]]