temp.

Author: janiversen

pymodbus/datastore/simulator.py

@@ -1,8 +1,10 @@
 """Pymodbus ModbusSimulatorContext."""
 import dataclasses
 import logging
+import random
 import struct
 import sys
+from datetime import datetime
 from typing import Callable, Dict
 
 
@@ -122,39 +124,45 @@ def __init__(self):
             },
         }
 
-    def handle_type_bits(self, registers, start, stop, value, action):
+    def handle_type_bits(self, registers, reg_count, start, stop, value, action):
         """Handle type bits.
 
         :meta private:
         """
         for i in range(start, stop):
+            if i >= reg_count:
+                raise RuntimeError(f"Error section \"{Label.type_bits}\" addr {start},  {stop} out of range")
             if registers[i].type != CELL_TYPE_NONE:
                 txt = f'ERROR Configuration invalid in section "{Label.type_bits}" register {i} already defined'
                 raise RuntimeError(txt)
             registers[i].value = value
             registers[i].type = CELL_TYPE_BIT
             registers[i].action = action
 
-    def handle_type_uint16(self, registers, start, stop, value, action):
+    def handle_type_uint16(self, registers, reg_count, start, stop, value, action):
         """Handle type uint16.
 
         :meta private:
         """
         for i in range(start, stop):
+            if i >= reg_count:
+                raise RuntimeError(f"Error section \"{Label.type_uint16}\" addr {start},  {stop} out of range")
             if registers[i].type != CELL_TYPE_NONE:
                 txt = f'ERROR Configuration invalid in section "{Label.type_uint16}" register {i} already defined'
                 raise RuntimeError(txt)
             registers[i].value = value
             registers[i].type = CELL_TYPE_UINT16
             registers[i].action = action
 
-    def handle_type_uint32(self, registers, start, stop, value, action):
+    def handle_type_uint32(self, registers, reg_count, start, stop, value, action):
         """Handle type uint32.
 
         :meta private:
         """
         regs = ModbusSimulatorContext.build_registers_from_value(value, True)
         for i in range(start, stop, 2):
+            if i+1 >= reg_count:
+                raise RuntimeError(f"Error section \"{Label.type_uint32}\" addr {start},  {stop} out of range")
             if registers[i].type != CELL_TYPE_NONE:
                 txt = f'ERROR Configuration invalid in section "{Label.type_uint32}" register {i} already defined'
                 raise RuntimeError(txt)
@@ -168,13 +176,15 @@ def handle_type_uint32(self, registers, start, stop, value, action):
             registers[j].value = regs[1]
             registers[j].type = CELL_TYPE_NEXT
 
-    def handle_type_float32(self, registers, start, stop, value, action):
+    def handle_type_float32(self, registers, reg_count, start, stop, value, action):
         """Handle type uint32.
 
         :meta private:
         """
         regs = ModbusSimulatorContext.build_registers_from_value(value, False)
         for i in range(start, stop, 2):
+            if i+1 >= reg_count:
+                raise RuntimeError(f"Error section \"{Label.type_float32}\" addr {start},  {stop} out of range")
             if registers[i].type != CELL_TYPE_NONE:
                 txt = f'ERROR Configuration invalid in section "{Label.type_float32}" register {i} already defined'
                 raise RuntimeError(txt)
@@ -188,7 +198,7 @@ def handle_type_float32(self, registers, start, stop, value, action):
             registers[j].value = regs[1]
             registers[j].type = CELL_TYPE_NEXT
 
-    def handle_type_string(self, registers, start, stop, value, action):
+    def handle_type_string(self, registers, reg_count, start, stop, value, action):
         """Handle type string.
 
         :meta private:
@@ -201,6 +211,8 @@ def handle_type_string(self, registers, start, stop, value, action):
             value = value.ljust(reg_len)
         for i in range(stop - start):
             inx = start + i
+            if i+1 >= reg_count:
+                raise RuntimeError(f"Error section \"{Label.type_start}\" addr {start},  {stop} out of range")
             if registers[inx].type != CELL_TYPE_NONE:
                 txt = f'ERROR Configuration invalid in section "{Label.type_string}" register {inx} already defined'
                 raise RuntimeError(txt)
@@ -261,26 +273,30 @@ def handle_setup_section(self, config, actions):
             entry[Label.action] = action
         return registers, offset, type_exception
 
-    def handle_invalid_address(self, registers, config):
+    def handle_invalid_address(self, registers, reg_count, config):
         """Handle invalid address"""
         for entry in Label.try_get(Label.invalid, config):
             if isinstance(entry, int):
                 entry = [entry, entry]
             for i in range(entry[0], entry[1] + 1):
+                if i >= reg_count:
+                    raise RuntimeError(f"Error section \"{Label.invalid}\" addr {entry} out of range")
                 if registers[i].type != CELL_TYPE_NONE:
                     txt = f'ERROR Configuration invalid in section "invalid" register {i} already defined'
                     raise RuntimeError(txt)
                 registers[i].type = CELL_TYPE_ILLEGAL
 
-    def handle_write_allowed(self, registers, config):
+    def handle_write_allowed(self, registers, reg_count, config):
         """Handle write allowed"""
         for entry in Label.try_get(Label.write, config):
             if isinstance(entry, int):
                 entry = [entry, entry]
             for i in range(entry[0], entry[1] + 1):
+                if i >= reg_count:
+                    raise RuntimeError(f"Error section \"{Label.write}\" addr {entry} out of range")
                 registers[i].access = True
 
-    def handle_types(self, registers, actions, config):
+    def handle_types(self, registers, actions, reg_count, config):
         """Handle the different types"""
         for section, type_entry in self.config_types.items():
             layout = Label.try_get(section, config)
@@ -291,13 +307,14 @@ def handle_types(self, registers, actions, config):
                     entry[Label.addr] = [entry[Label.addr], entry[Label.addr]]
                 type_entry[Label.method](
                     registers,
+                    reg_count,
                     entry[Label.addr][0],
                     entry[Label.addr][1] + 1,
                     entry.get(Label.value, type_entry[Label.value]),
                     actions[entry.get("action", type_entry[Label.action])],
                 )
 
-    def handle_repeat(self, registers, config):
+    def handle_repeat(self, registers, reg_count, config):
         """Handle repeat.
 
         :meta private:
@@ -310,6 +327,8 @@ def handle_repeat(self, registers, config):
             addr_to = Label.try_get(Label.repeat_to, entry)
             for inx in range(addr_to[0], addr_to[1] + 1):
                 copy_inx = copy_start if copy_inx >= copy_end else copy_inx + 1
+                if inx >= reg_count:
+                    raise RuntimeError(f"Error section \"{Label.repeat}\" entry {entry} out of range")
                 registers[inx] = dataclasses.replace(registers[copy_inx])
 
     def setup(self, config, actions, custom_actions) -> None:
@@ -323,11 +342,11 @@ def setup(self, config, actions, custom_actions) -> None:
             actions.update(custom_actions)
 
         registers, offset, typ_exc = self.handle_setup_section(config, actions)
-        self.handle_invalid_address(registers, config)
-        self.handle_write_allowed(registers, config)
-        self.handle_types(registers, actions, config)
-        self.handle_repeat(registers, config)
         reg_count = len(registers)
+        self.handle_invalid_address(registers, reg_count, config)
+        self.handle_write_allowed(registers, reg_count, config)
+        self.handle_types(registers, actions, reg_count, config)
+        self.handle_repeat(registers, reg_count, config)
         for i in range(reg_count):
             if registers[i].type == CELL_TYPE_NONE:
                 registers[i].type = CELL_TYPE_ILLEGAL
@@ -484,7 +503,7 @@ def getValues(self, func_code, address, count=1):  # pylint: disable=invalid-nam
             for i in range(real_address, real_address + count):
                 reg = self.registers[i]
                 if reg.action:
-                    reg.action(i)
+                    reg.action(i, reg)
                 result.append(reg.value)
         else:
             # bit access
@@ -494,7 +513,7 @@ def getValues(self, func_code, address, count=1):  # pylint: disable=invalid-nam
             for i in range(real_address, real_address + reg_count):
                 reg = self.registers[i]
                 if reg.action:
-                    reg.action(i)
+                    reg.action(i, reg)
                 while count and bit_index < 16:
                     result.append(bool(reg.value & (2**bit_index)))
                     count -= 1
@@ -533,21 +552,66 @@ def setValues(self, func_code, address, values):  # pylint: disable=invalid-name
     # Internal action methods
     # --------------------------------------------
 
-    @classmethod
-    def action_random(cls, inx):
-        """Update with random value."""
+    def action_random(self, inx, cell):
+        """Update with random value.
 
-    @classmethod
-    def action_increment(cls, inx):
-        """Increment value reset with overflow."""
+        :meta private:
+        """
+        if cell.type == CELL_TYPE_BIT:
+            self.registers[inx].value = random.randint(0, 65536)
+        elif cell.type == CELL_TYPE_FLOAT32:
+            regs = self.build_registers_from_value(random.uniform(0.0, 100.0), False)
+            self.registers[inx].value = regs[0]
+            self.registers[inx+1].value = regs[1]
+        elif cell.type == CELL_TYPE_UINT16:
+            self.registers[inx].value = random.randint(0, 65536)
+        elif cell.type == CELL_TYPE_UINT32:
+            regs = self.build_registers_from_value(random.uniform(0.0, 100.0), True)
+            self.registers[inx].value = regs[0]
+            self.registers[inx+1].value = regs[1]
+
+    def action_increment(self, inx, cell):
+        """Increment value reset with overflow.
 
-    @classmethod
-    def action_timestamp(cls, inx):
-        """Set current time."""
+        :meta private:
+        """
+        if cell.type == CELL_TYPE_BIT:
+            self.registers[inx].value += 1
+        elif cell.type == CELL_TYPE_FLOAT32:
+            value = self.build_value_from_registers(self.registers[inx : inx+2])
+            value += 1.0
+            regs = self.build_registers_from_value(value, False)
+            self.registers[inx].value = regs[0]
+            self.registers[inx+1].value = regs[1]
+        elif cell.type == CELL_TYPE_UINT16:
+            self.registers[inx].value += 1
+        elif cell.type == CELL_TYPE_UINT32:
+            value = self.build_value_from_registers(self.registers[inx : inx+2])
+            value += 1
+            regs = self.build_registers_from_value(value, True)
+            self.registers[inx].value = regs[0]
+            self.registers[inx+1].value = regs[1]
+
+    def action_timestamp(self, inx, _cell):
+        """Set current time.
 
-    @classmethod
-    def action_reset(cls, inx):
-        """Reboot server."""
+        :meta private:
+        """
+        x = datetime.now()
+        self.registers[inx].value = x.tm_year - 1900
+        self.registers[inx+1].value = x.tm_mon
+        self.registers[inx+2].value = x.tm_mday
+        self.registers[inx+3].value = x.tm_wday
+        self.registers[inx+4].value = x.tm_hour
+        self.registers[inx+5].value = x.tm_min
+        self.registers[inx+1].value = x.tm_sec
+
+    def action_reset(self, _inx, _cell):
+        """Reboot server.
+
+        :meta private:
+        """
+        raise RuntimeError("RESET server")
 
     # --------------------------------------------
     # Internal helper methods