/usr/lib/python2.7/dist-packages/pymodbus/pdu.py is in python-pymodbus 1.3.2-1.
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 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 | '''
Contains base classes for modbus request/response/error packets
'''
from pymodbus.interfaces import Singleton
from pymodbus.exceptions import NotImplementedException
from pymodbus.constants import Defaults
from pymodbus.utilities import rtuFrameSize
from pymodbus.compat import iteritems, int2byte, byte2int
#---------------------------------------------------------------------------#
# Logging
#---------------------------------------------------------------------------#
import logging
_logger = logging.getLogger(__name__)
#---------------------------------------------------------------------------#
# Base PDU's
#---------------------------------------------------------------------------#
class ModbusPDU(object):
'''
Base class for all Modbus mesages
.. attribute:: transaction_id
This value is used to uniquely identify a request
response pair. It can be implemented as a simple counter
.. attribute:: protocol_id
This is a constant set at 0 to indicate Modbus. It is
put here for ease of expansion.
.. attribute:: unit_id
This is used to route the request to the correct child. In
the TCP modbus, it is used for routing (or not used at all. However,
for the serial versions, it is used to specify which child to perform
the requests against. The value 0x00 represents the broadcast address
(also 0xff).
.. attribute:: check
This is used for LRC/CRC in the serial modbus protocols
.. attribute:: skip_encode
This is used when the message payload has already been encoded.
Generally this will occur when the PayloadBuilder is being used
to create a complicated message. By setting this to True, the
request will pass the currently encoded message through instead
of encoding it again.
'''
def __init__(self, **kwargs):
''' Initializes the base data for a modbus request '''
self.transaction_id = kwargs.get('transaction', Defaults.TransactionId)
self.protocol_id = kwargs.get('protocol', Defaults.ProtocolId)
self.unit_id = kwargs.get('unit', Defaults.UnitId)
self.skip_encode = kwargs.get('skip_encode', False)
self.check = 0x0000
def encode(self):
''' Encodes the message
:raises: A not implemented exception
'''
raise NotImplementedException()
def decode(self, data):
''' Decodes data part of the message.
:param data: is a string object
:raises: A not implemented exception
'''
raise NotImplementedException()
@classmethod
def calculateRtuFrameSize(cls, buffer):
''' Calculates the size of a PDU.
:param buffer: A buffer containing the data that have been received.
:returns: The number of bytes in the PDU.
'''
if hasattr(cls, '_rtu_frame_size'):
return cls._rtu_frame_size
elif hasattr(cls, '_rtu_byte_count_pos'):
return rtuFrameSize(buffer, cls._rtu_byte_count_pos)
else: raise NotImplementedException(
"Cannot determine RTU frame size for %s" % cls.__name__)
class ModbusRequest(ModbusPDU):
''' Base class for a modbus request PDU '''
def __init__(self, **kwargs):
''' Proxy to the lower level initializer '''
ModbusPDU.__init__(self, **kwargs)
def doException(self, exception):
''' Builds an error response based on the function
:param exception: The exception to return
:raises: An exception response
'''
_logger.error("Exception Response F(%d) E(%d)" %
(self.function_code, exception))
return ExceptionResponse(self.function_code, exception)
class ModbusResponse(ModbusPDU):
''' Base class for a modbus response PDU
.. attribute:: should_respond
A flag that indicates if this response returns a result back
to the client issuing the request
.. attribute:: _rtu_frame_size
Indicates the size of the modbus rtu response used for
calculating how much to read.
'''
should_respond = True
def __init__(self, **kwargs):
''' Proxy to the lower level initializer '''
ModbusPDU.__init__(self, **kwargs)
#---------------------------------------------------------------------------#
# Exception PDU's
#---------------------------------------------------------------------------#
class ModbusExceptions(Singleton):
'''
An enumeration of the valid modbus exceptions
'''
IllegalFunction = 0x01
IllegalAddress = 0x02
IllegalValue = 0x03
SlaveFailure = 0x04
Acknowledge = 0x05
SlaveBusy = 0x06
MemoryParityError = 0x08
GatewayPathUnavailable = 0x0A
GatewayNoResponse = 0x0B
@classmethod
def decode(cls, code):
''' Given an error code, translate it to a
string error name.
:param code: The code number to translate
'''
values = dict((v, k) for k, v in iteritems(cls.__dict__)
if not k.startswith('__') and not callable(v))
return values.get(code, None)
class ExceptionResponse(ModbusResponse):
''' Base class for a modbus exception PDU '''
ExceptionOffset = 0x80
_rtu_frame_size = 5
def __init__(self, function_code, exception_code=None, **kwargs):
''' Initializes the modbus exception response
:param function_code: The function to build an exception response for
:param exception_code: The specific modbus exception to return
'''
ModbusResponse.__init__(self, **kwargs)
self.original_code = function_code
self.function_code = function_code | self.ExceptionOffset
self.exception_code = exception_code
def encode(self):
''' Encodes a modbus exception response
:returns: The encoded exception packet
'''
return int2byte(self.exception_code)
def decode(self, data):
''' Decodes a modbus exception response
:param data: The packet data to decode
'''
self.exception_code = byte2int(data[0])
def __str__(self):
''' Builds a representation of an exception response
:returns: The string representation of an exception response
'''
message = ModbusExceptions.decode(self.exception_code)
parameters = (self.function_code, self.original_code, message)
return "Exception Response(%d, %d, %s)" % parameters
class IllegalFunctionRequest(ModbusRequest):
'''
Defines the Modbus slave exception type 'Illegal Function'
This exception code is returned if the slave::
- does not implement the function code **or**
- is not in a state that allows it to process the function
'''
ErrorCode = 1
def __init__(self, function_code, **kwargs):
''' Initializes a IllegalFunctionRequest
:param function_code: The function we are erroring on
'''
ModbusRequest.__init__(self, **kwargs)
self.function_code = function_code
def decode(self, data):
''' This is here so this failure will run correctly
:param data: Not used
'''
pass
def execute(self, context):
''' Builds an illegal function request error response
:param context: The current context for the message
:returns: The error response packet
'''
return ExceptionResponse(self.function_code, self.ErrorCode)
#---------------------------------------------------------------------------#
# Exported symbols
#---------------------------------------------------------------------------#
__all__ = [
'ModbusRequest', 'ModbusResponse', 'ModbusExceptions',
'ExceptionResponse', 'IllegalFunctionRequest',
]
|