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]