/usr/lib/python3/dist-packages/pymodbus/events.py is in python3-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 | '''
Modbus Remote Events
------------------------------------------------------------
An event byte returned by the Get Communications Event Log function
can be any one of four types. The type is defined by bit 7
(the high-order bit) in each byte. It may be further defined by bit 6.
'''
from pymodbus.exceptions import NotImplementedException
from pymodbus.exceptions import ParameterException
from pymodbus.utilities import pack_bitstring, unpack_bitstring
class ModbusEvent(object):
def encode(self):
''' Encodes the status bits to an event message
:returns: The encoded event message
'''
raise NotImplementedException()
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
raise NotImplementedException()
class RemoteReceiveEvent(ModbusEvent):
''' Remote device MODBUS Receive Event
The remote device stores this type of event byte when a query message
is received. It is stored before the remote device processes the message.
This event is defined by bit 7 set to logic '1'. The other bits will be
set to a logic '1' if the corresponding condition is TRUE. The bit layout
is::
Bit Contents
----------------------------------
0 Not Used
2 Not Used
3 Not Used
4 Character Overrun
5 Currently in Listen Only Mode
6 Broadcast Receive
7 1
'''
def __init__(self, **kwargs):
''' Initialize a new event instance
'''
self.overrun = kwargs.get('overrun', False)
self.listen = kwargs.get('listen', False)
self.broadcast = kwargs.get('broadcast', False)
def encode(self):
''' Encodes the status bits to an event message
:returns: The encoded event message
'''
bits = [False] * 3
bits += [self.overrun, self.listen, self.broadcast, True]
packet = pack_bitstring(bits)
return packet
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
bits = unpack_bitstring(event)
self.overrun = bits[4]
self.listen = bits[5]
self.broadcast = bits[6]
class RemoteSendEvent(ModbusEvent):
''' Remote device MODBUS Send Event
The remote device stores this type of event byte when it finishes
processing a request message. It is stored if the remote device
returned a normal or exception response, or no response.
This event is defined by bit 7 set to a logic '0', with bit 6 set to a '1'.
The other bits will be set to a logic '1' if the corresponding
condition is TRUE. The bit layout is::
Bit Contents
-----------------------------------------------------------
0 Read Exception Sent (Exception Codes 1-3)
1 Slave Abort Exception Sent (Exception Code 4)
2 Slave Busy Exception Sent (Exception Codes 5-6)
3 Slave Program NAK Exception Sent (Exception Code 7)
4 Write Timeout Error Occurred
5 Currently in Listen Only Mode
6 1
7 0
'''
def __init__(self, **kwargs):
''' Initialize a new event instance
'''
self.read = kwargs.get('read', False)
self.slave_abort = kwargs.get('slave_abort', False)
self.slave_busy = kwargs.get('slave_busy', False)
self.slave_nak = kwargs.get('slave_nak', False)
self.write_timeout = kwargs.get('write_timeout', False)
self.listen = kwargs.get('listen', False)
def encode(self):
''' Encodes the status bits to an event message
:returns: The encoded event message
'''
bits = [self.read, self.slave_abort, self.slave_busy,
self.slave_nak, self.write_timeout, self.listen]
bits += [True, False]
packet = pack_bitstring(bits)
return packet
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
# todo fix the start byte count
bits = unpack_bitstring(event)
self.read = bits[0]
self.slave_abort = bits[1]
self.slave_busy = bits[2]
self.slave_nak = bits[3]
self.write_timeout = bits[4]
self.listen = bits[5]
class EnteredListenModeEvent(ModbusEvent):
''' Remote device Entered Listen Only Mode
The remote device stores this type of event byte when it enters
the Listen Only Mode. The event is defined by a content of 04 hex.
'''
value = 0x04
__encoded = b'\x04'
def encode(self):
''' Encodes the status bits to an event message
:returns: The encoded event message
'''
return self.__encoded
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
if event != self.__encoded:
raise ParameterException('Invalid decoded value')
class CommunicationRestartEvent(ModbusEvent):
''' Remote device Initiated Communication Restart
The remote device stores this type of event byte when its communications
port is restarted. The remote device can be restarted by the Diagnostics
function (code 08), with sub-function Restart Communications Option
(code 00 01).
That function also places the remote device into a 'Continue on Error'
or 'Stop on Error' mode. If the remote device is placed into 'Continue on
Error' mode, the event byte is added to the existing event log. If the
remote device is placed into 'Stop on Error' mode, the byte is added to
the log and the rest of the log is cleared to zeros.
The event is defined by a content of zero.
'''
value = 0x00
__encoded = b'\x00'
def encode(self):
''' Encodes the status bits to an event message
:returns: The encoded event message
'''
return self.__encoded
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
if event != self.__encoded:
raise ParameterException('Invalid decoded value')
|