+ Updated functionality for module.get

main.py

@@ -1,19 +1,18 @@
 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_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 = {
-    "port_name": "COM4",
+    "port_name": "COM6",
     "baud": NBusBaudrate.SPEED_921600,
     "parity": NBusParity.NONE,
     "timeout": 0.4,
     "request_attempts": 1,
-    "enable_log": True
+    "enable_log": False
 }
 
 
@@ -39,28 +38,42 @@ 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))
 
-        module.cmd_set_module_start()
-        sensing_element = module.get_device(1)
+        # 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_info(NBusInfoParam.INFO_FORMAT))
-        print(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())
 
 
 
     except Exception as Ex:
+        print("Error")
         print(str(Ex))
         print(Ex.args)

nbus_api/nbus_slave_device.py

@@ -2,8 +2,9 @@ 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_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
@@ -148,11 +149,35 @@ class NBusSlaveDevice(metaclass=ABCMeta):
                                                   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]))
+    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)
 
-        if parameter == NBusInfoParam.INFO_FORMAT:
+        binary_data = self.map_data_set(data)
+        _, *resp = self.__port.request_device(self.__module_address, self.__address, NBusCommand.CMD_SET_DATA, bytearray(binary_data))
 
-            self.data_format = NbusCommonParser.format_from_response(response)
-            return self.data_format
+        return resp

nbus_api/nbus_slave_module.py

@@ -9,6 +9,10 @@ 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
@@ -42,7 +46,6 @@ class NBusSlaveModule:
                                                 Module Get Commands
     ================================================================================================================
     """
-
     def cmd_get_echo(self, message: bytearray) -> bool:
         """
         Send Echo Command.
@@ -64,61 +67,76 @@ class NBusSlaveModule:
 
         return param_id, param_val
 
-    def cmd_get_all_params(self) -> list[Tuple[NBusParameterID, NBusParameterValue]]:
+    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)
 
-        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
+        return self.__params.copy()
 
-    def cmd_get_sensor_cnt(self) -> Annotated[int, Is[lambda value: 0 < value < 64]]:
+    def cmd_get_sensor_cnt(self) -> NBusSensorCount:
         _, *response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_SENSOR_CNT, bytearray([]))
-        return response[0]
+        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 = []
+        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))
+            data[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]))
+    def cmd_get_info(self):
+        response = self.__port.request_module(self.__module_addr, NBusCommand.CMD_GET_INFO, bytearray([]))
 
-        if parameter == NBusInfoParam.INFO_FORMAT:
-            begin_idx = 0
-            formats = []
+        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]
 
-            while begin_idx < resp_length:
-                device_id = response[begin_idx]
-                device_format = NbusCommonParser.format_from_response(response[begin_idx:begin_idx + 4])
+        return NBusInfo(module_name=name, module_type=typ, uuid=uuid, hw=hw, fw=fw, memory_id=mem_id)
 
-                self.__devices[device_id].data_format = device_format
-                formats.append((device_id, device_format))
-                begin_idx += 4
 
-            return formats
+    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):
-        pass
+        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
 
 
     """
@@ -127,12 +145,19 @@ class NBusSlaveModule:
     ================================================================================================================
     """
 
+    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([]))
-
-
+        self.__port.request_module(self.__module_addr, NBusCommand.CMD_SET_START, bytearray([]))
 
 

nbus_types/nbus_command_type.py

@@ -7,32 +7,24 @@ class NBusCommand(Enum):
     """
     # get
     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
-
+    CMD_GET_FORMAT = 0x0F
+    
     # set
+    CMD_SET_FIND = 0x20
     CMD_SET_STOP = 0x22
     CMD_SET_START = 0x23
     CMD_SET_PARAM = 0x24
+    CMD_SET_SLEEP = 0x26
+    CMD_SET_WAKEUP = 0x27
     CMD_SET_CALIBRATE = 0x28
+    CMD_SET_RESET = 0x29
     CMD_SET_STORE = 0x2A
     CMD_SET_DATA = 0x2B
-
-
-class NBusInfoParam(Enum):
-    """
-    Enum class for info command parameters.
-    """
-    INFO_GENERAL = 0xE1
-    INFO_FORMAT = 0xE2
+    CMD_SET_SYNC = 0x2C

nbus_types/nbus_exceptions/nbus_node_exception.py

@@ -29,6 +29,8 @@ class NBusErrorNodeType(Enum):
     DEVICE_BUSY = 0x06
     DEVICE_NOT_READY = 0x07
     PARAM_NOT_IMPLEMENTED = 0x10
+    ILLEGAL_FUNCTION_PARAM = 0x11
+    DEVICE_IS_READ_ONLY = 0x12
     GENERIC_SUBSLAVE_ERROR = 0x13
     ERR_SUBSLAVE_INIT_FAIL = 0x14
     SUBSLAVE_CUSTOM_ERR_1 = 0x1A

nbus_types/nbus_info_type.py

@@ -0,0 +1,34 @@
+from beartype import beartype
+from beartype.vale import Is
+from dataclasses import dataclass
+from typing import Annotated
+import numpy as np
+
+
+@beartype
+@dataclass(frozen=True)
+class NBusInfo:
+    """
+    Class for data format.
+
+    :ivar module_name: name of module
+    :ivar module_type: type of module
+    :ivar uuid: unique ID, generated by STM
+    :ivar hw: firmware version. MAJOR.MINOR
+    :ivar fw: memory ID number
+    :ivar memory_id: sensor count
+    """
+    module_name: Annotated[str, Is[lambda value: len(value) == 8]]
+    module_type: Annotated[str, Is[lambda value: len(value) == 3]]
+    uuid: Annotated[int, Is[lambda value: 0 <= value <= np.iinfo(np.uint32).max]]
+    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]]
+
+
+
+
+
+
+
+

nbus_types/nbus_parameter_type.py

@@ -20,3 +20,5 @@ class NBusParameterID(Enum):
     PARAM_RANGE = 5
     PARAM_RANGE0 = 6
     PARAM_FILTER = 7
+
+

nbus_types/nbus_sensor_count_type.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass
+from typing import Annotated
+
+from beartype import beartype
+from beartype.vale import Is
+
+
+@beartype
+@dataclass(frozen=True)
+class NBusSensorCount:
+    """
+    Class for sensor count type.
+
+    :ivar read_only_count: count of read-only devices
+    :ivar read_write_count: count of read-write devices
+    """
+    read_only_count: Annotated[int, Is[lambda value: 0 <= value <= 31]]
+    read_write_count: Annotated[int, Is[lambda value: 0 <= value <= 31]]

nbus_types/nbus_sensor_type.py

@@ -9,9 +9,10 @@ class NBusSensorType(Enum):
     TYPE_ACCELEROMETER = 0
     TYPE_GYROSCOPE = 1
     TYPE_MAGNETOMETER = 2
-    TYPE_TEMPERATURE = 3
-    TYPE_HUMIDITY = 4
-    TYPE_PRESSURE = 5
-    TYPE_HEART_RATE = 6
-    TYPE_DEVIATION_DISTANCE = 7
-    
+    TYPE_THERMOMETER = 3
+    TYPE_HYGROMETER = 4
+    TYPE_PRESSURE_GAUGE = 5
+    TYPE_HEART_RATE_MONITOR = 6
+    TYPE_LENGTH_GAUGE = 7
+    TYPE_LED_CONTROLLER = 8
+    TYPE_MOTOR_CONTROLLER = 9

nbus_types/nbus_status_type.py

@@ -0,0 +1,10 @@
+from enum import Enum
+
+
+class NBusStatusType(Enum):
+    """
+    Enum class for status type.
+    """
+    STATUS_NOT_SUPPORTED = 0xFF
+    STATUS_SUCCESS = 0
+    STATUS_FAIL = 1