diff --git a/src/mcbebop/models.py b/src/mcbebop/models.py new file mode 100644 index 0000000..155f90e --- /dev/null +++ b/src/mcbebop/models.py @@ -0,0 +1,94 @@ +"""Result models. + +Pydantic for outputs so the schema reaches the caller; `Annotated`/`Field` is +used for inputs. These are deliberately plain: a tool result is read by a model +that has not seen the drone, so every field says what it is in its description. +""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class ConnectionInfo(BaseModel): + connected: bool + target: str = Field(description="Address of the drone, or 'sim' for the built-in simulator.") + armed: bool = Field(description="Whether commands that can move the aircraft are currently unlocked.") + armed_until: str | None = Field(default=None, description="When the unlock expires, ISO 8601.") + telemetry_keys: int = Field(default=0, description="How many distinct values the drone has reported.") + last_update_age: float | None = Field( + default=None, description="Seconds since the most recent telemetry. Large means the link is stale." + ) + undecodable_events: int = Field( + default=0, description="Events whose command id is absent from Parrot's XML. Normally zero." + ) + + +class CommandSummary(BaseModel): + name: str + title: str = "" + tier: str = Field(description="observe, config, envelope or motion. motion and envelope need arm().") + direction: str = Field(description="to_drone if you can send it, from_drone if the drone emits it.") + bebop2: bool | None = Field( + description="Whether the XML says the Bebop 2 supports it. None means unstated." + ) + deprecated: bool = False + + +class ArgumentInfo(BaseModel): + name: str + type: str + doc: str = "" + enum_values: list[str] = Field(default_factory=list, description="Allowed names, for an enum argument.") + + +class CommandDetail(CommandSummary): + ids: list[int] = Field(description="project, class and command id, as sent on the wire.") + args: list[ArgumentInfo] = Field(default_factory=list) + doc: str = "" + buffer: str = Field(description="Which link buffer carries it: ack, non_ack or high_prio.") + confirmed_by: list[str] = Field( + default_factory=list, description="Events the drone emits to confirm it, if the XML declares any." + ) + + +class SendResult(BaseModel): + command: str + args: dict = Field(default_factory=dict) + acked: bool = Field(description="Whether the drone acknowledged receipt.") + confirmed: bool = Field(default=False, description="Whether the drone also emitted its confirming event.") + elapsed_ms: float = 0.0 + note: str = "" + + +class StateValue(BaseModel): + value: object + age: float = Field(description="Seconds since the drone last reported this. Old values may be stale.") + + +class Check(BaseModel): + name: str + ok: bool | None = Field(description="True, False, or None when the drone has not reported it.") + detail: str + + +class Preflight(BaseModel): + ready: bool = Field( + description="False if any blocking check failed. Advisory, not an airworthiness test." + ) + blocking: list[str] = Field(default_factory=list, description="Checks that would prevent a safe flight.") + advisory: list[str] = Field(default_factory=list, description="Worth knowing, but not disqualifying.") + checks: list[Check] = Field(default_factory=list) + + +class FileEntry(BaseModel): + name: str + size: int | None = None + is_dir: bool = False + + +class ArmState(BaseModel): + armed: bool + reason: str = "" + expires_at: str | None = None + note: str = "" diff --git a/src/mcbebop/server.py b/src/mcbebop/server.py index 4fd7456..14dbd55 100644 --- a/src/mcbebop/server.py +++ b/src/mcbebop/server.py @@ -8,6 +8,8 @@ through pyparrot, which prints from its receive thread. import logging import os import sys +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from fastmcp import FastMCP @@ -33,12 +35,25 @@ preflight_check() is the quickest way to learn whether the aircraft is healthy. def build_server(settings: Settings | None = None) -> FastMCP: """Build the server. Tests call this directly so they can inject settings.""" settings = settings or Settings() - mcp = FastMCP("mcbebop", instructions=INSTRUCTIONS, version=__version__) - mcp.settings_obj = settings # type: ignore[attr-defined] from mcbebop.tools import register_all - register_all(mcp, settings) + state_holder: dict = {} + + @asynccontextmanager + async def lifespan(server: FastMCP) -> AsyncIterator[None]: + try: + yield + finally: + # A session outlives any single tool call, so closing it is the + # server's job. Without this a stopped server can leave the drone + # streaming video and holding the one controller slot it has. + app_state = state_holder.get("app") + if app_state is not None: + await app_state.close() + + mcp = FastMCP("mcbebop", instructions=INSTRUCTIONS, version=__version__, lifespan=lifespan) + state_holder["app"] = register_all(mcp, settings) return mcp diff --git a/src/mcbebop/state.py b/src/mcbebop/state.py new file mode 100644 index 0000000..274ff2e --- /dev/null +++ b/src/mcbebop/state.py @@ -0,0 +1,73 @@ +"""Server-wide state: the drone session and the arming lock. + +Kept out of `tools/` deliberately. This module may use postponed annotations; +modules that define MCP tools may not, because FastMCP builds their schemas +from real annotation objects. +""" + +from __future__ import annotations + +import asyncio +import time +from dataclasses import dataclass, field + +from mcbebop.config import Settings + +SIM_TARGET = "sim" + + +@dataclass +class Arming: + """Whether commands that can move the aircraft are unlocked. + + Expiring is the point. A session that armed an hour ago to nudge the camera + should not still be able to spin motors, so this is checked on every send + rather than only at the moment of arming. + """ + + reason: str = "" + expires_at: float = 0.0 + + @property + def armed(self) -> bool: + return self.expires_at > time.monotonic() + + def arm(self, reason: str, minutes: int) -> None: + self.reason = reason + self.expires_at = time.monotonic() + minutes * 60 + + def disarm(self) -> None: + self.reason = "" + self.expires_at = 0.0 + + @property + def seconds_left(self) -> float: + return max(0.0, self.expires_at - time.monotonic()) + + +@dataclass +class AppState: + settings: Settings + session: object | None = None # DroneSession once connected + sim: object | None = None # FakeBebop when target is "sim" + target: str = "" + arming: Arming = field(default_factory=Arming) + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + @property + def connected(self) -> bool: + return self.session is not None and getattr(self.session, "connected", False) + + async def close(self) -> None: + if self.session is not None: + try: + await self.session.disconnect() + finally: + self.session = None + if self.sim is not None: + try: + self.sim.__exit__(None, None, None) + finally: + self.sim = None + self.target = "" + self.arming.disarm() diff --git a/src/mcbebop/tools/__init__.py b/src/mcbebop/tools/__init__.py index cc556e0..8494cd7 100644 --- a/src/mcbebop/tools/__init__.py +++ b/src/mcbebop/tools/__init__.py @@ -1,17 +1,20 @@ """Tool registration. -Each module exposes `register(mcp, settings)`; this is the only place that -knows the full set, so tool naming stays consistent. +The one place that knows the full set, so naming stays consistent across the +modules that would otherwise each invent their own. """ from fastmcp import FastMCP from mcbebop.config import Settings +from mcbebop.state import AppState +from mcbebop.tools import camera, command, connection, files, protocol, safety, state +from mcbebop.tools._common import set_state -def register_all(mcp: FastMCP, settings: Settings) -> None: - # Modules are added here as each stream lands. Stage 0 registers nothing, - # so the server starts and answers list_tools with an empty set. - modules: list = [] - for module in modules: +def register_all(mcp: FastMCP, settings: Settings) -> AppState: + app_state = AppState(settings=settings) + set_state(app_state) + for module in (connection, protocol, command, state, camera, files, safety): module.register(mcp, settings) + return app_state diff --git a/src/mcbebop/tools/_common.py b/src/mcbebop/tools/_common.py new file mode 100644 index 0000000..103f199 --- /dev/null +++ b/src/mcbebop/tools/_common.py @@ -0,0 +1,34 @@ +"""Helpers every tool module needs.""" + +from datetime import UTC, datetime, timedelta + +from mcbebop import errors +from mcbebop.state import AppState + +_STATE: AppState | None = None + + +def set_state(state: AppState) -> None: + """Install the shared state. Called once by `register_all`.""" + global _STATE + _STATE = state + + +def app() -> AppState: + if _STATE is None: # pragma: no cover - register_all always runs first + raise RuntimeError("server state was never installed") + return _STATE + + +def require_session(): + """The live session, or an error that says how to get one.""" + state = app() + if not state.connected: + raise errors.not_connected() + return state.session + + +def expiry_iso(seconds_left: float) -> str | None: + if seconds_left <= 0: + return None + return (datetime.now(UTC) + timedelta(seconds=seconds_left)).isoformat(timespec="seconds") diff --git a/src/mcbebop/tools/camera.py b/src/mcbebop/tools/camera.py new file mode 100644 index 0000000..569d401 --- /dev/null +++ b/src/mcbebop/tools/camera.py @@ -0,0 +1,116 @@ +"""Live video from the drone's camera.""" + +from pathlib import Path +from typing import Annotated, Any + +from fastmcp import Context, FastMCP +from fastmcp.tools import ToolResult +from fastmcp.utilities.types import Image +from mcp.types import TextContent +from pydantic import Field + +from mcbebop.config import Settings +from mcbebop.media import video +from mcbebop.protocol import xml_index +from mcbebop.tools._common import require_session + + +def _sender(session): + """Adapt a session to the plain callable `media.video` expects. + + The video module deliberately knows nothing about sessions or the protocol + index, so the wiring lives here. + """ + + async def send(name: str, args: dict[str, Any]): + return await session.send(xml_index.get(name), args) + + return send + + +def _stamp() -> str: + from datetime import UTC, datetime + + return datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"openWorldHint": True}) + async def camera_snapshot( + ctx: Context, + max_width: Annotated[ + int, Field(description="Downscale to this width before returning it.", ge=64, le=1920) + ] = 0, + exposure: Annotated[ + float | None, + Field(description="Exposure while capturing, -1.5 to 1.5. Restored afterwards.", ge=-1.5, le=1.5), + ] = None, + settle_seconds: Annotated[ + float, + Field( + description=( + "Discard this much video first. The opening frames reference parameter sets that " + "have not arrived yet, and the camera's exposure needs a moment to settle." + ), + ge=0, + le=20, + ), + ] = 4.0, + ) -> ToolResult: + """Take one frame from the live camera and return it as an image. + + Enables the video stream, grabs a frame, then turns the stream off + again. The stream is 856x480: the drone records 1080p internally but + only streams the lower resolution. + """ + session = require_session() + capture_dir = Path(settings.capture_dir) + sdp = video.write_sdp(capture_dir / "bebop.sdp") + out = capture_dir / f"frame-{_stamp()}.jpg" + sink = video.snapshot(sdp, out, settle_seconds=settle_seconds) + + async with video.StreamSession( + sink, _sender(session), exposure=exposure, read_state=session.values + ) as stream: + await stream.wait(settle_seconds + 20) + + if not out.exists() or out.stat().st_size == 0: + raise video.VideoUnavailable( + "no frame was captured. Is ffmpeg installed, and is the drone's camera unobstructed?" + ) + + raw = out.read_bytes() + shown = video.downscale(raw, max_width or settings.snapshot_max_width) + attitude = session.values(["AttitudeChanged", "AltitudeChanged", "GpsLocationChanged"]) + return ToolResult( + content=[ + TextContent(type="text", text=f"frame saved to {out} ({len(raw)} bytes)"), + Image(data=shown, format="jpeg").to_image_content(), + ], + structured_content={"path": str(out), "bytes": len(raw), "state_at_capture": attitude}, + ) + + @mcp.tool(annotations={"openWorldHint": True}) + async def camera_record( + ctx: Context, + seconds: Annotated[float, Field(description="How long to record for.", ge=1, le=300)] = 15.0, + exposure: Annotated[ + float | None, Field(description="Exposure while recording, -1.5 to 1.5.", ge=-1.5, le=1.5) + ] = None, + ) -> dict: + """Record the live camera to an MP4, copying the drone's H.264 without re-encoding.""" + session = require_session() + capture_dir = Path(settings.capture_dir) + sdp = video.write_sdp(capture_dir / "bebop.sdp") + out = capture_dir / f"video-{_stamp()}.mp4" + sink = video.record(sdp, out, seconds) + + async with video.StreamSession( + sink, _sender(session), exposure=exposure, read_state=session.values + ) as stream: + await stream.wait(seconds + 20) + + size = out.stat().st_size if out.exists() else 0 + if not size: + raise video.VideoUnavailable("nothing was recorded. Is ffmpeg installed?") + return {"path": str(out), "bytes": size, "seconds": seconds} diff --git a/src/mcbebop/tools/command.py b/src/mcbebop/tools/command.py new file mode 100644 index 0000000..137486e --- /dev/null +++ b/src/mcbebop/tools/command.py @@ -0,0 +1,90 @@ +"""The generic sender: any of Parrot's commands, gated by what it can do.""" + +from typing import Annotated, Any + +from fastmcp import Context, FastMCP +from pydantic import Field + +from mcbebop import errors +from mcbebop.config import Settings +from mcbebop.models import SendResult +from mcbebop.protocol import xml_index +from mcbebop.protocol.safety import ALWAYS_ALLOWED +from mcbebop.protocol.types import Direction, Tier +from mcbebop.tools._common import app, require_session + +LOCKED_TIERS = (Tier.MOTION, Tier.ENVELOPE) + + +def check_allowed(spec) -> None: + """Refuse a command the session has not earned the right to send. + + Landing and Emergency are exempt on purpose: if the aircraft is airborne, + refusing to bring it down is the more dangerous answer. + """ + if spec.full_name in ALWAYS_ALLOWED or spec.tier not in LOCKED_TIERS: + return + if not app().arming.armed: + raise errors.needs_arming(spec.full_name, str(spec.tier)) + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"openWorldHint": True}) + async def send_command( + ctx: Context, + name: Annotated[ + str, Field(description="Full name, e.g. 'ardrone3.PictureSettings.PictureFormatSelection'.") + ], + args: Annotated[ + dict[str, Any], + Field( + description=( + "Arguments by name. Enum arguments take the member name as a string. " + "Call command_info() for the names, types and allowed values." + ) + ), + ] = {}, # noqa: B006 - FastMCP reads the default into the schema; never mutated + wait_for_confirm: Annotated[ + bool, + Field( + description=( + "Also wait for the event Parrot says confirms this command, where one is declared. " + "Stronger evidence than the acknowledgement alone." + ) + ), + ] = True, + ) -> SendResult: + """Send any command the drone understands. + + Commands are classified by consequence. Observing and configuring work + straight away; anything that can move the aircraft or change its flight + envelope refuses until arm() is called. + """ + spec = xml_index.get(name) + if spec is None: + raise errors.unknown_command(name, [c.full_name for c in xml_index.search(name)][:5]) + if spec.direction is Direction.FROM_DRONE: + raise errors.not_sendable(name) + + check_allowed(spec) + session = require_session() + + note = "" + if xml_index.supports_bebop2(spec.support) is False: + note = f"The XML does not list the Bebop 2 as supporting this (support={spec.support!r})." + elif spec.deprecated: + note = "Parrot marks this command deprecated; the drone may ignore it." + + try: + result = await session.send(spec, dict(args), confirm=wait_for_confirm) + except ValueError as exc: + raise errors.bad_argument(name, str(exc)) from exc + + return SendResult( + command=spec.full_name, + args=dict(args), + acked=bool(result.get("acked")), + confirmed=bool(result.get("confirmed")), + elapsed_ms=float(result.get("elapsed_ms", 0.0)), + note=note, + ) diff --git a/src/mcbebop/tools/connection.py b/src/mcbebop/tools/connection.py new file mode 100644 index 0000000..ab4bf00 --- /dev/null +++ b/src/mcbebop/tools/connection.py @@ -0,0 +1,100 @@ +"""Opening and closing the link to the aircraft.""" + +import asyncio +from typing import Annotated + +from fastmcp import Context, FastMCP +from pydantic import Field + +from mcbebop import errors +from mcbebop.config import Settings +from mcbebop.models import ConnectionInfo +from mcbebop.state import SIM_TARGET +from mcbebop.tools._common import app, expiry_iso + + +def _info() -> ConnectionInfo: + state = app() + session = state.session + stats = session.link_stats() if state.connected else {} + return ConnectionInfo( + connected=state.connected, + target=state.target or "none", + armed=state.arming.armed, + armed_until=expiry_iso(state.arming.seconds_left), + telemetry_keys=stats.get("telemetry_keys", 0) or 0, + last_update_age=stats.get("last_event_age"), + undecodable_events=stats.get("undecodable_events", 0) or 0, + ) + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"openWorldHint": True}) + async def connect( + ctx: Context, + target: Annotated[ + str, + Field( + description=( + "Where to connect: an IP address, 'drone' for the configured address " + f"({settings.drone_ip}), or 'sim' to run a built-in simulator instead of the " + "aircraft. Use 'sim' to rehearse anything that would move a real drone." + ) + ), + ] = "drone", + ) -> ConnectionInfo: + """Open a session with the drone. Nothing else works until this succeeds. + + The drone serves one controller at a time, so close any phone app first. + This machine must already have joined the drone's own Wi-Fi network. + """ + state = app() + async with state.lock: + if state.connected: + raise errors.already_connected(state.target) + + from mcbebop.arsdk.session import DroneSession + from mcbebop.arsdk.types import HandshakeError + + if target == SIM_TARGET: + from mcbebop.sim import FakeBebop + + sim = FakeBebop() + sim.__enter__() + state.sim = sim + session = DroneSession(ip=sim.host, discovery_port=sim.discovery_port) + else: + ip = settings.drone_ip if target == "drone" else target + session = DroneSession(ip=ip) + + try: + await session.connect() + except HandshakeError as exc: + await state.close() + status = getattr(exc, "status", None) + raise errors.handshake_refused(status) if status else errors.not_connected() from exc + except OSError as exc: + await state.close() + raise errors.not_connected() from exc + + state.session = session + state.target = target if target == SIM_TARGET else getattr(session, "ip", target) + await session.request_full_state() + # The drone answers the state request as a burst over the next + # moment, so pause before reporting a count: returning zero keys + # from a healthy connection reads as a failure. + await asyncio.sleep(1.0) + return _info() + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def connection_status(ctx: Context) -> ConnectionInfo: + """Whether a session is open, how fresh its telemetry is, and whether it is armed.""" + return _info() + + @mcp.tool(annotations={"openWorldHint": False}) + async def disconnect(ctx: Context) -> ConnectionInfo: + """Close the session. Also disarms, and stops the drone's video if it was started.""" + state = app() + async with state.lock: + await state.close() + return _info() diff --git a/src/mcbebop/tools/files.py b/src/mcbebop/tools/files.py new file mode 100644 index 0000000..81c35b1 --- /dev/null +++ b/src/mcbebop/tools/files.py @@ -0,0 +1,91 @@ +"""Reading files off the drone over FTP, and its debug shell.""" + +import asyncio +from typing import Annotated, Literal + +from fastmcp import Context, FastMCP +from fastmcp.exceptions import ToolError +from pydantic import Field + +from mcbebop.config import Settings +from mcbebop.files import ftp, shell +from mcbebop.models import FileEntry +from mcbebop.tools._common import app + +AreaName = Literal["media", "flightplans", "logs"] + + +def _host(settings: Settings) -> str: + """FTP talks to the aircraft directly, so the simulator has nothing to serve.""" + state = app() + if state.target == "sim": + raise ToolError( + "File access talks to the real aircraft's FTP server; the simulator has no filesystem. " + "Connect to the drone to use this." + ) + return state.target or settings.drone_ip + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True}) + async def list_files( + ctx: Context, + area: Annotated[ + AreaName, + Field( + description=( + "media: photos and videos. flightplans: stored missions. " + "logs: the drone's own blackbox and debug files." + ) + ), + ] = "media", + path: Annotated[str, Field(description="Subdirectory within the area. Empty is its root.")] = "", + ) -> list[FileEntry]: + """List files on the drone. + + Read-only. The firmware-update channel is deliberately not reachable + through this tool. + """ + host = _host(settings) + entries = await asyncio.to_thread(ftp.list_dir, ftp.resolve_area(area), path, host=host) + return [FileEntry(name=e.name, size=e.size, is_dir=e.is_dir) for e in entries] + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True}) + async def fetch_file( + ctx: Context, + area: Annotated[AreaName, Field(description="Which area the path is in.")], + path: Annotated[str, Field(description="Path within the area, as shown by list_files.")], + max_mb: Annotated[int, Field(description="Refuse anything larger.", ge=1, le=512)] = 32, + ) -> dict: + """Download one file from the drone to this machine, and return where it landed.""" + host = _host(settings) + dest = await asyncio.to_thread( + ftp.fetch, + ftp.resolve_area(area), + path, + host=host, + capture_dir=settings.capture_dir, + max_bytes=max_mb * 1024 * 1024, + ) + return {"path": str(dest), "bytes": dest.stat().st_size} + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True}) + async def shell_read( + ctx: Context, + command: Annotated[str, Field(description=f"One of: {', '.join(sorted(shell.allowed_commands()))}.")], + args: Annotated[ + list[str], + Field(description="Arguments. Shell metacharacters are refused rather than escaped."), + ] = [], # noqa: B006 - read into the schema by FastMCP; never mutated + ) -> dict: + """Run one read-only command on the drone's debug shell. + + The shell only exists after the drone's debug mode is switched on with + four short presses of its power button, and it does not survive a + reboot. It is an unauthenticated root shell, which is why this tool + takes an allow-list of command names rather than trying to filter a + free-form line. + """ + host = _host(settings) + result = await asyncio.to_thread(shell.run, command, list(args), host=host) + return {"command": result.command, "stdout": result.stdout} diff --git a/src/mcbebop/tools/protocol.py b/src/mcbebop/tools/protocol.py new file mode 100644 index 0000000..a07f683 --- /dev/null +++ b/src/mcbebop/tools/protocol.py @@ -0,0 +1,93 @@ +"""Browsing Parrot's command set, so nothing has to be known in advance.""" + +from typing import Annotated, Literal + +from fastmcp import Context, FastMCP +from pydantic import Field + +from mcbebop import errors +from mcbebop.config import Settings +from mcbebop.models import ArgumentInfo, CommandDetail, CommandSummary +from mcbebop.protocol import xml_index + + +def _summary(spec) -> CommandSummary: + return CommandSummary( + name=spec.full_name, + title=spec.title, + tier=str(spec.tier), + direction=str(spec.direction), + bebop2=xml_index.supports_bebop2(spec.support), + deprecated=spec.deprecated, + ) + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def list_commands( + ctx: Context, + search: Annotated[ + str, Field(description="Case-insensitive substring of the name or title. Empty lists everything.") + ] = "", + direction: Annotated[ + Literal["any", "to_drone", "from_drone"], + Field(description="to_drone are commands you can send; from_drone are events the drone emits."), + ] = "any", + tier: Annotated[ + Literal["any", "observe", "config", "envelope", "motion"], + Field(description="Filter by what a command can do. motion and envelope require arm()."), + ] = "any", + bebop2_only: Annotated[ + bool, + Field(description="Hide commands the XML says the Bebop 2 does not support. Unstated ones stay."), + ] = True, + ) -> list[CommandSummary]: + """Search the 264 commands Parrot defines. Start here when you do not know the name.""" + out = [] + for spec in xml_index.all_commands(): + if search and search.lower() not in f"{spec.full_name} {spec.title}".lower(): + continue + if direction != "any" and str(spec.direction) != direction: + continue + if tier != "any" and str(spec.tier) != tier: + continue + if bebop2_only and xml_index.supports_bebop2(spec.support) is False: + continue + out.append(_summary(spec)) + return sorted(out, key=lambda c: c.name) + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def command_info( + ctx: Context, + name: Annotated[str, Field(description="Full name, e.g. 'ardrone3.MediaStreaming.VideoEnable'.")], + ) -> CommandDetail: + """Everything Parrot documents about one command: arguments, enum values, support, tier.""" + spec = xml_index.get(name) + if spec is None: + near = [c.full_name for c in xml_index.search(name)][:5] + raise errors.unknown_command(name, near) + + confirmed = [] + for exp in spec.expectations: + other = xml_index.by_ids(exp.ids) + label = other.full_name if other else "-".join(str(i) for i in exp.ids) + if exp.fields: + label += " where " + ", ".join(f"{k}={v}" for k, v in exp.fields.items()) + confirmed.append(label) + + return CommandDetail( + **_summary(spec).model_dump(), + ids=list(spec.ids), + doc=spec.doc, + buffer=str(spec.buffer), + confirmed_by=confirmed, + args=[ + ArgumentInfo( + name=a.name, + type=a.type, + doc=a.doc, + enum_values=[m.name for m in a.members], + ) + for a in spec.args + ], + ) diff --git a/src/mcbebop/tools/safety.py b/src/mcbebop/tools/safety.py new file mode 100644 index 0000000..6335081 --- /dev/null +++ b/src/mcbebop/tools/safety.py @@ -0,0 +1,44 @@ +"""Arming: the deliberate step before anything can move the aircraft.""" + +from typing import Annotated + +from fastmcp import Context, FastMCP +from pydantic import Field + +from mcbebop.config import Settings +from mcbebop.models import ArmState +from mcbebop.tools._common import app, expiry_iso + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"openWorldHint": False}) + async def arm( + ctx: Context, + reason: Annotated[ + str, + Field(description="Why, in your own words. Recorded, and shown back in connection_status."), + ], + minutes: Annotated[ + int, Field(description="How long the unlock lasts before it expires on its own.", ge=1, le=60) + ] = 0, + ) -> ArmState: + """Unlock commands that can move the aircraft or change its flight envelope. + + Only do this when the propellers are off, or the drone is somewhere it + is safe for it to move. The unlock expires on its own and is cleared by + disconnect. Landing and Emergency never need it. + """ + state = app() + state.arming.arm(reason, minutes or settings.arm_minutes) + return ArmState( + armed=True, + reason=reason, + expires_at=expiry_iso(state.arming.seconds_left), + note="Motion and envelope commands are unlocked until this expires.", + ) + + @mcp.tool(annotations={"openWorldHint": False}) + async def disarm(ctx: Context) -> ArmState: + """Re-lock motion commands immediately, without waiting for the timer.""" + app().arming.disarm() + return ArmState(armed=False, note="Motion and envelope commands are locked again.") diff --git a/src/mcbebop/tools/state.py b/src/mcbebop/tools/state.py new file mode 100644 index 0000000..5322254 --- /dev/null +++ b/src/mcbebop/tools/state.py @@ -0,0 +1,159 @@ +"""Reading what the drone reports about itself.""" + +import asyncio +from typing import Annotated + +from fastmcp import Context, FastMCP +from pydantic import Field + +from mcbebop.config import Settings +from mcbebop.models import Check, Preflight, StateValue +from mcbebop.tools._common import require_session + +# The drone says a position is unavailable by reporting this, not by omitting it. +GPS_UNAVAILABLE = 500.0 +SENSOR_PREFIX = "SensorsStatesListChanged[" + + +def _sensor_results(values: dict) -> dict[str, bool]: + """Pull the self-test results back out of the per-element keys. + + These arrive as one event per sensor, all carrying the same argument names, + which is why the session stores them under `Command[element]_arg`. + """ + out = {} + for key, value in values.items(): + if key.startswith(SENSOR_PREFIX) and key.endswith("_sensorState"): + out[key.split("[", 1)[1].split("]", 1)[0]] = bool(value) + return out + + +def register(mcp: FastMCP, settings: Settings) -> None: + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def get_state( + ctx: Context, + keys: Annotated[ + list[str], + Field( + description=( + "Which values to read, as '_' (e.g. 'BatteryStateChanged_percent'). " + "A bare prefix matches everything under it. Empty returns everything reported." + ) + ), + ] = [], # noqa: B006 - read into the schema by FastMCP; never mutated + ) -> dict[str, StateValue]: + """Read the drone's reported state, with the age of each value. + + Age matters: these are the last thing the drone said, not a fresh read. + A large age means the link went quiet, not that the value is current. + """ + session = require_session() + raw = session.state(keys or None) + return {k: StateValue(value=v["value"], age=round(v["age"], 2)) for k, v in raw.items()} + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def watch_state( + ctx: Context, + keys: Annotated[ + list[str], Field(description="Values to watch, same form as get_state. A bare prefix matches.") + ], + seconds: Annotated[float, Field(description="How long to watch for.", ge=0.5, le=60)] = 5.0, + ) -> dict[str, list]: + """Watch values change over a window, to see motion rather than a snapshot.""" + session = require_session() + start = session.state(keys or None) + await asyncio.sleep(seconds) + end = session.state(keys or None) + + changes: dict[str, list] = {} + for key, after in end.items(): + before = start.get(key, {}).get("value") + if before != after["value"]: + changes[key] = [before, after["value"]] + return changes or {"(no change)": [None, None]} + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False}) + async def preflight_check(ctx: Context) -> Preflight: + """A go/no-go summary: battery, link, sensor self-tests, motors, GPS and limits. + + Advisory. It reports what the drone says about itself and cannot see + whether the propellers are on, the weather, or what is in the way. + """ + session = require_session() + v = session.values() + checks: list[Check] = [] + blocking: list[str] = [] + advisory: list[str] = [] + + def add(name: str, ok: bool | None, detail: str, blocks: bool = False) -> None: + checks.append(Check(name=name, ok=ok, detail=detail)) + if ok is False: + (blocking if blocks else advisory).append(f"{name}: {detail}") + elif ok is None: + advisory.append(f"{name}: not reported") + + battery = v.get("BatteryStateChanged_percent") + add( + "battery", + None if battery is None else battery >= 30, + "not reported" if battery is None else f"{battery}%", + blocks=True, + ) + + rssi = v.get("WifiSignalChanged_rssi") + add("link", None if rssi is None else rssi > -75, "not reported" if rssi is None else f"{rssi} dBm") + + sensors = _sensor_results(v) + failed = sorted(n for n, ok in sensors.items() if not ok) + add( + "sensor self-tests", + None if not sensors else not failed, + "none reported" + if not sensors + else (f"failing: {', '.join(failed)}" if failed else f"{len(sensors)} ok"), + blocks=True, + ) + + motor = v.get("MotorErrorStateChanged_motorError") + motor = getattr(motor, "value", motor) + add( + "motors", None if motor is None else motor == "noError", str(motor or "not reported"), blocks=True + ) + + alert = getattr(v.get("AlertStateChanged_state"), "value", v.get("AlertStateChanged_state")) + add( + "alert state", + None if alert is None else alert == "none", + str(alert or "not reported"), + blocks=True, + ) + + flying = getattr(v.get("FlyingStateChanged_state"), "value", v.get("FlyingStateChanged_state")) + add("flying state", None if flying is None else flying == "landed", str(flying or "not reported")) + + fixed = v.get("GPSFixStateChanged_fixed") + sats = v.get("NumberOfSatelliteChanged_numberOfSatellite") + add( + "gps", + None if fixed is None else bool(fixed), + "no fix" + (f", {sats} satellites" if sats is not None else "") + if not fixed + else f"fix, {sats} satellites", + ) + + calib = v.get("MagnetoCalibrationRequiredState_required") + add( + "compass calibration", + None if calib is None else not calib, + "not reported" if calib is None else ("required" if calib else "not required"), + ) + + geofence = v.get("NoFlyOverMaxDistanceChanged_shouldNotFlyOver") + max_dist = v.get("MaxDistanceChanged_current") + add( + "geofence", + None if geofence is None else bool(geofence), + f"off, max distance {max_dist} m" if geofence == 0 else f"on at {max_dist} m", + ) + + return Preflight(ready=not blocking, blocking=blocking, advisory=advisory, checks=checks) diff --git a/tests/test_tools.py b/tests/test_tools.py new file mode 100644 index 0000000..2626ed9 --- /dev/null +++ b/tests/test_tools.py @@ -0,0 +1,131 @@ +"""The tools layer, driven through a real MCP client against the simulator.""" + +import pytest +from fastmcp import Client +from fastmcp.exceptions import ToolError + +from mcbebop.protocol import xml_index +from mcbebop.protocol.safety import ALWAYS_ALLOWED +from mcbebop.protocol.types import Direction, Tier +from mcbebop.server import build_server + + +def data(result): + return result.structured_content if result.structured_content is not None else result.data + + +@pytest.fixture +async def client(): + async with Client(build_server()) as c: + yield c + + +@pytest.fixture +async def connected(client): + await client.call_tool("connect", {"target": "sim"}) + yield client + await client.call_tool("disconnect", {}) + + +async def test_every_tool_is_registered(client): + names = {t.name for t in await client.list_tools()} + assert names == { + "connect", "disconnect", "connection_status", "list_commands", "command_info", + "send_command", "get_state", "watch_state", "preflight_check", "camera_snapshot", + "camera_record", "list_files", "fetch_file", "shell_read", "arm", "disarm", + } # fmt: skip + + +async def test_tools_refuse_before_connecting(client): + calls = ( + ("get_state", {}), + ("preflight_check", {}), + ("send_command", {"name": "common.Common.AllStates"}), + ) + for tool, args in calls: + with pytest.raises(ToolError, match=r"[Nn]ot connected"): + await client.call_tool(tool, args) + + +async def test_connect_reports_live_telemetry(connected): + info = data(await connected.call_tool("connection_status", {})) + assert info["connected"] and info["target"] == "sim" + assert info["telemetry_keys"] > 20, "the drone's state burst should have arrived" + assert info["last_update_age"] is not None and info["last_update_age"] < 10 + assert info["undecodable_events"] == 0, "every event the simulator sends should decode" + + +@pytest.mark.parametrize( + "name", + [c.full_name for c in xml_index.all_commands() + if c.direction is Direction.TO_DRONE and c.tier in (Tier.MOTION, Tier.ENVELOPE) + and c.full_name not in ALWAYS_ALLOWED], +) # fmt: skip +async def test_every_locked_command_is_refused_unarmed(connected, name): + """The gate must hold for all of them, not just the obvious ones.""" + with pytest.raises(ToolError, match="locked"): + await connected.call_tool("send_command", {"name": name}) + + +@pytest.mark.parametrize("name", sorted(ALWAYS_ALLOWED)) +async def test_landing_and_emergency_are_never_locked(connected, name): + """Refusing to bring down an airborne aircraft is the more dangerous answer.""" + result = data(await connected.call_tool("send_command", {"name": name})) + assert result["acked"] + + +async def test_arming_unlocks_and_disarming_relocks(connected): + takeoff = "ardrone3.Piloting.TakeOff" + with pytest.raises(ToolError, match="locked"): + await connected.call_tool("send_command", {"name": takeoff}) + + armed = data(await connected.call_tool("arm", {"reason": "test", "minutes": 1})) + assert armed["armed"] and armed["expires_at"] + assert data(await connected.call_tool("send_command", {"name": takeoff}))["acked"] + + await connected.call_tool("disarm", {}) + with pytest.raises(ToolError, match="locked"): + await connected.call_tool("send_command", {"name": takeoff}) + + +async def test_events_cannot_be_sent(connected): + with pytest.raises(ToolError, match="event the drone emits"): + await connected.call_tool("send_command", {"name": "common.CommonState.BatteryStateChanged"}) + + +async def test_unknown_command_suggests_alternatives(connected): + with pytest.raises(ToolError, match="No command named"): + await connected.call_tool("send_command", {"name": "ardrone3.Piloting.TakeOf"}) + + +async def test_preflight_sees_a_failing_sensor(connected): + """The simulator fails its magnetometer; a flat telemetry store would hide it.""" + report = data(await connected.call_tool("preflight_check", {})) + sensors = next(c for c in report["checks"] if c["name"] == "sensor self-tests") + assert sensors["ok"] is False and "magnetometer" in sensors["detail"] + assert report["ready"] is False + assert any("magnetometer" in b for b in report["blocking"]) + + +async def test_command_info_carries_the_confirming_event(connected): + info = data(await connected.call_tool("command_info", {"name": "ardrone3.MediaStreaming.VideoEnable"})) + assert info["ids"] == [1, 21, 0] + assert info["args"][0]["name"] == "enable" + assert any("VideoEnableChanged" in c for c in info["confirmed_by"]) + + +async def test_list_commands_filters(connected): + args = {"direction": "to_drone", "bebop2_only": False} + sendable = data(await connected.call_tool("list_commands", args)) + sendable = sendable.get("result", sendable) + # 101 in Parrot's XML, plus FlatTrim which the firmware accepts but the XML omits. + assert len(sendable) == 102 + motion = data(await connected.call_tool("list_commands", {"tier": "motion", "bebop2_only": False})) + motion = motion.get("result", motion) + assert all(c["tier"] == "motion" for c in motion) + + +async def test_file_tools_reject_the_simulator(connected): + """The simulator has no filesystem; saying so beats a confusing timeout.""" + with pytest.raises(ToolError, match="simulator has no filesystem"): + await connected.call_tool("list_files", {"area": "media"})