/usr/share/pyshared/pymodbus/client/common.py is in python-pymodbus 1.2.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | '''
Modbus Client Common
----------------------------------
This is a common client mixin that can be used by
both the synchronous and asynchronous clients to
simplify the interface.
'''
from pymodbus.bit_read_message import *
from pymodbus.bit_write_message import *
from pymodbus.register_read_message import *
from pymodbus.register_write_message import *
from pymodbus.diag_message import *
from pymodbus.file_message import *
from pymodbus.other_message import *
class ModbusClientMixin(object):
'''
This is a modbus client mixin that provides additional factory
methods for all the current modbus methods. This can be used
instead of the normal pattern of::
# instead of this
client = ModbusClient(...)
request = ReadCoilsRequest(1,10)
response = client.execute(request)
# now like this
client = ModbusClient(...)
response = client.read_coils(1, 10)
'''
def read_coils(self, address, count=1, **kwargs):
'''
:param address: The starting address to read from
:param count: The number of coils to read
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = ReadCoilsRequest(address, count, **kwargs)
return self.execute(request)
def read_discrete_inputs(self, address, count=1, **kwargs):
'''
:param address: The starting address to read from
:param count: The number of discretes to read
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = ReadDiscreteInputsRequest(address, count, **kwargs)
return self.execute(request)
def write_coil(self, address, value, **kwargs):
'''
:param address: The starting address to write to
:param value: The value to write to the specified address
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = WriteSingleCoilRequest(address, value, **kwargs)
return self.execute(request)
def write_coils(self, address, values, **kwargs):
'''
:param address: The starting address to write to
:param values: The values to write to the specified address
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = WriteMultipleCoilsRequest(address, values, **kwargs)
return self.execute(request)
def write_register(self, address, value, **kwargs):
'''
:param address: The starting address to write to
:param value: The value to write to the specified address
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = WriteSingleRegisterRequest(address, value, **kwargs)
return self.execute(request)
def write_registers(self, address, values, **kwargs):
'''
:param address: The starting address to write to
:param values: The values to write to the specified address
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = WriteMultipleRegistersRequest(address, values, **kwargs)
return self.execute(request)
def read_holding_registers(self, address, count=1, **kwargs):
'''
:param address: The starting address to read from
:param count: The number of registers to read
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = ReadHoldingRegistersRequest(address, count, **kwargs)
return self.execute(request)
def read_input_registers(self, address, count=1, **kwargs):
'''
:param address: The starting address to read from
:param count: The number of registers to read
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = ReadInputRegistersRequest(address, count, **kwargs)
return self.execute(request)
def readwrite_registers(self, *args, **kwargs):
'''
:param read_address: The address to start reading from
:param read_count: The number of registers to read from address
:param write_address: The address to start writing to
:param write_registers: The registers to write to the specified address
:param unit: The slave unit this request is targeting
:returns: A deferred response handle
'''
request = ReadWriteMultipleRegistersRequest(*args, **kwargs)
return self.execute(request)
#---------------------------------------------------------------------------#
# Exported symbols
#---------------------------------------------------------------------------#
__all__ = [ 'ModbusClientMixin' ]
|