nmt: use bytes instead of list for send_message and send_periodic

canopen/network.py

@@ -6,6 +6,7 @@
 from typing import Callable, Final, Optional, Union
 
 import can
+from can.typechecking import CanData
 
 from canopen.lss import LssMaster
 from canopen.nmt import NmtMaster
@@ -187,7 +188,7 @@ def create_node(
         self[node.id] = node
         return node
 
-    def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None:
+    def send_message(self, can_id: int, data: Optional[CanData], remote: bool = False) -> None:
         """Send a raw CAN message to the network.
 
         This method may be overridden in a subclass if you need to integrate
@@ -215,7 +216,7 @@ def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None:
         self.check()
 
     def send_periodic(
-        self, can_id: int, data: bytes, period: float, remote: bool = False
+        self, can_id: int, data: Optional[CanData], period: float, remote: bool = False
     ) -> PeriodicMessageTask:
         """Start sending a message periodically.
 
@@ -310,7 +311,7 @@ class PeriodicMessageTask:
     def __init__(
         self,
         can_id: int,
-        data: bytes,
+        data: Optional[CanData],
         period: float,
         bus,
         remote: bool = False,
@@ -339,7 +340,7 @@ def stop(self):
         """Stop transmission"""
         self._task.stop()
 
-    def update(self, data: bytes) -> None:
+    def update(self, data: CanData) -> None:
         """Update data of message
 
         :param data:

canopen/nmt.py

@@ -147,7 +147,7 @@ def send_command(self, code: int):
         super(NmtMaster, self).send_command(code)
         logger.info(
             "Sending NMT command 0x%X to node %d", code, self.id)
-        self.network.send_message(0, [code, self.id])
+        self.network.send_message(0, bytes([code, self.id]))
 
     def wait_for_heartbeat(self, timeout: float = 10):
         """Wait until a heartbeat message is received."""
@@ -190,7 +190,7 @@ def start_node_guarding(self, period: float):
         """
         if self._node_guarding_producer:
             self.stop_node_guarding()
-        self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, None, period, True)
+        self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, b'', period, True)
 
     def stop_node_guarding(self):
         """Stops the node guarding mechanism."""
@@ -225,7 +225,7 @@ def send_command(self, code: int) -> None:
 
         if self._state == 0:
             logger.info("Sending boot-up message")
-            self.network.send_message(0x700 + self.id, [0])
+            self.network.send_message(0x700 + self.id, b'\x00')
 
         # The heartbeat service should start on the transition
         # between INITIALIZING and PRE-OPERATIONAL state
@@ -259,7 +259,7 @@ def start_heartbeat(self, heartbeat_time_ms: int):
         if heartbeat_time_ms > 0:
             logger.info("Start the heartbeat timer, interval is %d ms", self._heartbeat_time_ms)
             self._send_task = self.network.send_periodic(
-                0x700 + self.id, [self._state], heartbeat_time_ms / 1000.0)
+                0x700 + self.id, bytes([self._state]), heartbeat_time_ms / 1000.0)
 
     def stop_heartbeat(self):
         """Stop the heartbeat service."""
@@ -270,7 +270,7 @@ def stop_heartbeat(self):
 
     def update_heartbeat(self):
         if self._send_task is not None:
-            self._send_task.update([self._state])
+            self._send_task.update(bytes([self._state]))
 
 
 class NmtError(Exception):