From 66a18ef707401bf4aa0b57ebd705928b9d12c057 Mon Sep 17 00:00:00 2001 From: Grant du Toit Date: Sun, 5 Jul 2026 15:40:34 +0000 Subject: [PATCH] Phase 3 Stage 2: rov_api FastAPI backend (core) - New rov_api package: HTTP-to-ROS2 bridge on port 8081 - GET /health: mission state, recording active, failsafe state - POST /abort: W3 RETURN TO SAFE, publishes Bool to /rov/mission/abort - POST /mission/start: MissionCommand START (recording no-go gate enforced) - POST /mission/stop: MissionCommand ABORT - Embedded rclpy node in background thread, uvicorn in main thread - Verified end-to-end: /abort -> failsafe_monitor FSM NORMAL -> HOLD_AND_RECOVER - Runs as argonaut-api.service (manual-start in dev, same policy as argonaut.service) - Deferred to Stage 2b: /return_budget, /backup/*, /mission/upload --- src/rov_api/package.xml | 28 ++ src/rov_api/resource/rov_api | 0 src/rov_api/rov_api/__init__.py | 0 src/rov_api/rov_api/api_node.py | 488 ++++++++++++++++++++++++++++++++ src/rov_api/setup.cfg | 4 + src/rov_api/setup.py | 26 ++ 6 files changed, 546 insertions(+) create mode 100644 src/rov_api/package.xml create mode 100644 src/rov_api/resource/rov_api create mode 100644 src/rov_api/rov_api/__init__.py create mode 100755 src/rov_api/rov_api/api_node.py create mode 100644 src/rov_api/setup.cfg create mode 100644 src/rov_api/setup.py diff --git a/src/rov_api/package.xml b/src/rov_api/package.xml new file mode 100644 index 0000000..f31bea3 --- /dev/null +++ b/src/rov_api/package.xml @@ -0,0 +1,28 @@ + + + + rov_api + 0.1.0 + + Argonaut 3 FastAPI backend — HTTP bridge from Cockpit widgets to the + ROS2 autonomy stack. Provides /health, /abort (W3 RETURN TO SAFE), + /mission/start, and /mission/stop endpoints on port 8081. + + Grant du Toit + Proprietary + + + ament_python + + + rclpy + std_msgs + rov_interfaces + + + + + ament_python + + diff --git a/src/rov_api/resource/rov_api b/src/rov_api/resource/rov_api new file mode 100644 index 0000000..e69de29 diff --git a/src/rov_api/rov_api/__init__.py b/src/rov_api/rov_api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/rov_api/rov_api/api_node.py b/src/rov_api/rov_api/api_node.py new file mode 100755 index 0000000..4bc823f --- /dev/null +++ b/src/rov_api/rov_api/api_node.py @@ -0,0 +1,488 @@ +#!/usr/bin/env python3 +""" +api_node.py — Argonaut 3 FastAPI Backend (Phase 3 Stage 2, core). + +Bridges operator-facing HTTP requests (from Cockpit DIY widgets) to the +ROS2 autonomy stack. Runs FastAPI/uvicorn in the main thread and an +embedded rclpy node in a background thread; the two communicate through +the RovApiNode instance held in shared module state. + +ARCHITECTURE +------------ + Cockpit widget (fetch) --HTTP--> FastAPI (uvicorn, main thread) + | + v + RovApiNode (rclpy, bg thread) + | + +-------------------------+--------------------------+ + | | | + publish /rov/mission/abort call /rov/mission/command cache latest: + (Bool, W3 RETURN TO SAFE) (MissionCommand srv) /rov/mission/status + START / ABORT /rov/recording/active + /rov/failsafe + +WHY THIS DESIGN +--------------- +FastAPI is NOT a ROS2 node. rclpy needs its own executor spinning to +service subscriptions and service clients. We therefore run rclpy.spin() +in a daemon background thread, and let the FastAPI request handlers call +thread-safe methods on the shared RovApiNode. rclpy publishers, service +clients, and subscription callbacks are safe to touch from another thread +as long as we do not call rclpy.spin() more than once. + +ENDPOINTS (core only — Stage 2b adds /return_budget, /backup/*, /mission/upload) +------------------------------------------------------------------------------- + GET /health -> mission state + recording active + failsafe state + POST /abort -> W3 RETURN TO SAFE (publishes Bool true to /rov/mission/abort) + POST /mission/start -> MissionCommand START + POST /mission/stop -> MissionCommand ABORT + +VERIFIED INTERFACES (read from source before writing — do not change without re-checking) +----------------------------------------------------------------------------------------- + /rov/mission/abort : std_msgs/Bool + failsafe_monitor._abort_callback sets flag_manual_abort=True + when msg.data is True. This is the vehicle-layer return-to-safe + path (DIR-5), correct for W3. + /rov/mission/command: rov_interfaces/srv/MissionCommand + Request : string command, string mission_id, string[] parameters + Response: bool success, string message + Valid commands (mission_executor): START, LOAD, PAUSE, RESUME, ABORT + /rov/mission/status : rov_interfaces/msg/MissionStatus (state, progress_percent, ...) + /rov/recording/active: std_msgs/Bool (recording_manager no-go gate) + /rov/failsafe : rov_interfaces/msg/FailsafeStatus (failsafe_state, assessment_state, ...) +""" + +import threading +from typing import Optional + +# ROS2 +import rclpy +from rclpy.node import Node +from std_msgs.msg import Bool +from rov_interfaces.msg import MissionStatus, FailsafeStatus +from rov_interfaces.srv import MissionCommand + +# FastAPI +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel +import uvicorn + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +# Port the FastAPI backend listens on. Matches the W3/W4 widget config +# constant FASTAPI_HOST = http://192.168.1.101:8081. +API_PORT = 8081 + +# Bind on all interfaces so Cockpit (on the laptop, over the tether/LAN) +# can reach the API. UFW is deferred to the pre-field hardening pass; +# until then there is no external exposure (Tailscale disabled, no proxy). +API_HOST = '0.0.0.0' + +# Timeout (seconds) to wait for the mission_executor service to become +# available before failing a mission command request. Kept short so the +# widget gets a fast, clear error rather than hanging. +SERVICE_WAIT_TIMEOUT_S = 2.0 + + +# --------------------------------------------------------------------------- +# Pydantic request models +# These define and validate the JSON body shape for POST endpoints. +# --------------------------------------------------------------------------- + +class MissionStartRequest(BaseModel): + """ + Body for POST /mission/start. + + mission_id : optional identifier echoed back in status. Defaults to + empty string if the widget does not supply one. + parameters : optional list of strings passed through to the mission + executor (e.g. operation mode 'tethered'/'untethered'). + The mission plan itself is loaded separately (DIR-3: + mission is uploaded beforehand, not at button-time). + """ + mission_id: str = '' + parameters: list[str] = [] + + +# --------------------------------------------------------------------------- +# ROS2 node +# --------------------------------------------------------------------------- + +class RovApiNode(Node): + """ + Embedded ROS2 node backing the FastAPI endpoints. + + Owns: + - a publisher on /rov/mission/abort (Bool) for W3 RETURN TO SAFE + - a service client on /rov/mission/command for mission START/STOP + - subscriptions caching the latest mission status, recording state, + and failsafe state, used to answer GET /health without blocking + + All cached values are plain Python types updated inside subscription + callbacks. Reads from the FastAPI thread are atomic (single attribute + reads of immutable values) so no explicit lock is required for them. + """ + + def __init__(self): + super().__init__('rov_api') + + # ------------------------------------------------------------------ + # Cached latest values for /health + # Initialised to sentinels indicating "no data received yet" so the + # health endpoint can report honestly before the stack publishes. + # ------------------------------------------------------------------ + + # Latest mission state integer (MissionStatus.STATE_*); -1 = no data + self._mission_state: int = -1 + + # Latest mission progress percent (0-100); 0 until data arrives + self._mission_progress: float = 0.0 + + # Latest recording-active flag; None = no data, True/False once known + self._recording_active: Optional[bool] = None + + # Latest failsafe state integer (FailsafeStatus.*); -1 = no data + self._failsafe_state: int = -1 + + # ------------------------------------------------------------------ + # Publisher — /rov/mission/abort (W3 RETURN TO SAFE) + # Publishing Bool(true) sets flag_manual_abort in failsafe_monitor, + # which drives the vehicle-layer return-to-safe action (DIR-5). + # ------------------------------------------------------------------ + self._abort_pub = self.create_publisher( + Bool, '/rov/mission/abort', 10) + + # ------------------------------------------------------------------ + # Service client — /rov/mission/command + # Used for mission START and STOP. mission_executor enforces the + # recording no-go gate (DIR-9) on START, so the API does not need + # to re-check recording here — a blocked START returns success=False + # with an explanatory message which we pass straight back to the widget. + # ------------------------------------------------------------------ + self._mission_cli = self.create_client( + MissionCommand, '/rov/mission/command') + + # ------------------------------------------------------------------ + # Subscriptions — cache latest values for /health + # ------------------------------------------------------------------ + + # Mission status from mission_executor (2 Hz) + self.create_subscription( + MissionStatus, '/rov/mission/status', + self._mission_status_cb, 10) + + # Recording active from recording_manager (1 Hz) — no-go gate signal + self.create_subscription( + Bool, '/rov/recording/active', + self._recording_cb, 10) + + # Failsafe status from failsafe_monitor (rate varies with state) + self.create_subscription( + FailsafeStatus, '/rov/failsafe', + self._failsafe_cb, 10) + + self.get_logger().info('RovApiNode started — HTTP bridge on :%d' % API_PORT) + + # ------------------------------------------------------------------ + # Subscription callbacks — update cached values + # ------------------------------------------------------------------ + + def _mission_status_cb(self, msg: MissionStatus): + """Cache the latest mission state and progress for /health.""" + self._mission_state = int(msg.state) + self._mission_progress = float(msg.progress_percent) + + def _recording_cb(self, msg: Bool): + """Cache the latest recording-active flag for /health.""" + self._recording_active = bool(msg.data) + + def _failsafe_cb(self, msg: FailsafeStatus): + """Cache the latest failsafe state for /health.""" + self._failsafe_state = int(msg.failsafe_state) + + # ------------------------------------------------------------------ + # Actions called from the FastAPI thread + # ------------------------------------------------------------------ + + def publish_abort(self): + """ + Publish Bool(true) to /rov/mission/abort — W3 RETURN TO SAFE. + + This is fire-and-forget: the failsafe monitor latches + flag_manual_abort on the rising edge, so a single publish is + sufficient. Called from the FastAPI request thread; create_publisher + publish is thread-safe with respect to the spinning executor. + """ + msg = Bool() + msg.data = True + self._abort_pub.publish(msg) + self.get_logger().warn('RETURN TO SAFE published to /rov/mission/abort (W3)') + + def call_mission_command(self, command: str, mission_id: str = '', + parameters: Optional[list] = None): + """ + Call the /rov/mission/command service synchronously and return + (success: bool, message: str). + + Because we are calling from the FastAPI thread (NOT the executor + thread), we cannot use the spin_until_future_complete pattern — + that would require spinning the executor, which is already being + spun in the background thread. Instead we use the blocking + call() convenience via a threading.Event on the future's callback. + + command : one of START / ABORT / PAUSE / RESUME / LOAD + mission_id : optional identifier + parameters : optional list of strings + """ + if parameters is None: + parameters = [] + + # Ensure the mission_executor service is actually up. If the stack + # is not running, fail fast with a clear message rather than hang. + if not self._mission_cli.wait_for_service(timeout_sec=SERVICE_WAIT_TIMEOUT_S): + return False, ( + 'mission_executor service /rov/mission/command unavailable — ' + 'is argonaut.service running?' + ) + + # Build the request per the verified .srv definition: + # string command, string mission_id, string[] parameters + req = MissionCommand.Request() + req.command = command + req.mission_id = mission_id + req.parameters = parameters + + # Fire the async request. We then block this (FastAPI) thread on a + # threading.Event that the future's done-callback sets. The future + # is completed by the background executor thread, so this is the + # correct cross-thread pattern (do NOT spin here). + future = self._mission_cli.call_async(req) + done_event = threading.Event() + + def _on_done(_fut): + # Called by the executor thread when the response arrives. + done_event.set() + + future.add_done_callback(_on_done) + + # Wait for the response, bounded by a timeout so a stuck service + # cannot hang the HTTP request indefinitely. + if not done_event.wait(timeout=SERVICE_WAIT_TIMEOUT_S + 3.0): + return False, 'mission command timed out waiting for response' + + # Retrieve the result. future.result() is safe now the callback fired. + result = future.result() + if result is None: + return False, 'mission command failed — no response from service' + + return bool(result.success), str(result.message) + + # ------------------------------------------------------------------ + # Health snapshot for GET /health + # ------------------------------------------------------------------ + + def health_snapshot(self) -> dict: + """ + Return a plain dict of the latest cached stack state. + + Values are read atomically (single immutable attribute reads) so + this is safe to call from the FastAPI thread without a lock. + """ + return { + 'mission_state': self._mission_state, # int, -1 = no data + 'mission_progress': self._mission_progress, # float 0-100 + 'recording_active': self._recording_active, # bool or None + 'failsafe_state': self._failsafe_state, # int, -1 = no data + } + + +# --------------------------------------------------------------------------- +# Shared node handle +# Populated by main() before uvicorn starts; read by endpoint handlers. +# --------------------------------------------------------------------------- + +# Module-level reference to the single RovApiNode instance. Set in main(). +# FastAPI endpoint functions access this to reach ROS2. It is created and +# assigned before uvicorn.run() is called, so it is never None during requests. +_node: Optional[RovApiNode] = None + + +# --------------------------------------------------------------------------- +# FastAPI application +# --------------------------------------------------------------------------- + +app = FastAPI( + title='Argonaut 3 API', + description='HTTP bridge from Cockpit widgets to the ROS2 autonomy stack.', + version='1.0.0', +) + +# CORS — Cockpit widgets run in the Cockpit app and POST cross-origin. +# Allow all origins for LAN/dev use. Tighten to the Cockpit origin at the +# pre-field hardening pass alongside UFW and the SSH password change. +app.add_middleware( + CORSMiddleware, + allow_origins=['*'], # dev: any origin; tighten before field + allow_credentials=False, + allow_methods=['*'], + allow_headers=['*'], +) + + +# ------------------------------------------------------------------ +# GET /health +# ------------------------------------------------------------------ + +@app.get('/health') +def health(): + """ + Return the current stack state snapshot. + + Used by widgets and for quick manual checks (curl). Reports mission + state, mission progress, recording-active flag, and failsafe state + from the latest cached ROS2 messages. Sentinel values (-1 / None) + indicate no data has been received on that topic yet. + """ + if _node is None: + # Should never happen — node is created before uvicorn starts. + return {'ok': False, 'error': 'ROS2 node not initialised'} + + snap = _node.health_snapshot() + return { + 'ok': True, + 'mission_state': snap['mission_state'], + 'mission_progress': snap['mission_progress'], + 'recording_active': snap['recording_active'], + 'failsafe_state': snap['failsafe_state'], + } + + +# ------------------------------------------------------------------ +# POST /abort — W3 RETURN TO SAFE +# ------------------------------------------------------------------ + +@app.post('/abort') +def abort(): + """ + W3 RETURN TO SAFE — vehicle-layer action (DIR-5). + + Publishes Bool(true) to /rov/mission/abort, which the failsafe monitor + latches as flag_manual_abort and drives the return-to-safe action. + + This is deliberately NOT the mission ABORT service — the DIR-5 intent + is a vehicle return-to-safe (follow breadcrumb path to the closest + safe place), not merely ending mission execution. + """ + if _node is None: + return {'ok': False, 'error': 'ROS2 node not initialised'} + + _node.publish_abort() + return { + 'ok': True, + 'action': 'return_to_safe', + 'message': 'RETURN TO SAFE command sent to vehicle.', + } + + +# ------------------------------------------------------------------ +# POST /mission/start +# ------------------------------------------------------------------ + +@app.post('/mission/start') +def mission_start(req: MissionStartRequest): + """ + Start the loaded mission plan via the MissionCommand START service. + + The mission plan must already be loaded (DIR-3: mission is uploaded + beforehand). mission_executor enforces the recording no-go gate + (DIR-9): if recording is not active, START returns success=False with + an explanatory message, which we pass straight back to the widget. + """ + if _node is None: + return {'ok': False, 'error': 'ROS2 node not initialised'} + + success, message = _node.call_mission_command( + command='START', + mission_id=req.mission_id, + parameters=req.parameters, + ) + return {'ok': success, 'message': message} + + +# ------------------------------------------------------------------ +# POST /mission/stop +# ------------------------------------------------------------------ + +@app.post('/mission/stop') +def mission_stop(): + """ + Stop the running mission via the MissionCommand ABORT service. + + This is the mission-layer stop (DIR-4) — it ends mission execution and + stops motion via mission_executor. It is distinct from W3 RETURN TO SAFE + (/abort), which is the vehicle-layer return-to-safe action. + + NOTE: mission_executor maps ABORT to STATE_ABORTED and stops motion. + A dedicated mission STOP (recording stop + clean end) is part of a later + stage; for now ABORT is the available clean-stop command. + """ + if _node is None: + return {'ok': False, 'error': 'ROS2 node not initialised'} + + success, message = _node.call_mission_command(command='ABORT') + return {'ok': success, 'message': message} + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + +def main(args=None): + """ + Initialise ROS2, start the embedded node spinning in a background + thread, then run uvicorn (FastAPI) in the main thread. + + Shutdown: uvicorn.run blocks until the process is signalled. On exit + we tear down rclpy cleanly. The background spin thread is a daemon so + it does not block process exit. + """ + global _node + + # Bring up ROS2 and create the bridge node BEFORE starting uvicorn, + # so endpoint handlers always have a valid _node. + rclpy.init(args=args) + _node = RovApiNode() + + # Spin the node in a daemon background thread. This services the + # subscriptions (health cache) and the mission service client futures. + def _spin(): + # rclpy.spin blocks, servicing callbacks, until shutdown. + try: + rclpy.spin(_node) + except Exception: + # On shutdown the context is torn down; swallow the resulting + # exception so the daemon thread exits quietly. + pass + + spin_thread = threading.Thread(target=_spin, daemon=True) + spin_thread.start() + + # Run the HTTP server in the main thread. This blocks until the process + # is stopped (Ctrl+C or systemd stop). + try: + uvicorn.run(app, host=API_HOST, port=API_PORT, log_level='info') + finally: + # Clean ROS2 shutdown. Guard against double-shutdown races. + _node.destroy_node() + try: + rclpy.shutdown() + except Exception: + pass + + +if __name__ == '__main__': + main() diff --git a/src/rov_api/setup.cfg b/src/rov_api/setup.cfg new file mode 100644 index 0000000..0026161 --- /dev/null +++ b/src/rov_api/setup.cfg @@ -0,0 +1,4 @@ +[develop] +script_dir=$base/lib/rov_api +[install] +install_scripts=$base/lib/rov_api diff --git a/src/rov_api/setup.py b/src/rov_api/setup.py new file mode 100644 index 0000000..05a307e --- /dev/null +++ b/src/rov_api/setup.py @@ -0,0 +1,26 @@ +from setuptools import setup + +package_name = 'rov_api' + +setup( + name=package_name, + version='0.1.0', + packages=[package_name], + data_files=[ + ('share/ament_index/resource_index/packages', + ['resource/' + package_name]), + ('share/' + package_name, ['package.xml']), + ], + install_requires=['setuptools'], + zip_safe=True, + maintainer='Grant du Toit', + maintainer_email='grant@symbytech.com', + description='Argonaut 3 FastAPI backend — HTTP bridge to the ROS2 autonomy stack.', + license='Proprietary', + entry_points={ + 'console_scripts': [ + # FastAPI backend node — HTTP bridge on :8081 for W3/W4 + mission control + 'api_node = rov_api.api_node:main', + ], + }, +)