From 38ef92ff87f18bd66f14eb56080a1b172653de35 Mon Sep 17 00:00:00 2001 From: Bastian Richter Date: Mon, 19 Oct 2026 11:23:00 +0200 Subject: [PATCH 1/2] multiple event listener handling --- README.md | 2 + cri_lib/__init__.py | 8 +- cri_lib/cri_controller.py | 334 ++++++++++++++++++++------------------ tests/test_cri_parser.py | 212 +++++++++++++++++++++--- 4 files changed, 381 insertions(+), 175 deletions(-) diff --git a/README.md b/README.md index 09d752b..b89e340 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,8 @@ In the top directory of this repository execute `pip install .` The library provides the `CRIController` class which is the main interface to controll the iRC. It handles the network connection and starts background threads for keep alive messages and processing of received messages. Movement functions directly return even though the move might not be finisched. If you want to wait for the move to finish, set the `wait_move_finished` parameter to `True`. Most function return a bool to check whether the execution was successful. +The low-level `CRIController.send_command(...)` method returns a `SentCommand` with a `message_id` and an `answer_event` (`None` when no answer was registered). + ### Typical procedure 1. Connect to iRC: `CRIController.connect(...)` Default IP is `192.168.3.11` with port `3920`. For using the simulation in the iRC desktop software, connect to `127.0.0.1`, most of the time the port in the simulation is `3921`, but can be different. Have a look at the log if you are unable to connect. Check whether connection was successfull via the returned bool. 2. If you want to control the robot, acquire active control via `CRIController.set_active_control(True)`. diff --git a/cri_lib/__init__.py b/cri_lib/__init__.py index 5aaf958..11debbf 100644 --- a/cri_lib/__init__.py +++ b/cri_lib/__init__.py @@ -2,7 +2,13 @@ .. include:: ../README.md """ -from .cri_controller import CRIClient, CRIConnector, CRIController, MotionType +from .cri_controller import ( + CRIClient, + CRIConnector, + CRIController, + MotionType, + SentCommand, +) from .cri_errors import ( CRICommandError, CRICommandTimeOutError, diff --git a/cri_lib/cri_controller.py b/cri_lib/cri_controller.py index edbc04d..30b5792 100644 --- a/cri_lib/cri_controller.py +++ b/cri_lib/cri_controller.py @@ -9,7 +9,7 @@ from pathlib import Path from queue import Empty, Queue from time import sleep, time -from typing import Any, Callable, Literal +from typing import Any, Callable, Literal, NamedTuple from .cri_errors import CRICommandError, CRICommandTimeOutError, CRIConnectionError from .cri_protocol_parser import CRIProtocolParser @@ -40,6 +40,13 @@ def _run_sync(coro): """Robot state message categories that must must be received before confirming fully connected.""" +class SentCommand(NamedTuple): + """The wire message ID and its optional answer event.""" + + message_id: int + answer_event: asyncio.Event | None + + class MotionType(Enum): """Robot Motion Type for Jogging""" @@ -94,8 +101,8 @@ def __init__(self) -> None: self.sent_command_counter_lock = threading.Lock() self.sent_command_counter = 0 self.answer_events_lock = threading.Lock() - self.answer_events: dict[str, asyncio.Event] = {} - self.error_messages: dict[str, str] = {} + self.answer_events: dict[str, list[asyncio.Event]] = {} + self.error_messages: dict[asyncio.Event, str] = {} self.status_callback: Callable | None = None @@ -183,16 +190,28 @@ def close(self) -> None: self.sock.close() - def _register_answer(self, answer_id: str) -> None: + def _register_answer(self, answer_id: str) -> asyncio.Event: + event = asyncio.Event() + with self.answer_events_lock: + self.answer_events.setdefault(answer_id, []).append(event) + return event + + def _remove_answer_event(self, event: asyncio.Event) -> None: with self.answer_events_lock: - self.answer_events[answer_id] = asyncio.Event() + for answer_id, events in list(self.answer_events.items()): + if event in events: + events.remove(event) + if not events: + del self.answer_events[answer_id] + break + self.error_messages.pop(event, None) def _send_command( self, command: str, register_answer: bool = False, fixed_answer_name: str | None = None, - ) -> int: + ) -> SentCommand: """Sends the given command to iRC. The method is marked private because technically it can be used to send control commands as well. @@ -204,8 +223,8 @@ def _send_command( Returns ------- - int - The sent message_id. + SentCommand + The sent message ID and its registered answer event, if any. Raises ------ @@ -228,28 +247,24 @@ def _send_command( message = f"CRISTART {command_counter} {command} CRIEND" - if register_answer: - with self.answer_events_lock: - if fixed_answer_name is not None: - self.answer_events[fixed_answer_name] = asyncio.Event() - else: - self.answer_events[str(command_counter)] = asyncio.Event() + answer_id = ( + fixed_answer_name + if fixed_answer_name is not None + else str(command_counter) + ) + answer_event = self._register_answer(answer_id) if register_answer else None try: with self.socket_write_lock: self.sock.sendall(message.encode()) logger.debug("Sent command: %s", message) - return command_counter + return SentCommand(command_counter, answer_event) except Exception as e: logger.exception("Failed to send command.") - if register_answer: - with self.answer_events_lock: - if fixed_answer_name is not None: - del self.answer_events[fixed_answer_name] - else: - del self.answer_events[str(command_counter)] + if answer_event is not None: + self._remove_answer_event(answer_event) self.connected = False raise CRIConnectionError("ConnectionLost") @@ -258,10 +273,7 @@ def _bg_alivejog_thread(self) -> None: Background Thread sending alivejog messages to keep connection alive. """ while self.connected: - if self._send_command("ALIVEJOG 0 0 0 0 0 0 0 0 0") is None: - logger.error("AliveJog Thread: Connection lost.") - self.connected = False - return + self._send_command("ALIVEJOG 0 0 0 0 0 0 0 0 0") sleep(self.jog_intervall) @@ -317,29 +329,36 @@ def _bg_receive_thread(self) -> None: def _wait_for_answer( self, - message_id: str | int, + message_id: str | int | SentCommand, timeout: float | None = DEFAULT, # type: ignore + answer_event: asyncio.Event | None = None, ) -> None | str: - return _run_sync(self._wait_for_answer_async(message_id, timeout)) + return _run_sync( + self._wait_for_answer_async(message_id, timeout, answer_event) + ) async def _wait_for_answer_async( self, - message_id: str | int, + message_id: str | int | SentCommand, timeout: float | None = DEFAULT, # type: ignore + answer_event: asyncio.Event | None = None, ) -> None | str: """Waits for an answer to a message. The answer event will be removed after the call, even if there was a timeout. Choose timeout accordingly. Parameters ---------- - message_id : int or str - message id of sent message of which an answer is expected + message_id : int, str, or SentCommand + Sent command result or message ID associated with the answer. timeout : float | DEFAULT | None timeout for wait in seconds. - `DEFAULT` uses `self.DEFAULT_ANSWER_TIMEOUT` - `None` will wait 24 hours. + answer_event : asyncio.Event | None + Exact event returned by ``_send_command`` or registered for a response key. + Returns ------- None | str @@ -352,33 +371,30 @@ async def _wait_for_answer_async( raised if no answer was received in given timeout """ - message_id = str(message_id) - with self.answer_events_lock: - if message_id not in self.answer_events: - return None - wait_event = self.answer_events[message_id] + if isinstance(message_id, SentCommand): + answer_event = message_id.answer_event + message_id = message_id.message_id + answer_id = str(message_id) + if answer_event is None: + return None if timeout is DEFAULT: timeout = self.DEFAULT_ANSWER_TIMEOUT if timeout is None: timeout = 24 * 3600 + error_msg = None try: - await wait_event_with_timeout(wait_event, timeout=timeout) + await wait_event_with_timeout(answer_event, timeout=timeout) + with self.answer_events_lock: + error_msg = self.error_messages.get(answer_event) except TimeoutError as ex: raise CRICommandTimeOutError( - f"Did not receive {message_id=} answer within {timeout=} s." + f"Did not receive {answer_id=} answer within {timeout=} s." ) from ex + finally: + self._remove_answer_event(answer_event) - # prevent deadlock through answer_events_lock - with self.answer_events_lock: - del self.answer_events[message_id] - - if message_id in self.error_messages: - error_msg = self.error_messages[message_id] - del self.error_messages[message_id] - return error_msg - else: - return None + return error_msg def _parse_message(self, message: str) -> None: """Internal function to parse a message. If an answer event is registered for a certain msg_id it is triggered.""" @@ -398,11 +414,12 @@ def _parse_message(self, message: str) -> None: with self.answer_events_lock: msg_id = notification["answer"] - if msg_id in self.answer_events: - if (error_msg := notification.get("error", None)) is not None: - self.error_messages[msg_id] = error_msg - - self.answer_events[msg_id].set() + events = self.answer_events.pop(msg_id, []) + error_msg = notification.get("error") + for event in events: + if error_msg is not None: + self.error_messages[event] = error_msg + event.set() async def wait_for_status_update_async(self, timeout: float | None = None) -> None: """Wait for next STATUS message. @@ -417,8 +434,8 @@ async def wait_for_status_update_async(self, timeout: float | None = None) -> No CRITimeoutError raised if no status update was received in given timeout """ - self._register_answer("status") - await self._wait_for_answer_async("status", timeout) + answer_event = self._register_answer("status") + await self._wait_for_answer_async("status", timeout, answer_event) def register_status_callback(self, callback: Callable | None) -> None: """Register a callback which is called every time a STATUS message was parsed to the state. @@ -475,10 +492,12 @@ async def get_board_temperatures_async( ------- list of motor controller PCB temperatures """ - self._send_command("SYSTEM GetBoardTemp", True, "info_boardtemp") + sent_command = self._send_command( + "SYSTEM GetBoardTemp", True, "info_boardtemp" + ) if ( error_msg := await self._wait_for_answer_async( - "info_boardtemp", timeout=timeout + sent_command, timeout=timeout ) ) is not None: raise CRICommandError(f"Error in GetBoardTemp command: {error_msg}") @@ -499,10 +518,12 @@ async def get_motor_temperatures_async( ------- list of motor temperatures """ - self._send_command("SYSTEM GetMotorTemp", True, "info_motortemp") + sent_command = self._send_command( + "SYSTEM GetMotorTemp", True, "info_motortemp" + ) if ( error_msg := await self._wait_for_answer_async( - "info_motortemp", timeout=timeout + sent_command, timeout=timeout ) ) is not None: raise CRICommandError(f"Error in GetMotorTemp command: {error_msg}") @@ -522,11 +543,11 @@ async def list_files_async(self, target_directory: str = "Programs") -> list[str """ command = f"CMD ListFiles {target_directory}" - self._send_command( + sent_command = self._send_command( command=command, register_answer=True, fixed_answer_name="info_filelist" ) if ( - error_msg := await self._wait_for_answer_async("info_filelist") + error_msg := await self._wait_for_answer_async(sent_command) ) is not None: raise CRICommandError(f"Error in ListFiles command: {error_msg}") # the obtained file list is cached in the parser @@ -592,15 +613,17 @@ def _bg_alivejog_thread(self) -> None: else: command = "ALIVEJOG 0 0 0 0 0 0 0 0 0" - if self._send_command(command) is None: - logger.error("AliveJog Thread: Connection lost.") - self.connected = False - return + self._send_command(command) sleep(self.jog_intervall) - def send_command(self, command, register_answer=False, fixed_answer_name=None): - """Wraps the superclass method to make it public.""" + def send_command( + self, + command: str, + register_answer: bool = False, + fixed_answer_name: str | None = None, + ) -> SentCommand: + """Send a raw command and return its message ID and optional answer event.""" return super()._send_command(command, register_answer, fixed_answer_name) async def reset_async(self) -> bool: @@ -662,15 +685,13 @@ async def set_active_control_async(self, active: bool) -> bool: `True` acquire active control `False` return active control """ - self._send_command( + sent_command = self._send_command( f"CMD SetActive {str(active).lower()}", True, f"Active_{str(active).lower()}", ) if ( - error_msg := await self._wait_for_answer_async( - f"Active_{str(active).lower()}" - ) + error_msg := await self._wait_for_answer_async(sent_command) ) is not None: logger.debug("Error in set active control command: %s", error_msg) return False @@ -753,15 +774,45 @@ async def get_referencing_info_async(self): `True` if request was successful `False` if request was not successful """ - self._send_command("CMD GetReferencingInfo", True, "info_referencing") + sent_command = self._send_command( + "CMD GetReferencingInfo", True, "info_referencing" + ) if ( - error_msg := await self._wait_for_answer_async("info_referencing") + error_msg := await self._wait_for_answer_async(sent_command) ) is not None: logger.debug("Error in GetReferencingInfo command: %s", error_msg) return False else: return True + async def _send_move_command_and_wait( + self, + command: str, + completion_response: str, + wait_move_finished: bool, + move_finished_timeout: float | None, + ) -> tuple[str | None, str | None]: + completion_event = ( + self._register_answer(completion_response) if wait_move_finished else None + ) + try: + sent_command = self._send_command(command, True) + command_error = await self._wait_for_answer_async( + sent_command, timeout=30.0 + ) + if command_error is not None or completion_event is None: + return command_error, None + + completion_error = await self._wait_for_answer_async( + completion_response, + timeout=move_finished_timeout, + answer_event=completion_event, + ) + return None, completion_error + finally: + if completion_event is not None: + self._remove_answer_event(completion_event) + async def move_joints_async( self, A1: float, @@ -816,24 +867,19 @@ async def move_joints_async( else: expected_response = "MOVETOEXECEND" - if wait_move_finished: - self._register_answer(expected_response) - - msg_id = self._send_command(command, True) - if ( - error_msg := await self._wait_for_answer_async(msg_id, timeout=30.0) - ) is not None: - logger.debug("Error in Move Joints command: %s", error_msg) + command_error, completion_error = await self._send_move_command_and_wait( + command, + expected_response, + wait_move_finished, + move_finished_timeout, + ) + if command_error is not None: + logger.debug("Error in Move Joints command: %s", command_error) return False - if wait_move_finished: - if ( - error_msg := await self._wait_for_answer_async( - expected_response, timeout=move_finished_timeout - ) - ) is not None: - logger.debug("Exec Error in Move Joints command: %s", error_msg) - return False + if completion_error is not None: + logger.debug("Exec Error in Move Joints command: %s", completion_error) + return False return True async def move_joints_relative_async( @@ -888,26 +934,21 @@ async def move_joints_relative_async( else: expected_response = "MOVETOEXECEND" - if wait_move_finished: - self._register_answer(expected_response) - - msg_id = self._send_command(command, True) - if ( - error_msg := await self._wait_for_answer_async(msg_id, timeout=30.0) - ) is not None: - logger.debug("Error in Move Joints command: %s", error_msg) + command_error, completion_error = await self._send_move_command_and_wait( + command, + expected_response, + wait_move_finished, + move_finished_timeout, + ) + if command_error is not None: + logger.debug("Error in Move Joints command: %s", command_error) return False - if wait_move_finished: - if ( - error_msg := await self._wait_for_answer_async( - expected_response, timeout=move_finished_timeout - ) - ) is not None: - logger.debug( - "Exec Error in Move Joints Relative command: %s", error_msg - ) - return False + if completion_error is not None: + logger.debug( + "Exec Error in Move Joints Relative command: %s", completion_error + ) + return False return True async def move_cartesian_async( @@ -968,24 +1009,19 @@ async def move_cartesian_async( else: expected_response = "MOVETOEXECEND" - if wait_move_finished: - self._register_answer(expected_response) - - msg_id = self._send_command(command, True) - if ( - error_msg := await self._wait_for_answer_async(msg_id, timeout=30.0) - ) is not None: - logger.debug("Error in Move Joints command: %s", error_msg) + command_error, completion_error = await self._send_move_command_and_wait( + command, + expected_response, + wait_move_finished, + move_finished_timeout, + ) + if command_error is not None: + logger.debug("Error in Move Joints command: %s", command_error) return False - if wait_move_finished: - if ( - error_msg := await self._wait_for_answer_async( - expected_response, timeout=move_finished_timeout - ) - ) is not None: - logger.debug("Exec Error in Move Cartesian command: %s", error_msg) - return False + if completion_error is not None: + logger.debug("Exec Error in Move Cartesian command: %s", completion_error) + return False return True @@ -1045,24 +1081,19 @@ async def move_base_relative_async( else: expected_response = "MOVETOEXECEND" - if wait_move_finished: - self._register_answer(expected_response) - - msg_id = self._send_command(command, True) - if ( - error_msg := await self._wait_for_answer_async(msg_id, timeout=30.0) - ) is not None: - logger.debug("Error in Move Joints command: %s", error_msg) + command_error, completion_error = await self._send_move_command_and_wait( + command, + expected_response, + wait_move_finished, + move_finished_timeout, + ) + if command_error is not None: + logger.debug("Error in Move Joints command: %s", command_error) return False - if wait_move_finished: - if ( - error_msg := await self._wait_for_answer_async( - expected_response, timeout=move_finished_timeout - ) - ) is not None: - logger.debug("Exec Error in Move BaseRelative command: %s", error_msg) - return False + if completion_error is not None: + logger.debug("Exec Error in Move BaseRelative command: %s", completion_error) + return False return True @@ -1122,24 +1153,19 @@ async def move_tool_relative_async( else: expected_response = "MOVETOEXECEND" - if wait_move_finished: - self._register_answer(expected_response) - - msg_id = self._send_command(command, True) - if ( - error_msg := await self._wait_for_answer_async(msg_id, timeout=30.0) - ) is not None: - logger.debug("Error in Move Joints command: %s", error_msg) + command_error, completion_error = await self._send_move_command_and_wait( + command, + expected_response, + wait_move_finished, + move_finished_timeout, + ) + if command_error is not None: + logger.debug("Error in Move Joints command: %s", command_error) return False - if wait_move_finished: - if ( - error_msg := await self._wait_for_answer_async( - expected_response, timeout=move_finished_timeout - ) - ) is not None: - logger.debug("Exec Error in Move BaseTool command: %s", error_msg) - return False + if completion_error is not None: + logger.debug("Exec Error in Move BaseTool command: %s", completion_error) + return False return True diff --git a/tests/test_cri_parser.py b/tests/test_cri_parser.py index 98962be..c65fded 100644 --- a/tests/test_cri_parser.py +++ b/tests/test_cri_parser.py @@ -1,9 +1,11 @@ +import asyncio import copy import threading import pytest from cri_lib import ( + CRICommandTimeOutError, CRIController, CRIProtocolParser, ErrorStates, @@ -20,6 +22,7 @@ RobotMode, RobotState, RunState, + SentCommand, ) @@ -33,6 +36,12 @@ def robot_state_equal(actual: RobotState, expected: RobotState) -> bool: return actual == expected_copy +def register_test_event(controller: CRIController, answer_id: str) -> asyncio.Event: + event = asyncio.Event() + controller.answer_events.setdefault(answer_id, []).append(event) + return event + + def test_parse_state(): """This tests parsing the STATUS message with the assumption that all possible axes are present""" @@ -668,21 +677,186 @@ def test_parse_cmdack(): test_message = "CRISTART 1234 CMDACK 123 CRIEND" controller = CRIController() - controller.answer_events["123"] = threading.Event() + event = register_test_event(controller, "123") controller._parse_message(test_message) - assert controller.answer_events["123"].is_set() + assert event.is_set() + + +def test_status_notification_wakes_all_registered_waiters(): + controller = CRIController() + first_event = controller._register_answer("status") + second_event = controller._register_answer("status") + + controller._parse_message("CRISTART 1234 STATUS MODE joint CRIEND") + + assert first_event is not second_event + assert first_event.is_set() + assert second_event.is_set() + + +def test_send_command_returns_its_answer_event(monkeypatch): + controller = CRIController() + + class ImmediateReplySocket: + sent_message = b"" + + def sendall(self, message: bytes) -> None: + self.sent_message = message + controller._parse_message("CRISTART 1234 CMDACK 0 CRIEND") + + fake_socket = ImmediateReplySocket() + monkeypatch.setattr(controller, "sock", fake_socket) + controller.connected = True + + sent_command = controller._send_command("CMD Test", register_answer=True) + unregistered_command = controller._send_command("CMD FireAndForget") + + assert sent_command.message_id == 0 + assert sent_command.answer_event is not None + assert sent_command.answer_event.is_set() + assert fake_socket.sent_message == b"CRISTART 1 CMD FireAndForget CRIEND" + assert unregistered_command.message_id == 1 + assert unregistered_command.answer_event is None + + +def test_concurrent_status_waiters_complete_and_clean_up(): + async def scenario(): + controller = CRIController() + callbacks = [] + controller.register_status_callback(callbacks.append) + waiters = [ + asyncio.create_task(controller.wait_for_status_update_async(timeout=1.0)) + for _ in range(2) + ] + + for _ in range(100): + if len(controller.answer_events.get("status", [])) == 2: + break + await asyncio.sleep(0.001) + + assert len(controller.answer_events["status"]) == 2 + controller._parse_message("CRISTART 1234 STATUS MODE joint CRIEND") + await asyncio.gather(*waiters) + + assert controller.answer_events == {} + assert len(callbacks) == 1 + + asyncio.run(scenario()) + + +def test_shared_error_is_returned_to_each_waiter(): + async def scenario(): + controller = CRIController() + first_event = controller._register_answer("123") + second_event = controller._register_answer("123") + waiters = [ + asyncio.create_task( + controller._wait_for_answer_async("123", answer_event=event) + ) + for event in (first_event, second_event) + ] + + controller._parse_message( + "CRISTART 1234 CMDERROR 123 There was an exception CRIEND" + ) + + assert await asyncio.gather(*waiters) == [ + "There was an exception", + "There was an exception", + ] + assert controller.answer_events == {} + assert controller.error_messages == {} + + asyncio.run(scenario()) + + +def test_cancelled_waiter_does_not_remove_other_registration(): + async def scenario(): + controller = CRIController() + cancelled_event = controller._register_answer("123") + remaining_event = controller._register_answer("123") + cancelled_waiter = asyncio.create_task( + controller._wait_for_answer_async("123", answer_event=cancelled_event) + ) + remaining_waiter = asyncio.create_task( + controller._wait_for_answer_async("123", answer_event=remaining_event) + ) + await asyncio.sleep(0) + cancelled_waiter.cancel() + try: + await cancelled_waiter + except asyncio.CancelledError: + pass + + assert controller.answer_events["123"] == [remaining_event] + controller._parse_message("CRISTART 1234 CMDACK 123 CRIEND") + assert await remaining_waiter is None + assert controller.answer_events == {} + + asyncio.run(scenario()) + + +def test_timeout_removes_fixed_key_event_by_identity(): + async def scenario(): + controller = CRIController() + event = controller._register_answer("info_filelist") + + with pytest.raises(CRICommandTimeOutError): + await controller._wait_for_answer_async( + SentCommand(42, event), timeout=0.0 + ) + + assert controller.answer_events == {} + + asyncio.run(scenario()) + + +def test_concurrent_move_waiters_share_completion_notification(monkeypatch): + async def scenario(): + controller = CRIController() + next_message_id = iter((10, 11)) + + def send_command(command: str, register_answer: bool) -> SentCommand: + message_id = next(next_message_id) + answer_event = controller._register_answer(str(message_id)) + controller._parse_message( + f"CRISTART 1234 CMDACK {message_id} CRIEND" + ) + return SentCommand(message_id, answer_event) + + monkeypatch.setattr(controller, "_send_command", send_command) + waiters = [ + asyncio.create_task( + controller._send_move_command_and_wait( + "CMD Move Joint", "MOVETOEXECEND", True, 1.0 + ) + ) + for _ in range(2) + ] + + for _ in range(100): + if len(controller.answer_events.get("MOVETOEXECEND", [])) == 2: + break + await asyncio.sleep(0.001) + + assert len(controller.answer_events["MOVETOEXECEND"]) == 2 + controller._parse_message("CRISTART 1234 MOVETOEXECEND 0 0 CRIEND") + assert await asyncio.gather(*waiters) == [(None, None), (None, None)] + assert controller.answer_events == {} + + asyncio.run(scenario()) def test_parse_cmderror(): test_message = "CRISTART 1234 CMDERROR 123 There was an exception CRIEND" controller = CRIController() - controller.answer_events["123"] = threading.Event() + event = register_test_event(controller, "123") controller._parse_message(test_message) - assert controller.answer_events["123"].is_set() - assert controller.error_messages["123"] == "There was an exception" + assert event.is_set() + assert controller.error_messages[event] == "There was an exception" def test_info_referencinginfo(): @@ -746,23 +920,21 @@ def test_parse_execend(): test_message = "CRISTART 1234 EXECEND 0 0 CRIEND" controller = CRIController() - controller.answer_events["EXECEND"] = threading.Event() + event = register_test_event(controller, "EXECEND") controller._parse_message(test_message) - assert controller.answer_events["EXECEND"].is_set() + assert event.is_set() def test_parse_execerror(): test_message = "CRISTART 67 EXECERROR 0 0 PGLinear exception: 'Out of reach' CRIEND" controller = CRIController() - controller.answer_events["EXECEND"] = threading.Event() + event = register_test_event(controller, "EXECEND") controller._parse_message(test_message) - assert controller.answer_events["EXECEND"].is_set() - assert ( - controller.error_messages["EXECEND"] == "0 0 PGLinear exception: 'Out of reach'" - ) + assert event.is_set() + assert controller.error_messages[event] == "0 0 PGLinear exception: 'Out of reach'" def test_parse_can_brdige(): @@ -777,10 +949,10 @@ def test_parse_can_brdige(): } controller = CRIController() - controller.answer_events["CAN"] = threading.Event() + event = register_test_event(controller, "CAN") controller._parse_message(test_message) - assert controller.answer_events["CAN"].is_set() + assert event.is_set() assert controller.can_queue.get_nowait() == can_message @@ -807,10 +979,10 @@ def test_info_boardtemps(): ] controller = CRIController() - controller.answer_events["info_boardtemp"] = threading.Event() + event = register_test_event(controller, "info_boardtemp") controller._parse_message(test_message) - assert controller.answer_events["info_boardtemp"].is_set() + assert event.is_set() assert controller.robot_state.board_temps == pytest.approx(board_temps) @@ -837,10 +1009,10 @@ def test_info_motortemps(): ] controller = CRIController() - controller.answer_events["info_motortemp"] = threading.Event() + event = register_test_event(controller, "info_motortemp") controller._parse_message(test_message) - assert controller.answer_events["info_motortemp"].is_set() + assert event.is_set() assert controller.robot_state.motor_temps == pytest.approx(motor_temps) @@ -850,9 +1022,9 @@ def test_list_files(): ) controller = CRIController() - controller.answer_events["info_filelist"] = threading.Event() + event = register_test_event(controller, "info_filelist") controller._parse_message(test_message) - assert controller.answer_events["info_filelist"].is_set() + assert event.is_set() assert "FirstFile.xml" in controller.file_list assert "SecondFile.txt" in controller.file_list From 57f294b791f58c3b0fea1adb4a39ff8533f3add0 Mon Sep 17 00:00:00 2001 From: Bastian Richter Date: Mon, 19 Oct 2026 11:36:15 +0200 Subject: [PATCH 2/2] fixed formatting and one none check --- cri_lib/cri_controller.py | 45 +++++++++++++++------------------------ tests/test_cri_parser.py | 8 ++----- 2 files changed, 19 insertions(+), 34 deletions(-) diff --git a/cri_lib/cri_controller.py b/cri_lib/cri_controller.py index 30b5792..89f9de5 100644 --- a/cri_lib/cri_controller.py +++ b/cri_lib/cri_controller.py @@ -248,9 +248,7 @@ def _send_command( message = f"CRISTART {command_counter} {command} CRIEND" answer_id = ( - fixed_answer_name - if fixed_answer_name is not None - else str(command_counter) + fixed_answer_name if fixed_answer_name is not None else str(command_counter) ) answer_event = self._register_answer(answer_id) if register_answer else None @@ -333,9 +331,7 @@ def _wait_for_answer( timeout: float | None = DEFAULT, # type: ignore answer_event: asyncio.Event | None = None, ) -> None | str: - return _run_sync( - self._wait_for_answer_async(message_id, timeout, answer_event) - ) + return _run_sync(self._wait_for_answer_async(message_id, timeout, answer_event)) async def _wait_for_answer_async( self, @@ -414,12 +410,13 @@ def _parse_message(self, message: str) -> None: with self.answer_events_lock: msg_id = notification["answer"] - events = self.answer_events.pop(msg_id, []) - error_msg = notification.get("error") - for event in events: - if error_msg is not None: - self.error_messages[event] = error_msg - event.set() + if msg_id is not None: + events = self.answer_events.pop(msg_id, []) + error_msg = notification.get("error") + for event in events: + if error_msg is not None: + self.error_messages[event] = error_msg + event.set() async def wait_for_status_update_async(self, timeout: float | None = None) -> None: """Wait for next STATUS message. @@ -492,9 +489,7 @@ async def get_board_temperatures_async( ------- list of motor controller PCB temperatures """ - sent_command = self._send_command( - "SYSTEM GetBoardTemp", True, "info_boardtemp" - ) + sent_command = self._send_command("SYSTEM GetBoardTemp", True, "info_boardtemp") if ( error_msg := await self._wait_for_answer_async( sent_command, timeout=timeout @@ -518,9 +513,7 @@ async def get_motor_temperatures_async( ------- list of motor temperatures """ - sent_command = self._send_command( - "SYSTEM GetMotorTemp", True, "info_motortemp" - ) + sent_command = self._send_command("SYSTEM GetMotorTemp", True, "info_motortemp") if ( error_msg := await self._wait_for_answer_async( sent_command, timeout=timeout @@ -546,9 +539,7 @@ async def list_files_async(self, target_directory: str = "Programs") -> list[str sent_command = self._send_command( command=command, register_answer=True, fixed_answer_name="info_filelist" ) - if ( - error_msg := await self._wait_for_answer_async(sent_command) - ) is not None: + if (error_msg := await self._wait_for_answer_async(sent_command)) is not None: raise CRICommandError(f"Error in ListFiles command: {error_msg}") # the obtained file list is cached in the parser with self.parser.file_list_lock: @@ -690,9 +681,7 @@ async def set_active_control_async(self, active: bool) -> bool: True, f"Active_{str(active).lower()}", ) - if ( - error_msg := await self._wait_for_answer_async(sent_command) - ) is not None: + if (error_msg := await self._wait_for_answer_async(sent_command)) is not None: logger.debug("Error in set active control command: %s", error_msg) return False else: @@ -777,9 +766,7 @@ async def get_referencing_info_async(self): sent_command = self._send_command( "CMD GetReferencingInfo", True, "info_referencing" ) - if ( - error_msg := await self._wait_for_answer_async(sent_command) - ) is not None: + if (error_msg := await self._wait_for_answer_async(sent_command)) is not None: logger.debug("Error in GetReferencingInfo command: %s", error_msg) return False else: @@ -1092,7 +1079,9 @@ async def move_base_relative_async( return False if completion_error is not None: - logger.debug("Exec Error in Move BaseRelative command: %s", completion_error) + logger.debug( + "Exec Error in Move BaseRelative command: %s", completion_error + ) return False return True diff --git a/tests/test_cri_parser.py b/tests/test_cri_parser.py index c65fded..27162d9 100644 --- a/tests/test_cri_parser.py +++ b/tests/test_cri_parser.py @@ -803,9 +803,7 @@ async def scenario(): event = controller._register_answer("info_filelist") with pytest.raises(CRICommandTimeOutError): - await controller._wait_for_answer_async( - SentCommand(42, event), timeout=0.0 - ) + await controller._wait_for_answer_async(SentCommand(42, event), timeout=0.0) assert controller.answer_events == {} @@ -820,9 +818,7 @@ async def scenario(): def send_command(command: str, register_answer: bool) -> SentCommand: message_id = next(next_message_id) answer_event = controller._register_answer(str(message_id)) - controller._parse_message( - f"CRISTART 1234 CMDACK {message_id} CRIEND" - ) + controller._parse_message(f"CRISTART 1234 CMDACK {message_id} CRIEND") return SentCommand(message_id, answer_event) monkeypatch.setattr(controller, "_send_command", send_command)