Datastore
Datastore is responsible for managing registers for a server.
Datastore classes
- class pymodbus.datastore.ModbusSparseDataBlock(values=None, mutable=True)
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 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.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)
Unless ‘mutable’ is set to True during initialization, the datablock cannot be altered with setValues (new datablocks cannot be added)
- classmethod create(values=None)
Create sparse datastore.
Use setValues to initialize registers.
- Parameters:
values – Either a list or a dictionary of values
- Returns:
An initialized datastore
- reset()
Reset the store to the initially provided defaults.
- validate(address, count=1)
Check to see if the request is in range.
- Parameters:
address – The starting address
count – The number of values to test for
- Returns:
True if the request in within range, False otherwise
- getValues(address, count=1)
Return the requested values of the datastore.
- Parameters:
address – The starting address
count – The number of values to retrieve
- Returns:
The requested values from a:a+c
- setValues(address, values, use_as_default=False)
Set the requested values of the datastore.
- Parameters:
address – The starting address
values – The new values to be set
use_as_default – Use the values as default
- Raises:
- class pymodbus.datastore.ModbusSlaveContext(*_args, **kwargs)
Create a modbus data model with data stored in a block.
- Parameters:
di – discrete inputs initializer ModbusDataBlock
co – coils initializer ModbusDataBlock
hr – holding register initializer ModbusDataBlock
ir – input registers initializer ModbusDataBlock
zero_mode –
Not add one to address
When True, a request for address zero to n will map to datastore address zero to n.
When False, a request for address zero to n will map to datastore address one to n+1, based on section 4.4 of specification.
Default is False.
- reset()
Reset all the datastores to their default values.
- validate(fc_as_hex, address, count=1)
Validate the request to make sure it is in range.
- Parameters:
fc_as_hex – The function we are working with
address – The starting address
count – The number of values to test
- Returns:
True if the request in within range, False otherwise
- getValues(fc_as_hex, address, count=1)
Get count values from datastore.
- Parameters:
fc_as_hex – The function we are working with
address – The starting address
count – The number of values to retrieve
- Returns:
The requested values from a:a+c
- setValues(fc_as_hex, address, values)
Set the datastore with the supplied values.
- Parameters:
fc_as_hex – The function we are working with
address – The starting address
values – The new values to be set
- register(function_code, fc_as_hex, datablock=None)
Register a datablock with the slave context.
- Parameters:
function_code – function code (int)
fc_as_hex – string representation of function code (e.g “cf” )
datablock – datablock to associate with this function code
- class pymodbus.datastore.ModbusServerContext(slaves=None, single=True)
This represents a master collection of slave contexts.
If single is set to true, it will be treated as a single context so every slave_id returns the same context. If single is set to false, it will be interpreted as a collection of slave contexts.
- slaves()
Define slaves.
- class pymodbus.datastore.ModbusSimulatorContext(config: dict[str, Any], custom_actions: dict[str, Callable] | None)
Modbus simulator.
- Parameters:
config – A dict with structure as shown below.
actions – A dict with “<name>”: <function> structure.
- Raises:
RuntimeError – if json contains errors (msg explains what)
It builds and maintains a virtual copy of a device, with simulation of device specific functions.
The device is described in a dict, user supplied actions will be added to the builtin actions.
It is used in conjunction with a pymodbus server.
Example:
store = ModbusSimulatorContext(<config dict>, <actions dict>) StartAsyncTcpServer(<host>, context=store) Now the server will simulate the defined device with features like: - invalid addresses - write protected addresses - optional control of access for string, uint32, bit/bits - builtin actions for e.g. reset/datetime, value increment by read - custom actions
Description of the json file or dict to be supplied:
{ "setup": { "di size": 0, --> Size of discrete input block (8 bit) "co size": 0, --> Size of coils block (8 bit) "ir size": 0, --> Size of input registers block (16 bit) "hr size": 0, --> Size of holding registers block (16 bit) "shared blocks": True, --> share memory for all blocks (largest size wins) "defaults": { "value": { --> Initial values (can be overwritten) "bits": 0x01, "uint16": 122, "uint32": 67000, "float32": 127.4, "string": " ", }, "action": { --> default action (can be overwritten) "bits": None, "uint16": None, "uint32": None, "float32": None, "string": None, }, }, "type exception": False, --> return IO exception if read/write on non boundary }, "invalid": [ --> List of invalid addresses, IO exception returned 51, --> single register [78, 99], --> start, end registers, repeated as needed ], "write": [ --> allow write, efault is ReadOnly [5, 5] --> start, end bytes, repeated as needed ], "bits": [ --> Define bits (1 register == 2 bytes) [30, 31], --> start, end registers, repeated as needed {"addr": [32, 34], "value": 0xF1}, --> with value {"addr": [35, 36], "action": "increment"}, --> with action {"addr": [37, 38], "action": "increment", "value": 0xF1} --> with action and value {"addr": [37, 38], "action": "increment", "kwargs": {"min": 0, "max": 100}} --> with action with arguments ], "uint16": [ --> Define uint16 (1 register == 2 bytes) --> same as type_bits ], "uint32": [ --> Define 32 bit integers (2 registers == 4 bytes) --> same as type_bits ], "float32": [ --> Define 32 bit floats (2 registers == 4 bytes) --> same as type_bits ], "string": [ --> Define strings (variable number of registers (each 2 bytes)) [21, 22], --> start, end registers, define 1 string {"addr": 23, 25], "value": "ups"}, --> with value {"addr": 26, 27], "action": "user"}, --> with action {"addr": 28, 29], "action": "", "value": "user"} --> with action and value ], "repeat": [ --> allows to repeat section e.g. for n devices {"addr": [100, 200], "to": [50, 275]} --> Repeat registers 100-200 to 50+ until 275 ] }
- get_text_register(register)
Get raw register.
- classmethod build_registers_from_value(value, is_int)
Build registers from int32 or float32.
- classmethod build_value_from_registers(registers, is_int)
Build int32 or float32 value from registers.