mil_usb_to_can.sub9: Use electrical_protocol packets for communicatio…

…n with sub9 USB to CAN board

mil_common/drivers/electrical_protocol/electrical_protocol/packet.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import struct
-from dataclasses import dataclass
+from dataclasses import dataclass, fields
 from enum import Enum
 from functools import lru_cache
 from typing import ClassVar, TypeVar, get_type_hints
@@ -123,6 +123,29 @@ def __post_init__(self):
                 and issubclass(field_type, Enum)
             ):
                 setattr(self, name, field_type(self.__dict__[name]))
+        if not self.payload_format.startswith(("<", ">", "=", "!")):
+            raise ValueError(
+                "The payload format does not start with a standard size character: ('<', '>', '!', '=').",
+            )
+        available_chars: dict[type, list[str]] = {
+            bool: ["c", "b", "B", "?"],
+            int: ["b", "B", "h", "H", "i", "I", "l", "L", "q", "Q"],
+            float: ["f", "d"],
+        }
+        stripped_format = self.payload_format.lstrip("<>=!@")
+        for i, field in enumerate(fields(self)):
+            if field.type not in available_chars:
+                continue
+            chars = available_chars[field.type]
+            if i >= len(stripped_format):
+                raise ValueError(
+                    f"The payload format for the packet is too short to support all dataclass fields; expected: {len(fields(self))}, found: {len(self.payload_format)}.",
+                )
+            represented_char = stripped_format[i]
+            if represented_char not in chars:
+                raise ValueError(
+                    f"The type of {field.name} in the payload format is '{represented_char}', which does not correspond to its dataclass type of {field.type}.",
+                )
 
     @classmethod
     def _calculate_checksum(cls, data: bytes) -> tuple[int, int]:
@@ -147,11 +170,12 @@ def __bytes__(self):
         return data + struct.pack("<BB", *checksum)
 
     def __len__(self) -> int:
-        return struct.calcsize(f"<BBBBH{self.payload_format}BB")
+        return self.__class__._expected_len()
 
     @classmethod
     def _expected_len(cls) -> int:
-        return struct.calcsize(f"<BBBBH{cls.payload_format}BB")
+        # We cannot use one calcsize since payload_format should start with a standard size character
+        return struct.calcsize("<BBBBHBB") + struct.calcsize(cls.payload_format)
 
     @classmethod
     def from_bytes(

mil_common/drivers/mil_usb_to_can/mil_usb_to_can/sub9/__init__.py

@@ -1,2 +1 @@
 from .device import CANDeviceHandle, SimulatedCANDeviceHandle
-from .packet import AckPacket, ChecksumException, NackPacket, Packet

mil_common/drivers/mil_usb_to_can/mil_usb_to_can/sub9/device.py

@@ -2,8 +2,9 @@
 
 from typing import TYPE_CHECKING
 
+from electrical_protocol import Packet
+
 if TYPE_CHECKING:
-    from .packet import Packet
     from .sub9_driver import SimulatedUSBtoCANStream, USBtoCANDriver
 
 

mil_common/drivers/mil_usb_to_can/mil_usb_to_can/sub9/example.py

@@ -3,39 +3,39 @@
 from dataclasses import dataclass
 
 import rospy
+from electrical_protocol import Packet
 from rospy_tutorials.srv import AddTwoInts
 from std_srvs.srv import Trigger, TriggerRequest, TriggerResponse
 
 from .device import CANDeviceHandle, SimulatedCANDeviceHandle
-from .packet import Packet
 
 
 @dataclass
 class ExampleEchoRequestPacket(
     Packet,
-    msg_id=0x99,
+    class_id=0x99,
     subclass_id=0x00,
-    payload_format="10s",
+    payload_format="<10s",
 ):
     my_special_string: bytes
 
 
 @dataclass
 class ExampleEchoResponsePacket(
     Packet,
-    msg_id=0x99,
+    class_id=0x99,
     subclass_id=0x01,
-    payload_format="10s",
+    payload_format="<10s",
 ):
     my_special_string: bytes
 
 
 @dataclass
 class ExampleAdderRequestPacket(
     Packet,
-    msg_id=0x99,
+    class_id=0x99,
     subclass_id=0x02,
-    payload_format="BB",
+    payload_format="<BB",
 ):
     num_one: int
     num_two: int
@@ -44,9 +44,9 @@ class ExampleAdderRequestPacket(
 @dataclass
 class ExampleAdderResponsePacket(
     Packet,
-    msg_id=0x99,
+    class_id=0x99,
     subclass_id=0x03,
-    payload_format="B",
+    payload_format="<B",
 ):
     response: int
 

mil_common/drivers/mil_usb_to_can/mil_usb_to_can/sub9/sub9_driver.py

@@ -1,4 +1,5 @@
 #!/usr/bin/python3
+
 from __future__ import annotations
 
 import importlib
@@ -8,11 +9,12 @@
 
 import rospy
 import serial
+from electrical_protocol import Packet
+from electrical_protocol.packet import SYNC_CHAR_1
 from mil_misc_tools.serial_tools import SimulatedSerial
 from serial import SerialException
 
 from mil_usb_to_can.sub9.device import CANDeviceHandle, SimulatedCANDeviceHandle
-from mil_usb_to_can.sub9.packet import SYNC_CHAR_1, Packet
 
 if TYPE_CHECKING:
     HandlePacketListing = tuple[