administrator/rov-autonomy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 3 Stage 2: rov_api FastAPI backend (core)

Browse Source 
- 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
This commit is contained in:
Grant du Toit 2026-07-05 15:40:34 +00:00
parent 06403533f2
commit 66a18ef707
6 changed files with 546 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/rov_api/package.xml Normal file
View File

@ -0,0 +1,28 @@
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
<name>rov_api</name>
<version>0.1.0</version>
<description>
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.
</description>
<maintainer email="grant@symbytech.com">Grant du Toit</maintainer>
<license>Proprietary</license>
<!-- Build tooling for a Python ament package -->
<buildtool_depend>ament_python</buildtool_depend>
<!-- ROS2 runtime dependencies -->
<exec_depend>rclpy</exec_depend>
<exec_depend>std_msgs</exec_depend>
<exec_depend>rov_interfaces</exec_depend>
<!-- FastAPI/uvicorn are pip-installed system-wide (not rosdep keys);
documented in the deploy notes, not declared here as ament deps. -->
<export>
<build_type>ament_python</build_type>
</export>
</package>

0
src/rov_api/resource/rov_api Normal file
View File

0
src/rov_api/rov_api/__init__.py Normal file
View File

488
src/rov_api/rov_api/api_node.py Executable file
View File

@ -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()

4
src/rov_api/setup.cfg Normal file
View File

@ -0,0 +1,4 @@
[develop]
script_dir=$base/lib/rov_api
[install]
install_scripts=$base/lib/rov_api

26
src/rov_api/setup.py Normal file
View File

@ -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',
],
},
)