Merge pull request #3 from facts-engineering/signed-support

Add Server support for signed integers

examples/rtu_server.py

@@ -33,21 +33,26 @@
 mb_server = RTUServer(
     comm,
     unit_addr=1,
-    number_coils=32,
-    number_input_registers=255,
-    number_discrete_inputs=16,
+    number_coils=20,
+    number_input_registers=0xFF,
+    number_discrete_inputs=0x10,
     number_holding_registers=10,
 )
 
-mb_server.input_registers = list(range(255))  # set input register value to their address
+mb_server.input_registers[0:] = list(range(0xFF)) # set input registers 0-255 to 0-255
+
 mb_server.discrete_inputs[5] = True  # set input register 5 to True
+
+mb_server.holding_registers.signed[1] = True # set holding register 1 to use 16-bit signed values
+
 count = 0
 
 while True:
 
     mb_server.poll(timeout=.5) # Regularly poll the modbus server to handle incoming requests
     mb_server.discrete_inputs[0] = switch.value  # set discrete input 0 to switch value
     mb_server.holding_registers[0] = count  # set holding register 0 to count value
+    mb_server.holding_registers[1] = -count  # set holding register 1 to negative count value
     led.value = mb_server.coils[0]  # set led to output value
 
     count += 1

examples/tcp_server.py

@@ -4,14 +4,15 @@
     This example shows how to configure a Modbus TCP server.
 
 	Written by FACTS Engineering
-	Copyright (c) 2021 FACTS Engineering, LLC
+	Copyright (c) 2023 FACTS Engineering, LLC
 	Licensed under the MIT license.
 
 """
 
 import board
 import busio
 import digitalio
+import p1am_200_helpers as helpers # For P1AM-ETH
 from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K
 import adafruit_wiznet5k.adafruit_wiznet5k_socket as socket
 from uModBus.tcp import TCPServer
@@ -21,9 +22,13 @@
 led.switch_to_output()
 switch = digitalio.DigitalInOut(board.SWITCH)
 
-cs = digitalio.DigitalInOut(board.D5)
-spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO)
-eth = WIZNET5K(spi_bus, cs, is_dhcp=False)
+# For P1AM-ETH
+eth = helpers.get_ethernet(False) # DHCP False
+
+# For generic ethernet
+# cs = digitalio.DigitalInOut(board.D5)
+# spi_bus = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO)
+# eth = WIZNET5K(spi_bus, cs, is_dhcp=False)
 
 IP_ADDRESS = (192, 168, 1, 177)
 SUBNET_MASK = (255, 255, 248, 0)
@@ -42,8 +47,12 @@
     number_holding_registers=10,
 )
 
-mb_server.input_registers = list(range(0xFF))
+mb_server.input_registers[0:] = list(range(0xFF)) # set input registers 0-255 to 0-255
+
 mb_server.discrete_inputs[5] = True
+
+mb_server.holding_registers.signed[1] = True # set holding register 1 to use 16-bit signed values
+
 count = 0
 
 while True:
@@ -54,6 +63,7 @@
         pass # Ignore errors in case the client disconnects mid-poll
     mb_server.discrete_inputs[0] = switch.value  # set discrete input 0 to switch value
     mb_server.holding_registers[0] = count  # set holding register 0 to count value
+    mb_server.holding_registers[1] = -count  # set holding register 1 to count value
     led.value = mb_server.coils[0]  # set led to output value
 
     count += 1

uModbus/common.py

@@ -117,10 +117,10 @@ def __init__(self, unit_addr=None, *, number_coils=None, number_discrete_inputs=
             self.discrete_inputs = [0] * number_discrete_inputs
 
         if number_input_registers is not None:
-            self.input_registers = [0] * number_input_registers
+            self.input_registers = _ValueRegisters(number_input_registers)
 
         if number_holding_registers is not None:
-            self.holding_registers = [0] * number_holding_registers
+            self.holding_registers = _ValueRegisters(number_holding_registers)
 
 
     def handle_request(self, data):
@@ -148,9 +148,10 @@ def handle_request(self, data):
                 return
 
             if function_code == Const.READ_HOLDING_REGISTERS:
-                data = self.holding_registers[address:address+quantity]
+                data = self.holding_registers.raw[address:address+quantity]
             else:
-                data = self.input_registers[address:address+quantity]
+                data = self.input_registers.raw[address:address+quantity]
+            data = b''.join(data)
 
         elif function_code == Const.WRITE_SINGLE_COIL:
             quantity = None
@@ -170,7 +171,7 @@ def handle_request(self, data):
             if not self._within_limits(function_code, quantity, address):
                 self.send_exception(function_code, Const.ILLEGAL_DATA_ADDRESS)
                 return
-            self.holding_registers[address] = self.data_as_registers(data, 1)[0]
+            self.holding_registers.raw[address] = data
             # all values allowed
 
         elif function_code == Const.WRITE_MULTIPLE_COILS:
@@ -193,7 +194,7 @@ def handle_request(self, data):
             if len(data) != quantity * 2:
                 self.send_exception(function_code, Const.ILLEGAL_DATA_ADDRESS)
                 raise ModbusException(function_code, Const.ILLEGAL_DATA_VALUE, self)
-            self.holding_registers[address:address+quantity] = [i for i in self.data_as_registers(data, quantity)]
+            self.holding_registers.raw[address:address+quantity] = [data[i:i+2] for i in range(0, quantity*2, 2)]
 
         else:
             # Not implemented functions
@@ -229,14 +230,6 @@ def data_as_bits(self, data, quantity):
                 if len(bits) == quantity:
                     return bits
 
-    def data_as_registers(self, data, quantity, signed=False):
-        if quantity is not None:
-            qty = quantity  
-        else: 
-            qty = 1
-        fmt = ('h' if signed else 'H') * qty
-        return struct.unpack('>' + fmt, data)
-
     def _within_limits(self, function_code, quantity, address):
 
         if function_code == Const.READ_DISCRETE_INPUTS:
@@ -269,3 +262,62 @@ def __init__(self, function_code, exception_code, instance):
         instance.send_exception_response(instance.unit_addr, function_code, exception_code)
         self.function_code = function_code
         self.exception_code = exception_code
+
+
+class _ValueRegisters():
+    def __init__(self, length):
+        self.raw = [bytes(2)] * length
+        self.signed = [False] * length
+        self.byteswap = [False] * length
+
+    def __len__(self):
+        return len(self.raw)
+
+    def __setitem__(self, index, value):
+        if isinstance(index, int):
+            self._set_value(index, value)
+        elif isinstance(index, slice):
+            start = index.start
+            if start is None:
+                start = 0
+            end = index.stop
+            if end is None:
+                end = len(self)
+
+            for i in range(start, end):
+                self._set_value(i, value[i-start])
+
+        else:
+            raise TypeError('Index must be an integer or slice')
+
+
+    def __getitem__(self, index):
+        if isinstance(index, int):
+            return self._get_value(index)
+        elif isinstance(index, slice):
+            start = index.start
+            if start is None:
+                start = 0
+            end = index.stop
+            if end is None:
+                end = len(self)
+
+            return [self._get_value(i) for i in range(start, end)]
+        else:
+            raise TypeError('Index must be an integer or slice')
+
+    def _set_value(self, index, value):
+        format = '<' if self.byteswap[index] else '>'
+        format += 'h' if self.signed[index] else 'H'
+
+        try:
+            self.raw[index] = struct.pack(format, value)
+        except OverflowError:
+            raise OverflowError(f'Address {index} value {value} must be between {((-32768 if self.signed[index] else 0))} and {(32767 if self.signed[index] else 65535)}')
+
+
+    def _get_value(self, index):
+        format = '<' if self.byteswap[index] else '>'
+        format += 'h' if self.signed[index] else 'H'
+
+        return struct.unpack(format, self.raw[index])[0]

uModbus/functions.py

@@ -97,16 +97,7 @@ def response(function_code, request_register_addr, request_register_qty, request
         return struct.pack('>BB' + fmt, function_code, ((len(value_list) - 1) // 8) + 1, *output_value)
 
     elif function_code in [Const.READ_HOLDING_REGISTERS, Const.READ_INPUT_REGISTER]:
-        quantity = len(value_list)
-
-        if signed == True or signed == False:
-            fmt = ('h' if signed else 'H') * quantity
-        else:
-            fmt = ''
-            for s in signed:
-                fmt += 'h' if s else 'H'
-
-        return struct.pack('>BB' + fmt, function_code, quantity * 2, *value_list)
+        return struct.pack('>BB', function_code, len(value_list)) + value_list
 
     elif function_code in [Const.WRITE_SINGLE_COIL, Const.WRITE_SINGLE_REGISTER]:
         return struct.pack('>BHBB', function_code, request_register_addr, *request_data)