From be46488a3a57018d3dca227e86a50617713c2921 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Badenes?= Date: Fri, 28 Aug 2026 11:21:55 -0300 Subject: [PATCH] Report the robot's real status, and only when it changes _on_connect published an unconditional "online" status message, so a robot was marked online in InOrbit whenever its MQTT session came up -- whether or not the robot itself was reachable. Restarting a connector for a dead robot therefore ended its offline period, and nothing corrected it: with the robot marked online there is no discrepancy for InOrbit to resolve via get_state. Take the status from the online status callback instead, sharing the resolution (and its default-to-online fallback for a missing or failing callback) with _handle_get_state. That callback is only useful on connect if it is registered before connect(), and a RobotSessionPool connects a session as soon as it builds it, leaving no window to call RobotSession.set_online_status_callback() in time. Add RobotSessionFactory.set_online_status_callback(), mirroring register_command_callback, so the callback is attached at build time and covers the first connection. Add publish_status(), which publishes only when the status changed, so a caller that evaluates the robot's health on a loop can hand the result over on every iteration and produce one message per transition. This is the only way to report a robot offline while its session stays connected: InOrbit asks for state only when it already has the robot offline, and the session answers InOrbit's pings for as long as the process is alive. disconnect() goes through it too, so ending a session does not re-report a robot that was already offline. Connecting still reports the status directly, without that de-duplication: the will may have been published while the session was down, so InOrbit's view of the robot is unknown at that point. get_state is likewise always answered -- InOrbit asked, so it gets an answer. Co-Authored-By: Claude Opus 5 (1M context) --- inorbit_edge/robot.py | 98 ++++++++-- .../tests/test_robot_session_callbacks.py | 169 ++++++++++++++++++ .../tests/test_robot_session_factory.py | 41 +++++ 3 files changed, 295 insertions(+), 13 deletions(-) diff --git a/inorbit_edge/robot.py b/inorbit_edge/robot.py index 097b0df..7d58ade 100644 --- a/inorbit_edge/robot.py +++ b/inorbit_edge/robot.py @@ -446,6 +446,9 @@ def __init__(self, robot_id, robot_name, api_key=None, **kwargs) -> None: # Callback for determining robot online status self._online_status_callback = None + # Last online status reported to InOrbit, None until one is sent + self._last_published_online_status = None + # Odometry accumulator for estimating odometry data when it is not available. self._distance_accumulator = RobotDistanceAccumulator( estimate_distance_linear=kwargs.get("estimate_distance_linear", True), @@ -616,10 +619,14 @@ def _on_connect(self, client, userdata, flags, reason_code, properties): ) return - # Send robot online status (best effort) + # Send robot status (best effort) # If this fails and InOrbit is getting system stats data, it will detect # the discrepancy and request a status update via get_state. - self._send_robot_status(online=True) + # A connected session means the connector is up, not that the robot is, so + # the status comes from the online status callback. Sent directly rather + # than through publish_status(): the will may have been published while + # this session was down, so InOrbit's view is unknown at this point. + self._send_robot_status(online=self._get_online_status()) # Subscribe to interesting topics self.client.subscribe( @@ -831,14 +838,7 @@ def _handle_get_state(self): If the robot is offline, the next pose is not accumulated for odometry estimation. """ - is_online = True # Default assumption - - if self._online_status_callback: - try: - is_online = self._online_status_callback() - except Exception as e: - self.logger.error(f"Online status callback failed: {e}") - # Fall back to default (True) on callback error + is_online = self._get_online_status() self._send_robot_status(online=is_online) self.logger.debug( @@ -849,6 +849,23 @@ def _handle_get_state(self): if not is_online: self._distance_accumulator.discard_next_delta() + def _get_online_status(self) -> bool: + """Return the robot's online status, as reported by the online status + callback. + + Defaults to True when no callback was registered or the callback fails, so + a robot is never reported offline because of a missing or broken health + check. @see set_online_status_callback. + """ + if not self._online_status_callback: + return True # Default assumption + + try: + return bool(self._online_status_callback()) + except Exception as e: + self.logger.error(f"Online status callback failed: {e}") + return True # Fall back to default (True) on callback error + def _start_cameras_streaming(self): """Start streaming on all registered cameras""" with self.camera_streaming_mutex: @@ -1129,7 +1146,12 @@ def register_command_callback(self, callback): def set_online_status_callback(self, callback): """Set callback to determine robot online status. - Called on a state request from InOrbit. @see _handle_get_state. + Called on a state request from InOrbit (@see _handle_get_state) and + whenever the MQTT session connects (@see _on_connect). Because a + RobotSessionPool connects a session as soon as it builds it, setting the + callback here only affects later reconnections; use + RobotSessionFactory.set_online_status_callback() to have it registered + before the first connection. Args: callback: A callable that returns bool indicating if robot is online. @@ -1173,6 +1195,26 @@ def _resend_modules(self): qos=1, ) + def publish_status(self, online: bool) -> None: + """Publish the robot's online status to InOrbit, if it changed. + + A status equal to the last one published is skipped, so a caller that + evaluates the robot's health on a loop can hand the result over on every + iteration and produce one message per transition. + + This is the only way to report a robot offline while its session stays + connected: InOrbit asks for state (@see _handle_get_state) only when it + already has the robot offline, and the session answers InOrbit's pings for + as long as the process is alive. + + Args: + online (bool): True if the robot is online, False otherwise. + """ + if online == self._last_published_online_status: + return + + self._send_robot_status(online=online) + def _send_robot_status(self, online=True): """Send robot online/offline status (best effort, non-blocking). @@ -1193,6 +1235,7 @@ def _send_robot_status(self, online=True): qos=1, retain=True, ) + self._last_published_online_status = online self.logger.debug(f"{status_str.capitalize()} status sent successfully") except Exception as e: self.logger.debug(f"{status_str.capitalize()} status failed: {e}") @@ -1287,8 +1330,9 @@ def disconnect(self): ) # Send offline status (best effort, non-blocking) - # InOrbit will detect offline via data absence if this fails - self._send_robot_status(online=False) + # InOrbit will detect offline via data absence if this fails. Skipped when + # the robot was already reported offline. + self.publish_status(online=False) # TODO: Unsubscribe from topics @@ -1818,6 +1862,7 @@ def __init__(self, **robot_session_kw_args): self.robot_session_kw_args = robot_session_kw_args self.command_callbacks = [] self.commands_paths_rules = [] + self.online_status_callback = None def build(self, robot_id, robot_name="", **robot_config): """Builds a RobotSession object using the provided id and name. @@ -1838,6 +1883,13 @@ def c(*args): for command_callback in self.command_callbacks: session.register_command_callback(build_callback(command_callback)) + if self.online_status_callback: + # Not build_callback(): that wrapper discards the return value, and + # the online status callback is read for its result. + session.set_online_status_callback( + lambda: self.online_status_callback(robot_id) + ) + for path, exec_name_regex in self.commands_paths_rules: session.register_commands_path(path, exec_name_regex) return session @@ -1851,6 +1903,26 @@ def register_command_callback(self, callback): self.command_callbacks.append(callback) + def set_online_status_callback(self, callback): + """Set the callback used to determine online status on all robot sessions + created by this factory. + + Unlike RobotSession.set_online_status_callback(), the callback registered + here is set on each session before it is connected, so it is also used for + the status published on the first connection. It receives the robot id: + + factory.set_online_status_callback(lambda robot_id: ...) + + Args: + callback: A callable taking the robot id and returning a bool, True if + the robot is online. + """ + if not callable(callback): + # Don't do anything if callback is not a valid function + return + + self.online_status_callback = callback + def register_commands_path(self, path="./user_scripts", exec_name_regex=r".*"): """Registers executable commands that handle InOrbit custom command actions. Use `exec_name_regex` and `path` to customize which executables can be diff --git a/inorbit_edge/tests/test_robot_session_callbacks.py b/inorbit_edge/tests/test_robot_session_callbacks.py index f0838b3..d3071af 100644 --- a/inorbit_edge/tests/test_robot_session_callbacks.py +++ b/inorbit_edge/tests/test_robot_session_callbacks.py @@ -92,6 +92,175 @@ def test_responds_to_get_state_with_callback_online_status( ) +def test_sends_online_status_on_connect_without_callback( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that connecting reports the robot online when no callback is set.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session._on_connect(None, None, None, 0, None) + + robot_session.client.publish.assert_any_call( + "r/id_123/state", + "1|robot_apikey_123|{}.edgesdk_py|name_123".format(get_module_version()), + qos=1, + retain=True, + ) + + +def _state_publishes(robot_session): + """Return the status messages published on the robot's state topic.""" + return [ + call.args[1] + for call in robot_session.client.publish.call_args_list + if call.args and call.args[0] == "r/id_123/state" + ] + + +def _status(online): + """Return the status message the session publishes for the given status.""" + return "{}|robot_apikey_123|{}.edgesdk_py|name_123".format( + "1" if online else "0", get_module_version() + ) + + +def test_sends_offline_status_on_connect_when_callback_reports_offline( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that connecting reports offline when the callback says so. + + A connected session means the connector is up, not the robot. This is sent + without de-duplication: the will may have been published while the session was + down, so InOrbit's view of the robot is unknown on connect. + """ + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session.set_online_status_callback(lambda: False) + + robot_session._on_connect(None, None, None, 0, None) + + assert _state_publishes(robot_session) == [_status(online=False)] + + +def test_publish_status_skips_unchanged_status( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that publish_status only publishes on a change.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session.set_online_status_callback(lambda: False) + robot_session._on_connect(None, None, None, 0, None) + + # Already reported offline on connect: nothing to add + robot_session.publish_status(online=False) + assert _state_publishes(robot_session) == [_status(online=False)] + + # The robot came back: report it, once + robot_session.publish_status(online=True) + robot_session.publish_status(online=True) + assert _state_publishes(robot_session) == [ + _status(online=False), + _status(online=True), + ] + + +def test_disconnect_skips_offline_status_when_already_offline( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that ending a session does not re-report an offline robot. + + Otherwise stopping the agent would move the offline timestamp of a robot that + has been offline all along. + """ + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session.set_online_status_callback(lambda: False) + robot_session._on_connect(None, None, None, 0, None) + + # The mocked client stays "connected" unless told otherwise + robot_session.client.is_connected.return_value = False + robot_session.disconnect() + + # Only the status reported on connect + assert _state_publishes(robot_session) == [_status(online=False)] + + +def test_disconnect_reports_offline_status_when_online( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that ending a session reports an online robot as offline.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session._on_connect(None, None, None, 0, None) + + # The mocked client stays "connected" unless told otherwise + robot_session.client.is_connected.return_value = False + robot_session.disconnect() + + assert _state_publishes(robot_session) == [ + _status(online=True), + _status(online=False), + ] + + +def test_responds_to_get_state_even_when_status_is_unchanged( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that a get_state request is always answered. + + InOrbit asked, so it gets an answer even if that status was already published: + the de-duplication only covers statuses the robot volunteers. + """ + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + robot_session.set_online_status_callback(lambda: False) + robot_session._on_connect(None, None, None, 0, None) + + robot_session._handle_in_cmd(b"get_state") + + # The one reported on connect, and the answer to the request + assert _state_publishes(robot_session) == [ + _status(online=False), + _status(online=False), + ] + + +def test_connect_status_handles_callback_exception( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that a failing callback falls back to online on connect.""" + robot_session = RobotSession( + robot_id="id_123", robot_name="name_123", api_key="apikey_123" + ) + robot_session.connect() + + def failing_callback(): + raise RuntimeError("Test error") + + robot_session.set_online_status_callback(failing_callback) + + robot_session._on_connect(None, None, None, 0, None) + + robot_session.client.publish.assert_any_call( + "r/id_123/state", + "1|robot_apikey_123|{}.edgesdk_py|name_123".format(get_module_version()), + qos=1, + retain=True, + ) + + def test_get_state_handles_callback_exception( mock_mqtt_client, mock_inorbit_api, mock_sleep ): diff --git a/inorbit_edge/tests/test_robot_session_factory.py b/inorbit_edge/tests/test_robot_session_factory.py index 38c6683..42461bc 100644 --- a/inorbit_edge/tests/test_robot_session_factory.py +++ b/inorbit_edge/tests/test_robot_session_factory.py @@ -3,6 +3,7 @@ from unittest.mock import MagicMock from paho.mqtt.client import MQTTMessage +from inorbit_edge import get_module_version from inorbit_edge.robot import RobotSessionFactory from inorbit_edge.inorbit_pb2 import CustomScriptCommandMessage from inorbit_edge.tests.utils.helpers import test_robot_session_connect_helper @@ -95,6 +96,46 @@ def test_built_robot_session_executes_command_callback_on_message( _test_command_handler_helper(another_command_handler) +def test_built_robot_session_reports_online_status_on_first_connect( + mock_mqtt_client, mock_inorbit_api, mock_sleep +): + """Test that a factory-level online status callback applies on connect. + + Sessions built by a RobotSessionPool are connected as soon as they are + created, so a callback registered on the session afterwards would be too late + for the status published on the first connection. + """ + robot_session_factory = RobotSessionFactory(api_key="apikey_123") + robot_session_factory.set_online_status_callback( + lambda robot_id: robot_id != "id_123" + ) + + robot_session = robot_session_factory.build("id_123", "name_123") + robot_session.connect() + robot_session._on_connect(None, None, None, 0, None) + + # The callback is called with the robot id, and reports "id_123" offline, so + # that is the status published on connect + assert robot_session._online_status_callback() is False + robot_session.client.publish.assert_any_call( + "r/id_123/state", + "0|robot_apikey_123|{}.edgesdk_py|name_123".format(get_module_version()), + qos=1, + retain=True, + ) + + +def test_factory_online_status_callback_ignores_non_callable(mock_mqtt_client): + """Test that set_online_status_callback ignores non-callable values.""" + robot_session_factory = RobotSessionFactory(api_key="apikey_123") + + robot_session_factory.set_online_status_callback("not_callable") + + assert robot_session_factory.online_status_callback is None + robot_session = robot_session_factory.build("id_123", "name_123") + assert robot_session._online_status_callback is None + + def _test_command_handler_helper(command_handler): command_handler.assert_called_once() call_args, call_kwargs = command_handler.call_args_list[0]