diff --git a/pymodbus/datastore/store.py b/pymodbus/datastore/store.py index f58d93b83..d27252bd1 100644 --- a/pymodbus/datastore/store.py +++ b/pymodbus/datastore/store.py @@ -48,7 +48,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from typing import Any, Iterable +from typing import Any, Dict, Generic, Iterable, TypeVar from pymodbus.exceptions import ParameterException @@ -56,7 +56,9 @@ # ---------------------------------------------------------------------------# # Datablock Storage # ---------------------------------------------------------------------------# -class BaseModbusDataBlock(ABC): + +V = TypeVar('V', list, Dict[int, Any]) +class BaseModbusDataBlock(ABC, Generic[V]): """Base class for a modbus datastore. Derived classes must create the following fields: @@ -71,7 +73,7 @@ class BaseModbusDataBlock(ABC): reset(self) """ - values: dict | list + values: V address: int default_value: Any @@ -119,7 +121,7 @@ def __iter__(self): return enumerate(self.values, self.address) -class ModbusSequentialDataBlock(BaseModbusDataBlock): +class ModbusSequentialDataBlock(BaseModbusDataBlock[list]): """Creates a sequential modbus datastore.""" def __init__(self, address, values): @@ -192,42 +194,43 @@ def setValues(self, address, values): self.values[start : start + len(values)] = values -class ModbusSparseDataBlock(BaseModbusDataBlock): - """Create a sparse modbus datastore. +class ModbusSparseDataBlock(BaseModbusDataBlock[Dict[int, Any]]): + """A sparse modbus datastore. E.g Usage. sparse = ModbusSparseDataBlock({10: [3, 5, 6, 8], 30: 1, 40: [0]*20}) - This would create a datablock with 3 blocks starting at - offset 10 with length 4 , 30 with length 1 and 40 with length 20 + This would create a datablock with 3 blocks + One starts at offset 10 with length 4, one at 30 with length 1, and one at 40 with length 20 sparse = ModbusSparseDataBlock([10]*100) Creates a sparse datablock of length 100 starting at offset 0 and default value of 10 - sparse = ModbusSparseDataBlock() --> Create Empty datablock + sparse = ModbusSparseDataBlock() --> Create empty datablock sparse.setValues(0, [10]*10) --> Add block 1 at offset 0 with length 10 (default value 10) sparse.setValues(30, [20]*5) --> Add block 2 at offset 30 with length 5 (default value 20) - if mutable is set to True during initialization, the datablock can not be altered with - setValues (new datablocks can not be added) + Unless 'mutable' is set to True during initialization, the datablock cannot be altered with + setValues (new datablocks cannot be added) """ def __init__(self, values=None, mutable=True): """Initialize a sparse datastore. - Will only answer to addresses - registered, either initially here, or later via setValues() + Will only answer to addresses registered, + either initially here, or later via setValues() :param values: Either a list or a dictionary of values - :param mutable: The data-block can be altered later with setValues(i.e add more blocks) + :param mutable: Whether the data-block can be altered later with setValues (i.e add more blocks) + + If values is a list, a sequential datablock will be created. - If values are list , This is as good as sequential datablock. - Values as dictionary should be in {offset: } format, if values - is a list, a sparse datablock is created starting at offset with the length of values. - If values is a integer, then the value is set for the corresponding offset. + If values is a dictionary, it should be in {offset: } format + For each list, a sparse datablock is created, starting at 'offset' with the length of the list + For each integer, the value is set for the corresponding offset. """ - self.values: dict[int, Any] = {} + self.values = {} self._process_values(values) self.mutable = mutable self.default_value = self.values.copy()