+ Resolved functional cross-dependencies between modules and devices

README.md

@@ -2,4 +2,4 @@
 
 This project provides a Python client API for the nBus protocol over a serial port. 
 It offers a high-level, strictly object-oriented approach, ensuring type safety, clear error handling
-and maintainability.
+and maintainability. It uses The Beatype runtime typechecking.

main.py

@@ -1,7 +1,10 @@
 from nbus_hal.nbus_serial.serial_port import *
 from nbus_api.nbus_slave_module import NBusSlaveModule
-from nbus_types.nbus_address_type import NBusModuleAddress
-from nbus_hal.nbus_serial.serial_types import *
+from nbus_api.nbus_slave_device import NBusSlaveDevice
+from nbus_types.nbus_address_type import NBusModuleAddress, NBusDeviceAddress
+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
 
 # example config
 config = {
@@ -13,15 +16,51 @@ config = {
     "enable_log": True
 }
 
+
+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.cmd_set_module_start()
+        sensing_element = module.get_device(1)
+
+        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_info(NBusInfoParam.INFO_FORMAT))
+        print(module.cmd_get_data())
+
 
 
-        port = NBusSerialPort(NBusSerialConfig(**config))
-        nbus = NBusSlaveModule(port, NBusModuleAddress(5))
-        print(nbus.cmd_get_echo_module(bytearray([1,2,3])))
 
     except Exception as Ex:
         print(str(Ex))
+        print(Ex.args)

nbus_api/nbus_common_parser.py

@@ -0,0 +1,97 @@
+import struct
+from typing import Tuple, Any
+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.
+    These methods should be called from the inside of API modules.
+    """
+
+    @staticmethod
+    def format_from_response(response: list[int]) -> NBusDataFormat:
+        """
+        Parse format from response.
+
+        Note: byte representation is:
+        | device_address | unit_multiplier, sign | value_multiplier | samples, byte_length |
+
+        :param response: response from NBus
+        :return: data format
+        """
+        sign = bool(response[1] & 0x01)
+
+        unit_mul = response[1] >> 1
+        if unit_mul >= 64:
+            unit_mul -= 128
+
+        value_mul = response[2]
+        if value_mul >= 128:
+            value_mul -= 256
+
+        byte_len = response[3] & 0x0F
+        samples = (response[3] & 0xF0) >> 4
+
+        return NBusDataFormat(sign=sign, unit_multiplier=unit_mul, value_multiplier=value_mul,
+                              byte_length=byte_len, samples=samples)
+
+    @staticmethod
+    def data_from_response(nbus_device: Any, response: list[int]) -> Tuple[list[NBusDataValue], int]:
+        """
+        Parse data from response.
+        :param nbus_device: device to parse for
+        :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)
+        half_max_uint = max_uint >> 1
+
+        values = []
+
+        for s in range(nbus_device.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)
+                # compose value little endian
+
+            if nbus_device.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
+
+            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
+
+    @staticmethod
+    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

nbus_api/nbus_slave_device.py

@@ -1,7 +1,13 @@
+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, NBusInfoParam
 from nbus_types.nbus_data_fomat import *
 from nbus_types.nbus_parameter_type import *
-from nbus_types.nbus_address_type import NBusDeviceAddress
+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
@@ -9,9 +15,10 @@ 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 = None
-
+        self.__parameters = {}
 
     @property
     def address(self):
@@ -33,14 +40,20 @@ class NBusSlaveDevice(metaclass=ABCMeta):
     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
+
     """
     ================================================================================================================
-                                                        Device Interface
+                                                    Abstract Methods
     ================================================================================================================
     """
 
     @abstractmethod
-    def data_params_loaded(self) -> bool:
+    def data_parameters_loaded(self) -> bool:
         """
         Verify that all necessary parameters are loaded
         before performing data get/set conversion.
@@ -50,7 +63,7 @@ class NBusSlaveDevice(metaclass=ABCMeta):
         pass
 
     @abstractmethod
-    def map_param_get(self, param_id: NBusParameterType, param_value: int) -> NBusParameterValue:
+    def map_parameter_get(self, param_id: NBusParameterID, param_value: int) -> NBusParameterValue:
         """
         Convert a parameter from cmd_get_param() to its engineering range.
 
@@ -61,7 +74,7 @@ class NBusSlaveDevice(metaclass=ABCMeta):
         pass
 
     @abstractmethod
-    def map_param_set(self, param_id: NBusParameterType, param_value: NBusParameterValue) -> int:
+    def map_parameter_set(self, param_id: NBusParameterID, param_value: NBusParameterValue) -> int:
         """
         Convert a parameter to its binary range for cmd_set_data().
 
@@ -90,3 +103,56 @@ class NBusSlaveDevice(metaclass=ABCMeta):
         :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_info(self, parameter: NBusInfoParam):
+        _, *response = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_GET_INFO,
+                                                  bytearray([parameter.value]))
+
+        if parameter == NBusInfoParam.INFO_FORMAT:
+
+            self.data_format = NbusCommonParser.format_from_response(response)
+            return self.data_format

nbus_api/nbus_slave_module.py

@@ -1,5 +1,14 @@
+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
 
 
 @beartype
@@ -10,12 +19,119 @@ class NBusSlaveModule:
         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]
 
-    def cmd_get_echo_module(self, message: bytearray) -> bool:
-        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_ECHO, message)
+
+
+    """
+    ================================================================================================================
+                                                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) -> list[Tuple[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)
+
+        params = []
+
+        for param_id, param_val_raw in params_raw:
+            param_val = self._map_param_get(param_id, param_val_raw)
+
+            params.append((param_id, param_val))
+            self.__params[param_id] = param_val
+
+        return params
+
+    def cmd_get_sensor_cnt(self) -> Annotated[int, Is[lambda value: 0 < value < 64]]:
+        _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_CNT, bytearray([]))
+        return response[0]
+
+    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.append((device_id, values))
+            begin_idx += offset + 1
+
+        return data
+
+    def cmd_get_info(self, parameter: NBusInfoParam):
+        resp_length, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_INFO,
+                                                            bytearray([parameter.value]))
+
+        if parameter == NBusInfoParam.INFO_FORMAT:
+            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.append((device_id, device_format))
+                begin_idx += 4
+
+            return formats
+
+
+
+
+    def cmd_get_sensor_type(self):
+        pass
+
+
+    """
+    ================================================================================================================
+                                                Module Set Commands
+    ================================================================================================================
+    """
+
+    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_serial/serial_port.py

@@ -97,7 +97,7 @@ class NBusSerialPort(NBusPort):
         :param long_answer: delay in s for longer answer
         :return: | payload length | payload |
         """
-        return self._request_response(module_addr.value, 0, command.value, data, long_answer)
+        return self._request_response(module_addr, 0, command.value, data, long_answer)
 
     def request_device(self, module_addr: NBusModuleAddress, device_address: NBusDeviceAddress,
                        command: NBusCommand, data: bytearray, long_answer: NBusDelay = 0.0) -> bytearray:
@@ -110,7 +110,7 @@ class NBusSerialPort(NBusPort):
         :param long_answer: delay in s for longer answer
         :return: | payload length | payload |
         """
-        return self._request_response(module_addr.value, device_address.value, command.value, data, long_answer)
+        return self._request_response(module_addr, device_address, command.value, data, long_answer)
 
     """
     ================================================================================================================

nbus_types/nbus_address_type.py

@@ -1,39 +1,12 @@
-from enum import Enum
 from typing import Annotated
-from beartype import beartype
 from beartype.vale import Is
 
+"""
+Typedef for NBus module address.
+"""
+NBusModuleAddress = Annotated[int, Is[lambda x: 1 <= x <= 127]]
 
-@beartype
-class NBusModuleAddress:
-    """
-    Class for NBus module address type.
-    """
-    def __init__(self, address: Annotated[int, Is[lambda x: 1 <= x <= 127]]):
-        self.__addr = address
-
-    @property
-    def value(self):
-        return self.__addr
-
-
-class NBusDeviceIO(Enum):
-    """
-    Enum class for device data direction.
-    """
-    OUTPUT = 0
-    INPUT = 1
-
-
-@beartype
-class NBusDeviceAddress:
-    """
-    Class for NBus device address.
-    """
-    def __init__(self, io: NBusDeviceIO, address: Annotated[int, Is[lambda value: 1 <= value < 31]]):
-        self.__io = io
-        self.__addr = address
-
-    @property
-    def value(self):
-        return self.__addr + 128 * self.__io.value
+"""
+Typedef for NBus device address.
+"""
+NBusDeviceAddress = Annotated[int, Is[lambda value: (1 <= value <= 31) or (129 <= value <= 159)]]

nbus_types/nbus_command_type.py

@@ -6,26 +6,28 @@ class NBusCommand(Enum):
     Enum class for valid nBus commands.
     """
     # get
-    CMD_ECHO = 0x01
-    CMD_STOP = 0x02
-    CMD_START = 0x03
-    CMD_PARAM = 0x04
-    CMD_SENSOR_CNT = 0x05
-    CMD_SLEEP = 0x06
-    CMD_WAKEUP = 0x07
-    CMD_CALIBRATE = 0x08
-    CMD_RESET = 0x09
-    CMD_STORE = 0x0A
-    CMD_DATA = 0x0B
-    CMD_SYNC = 0x0C
-    CMD_SENSOR_TYPE = 0x0D
-    CMD_INFO = 0x0E
+    CMD_GET_ECHO = 0x01
+    CMD_GET_STOP = 0x02
+    CMD_GET_START = 0x03
+    CMD_GET_PARAM = 0x04
+    CMD_GET_SENSOR_CNT = 0x05
+    CMD_GET_SLEEP = 0x06
+    CMD_GET_WAKEUP = 0x07
+    CMD_GET_CALIBRATE = 0x08
+    CMD_GET_RESET = 0x09
+    CMD_GET_STORE = 0x0A
+    CMD_GET_DATA = 0x0B
+    CMD_GET_SYNC = 0x0C
+    CMD_GET_SENSOR_TYPE = 0x0D
+    CMD_GET_INFO = 0x0E
 
     # set
-    CMD_PARAM_SET = 0x24
-    CMD_CALIBRATE_SET = 0x28
-    CMD_STORE_SET = 0x2A
-    CMD_DATA_SET = 0x2B
+    CMD_SET_STOP = 0x22
+    CMD_SET_START = 0x23
+    CMD_SET_PARAM = 0x24
+    CMD_SET_CALIBRATE = 0x28
+    CMD_SET_STORE = 0x2A
+    CMD_SET_DATA = 0x2B
 
 
 class NBusInfoParam(Enum):

nbus_types/nbus_parameter_type.py

@@ -7,7 +7,7 @@ Typedef for nBus parameter value.
 NBusParameterValue = Union[float, int]
 
 
-class NBusParameterType(Enum):
+class NBusParameterID(Enum):
     """
     Enum class for nBus parameter types.
     """