Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve datastore documentation #2056

Merged
merged 5 commits into from
Feb 25, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 22 additions & 19 deletions pymodbus/datastore/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,15 +48,17 @@
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


# ---------------------------------------------------------------------------#
# 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:
Expand All @@ -71,7 +73,7 @@ class BaseModbusDataBlock(ABC):
reset(self)
"""

values: dict | list
values: V
address: int
default_value: Any

Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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: <values>} 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: <int | list>} 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()
Expand Down