Merge pull request #3 from phettberg/adapt_lss_handling

Adapt lss handling

Author: semiversus

.github/workflows/workflow.yml

@@ -34,7 +34,7 @@ jobs:
         password: ${{ secrets.PYPI_API_TOKEN }}
 
   test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
     strategy:
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10']
@@ -51,4 +51,4 @@ jobs:
         pip install -r requirements_dev.txt -e .
     - name: Test with pytest
       run: |
-        pytest -v
+        pytest -v

CHANGELOG.md

@@ -1,3 +1,8 @@
+# 0.5.0
+
+* fix LSS baudrate configuration handling
+* provide LSS store configuration functionality (bitrate and node id)
+
 # 0.4.12
 
 * update object dictionary on type change in PDOs

src/durand/services/lss.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Callable, List, Optional
 from enum import IntEnum
 import logging
 import struct
@@ -18,24 +18,49 @@ class LSSState(IntEnum):
 
 
 class LSSSlave:
+    CiA_bit_timing_table = {
+        0: 1000000,
+        1: 800000,
+        2: 500000,
+        3: 250000,
+        4: 125000,
+        6: 50000,
+        7: 20000,
+        8: 10000,
+    }
+
     def __init__(self, node: "Node"):
         self._node = node
 
         self._state = LSSState.WAITING
 
         self._received_selective_address: List[Optional[int]] = [None] * 4
         self._remote_responder_address: List[Optional[int]] = [None] * 6
-        self._fastscan_state = 0
+        self._fastscan_state: int = 0
 
-        self._pending_baudrate = None
-        self._change_baudrate_cb = None
+        self._pending_baudrate: int | None = None
+        self._change_baudrate_cb: Callable[[int, float], None] | None = None
+
+        self._store_configuration_cb: Callable[[int, int], None] | None = None
 
         node.network.add_subscription(cob_id=0x7E5, callback=self.handle_msg)
         node.nmt.state_callbacks.add(self.on_nmt_state_update)
 
-    def set_baudrate_change_callback(self, cb):
+    def set_baudrate_change_callback(self, cb: Callable[[int, float], None]):
+        """Defines a callback function to change bitrate. The delay is obtained from
+        the `activate_bit_timing` command.
+        This callback is called after the delay specified in the command and should
+        change the bitrate of the network after waiting for the delay again.
+        :param cb: callback function with signature (baudrate, delay)
+        """
         self._change_baudrate_cb = cb
 
+    def set_store_configuration_callback(self, cb: Callable[[int, int], None]):
+        """Defines a callback function to store new bitrate and node id persistently.
+        :param cb: callback function with signature (baudrate, node_id)
+        """
+        self._store_configuration_cb = cb
+
     def on_nmt_state_update(self, state: StateEnum):
         if state == StateEnum.INITIALISATION:
             self._state = LSSState.WAITING
@@ -124,18 +149,17 @@ def cmd_configure_node_id(self, msg: bytes):
         )
 
     def cmd_configure_bit_timing(self, msg: bytes):
-        valid_table_entries = (0, 1, 2, 3, 4, 6, 7, 8)
         selector, index = msg[1:3]
 
         if (
             selector != 0
-            or index not in valid_table_entries
+            or index not in self.CiA_bit_timing_table
             or self._change_baudrate_cb is None
         ):
             self._node.network.send(0x7E4, b"\x13\x01" + bytes(6))
             return
 
-        self._pending_baudrate = index
+        self._pending_baudrate = self.CiA_bit_timing_table[index]
         self._node.network.send(0x7E4, b"\x13\x00" + bytes(6))
 
     def cmd_activate_bit_timing(self, msg: bytes):
@@ -146,13 +170,19 @@ def cmd_activate_bit_timing(self, msg: bytes):
 
     def _change_baudrate(self, delay: float):
         if self._change_baudrate_cb:
-            self._change_baudrate_cb(self._pending_baudrate)
+            self._change_baudrate_cb(self._pending_baudrate, delay)
         self._pending_baudrate = None
-        get_scheduler().add(delay, self._node.nmt.reset)
 
     def cmd_store_configuration(self, _msg: bytes):
-        # store configuration is not supported
-        self._node.network.send(0x7E4, b"\x17\x01" + bytes(6))
+        if self._store_configuration_cb is not None:
+            self._store_configuration_cb(self._pending_baudrate, self._node.nmt.pending_node_id)
+            result = 0  # 0x00 = successfully completed
+        else:
+            result = 1  # 0x01 = not supported
+
+        self._node.network.send(
+            0x7E4, b"\x17" + result.to_bytes(1, "little") + bytes(6)
+        )
 
     def cmd_identify_remote_responders(self, msg: bytes):
         index = msg[0] - 0x46

tests/example_based/test_lss.py

@@ -1,8 +1,22 @@
 """ Testing LSS service """
 
-from durand import Node
+from durand import Node, scheduler
 
-from ..mock_network import MockNetwork, TxMsg, RxMsg
+from ..mock_network import MockNetwork, RxMsg, TxMsg
+
+
+class MockScheduler(scheduler.AbstractScheduler):
+    def add(self, delay: float, callback, args=(), kwargs=None) -> scheduler.TEntry:
+        if kwargs is None:
+            kwargs = {}
+        callback(*args, **kwargs)
+
+    def cancel(self, entry: scheduler.TEntry):
+        ...
+
+    @property
+    def lock(self):
+        ...
 
 
 def test_global_selection():
@@ -38,3 +52,59 @@ def test_global_selection():
             TxMsg(0x581, "43 00 10 00 00 00 00 00"),  # receive the acknowledge
         ]
     )
+
+
+def test_configuration():
+    """Test example starts with a node with node id 0x01.
+    After the "switch state global" request to set every responder into configuration mode,
+    the node id is set to 2.
+
+    Furthermore, the baudrate is set to 500 kbit/s and the configuration is stored.
+    When the configuration is stored, the node sends an acknowledge message.
+
+    At last, activate bit timing is sent to the node. It is the responsibility of its callback
+    to reinitialize the network with the new baudrate after a delay.
+    The callback is tested with the baudrate and delay parameters.
+    """
+
+    scheduler_ = MockScheduler()
+    scheduler.set_scheduler(scheduler_)
+
+    network = MockNetwork()
+
+    # create the node
+    node = Node(network, node_id=0x01)
+
+    def baudrate_change_callback(baudrate: int, delay: float):
+        # This callback is supposed to reinitialize the network with the new baudrate
+        # after the delay
+        # In this test, we just check if the callback is called with the correct parameters
+        assert baudrate == 500_000
+        assert delay == 1
+
+    node.lss.set_baudrate_change_callback(baudrate_change_callback)
+
+    def store_configuration_callback(baudrate: int, node_id: int):
+        assert baudrate == 500_000
+        assert node_id == 2
+
+    node.lss.set_store_configuration_callback(store_configuration_callback)
+
+    network.test(
+        [
+            RxMsg(
+                0x7E5, "04 01 00 00 00 00 00 00"
+            ),  # switch state global to configuration state
+
+            RxMsg(0x7E5, "11 02 00 00 00 00 00 00"),  # set node id to 2
+            TxMsg(0x7E4, "11 00 00 00 00 00 00 00"),  # receive the acknowledge (success)
+
+            RxMsg(0x7E5, "13 00 02 00 00 00 00 00"),  # set baudrate to 500 kbit/s
+            TxMsg(0x7E4, "13 00 00 00 00 00 00 00"),  # receive the acknowledge (success)
+
+            RxMsg(0x7E5, "17 00 00 00 00 00 00 00"),  # store configuration
+            TxMsg(0x7E4, "17 00 00 00 00 00 00 00"),  # receive the acknowledge (success)
+
+            RxMsg(0x7E5, "15 E8 03 00 00 00 00 00"),  # activate bit timing
+        ]
+    )