Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _recv(self, size):
""" Reads data from the underlying descriptor
:param size: The number of bytes to read
:return: The bytes read
"""
raise NotImplementedException("Method not implemented by derived class")
def connect(self):
""" Connect to the modbus remote host
:returns: True if connection succeeded, False otherwise
"""
raise NotImplementedException("Method not implemented by derived class")
def addToFrame(self, message):
""" Add the next message to the frame buffer
This should be used before the decoding while loop to add the received
data to the buffer handle.
:param message: The most recent packet
"""
raise NotImplementedException(
"Method not implemented by derived class")
def decode(self, event):
''' Decodes the event message to its status bits
:param event: The event to decode
'''
raise NotImplementedException()
def lookupPduClass(self, function_code):
""" Use `function_code` to determine the class of the PDU.
:param function_code: The function code specified in a frame.
:returns: The class of the PDU that has a matching `function_code`.
"""
raise NotImplementedException(
"Method not implemented by derived class")
raise NotImplementedException(
"Unhandled class for register corruption; not implemented" )
elif self.what == "protocol":
message.protocol_id ^= 0xFFFF
packet = super( EvilFramerCorruptResponse, self ).buildPacket( message )
message.protocol_id ^= 0xFFFF
elif self.what == "unit":
message.unit_id ^= 0xFF
packet = super( EvilFramerCorruptResponse, self ).buildPacket( message )
message.unit_id ^= 0xFF
elif self.what == "function":
message.function_code ^= 0xFF
packet = super( EvilFramerCorruptResponse, self ).buildPacket( message )
message.function_code ^= 0xFF
else:
raise NotImplementedException(
"Unknown corruption specified; not implemented" )
except Exception:
log.info( "Could not build corrupt packet: %s", traceback.format_exc() )
return packet
def getValues(self, fx, address, count=1):
""" Get `count` values from datastore
:param fx: The function we are working with
:param address: The starting address
:param count: The number of values to retrieve
:returns: The requested values from a:a+c
"""
raise NotImplementedException("get context values")
def encode(self):
""" Encodes the message
:raises: A not implemented exception
"""
raise NotImplementedException()
def isFrameReady(self):
""" Check if we should continue decode logic
This is meant to be used in a while loop in the decoding phase to let
the decoder know that there is still data in the buffer.
:returns: True if ready, False otherwise
"""
raise NotImplementedException(
"Method not implemented by derived class")