sdo: Define constants for all standardized abort codes (#596)

* Define constants for all standardized SDO abort codes.
* Type annotations for constants (Final).
* Use constants in SDO client and server.
* Keep literals in tests, but comment and format.

Author: acolomb

canopen/sdo/client.py

@@ -96,11 +96,11 @@ def request_response(self, sdo_request):
             except SdoCommunicationError as e:
                 retries_left -= 1
                 if not retries_left:
-                    self.abort(0x0504_0000)
+                    self.abort(ABORT_TIMED_OUT)
                     raise
                 logger.warning(str(e))
 
-    def abort(self, abort_code=0x0800_0000):
+    def abort(self, abort_code=ABORT_GENERAL_ERROR):
         """Abort current transfer."""
         request = bytearray(8)
         request[0] = REQUEST_ABORTED
@@ -309,10 +309,10 @@ def read(self, size=-1):
         response = self.sdo_client.request_response(request)
         res_command, = struct.unpack_from("B", response)
         if res_command & 0xE0 != RESPONSE_SEGMENT_UPLOAD:
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         if res_command & TOGGLE_BIT != self._toggle:
-            self.sdo_client.abort(0x0503_0000)
+            self.sdo_client.abort(ABORT_TOGGLE_NOT_ALTERNATED)
             raise SdoCommunicationError("Toggle bit mismatch")
         length = 7 - ((res_command >> 1) & 0x7)
         if res_command & NO_MORE_DATA:
@@ -371,7 +371,7 @@ def __init__(self, sdo_client, index, subindex=0, size=None, force_segment=False
             response = sdo_client.request_response(request)
             res_command, = struct.unpack_from("B", response)
             if res_command != RESPONSE_DOWNLOAD:
-                self.sdo_client.abort(0x0504_0001)
+                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X}")
         else:
@@ -400,7 +400,7 @@ def write(self, b):
             response = self.sdo_client.request_response(request)
             res_command, = struct.unpack_from("B", response)
             if res_command & 0xE0 != RESPONSE_DOWNLOAD:
-                self.sdo_client.abort(0x0504_0001)
+                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X}")
             bytes_sent = len(b)
@@ -425,7 +425,7 @@ def write(self, b):
             response = self.sdo_client.request_response(request)
             res_command, = struct.unpack("B", response[0:1])
             if res_command & 0xE0 != RESPONSE_SEGMENT_DOWNLOAD:
-                self.sdo_client.abort(0x0504_0001)
+                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X} "
                     f"(expected 0x{RESPONSE_SEGMENT_DOWNLOAD:02X})")
@@ -499,7 +499,7 @@ def __init__(self, sdo_client, index, subindex=0, request_crc_support=True):
         res_command, res_index, res_subindex = SDO_STRUCT.unpack_from(response)
         if res_command & 0xE0 != RESPONSE_BLOCK_UPLOAD:
             self._error = True
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         # Check that the message is for us
         if res_index != index or res_subindex != subindex:
@@ -556,7 +556,7 @@ def read(self, size=-1):
             if self._done:
                 if self._server_crc != self._crc.final():
                     self._error = True
-                    self.sdo_client.abort(0x0504_0004)
+                    self.sdo_client.abort(ABORT_CRC_ERROR)
                     raise SdoCommunicationError("CRC is not OK")
                 logger.info("CRC is OK")
         self.pos += len(data)
@@ -576,7 +576,7 @@ def _retransmit(self):
                 self._ackseq = seqno
                 return response
         self._error = True
-        self.sdo_client.abort(0x0504_0000)
+        self.sdo_client.abort(ABORT_TIMED_OUT)
         raise SdoCommunicationError("Some data was lost and could not be retransmitted")
 
     def _ack_block(self):
@@ -591,16 +591,16 @@ def _end_upload(self):
         try:
             response = self.sdo_client.read_response()
         except SdoCommunicationError:
-            self.abort(0x0504_0000)
+            self.abort(ABORT_TIMED_OUT)
             raise
         res_command, self._server_crc = struct.unpack_from("<BH", response)
         if res_command & 0xE0 != RESPONSE_BLOCK_UPLOAD:
             self._error = True
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         if res_command & 0x3 != END_BLOCK_TRANSFER:
             self._error = True
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError("Server did not end transfer as expected")
         # Return number of bytes not used in last message
         return (res_command >> 2) & 0x7
@@ -670,7 +670,7 @@ def __init__(self, sdo_client, index, subindex=0, size=None, request_crc_support
         response = sdo_client.request_response(request)
         res_command, res_index, res_subindex = SDO_STRUCT.unpack_from(response)
         if res_command & 0xE0 != RESPONSE_BLOCK_DOWNLOAD:
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError(
                 f"Unexpected response 0x{res_command:02X}")
         # Check that the message is for us
@@ -755,15 +755,15 @@ def _block_ack(self):
         try:
             response = self.sdo_client.read_response()
         except SdoCommunicationError:
-            self.sdo_client.abort(0x0504_0000)
+            self.sdo_client.abort(ABORT_TIMED_OUT)
             raise
         res_command, ackseq, blksize = struct.unpack_from("BBB", response)
         if res_command & 0xE0 != RESPONSE_BLOCK_DOWNLOAD:
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError(
                 f"Unexpected response 0x{res_command:02X}")
         if res_command & 0x3 != BLOCK_TRANSFER_RESPONSE:
-            self.sdo_client.abort(0x0504_0001)
+            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
             raise SdoCommunicationError("Server did not respond with a "
                                         "block download response")
         if ackseq != self._blksize:

canopen/sdo/constants.py

@@ -1,4 +1,6 @@
 import struct
+from typing import Final
+
 
 # Command, index, subindex
 SDO_STRUCT = struct.Struct("<BHB")
@@ -31,3 +33,36 @@
 NO_MORE_DATA = 0x1
 NO_MORE_BLOCKS = 0x80
 TOGGLE_BIT = 0x10
+
+# Abort codes
+ABORT_TOGGLE_NOT_ALTERNATED: Final[int] = 0x0503_0000
+ABORT_TIMED_OUT: Final[int] = 0x0504_0000
+ABORT_INVALID_COMMAND_SPECIFIER: Final[int] = 0x0504_0001
+ABORT_INVALID_BLOCK_SIZE: Final[int] = 0x0504_0002
+ABORT_INVALID_SEQUENCE_NUMBER: Final[int] = 0x0504_0003
+ABORT_CRC_ERROR: Final[int] = 0x0504_0004
+ABORT_OUT_OF_MEMORY: Final[int] = 0x0504_0005
+ABORT_UNSUPPORTED_ACCESS: Final[int] = 0x0601_0000
+ABORT_READ_WRITEONLY: Final[int] = 0x0601_0001
+ABORT_WRITE_READONLY: Final[int] = 0x0601_0002
+ABORT_NOT_IN_OD: Final[int] = 0x0602_0000
+ABORT_PDO_CANNOT_MAP: Final[int] = 0x0604_0041
+ABORT_PDO_LENGTH_EXCEEDED: Final[int] = 0x0604_0042
+ABORT_PARAMETER_INCOMPATIBLE: Final[int] = 0x0604_0043
+ABORT_INTERNAL_INCOMPATIBILITY: Final[int] = 0x0604_0047
+ABORT_HARDWARE_ERROR: Final[int] = 0x0606_0000
+ABORT_LENGTH_NOT_MATCHED: Final[int] = 0x0607_0010
+ABORT_LENGTH_TOO_HIGH: Final[int] = 0x0607_0012
+ABORT_LENGTH_TOO_LOW: Final[int] = 0x0607_0013
+ABORT_NO_SUBINDEX: Final[int] = 0x0609_0011
+ABORT_INVALID_VALUE: Final[int] = 0x0609_0030  # download only
+ABORT_VALUE_TOO_HIGH: Final[int] = 0x0609_0031  # download only
+ABORT_VALUE_TOO_LOW: Final[int] = 0x0609_0032  # download only
+ABORT_MAXIMUM_LESS_THAN_MINIMUM: Final[int] = 0x0609_0036
+ABORT_NO_SDO_CONNECTION: Final[int] = 0x060A_0023
+ABORT_GENERAL_ERROR: Final[int] = 0x0800_0000
+ABORT_STORE_APPLICATION: Final[int] = 0x0800_0020
+ABORT_APPLICATION_LOCAL_CONTROL: Final[int] = 0x0800_0021
+ABORT_APPLICATION_DEVICE_STATE: Final[int] = 0x0800_0022
+ABORT_OD_GENERATION: Final[int] = 0x0800_0023
+ABORT_NO_DATA_AVAILABLE: Final[int] = 0x0800_0024

canopen/sdo/server.py

@@ -48,11 +48,11 @@ def on_request(self, can_id, data, timestamp):
             elif ccs == REQUEST_ABORTED:
                 self.request_aborted(data)
             else:
-                self.abort(0x05040001)
+                self.abort(ABORT_INVALID_COMMAND_SPECIFIER)
         except SdoAbortedError as exc:
             self.abort(exc.code)
         except KeyError as exc:
-            self.abort(0x06020000)
+            self.abort(ABORT_NOT_IN_OD)
         except Exception as exc:
             self.abort()
             logger.exception(exc)
@@ -68,7 +68,7 @@ def init_upload(self, request):
         size = len(data)
         if size == 0:
             logger.info("No content to upload for 0x%04X:%02X", index, subindex)
-            self.abort(0x0800_0024)
+            self.abort(ABORT_NO_DATA_AVAILABLE)
             return
         elif size <= 4:
             logger.info("Expedited upload for 0x%04X:%02X", index, subindex)
@@ -87,7 +87,7 @@ def init_upload(self, request):
     def segmented_upload(self, command):
         if command & TOGGLE_BIT != self._toggle:
             # Toggle bit mismatch
-            raise SdoAbortedError(0x05030000)
+            raise SdoAbortedError(ABORT_TOGGLE_NOT_ALTERNATED)
         data = self._buffer[:7]
         size = len(data)
 
@@ -129,7 +129,7 @@ def block_download(self, data):
         self._index = index
         self._subindex = subindex
         logger.error("Block download is not supported")
-        self.abort(0x05040001)
+        self.abort(ABORT_INVALID_COMMAND_SPECIFIER)
 
     def init_download(self, request):
         # TODO: Check if writable (now would fail on end of segmented downloads)
@@ -160,7 +160,7 @@ def init_download(self, request):
     def segmented_download(self, command, request):
         if command & TOGGLE_BIT != self._toggle:
             # Toggle bit mismatch
-            raise SdoAbortedError(0x05030000)
+            raise SdoAbortedError(ABORT_TOGGLE_NOT_ALTERNATED)
         last_byte = 8 - ((command >> 1) & 0x7)
         self._buffer.extend(request[1:last_byte])
 
@@ -183,7 +183,7 @@ def segmented_download(self, command, request):
     def send_response(self, response):
         self.network.send_message(self.tx_cobid, response)
 
-    def abort(self, abort_code=0x08000000):
+    def abort(self, abort_code=ABORT_GENERAL_ERROR):
         """Abort current transfer."""
         data = struct.pack("<BHBL", RESPONSE_ABORTED,
                            self._index, self._subindex, abort_code)

test/test_local.py

@@ -116,11 +116,11 @@ def test_nmt_state_initializing_to_preoper(self):
         self.assertEqual(state, 'PRE-OPERATIONAL')
 
     def test_receive_abort_request(self):
-        self.remote_node.sdo.abort(0x05040003)
+        self.remote_node.sdo.abort(0x0504_0003)  # Invalid sequence number
         # Line below is just so that we are sure the client have received the abort
         # before we do the check
         time.sleep(0.1)
-        self.assertEqual(self.local_node.sdo.last_received_error, 0x05040003)
+        self.assertEqual(self.local_node.sdo.last_received_error, 0x0504_0003)
 
     def test_start_remote_node(self):
         self.remote_node.nmt.state = 'OPERATIONAL'