From ef8c4658e0cb2d60736d14cc44a5ab2eb51571ac Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 2 Oct 2026 00:07:43 -0600 Subject: [PATCH] Protocol layer: XML index, codec, safety tiers, expectations Parse Parrot's common.xml and ardrone3.xml into CommandSpec objects rather than transcribing 264 commands by hand. Direction is derived from comment result=/triggered= with a class-name-suffix fallback for the 31 commands that state neither, giving 101 to-drone and 163 from-drone. ardrone3.Piloting.FlatTrim is injected: the firmware accepts it but neither the vendored snapshot nor upstream defines it. codec encodes the echoes. 90 tests, counts asserted so a bad parse fails loudly. --- src/mcbebop/protocol/codec.py | 231 +++++++++++++++++++ src/mcbebop/protocol/expectations.py | 89 ++++++++ src/mcbebop/protocol/safety.py | 96 ++++++++ src/mcbebop/protocol/xml_index.py | 325 +++++++++++++++++++++++++++ tests/test_codec.py | 244 ++++++++++++++++++++ tests/test_expectations.py | 113 ++++++++++ tests/test_protocol_index.py | 222 ++++++++++++++++++ tests/test_safety.py | 131 +++++++++++ 8 files changed, 1451 insertions(+) create mode 100644 src/mcbebop/protocol/codec.py create mode 100644 src/mcbebop/protocol/expectations.py create mode 100644 src/mcbebop/protocol/safety.py create mode 100644 src/mcbebop/protocol/xml_index.py create mode 100644 tests/test_codec.py create mode 100644 tests/test_expectations.py create mode 100644 tests/test_protocol_index.py create mode 100644 tests/test_safety.py diff --git a/src/mcbebop/protocol/codec.py b/src/mcbebop/protocol/codec.py new file mode 100644 index 0000000..dfeb6d8 --- /dev/null +++ b/src/mcbebop/protocol/codec.py @@ -0,0 +1,231 @@ +"""Turning arguments into bytes and bytes back into telemetry. + +Two things here cost real time if you get them from the wrong source. + +The command header is ` struct.Struct: + if arg.is_enum: + return struct.Struct(_ENUM_FORMAT) + if arg.is_bitfield: + # bitfield::, e.g. bitfield:u8:MyFlags. The width + # gives the wire size; the value is a mask of member positions. Neither + # common.xml nor ardrone3.xml uses one, but Parrot's other projects do + # and an index loaded from those must not fall over. + _, _, rest = arg.type.partition(":") + width, _, _ = rest.partition(":") + if width not in _FORMATS: + raise ValueError(f"argument {arg.name!r} has unknown bitfield width {width!r}") + return struct.Struct(_FORMATS[width]) + if arg.type in _FORMATS: + return struct.Struct(_FORMATS[arg.type]) + raise ValueError(f"argument {arg.name!r} has unsupported type {arg.type!r}") + + +def _member_value(arg: ArgSpec, value: Any) -> int: + """Resolve an enum argument to its wire value. + + Names are preferred: an agent writing `state="takingoff"` cannot silently + pick the wrong member the way an off-by-one integer can. Integers are still + accepted because captured traffic and replay give us positions. + """ + if isinstance(value, bool): + raise ValueError(f"argument {arg.name!r} is an enum; pass a member name, not a bool") + if isinstance(value, int): + if arg.members and not any(member.value == value for member in arg.members): + allowed = ", ".join(f"{m.value}={m.name}" for m in arg.members) + raise ValueError(f"argument {arg.name!r} has no member with value {value}; allowed: {allowed}") + return value + if isinstance(value, str): + for member in arg.members: + if member.name == value: + return member.value + allowed = ", ".join(member.name for member in arg.members) or "(none declared)" + raise ValueError(f"argument {arg.name!r} has no member named {value!r}; allowed: {allowed}") + raise ValueError( + f"argument {arg.name!r} is an enum; pass a member name or an int, not {type(value).__name__}" + ) + + +def _bitfield_value(arg: ArgSpec, value: Any) -> int: + if isinstance(value, bool): + raise ValueError(f"argument {arg.name!r} is a bitfield; pass member names or an int mask") + if isinstance(value, int): + return value + if isinstance(value, str): + names: Iterable[str] = (value,) + elif isinstance(value, Iterable): + names = value + else: + raise ValueError(f"argument {arg.name!r} is a bitfield; pass member names or an int mask") + + mask = 0 + lookup = {member.name: member.value for member in arg.members} + for name in names: + if name not in lookup: + allowed = ", ".join(lookup) or "(none declared)" + raise ValueError(f"argument {arg.name!r} has no member named {name!r}; allowed: {allowed}") + mask |= 1 << lookup[name] + return mask + + +def _encode_one(arg: ArgSpec, value: Any) -> bytes: + if arg.type == "string": + if not isinstance(value, str): + raise ValueError(f"argument {arg.name!r} is a string, got {type(value).__name__}") + if "\x00" in value: + # The wire format has no length prefix, so an embedded NUL would + # truncate the value and shift every argument after it. + raise ValueError(f"argument {arg.name!r} contains a NUL byte, which terminates the string") + return value.encode("utf-8") + b"\x00" + + packer = _struct_for(arg) + + if arg.is_enum: + return packer.pack(_member_value(arg, value)) + if arg.is_bitfield: + return packer.pack(_bitfield_value(arg, value)) + + if arg.type in ("float", "double"): + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"argument {arg.name!r} is a {arg.type}, got {type(value).__name__}") + number: int | float = float(value) + else: + if isinstance(value, bool): + # True would pack as 1 and read as deliberate. It usually is, but + # an accidental bool here has been a real bug, so say so. + raise ValueError(f"argument {arg.name!r} is a {arg.type}; pass 1 or 0 rather than a bool") + if not isinstance(value, int): + raise ValueError(f"argument {arg.name!r} is a {arg.type}, got {type(value).__name__}") + number = value + + try: + return packer.pack(number) + except struct.error as exc: + raise ValueError(f"argument {arg.name!r} ({arg.type}) cannot hold {value!r}: {exc}") from exc + + +def encode_args(spec: CommandSpec, args: Mapping[str, Any] | None = None) -> bytes: + """Pack a command's arguments in declaration order. + + Order is the XML's, not the caller's: the wire format has no field names. + """ + given = dict(args or {}) + unknown = [name for name in given if not any(arg.name == name for arg in spec.args)] + if unknown: + expected = ", ".join(arg.name for arg in spec.args) or "(none)" + raise ValueError( + f"{spec.full_name} has no argument {', '.join(sorted(unknown))}; expected: {expected}" + ) + + out = bytearray() + for arg in spec.args: + if arg.name not in given: + raise ValueError(f"{spec.full_name} requires argument {arg.name!r} ({arg.type})") + out += _encode_one(arg, given[arg.name]) + return bytes(out) + + +def encode_command(spec: CommandSpec, args: Mapping[str, Any] | None = None) -> bytes: + """Header plus arguments: the full command payload of one ARNetwork frame.""" + return HEADER.pack(*spec.ids) + encode_args(spec, args) + + +def _decode_one(arg: ArgSpec, payload: bytes, offset: int) -> tuple[Any, int]: + if arg.type == "string": + end = payload.find(b"\x00", offset) + if end < 0: + raise ValueError(f"argument {arg.name!r} is an unterminated string") + return payload[offset:end].decode("utf-8", errors="replace"), end + 1 + + unpacker = _struct_for(arg) + if offset + unpacker.size > len(payload): + raise ValueError( + f"argument {arg.name!r} ({arg.type}) needs {unpacker.size} bytes " + f"but only {len(payload) - offset} remain" + ) + (value,) = unpacker.unpack_from(payload, offset) + offset += unpacker.size + + if arg.is_enum: + for member in arg.members: + if member.value == value: + # Hand back the name: a telemetry key reading "takingoff" is + # worth more to the reader than a 3. + return member.name, offset + return value, offset # a member this XML snapshot does not know about + return value, offset + + +def decode_args(spec: CommandSpec, payload: bytes, offset: int = 0) -> dict[str, Any]: + """Decode arguments into `_` keys. + + The prefix is the command name alone, not the dotted path, because that is + the shape of every telemetry key in the captures we already have. + """ + values: dict[str, Any] = {} + for arg in spec.args: + value, offset = _decode_one(arg, payload, offset) + values[f"{spec.event_key_prefix}_{arg.name}"] = value + return values + + +def decode_event( + payload: bytes, table: ProtocolIndex | None = None +) -> tuple[tuple[int, int, int], dict[str, Any]]: + """Decode one command payload, header included. + + An unknown id triple raises rather than returning an empty dict. The XML is + demonstrably incomplete (FlatTrim is missing from it), so an event we + cannot name is evidence worth surfacing, not noise to drop. The receive + loop is expected to catch this, count it and carry on. + + `table` exists so a test or a replay can decode against a pinned index + rather than the process-wide one. + """ + if len(payload) < HEADER.size: + raise ValueError( + f"command payload is {len(payload)} bytes, too short for a {HEADER.size}-byte header" + ) + ids = HEADER.unpack_from(payload, 0) + spec = (table or xml_index.index()).by_ids(ids) + if spec is None: + raise ValueError(f"no command with ids {ids}; it is not in the vendored ARSDK XML") + return ids, decode_args(spec, payload, HEADER.size) diff --git a/src/mcbebop/protocol/expectations.py b/src/mcbebop/protocol/expectations.py new file mode 100644 index 0000000..cd62088 --- /dev/null +++ b/src/mcbebop/protocol/expectations.py @@ -0,0 +1,89 @@ +"""Parsing Parrot's `` blocks. + +A sendable command's ack only proves the frame arrived. The expectations block +names the event the drone emits when it has actually done the thing, which is +what lets `send_command` report "it took off" rather than "the radio worked". +82 of the 102 sendable commands carry one. + +The grammar is not documented anywhere; it is read off the two XML files: + + #-- a reference + #0-3-2(name: this.name) with field constraints + #0-3-7(automatic: this.automatic) #0-3-6 whitespace: expect BOTH + #1-4-3(state: inProgress) |#1-4-3(state: pending) `|`: an ALTERNATIVE + +A field value is either a literal (always an enum member name in these files) +or `this.`, meaning the event echoes the argument that was sent. +`` wraps the two cases where the event arrives after an unbounded +scan rather than promptly, so a caller must not time out on it the same way. +""" + +from __future__ import annotations + +import re +import xml.etree.ElementTree as ET + +from .types import Expectation + +# The `|` is glued to the `#` in the XML, but allow a space so a hand-edited +# file still parses. Field lists contain spaces, so scan rather than split. +_REFERENCE = re.compile( + r"(?P\|\s*)?#(?P\d+)-(?P\d+)-(?P\d+)\s*(?:\((?P[^)]*)\))?" +) + + +def _parse_fields(text: str | None) -> dict[str, str]: + if not text: + return {} + fields: dict[str, str] = {} + for part in text.split(","): + if not part.strip(): + continue + key, sep, value = part.partition(":") + if not sep: + raise ValueError(f"expectation field {part.strip()!r} is not 'name: value'") + fields[key.strip()] = value.strip() + return fields + + +def parse_expectation_text(text: str, *, delayed: bool = False) -> tuple[Expectation, ...]: + """Parse one `` or `` body into top-level expectations. + + Alternatives are folded into the `alternatives` tuple of the expectation + they follow, so the caller sees "these N must all happen" at the top level + and "any one of these" underneath. + """ + primaries: list[tuple[tuple[int, int, int], dict[str, str], list[Expectation]]] = [] + + for match in _REFERENCE.finditer(text): + ids = ( + int(match.group("project")), + int(match.group("klass")), + int(match.group("command")), + ) + fields = _parse_fields(match.group("fields")) + if match.group("alt"): + if not primaries: + raise ValueError(f"expectation {text.strip()!r} starts with an alternative") + primaries[-1][2].append(Expectation(ids=ids, fields=fields, delayed=delayed)) + else: + primaries.append((ids, fields, [])) + + if not primaries and text.strip(): + raise ValueError(f"expectation block {text.strip()!r} contains no #p-c-m reference") + + return tuple( + Expectation(ids=ids, fields=fields, alternatives=tuple(alts), delayed=delayed) + for ids, fields, alts in primaries + ) + + +def expectations_for(cmd: ET.Element) -> tuple[Expectation, ...]: + """Collect every expectation declared on one `` element.""" + found: list[Expectation] = [] + for block in cmd.findall("expectations"): + for body in block: + if body.tag not in ("immediate", "delayed"): + raise ValueError(f"unknown expectations child <{body.tag}>") + found.extend(parse_expectation_text(body.text or "", delayed=body.tag == "delayed")) + return tuple(found) diff --git a/src/mcbebop/protocol/safety.py b/src/mcbebop/protocol/safety.py new file mode 100644 index 0000000..c62ab7a --- /dev/null +++ b/src/mcbebop/protocol/safety.py @@ -0,0 +1,96 @@ +"""Which commands need an unlock before they are sent. + +The tier is derived from the command's class rather than from a hand-curated +list of 101 names, because Parrot groups commands by consequence already and a +class-level rule keeps working when a firmware turns out to accept something +the XML does not list. Where a class is mixed, the exceptions are named. + +`xml_index.py` stamps the result onto every `CommandSpec`; `tools/command.py` +enforces it. +""" + +from __future__ import annotations + +from .types import CommandSpec, Direction, Tier + +# Refusing to land an airborne aircraft is the more dangerous answer: the +# battery runs out either way, and an uncommanded descent lands it somewhere +# nobody chose. These two bypass the arming gate for that reason alone. +ALWAYS_ALLOWED: frozenset[str] = frozenset( + { + "ardrone3.Piloting.Landing", + "ardrone3.Piloting.Emergency", + } +) + +# Whole classes that can spin motors or move the airframe. +_MOTION_CLASSES: frozenset[str] = frozenset( + { + "ardrone3.Piloting", # TakeOff, PCMD, Landing, moveBy, the POI family, FlatTrim + "ardrone3.Animations", # Flip + "common.Animations", # Start/Stop/StopAll + "common.Mavlink", # Start/Pause/Stop RUN A STORED FLIGHT PLAN. Not config. + "common.Calibration", # magneto calibration asks the pilot to rotate a powered aircraft + } +) + +# Whole classes that change what the aircraft is allowed to do, or the radio +# link we depend on to tell it anything. Survivable, but not reversible from +# the air, and a bad Wifi change ends the session. +_ENVELOPE_CLASSES: frozenset[str] = frozenset( + { + "ardrone3.PilotingSettings", # max altitude/tilt/distance, geofence, banked turn + "ardrone3.SpeedSettings", # vertical and rotation speed limits, hull protection + "ardrone3.GPSSettings", # home point and return-home behaviour + "ardrone3.Network", # Wifi scan and auth channel + "ardrone3.NetworkSettings", # Wifi channel selection and security + "common.Network", # Disconnect + "common.WifiSettings", # outdoor mode changes the legal channel set + "common.FlightPlanSettings", # return home on disconnect + "common.OverHeat", # SwitchOff, Ventilate + "common.Charger", # charge rate affects the battery we fly on + "common.Factory", # Reset + } +) + +# Named exceptions inside otherwise harmless classes. +_ENVELOPE_COMMANDS: frozenset[str] = frozenset( + { + "common.Settings.Country", # changes Wifi band and channel: can drop the link + "common.Settings.AutoCountry", + "common.Settings.Reset", # resets every setting, including the envelope + "common.Common.Reboot", + "common.Controller.isPiloting", # tells the drone a pilot has the sticks + } +) + +# Asking the drone to dump what it already knows. +_OBSERVE_COMMANDS: frozenset[str] = frozenset( + { + "common.Common.AllStates", + "common.Settings.AllSettings", + } +) + + +def tier_for(spec: CommandSpec) -> Tier: + """Classify a command by what sending it can do. + + Events are OBSERVE: they arrive unbidden and sending one is not a thing we + can do, so the arming gate never sees them. Everything sendable that is not + named above is CONFIG, which is the honest default here because the + remaining 31 are camera, media, clock and accessory settings. + """ + if spec.direction is Direction.FROM_DRONE: + return Tier.OBSERVE + + full = spec.full_name + klass = f"{spec.project}.{spec.klass}" + + if full in _OBSERVE_COMMANDS: + return Tier.OBSERVE + if klass in _MOTION_CLASSES: + return Tier.MOTION + if klass in _ENVELOPE_CLASSES or full in _ENVELOPE_COMMANDS: + return Tier.ENVELOPE + return Tier.CONFIG diff --git a/src/mcbebop/protocol/xml_index.py b/src/mcbebop/protocol/xml_index.py new file mode 100644 index 0000000..97d89d1 --- /dev/null +++ b/src/mcbebop/protocol/xml_index.py @@ -0,0 +1,325 @@ +"""The command table, read from Parrot's own XML instead of transcribed. + +Transcribing 264 commands by hand is how you end up with an encoder that is +subtly wrong about one argument and nobody notices until an aircraft does +something unexpected. So the XML in `arsdk-xml/` is the single source: this +module parses it once, derives direction, buffer, tier and expectations from +it, and hands out frozen `CommandSpec` objects. + +Several properties of those files are load-bearing and not obvious, so they +are called out at the point where the code relies on them. The short version: +attribute order is unstable, ids have gaps, document order is not id order, +and direction is not stated anywhere. +""" + +from __future__ import annotations + +import functools +import os +import xml.etree.ElementTree as ET +from dataclasses import replace +from pathlib import Path + +from .expectations import expectations_for +from .safety import tier_for +from .types import BEBOP2_PRODUCT_ID, ArgSpec, Buffer, CommandSpec, Direction, EnumSpec, Tier + +XML_FILES = ("common.xml", "ardrone3.xml") +XML_DIR_ENV = "MCBEBOP_ARSDK_XML" + +# The firmware on the aircraft these notes were taken from. `support` entries +# like `090c:4.3.0` state a MINIMUM firmware, so answering "does the Bebop 2 +# support this" needs a version to compare against. +BEBOP2_FIRMWARE = (4, 7, 1) + +# Parrot's element is . pyparrot renamed it to to get around +# its `untangle` parser; accept both so a copy taken from either source loads. +_CLASS_TAGS = ("class", "myclass") + + +def _clean(text: str | None) -> str: + """Tidy a doc string without losing the author's paragraph breaks. + + Line breaks are stored as a literal backslash-n inside attribute values, + and continuation lines carry the XML file's own tab indentation, which is + meaningless to a reader. + """ + if not text: + return "" + lines = [" ".join(part.split()) for part in text.replace("\\n", "\n").split("\n")] + return "\n".join(lines).strip() + + +def _version_tuple(text: str) -> tuple[int, ...]: + """Parse `4.3.0` into comparable integers. + + Never compare these as strings: `2.0.29` sorts before `2.0.3` lexically, + which is backwards, and that version really does appear in common.xml. + """ + parts: list[int] = [] + for piece in text.strip().split("."): + try: + parts.append(int(piece)) + except ValueError: + # Unparseable version: treat the whole requirement as unmet rather + # than guessing, which is what the caller's `False` means. + return (1 << 30,) + return tuple(parts) + + +def supports_bebop2(support: str | None, firmware: tuple[int, ...] = BEBOP2_FIRMWARE) -> bool | None: + """Does this `support` field cover the Bebop 2? + + `None` means the XML does not say, which is NOT the same as unsupported: + 38 commands say nothing and plenty of them work. Callers must not refuse a + command on an unknown. + """ + if support is None: + return None + text = support.strip() + if not text: + return None + + for token in (t.strip() for t in text.split(";")): + if not token: + continue + if token == "drones": + return True + if token == "none": + return False + product, _, version = token.partition(":") + if product.strip().lower() != BEBOP2_PRODUCT_ID: + continue + if not version: + return True + return _version_tuple(version) <= tuple(firmware) + return False + + +def direction_of(klass: str, comment: ET.Element | None) -> Direction: + """Work out which way a command travels. + + There is no direction attribute. `` describes what + happens when you send it, `triggered=` describes when the drone emits it, + and no command carries both. 31 commands carry neither, and for those the + class-name suffix is the only signal available. That fallback would be + wrong for `common.Controller`, which is mixed, but both of its commands + happen to state result/triggered so the fallback never sees them. + """ + if comment is not None: + if "result" in comment.attrib: + return Direction.TO_DRONE + if "triggered" in comment.attrib: + return Direction.FROM_DRONE + return Direction.FROM_DRONE if klass.endswith(("State", "Event")) else Direction.TO_DRONE + + +def _buffer_of(cmd: ET.Element) -> Buffer: + raw = cmd.get("buffer") + if raw is None: + return Buffer.ACK # 248 commands; the acknowledged buffer is the default + if raw == "NON_ACK": + return Buffer.NON_ACK # 15 commands, the piloting stream among them + if raw == "HIGH_PRIO": + return Buffer.HIGH_PRIO # exactly one: ardrone3.Piloting.Emergency + raise ValueError(f"unknown buffer {raw!r} on {cmd.get('name')}") + + +def _parse_args(cmd: ET.Element) -> tuple[ArgSpec, ...]: + args: list[ArgSpec] = [] + for arg in cmd.findall("arg"): + # carries only name and type, and only name. An enum + # member's wire value is its 0-based position among its siblings, which + # is why these are read in document order and never sorted. + members = tuple( + EnumSpec(name=member.get("name", ""), value=position, doc=_clean(member.text)) + for position, member in enumerate(arg.findall("enum")) + ) + args.append( + ArgSpec( + name=arg.get("name", ""), + type=arg.get("type", ""), + doc=_clean(arg.text), + members=members, + ) + ) + return tuple(args) + + +def _parse_command(project: str, project_id: int, klass: str, class_id: int, cmd: ET.Element) -> CommandSpec: + comment = cmd.find("comment") + if comment is not None: + title = _clean(comment.get("title")) + doc = _clean(comment.get("desc")) + support = comment.get("support") + else: + # Six commands have no and carry a bare text node instead, + # e.g. ardrone3.SettingsState.CPUID. The text is the description. + title = "" + doc = _clean(cmd.text) + support = None + + spec = CommandSpec( + project=project, + klass=klass, + name=cmd.get("name", ""), + # Read ids from the attributes. They have gaps and document order is + # not id order, so position tells you nothing. + ids=(project_id, class_id, int(cmd.get("id", ""))), + args=_parse_args(cmd), + buffer=_buffer_of(cmd), + direction=direction_of(klass, comment), + support=support, + deprecated=cmd.get("deprecated") == "true", + title=title, + doc=doc, + expectations=expectations_for(cmd), + ) + return replace(spec, tier=tier_for(spec)) + + +def xml_dir() -> Path: + """Find the vendored XML. + + It lives at the repository root rather than inside the package, so look + package-relative first (in case it is ever vendored in) and then walk up. + """ + override = os.environ.get(XML_DIR_ENV) + if override: + path = Path(override).expanduser() + if all((path / name).is_file() for name in XML_FILES): + return path + raise FileNotFoundError(f"{XML_DIR_ENV}={override!r} does not contain {', '.join(XML_FILES)}") + + here = Path(__file__).resolve() + for base in (here.parent, *here.parents): + candidate = base / "arsdk-xml" + if all((candidate / name).is_file() for name in XML_FILES): + return candidate + raise FileNotFoundError( + f"cannot find arsdk-xml/ containing {', '.join(XML_FILES)} above {here}; " + f"set {XML_DIR_ENV} to its directory" + ) + + +def _flat_trim(takeoff: CommandSpec) -> CommandSpec: + """Hand-add `ardrone3.Piloting.FlatTrim`. + + Neither the vendored XML nor upstream arsdk-xml defines this command, yet + the firmware accepts it and FreeFlight sends it before every takeoff to + re-zero the accelerometers. Leaving it out would mean the only way to send + it is to bypass this index, so it is injected here with the ids read off + the live aircraft (1, 0, 0) and the class ids taken from its sibling so a + future renumbering cannot silently desync. + """ + project_id, class_id, _ = takeoff.ids + spec = CommandSpec( + project=takeoff.project, + klass=takeoff.klass, + name="FlatTrim", + ids=(project_id, class_id, 0), + args=(), + buffer=Buffer.ACK, + direction=Direction.TO_DRONE, + tier=Tier.MOTION, + support=None, + title="Flat trim", + doc=( + "Re-zero the horizontal reference from the accelerometers. Send it with the aircraft " + "level and still, before takeoff.\n" + "Hand-added: this command is missing from Parrot's XML (both the vendored Bebop-era " + "snapshot and upstream master) but the firmware accepts it, so it has no , no " + "declared support and no expectations to parse." + ), + ) + return replace(spec, tier=tier_for(spec)) + + +class _Index: + """The parsed table. Satisfies `types.ProtocolIndex`.""" + + def __init__(self, commands: list[CommandSpec]) -> None: + self._ordered = commands + self._by_name = {command.full_name: command for command in commands} + self._by_ids = {command.ids: command for command in commands} + if len(self._by_name) != len(commands) or len(self._by_ids) != len(commands): + raise ValueError("duplicate command name or id triple in the XML") + + def get(self, full_name: str) -> CommandSpec | None: + return self._by_name.get(full_name) + + def by_ids(self, ids: tuple[int, int, int]) -> CommandSpec | None: + return self._by_ids.get(tuple(ids)) # type: ignore[arg-type] + + def search(self, text: str) -> list[CommandSpec]: + """Substring search over name, title and description. + + Ranked so that a name match beats a prose match, because an agent that + searches "takeoff" wants the command, not the 9 events that mention it. + """ + needle = text.strip().lower() + if not needle: + return [] + scored: list[tuple[int, str, CommandSpec]] = [] + for command in self._ordered: + if needle == command.full_name.lower() or needle == command.name.lower(): + rank = 0 + elif needle in command.name.lower(): + rank = 1 + elif needle in command.full_name.lower(): + rank = 2 + elif needle in command.title.lower(): + rank = 3 + elif needle in command.doc.lower() or any(needle in a.name.lower() for a in command.args): + rank = 4 + else: + continue + scored.append((rank, command.full_name, command)) + scored.sort(key=lambda item: (item[0], item[1])) + return [command for _, _, command in scored] + + def all(self) -> list[CommandSpec]: + return list(self._ordered) + + +@functools.cache +def index() -> _Index: + """Parse the XML once per process and keep it.""" + directory = xml_dir() + commands: list[CommandSpec] = [] + + for filename in XML_FILES: + project = ET.parse(directory / filename).getroot() + project_name = project.get("name", "") + project_id = int(project.get("id", "")) + for tag in _CLASS_TAGS: + for klass in project.findall(tag): + class_name = klass.get("name", "") + class_id = int(klass.get("id", "")) + for cmd in klass.findall("cmd"): + commands.append(_parse_command(project_name, project_id, class_name, class_id, cmd)) + + built = _Index(commands) + takeoff = built.get("ardrone3.Piloting.TakeOff") + if takeoff is None: + raise ValueError("ardrone3.Piloting.TakeOff missing: the XML is not the expected ARSDK table") + flat_trim = _flat_trim(takeoff) + if built.by_ids(flat_trim.ids) is not None: + raise ValueError(f"{flat_trim.ids} is already taken; FlatTrim must not be injected over it") + return _Index([*commands, flat_trim]) + + +def get(full_name: str) -> CommandSpec | None: + return index().get(full_name) + + +def by_ids(ids: tuple[int, int, int]) -> CommandSpec | None: + return index().by_ids(ids) + + +def search(text: str) -> list[CommandSpec]: + return index().search(text) + + +def all_commands() -> list[CommandSpec]: + return index().all() diff --git a/tests/test_codec.py b/tests/test_codec.py new file mode 100644 index 0000000..8fbefbb --- /dev/null +++ b/tests/test_codec.py @@ -0,0 +1,244 @@ +"""Wire-format tests. The byte counts here are the whole point. + +Two of them exist because pyparrot gets them wrong and pyparrot is the obvious +thing to check against: the command id is 16 bits, and an enum is 32. +""" + +import struct + +import pytest + +from mcbebop.protocol import codec, xml_index +from mcbebop.protocol.types import ArgSpec, CommandSpec, EnumSpec + + +def test_header_is_four_bytes_with_a_16_bit_command_id(): + assert codec.HEADER.size == 4 + assert codec.HEADER.pack(1, 0, 2) == b"\x01\x00\x02\x00" + # An id above 255 must survive, even though no current command has one. + assert codec.HEADER.unpack(codec.HEADER.pack(1, 4, 300)) == (1, 4, 300) + + +def test_encode_command_is_header_plus_args(): + spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable") + assert codec.encode_command(spec, {"enable": 1}) == b"\x01\x15\x00\x00\x01" + + +def test_no_arg_command_encodes_to_nothing(): + assert codec.encode_args(xml_index.get("ardrone3.Piloting.TakeOff")) == b"" + assert codec.encode_args(xml_index.get("ardrone3.Piloting.FlatTrim"), {}) == b"" + + +def test_enum_is_four_bytes_little_endian(): + spec = xml_index.get("ardrone3.Network.WifiScan") + assert codec.encode_args(spec, {"band": "5ghz"}) == b"\x01\x00\x00\x00" + assert codec.encode_args(spec, {"band": "2_4ghz"}) == b"\x00\x00\x00\x00" + # An int position is accepted too, for replaying captured traffic. + assert codec.encode_args(spec, {"band": 2}) == b"\x02\x00\x00\x00" + + +def test_enum_decodes_back_to_the_member_name(): + spec = xml_index.get("ardrone3.Network.WifiScan") + assert codec.decode_args(spec, b"\x01\x00\x00\x00") == {"WifiScan_band": "5ghz"} + + +def test_unknown_enum_position_decodes_to_the_number(): + # A firmware member this XML snapshot predates must not raise. + spec = xml_index.get("ardrone3.Network.WifiScan") + assert codec.decode_args(spec, b"\x63\x00\x00\x00") == {"WifiScan_band": 99} + + +def test_string_is_nul_terminated(): + spec = xml_index.get("common.Settings.ProductName") + assert codec.encode_args(spec, {"name": "Bebop"}) == b"Bebop\x00" + assert codec.decode_args(spec, b"Bebop\x00") == {"ProductName_name": "Bebop"} + + +def test_string_arguments_can_be_followed_by_more_arguments(): + # A variable-length field mid-payload is where a fixed-width read goes + # wrong: everything after the string shifts. + spec = xml_index.get("ardrone3.NetworkState.WifiScanListChanged") + assert [a.type for a in spec.args] == ["string", "i16", "enum", "u8"] + payload = b"MyNet\x00" + struct.pack(" CommandSpec: + """Parrot's other projects use bitfields; these two files do not.""" + arg = ArgSpec( + name="flags", + type=f"bitfield:{width}:Flags", + members=(EnumSpec("first", 0), EnumSpec("second", 1), EnumSpec("third", 2)), + ) + return CommandSpec(project="x", klass="Y", name="Z", ids=(9, 9, 9), args=(arg,)) + + +def test_bitfield_takes_member_names_or_a_mask(): + spec = _bitfield_spec("u8") + assert codec.encode_args(spec, {"flags": ["first", "third"]}) == b"\x05" + assert codec.encode_args(spec, {"flags": "second"}) == b"\x02" + assert codec.encode_args(spec, {"flags": 0xFF}) == b"\xff" + assert codec.decode_args(spec, b"\x05") == {"Z_flags": 5} + + +def test_bitfield_width_sets_the_wire_size(): + assert len(codec.encode_args(_bitfield_spec("u16"), {"flags": 1})) == 2 + assert len(codec.encode_args(_bitfield_spec("u32"), {"flags": 1})) == 4 + with pytest.raises(ValueError, match="unknown bitfield width 'u7'"): + codec.encode_args(_bitfield_spec("u7"), {"flags": 1}) + + +def test_bitfield_rejects_an_unknown_member(): + with pytest.raises(ValueError, match="no member named 'fourth'"): + codec.encode_args(_bitfield_spec("u8"), {"flags": ["fourth"]}) + + +def test_unsupported_type_is_reported(): + blob = ArgSpec(name="a", type="blob") + spec = CommandSpec(project="x", klass="Y", name="Z", ids=(9, 9, 8), args=(blob,)) + with pytest.raises(ValueError, match="unsupported type 'blob'"): + codec.encode_args(spec, {"a": 1}) + + +def test_every_type_in_the_xml_is_encodable(): + declared = {a.type for c in xml_index.all_commands() for a in c.args} + for type_name in declared: + if type_name in ("string", "enum"): + continue + assert struct.calcsize(codec._FORMATS[type_name]) > 0 diff --git a/tests/test_expectations.py b/tests/test_expectations.py new file mode 100644 index 0000000..f601a69 --- /dev/null +++ b/tests/test_expectations.py @@ -0,0 +1,113 @@ +"""The expectations grammar is undocumented, so the examples are the spec.""" + +import xml.etree.ElementTree as ET + +import pytest + +from mcbebop.protocol import xml_index +from mcbebop.protocol.expectations import expectations_for, parse_expectation_text +from mcbebop.protocol.types import Direction + + +def test_every_block_is_parsed_and_only_sendable_commands_carry_one(): + with_expectations = [c for c in xml_index.all_commands() if c.expectations] + assert len(with_expectations) == 82 + assert all(c.direction is Direction.TO_DRONE for c in with_expectations) + + +def test_delayed_blocks(): + delayed = [c.full_name for c in xml_index.all_commands() if any(e.delayed for e in c.expectations)] + assert sorted(delayed) == ["ardrone3.Network.WifiScan", "ardrone3.Network.WifiAuthChannel"][::-1] + + +def test_bare_reference(): + (expectation,) = parse_expectation_text("#0-3-0") + assert expectation.ids == (0, 3, 0) + assert expectation.fields == {} + assert expectation.alternatives == () + assert expectation.delayed is False + + +def test_this_dot_argname_means_the_drone_echoes_what_was_sent(): + (expectation,) = xml_index.get("common.Settings.ProductName").expectations + assert expectation.ids == (0, 3, 2) + assert expectation.fields == {"name": "this.name"} + + +def test_whitespace_separated_references_are_all_expected(): + expectations = xml_index.get("common.Settings.AutoCountry").expectations + assert len(expectations) == 2 + assert expectations[0].ids == (0, 3, 7) + assert expectations[0].fields == {"automatic": "this.automatic"} + assert expectations[1].ids == (0, 3, 6) + assert all(e.alternatives == () for e in expectations) + + +def test_pipe_prefixed_references_are_alternatives(): + (expectation,) = xml_index.get("ardrone3.Piloting.NavigateHome").expectations + assert expectation.ids == (1, 4, 3) + assert expectation.fields == {"state": "inProgress", "reason": "userRequest"} + assert len(expectation.alternatives) == 2 + assert [a.fields["state"] for a in expectation.alternatives] == ["pending", "available"] + assert all(a.ids == (1, 4, 3) for a in expectation.alternatives) + + +def test_alternatives_can_name_different_commands(): + # StopPilotedPOI is confirmed by whichever POI event version is in play. + (expectation,) = xml_index.get("ardrone3.Piloting.StopPilotedPOI").expectations + assert expectation.ids == (1, 4, 14) + assert [a.ids for a in expectation.alternatives] == [(1, 4, 22)] + + +def test_literal_enum_member_values(): + (expectation,) = xml_index.get("ardrone3.Piloting.StartPilotedPOI").expectations + assert expectation.fields["status"] == "RUNNING" + assert expectation.fields["latitude"] == "this.latitude" + assert expectation.alternatives[0].fields["status"] == "PENDING" + + +def test_takeoff_expects_two_states_not_one_of_two(): + expectations = xml_index.get("ardrone3.Piloting.TakeOff").expectations + assert [e.fields["state"] for e in expectations] == ["motor_ramping", "takingoff"] + + +def test_delayed_flag_reaches_the_alternatives_too(): + parsed = parse_expectation_text("#1-14-0 |#1-14-1", delayed=True) + assert parsed[0].delayed is True + assert parsed[0].alternatives[0].delayed is True + + +def test_spaced_pipe_still_parses(): + (expectation,) = parse_expectation_text("#0-1-2 | #0-1-3") + assert [a.ids for a in expectation.alternatives] == [(0, 1, 3)] + + +def test_empty_block_is_empty_not_an_error(): + assert parse_expectation_text("\n\t\t\t") == () + + +def test_leading_alternative_is_rejected(): + with pytest.raises(ValueError, match="starts with an alternative"): + parse_expectation_text("|#0-1-2") + + +def test_text_without_a_reference_is_rejected(): + with pytest.raises(ValueError, match="no #p-c-m reference"): + parse_expectation_text("ProductNameChanged") + + +def test_malformed_field_is_rejected(): + with pytest.raises(ValueError, match="is not 'name: value'"): + parse_expectation_text("#0-3-2(name)") + + +def test_unknown_expectations_child_is_rejected(): + cmd = ET.fromstring( + "#0-0-0" + ) + with pytest.raises(ValueError, match="unknown expectations child"): + expectations_for(cmd) + + +def test_commands_without_a_block_have_an_empty_tuple(): + assert xml_index.get("ardrone3.Piloting.PCMD").expectations == () diff --git a/tests/test_protocol_index.py b/tests/test_protocol_index.py new file mode 100644 index 0000000..b729812 --- /dev/null +++ b/tests/test_protocol_index.py @@ -0,0 +1,222 @@ +"""The index is derived, not transcribed, so these pin the derivation. + +Every count here was read off the two vendored XML files. A bad parse that +still produces plausible-looking objects is the failure mode worth catching, +so the numbers are asserted rather than the shapes. +""" + +import collections +import shutil + +import pytest + +from mcbebop.protocol import xml_index +from mcbebop.protocol.types import Buffer, Direction, Tier + +TOTAL_IN_XML = 264 +TOTAL_WITH_FLAT_TRIM = 265 + + +def test_command_count(): + assert len(xml_index.all_commands()) == TOTAL_WITH_FLAT_TRIM + from_xml = [c for c in xml_index.all_commands() if c.name != "FlatTrim"] + assert len(from_xml) == TOTAL_IN_XML + + +def test_direction_split(): + counts = collections.Counter(c.direction for c in xml_index.all_commands() if c.name != "FlatTrim") + assert counts[Direction.TO_DRONE] == 101 + assert counts[Direction.FROM_DRONE] == 163 + assert sum(counts.values()) == TOTAL_IN_XML + + +def test_direction_fallback_count(): + """31 commands state neither result nor triggered and fall back to the suffix.""" + from xml.etree import ElementTree + + directory = xml_index.xml_dir() + neither = 0 + for name in xml_index.XML_FILES: + project = ElementTree.parse(directory / name).getroot() + for klass in project.iter("class"): + for cmd in klass.findall("cmd"): + comment = cmd.find("comment") + attrs = comment.attrib if comment is not None else {} + assert not ("result" in attrs and "triggered" in attrs), cmd.get("name") + if "result" not in attrs and "triggered" not in attrs: + neither += 1 + assert neither == 31 + + +def test_mixed_controller_class_is_not_decided_by_suffix(): + # common.Controller holds one of each, which is why the suffix rule alone + # would be wrong there. Both state result/triggered, so it never applies. + assert xml_index.get("common.Controller.isPiloting").direction is Direction.TO_DRONE + assert xml_index.get("common.Controller.PeerStateChanged").direction is Direction.FROM_DRONE + + +def test_buffers(): + counts = collections.Counter(c.buffer for c in xml_index.all_commands() if c.name != "FlatTrim") + assert counts[Buffer.ACK] == 248 + assert counts[Buffer.NON_ACK] == 15 + assert counts[Buffer.HIGH_PRIO] == 1 + high_prio = [c.full_name for c in xml_index.all_commands() if c.buffer is Buffer.HIGH_PRIO] + assert high_prio == ["ardrone3.Piloting.Emergency"] + + +def test_deprecated_count(): + assert sum(1 for c in xml_index.all_commands() if c.deprecated) == 39 + + +def test_expectation_block_count(): + blocks = sum(1 for c in xml_index.all_commands() if c.expectations) + assert blocks == 82 + assert all(c.direction is Direction.TO_DRONE for c in xml_index.all_commands() if c.expectations) + + +def test_ids_are_unique_and_lookups_agree(): + for command in xml_index.all_commands(): + assert xml_index.by_ids(command.ids) is command + assert xml_index.get(command.full_name) is command + assert xml_index.get("ardrone3.Piloting.Nope") is None + assert xml_index.by_ids((9, 9, 9)) is None + + +def test_known_ids_match_the_live_aircraft(): + # Read off real traffic, see docs/notes/protocol.md. + assert xml_index.get("common.Common.AllStates").ids == (0, 4, 0) + assert xml_index.get("common.Settings.AllSettings").ids == (0, 2, 0) + assert xml_index.get("ardrone3.Piloting.TakeOff").ids == (1, 0, 1) + assert xml_index.get("ardrone3.Piloting.PCMD").ids == (1, 0, 2) + + +def test_command_ids_have_gaps_and_document_order_is_not_id_order(): + """Never infer an id from a command's position in the file.""" + piloting = [c for c in xml_index.all_commands() if f"{c.project}.{c.klass}" == "ardrone3.Piloting"] + in_document_order = [c.ids[2] for c in piloting] + assert in_document_order != sorted(in_document_order) + + # ardrone3.PilotingState runs 1..22 with no id 0, so counting is not enough. + state = [ + c.ids[2] for c in xml_index.all_commands() if f"{c.project}.{c.klass}" == "ardrone3.PilotingState" + ] + assert len(state) < max(state) + 1 + + +def test_flat_trim_is_injected(): + flat_trim = xml_index.get("ardrone3.Piloting.FlatTrim") + assert flat_trim is not None + assert flat_trim.ids == (1, 0, 0) + assert flat_trim.args == () + assert flat_trim.buffer is Buffer.ACK + assert flat_trim.direction is Direction.TO_DRONE + assert flat_trim.tier is Tier.MOTION + assert flat_trim.support is None + assert "missing from Parrot's XML" in flat_trim.doc + + +def test_commands_without_a_comment_keep_their_bare_text_as_doc(): + bare = [ + "common.CalibrationState.PitotCalibrationStateChanged", + "common.ARLibsVersionsState.ControllerLibARCommandsVersion", + "common.ARLibsVersionsState.SkyControllerLibARCommandsVersion", + "common.ARLibsVersionsState.DeviceLibARCommandsVersion", + "ardrone3.SettingsState.CPUID", + "ardrone3.MediaStreamingState.VideoStreamModeChanged", + ] + for name in bare: + command = xml_index.get(name) + assert command.title == "" + assert command.doc, name + assert xml_index.get("ardrone3.SettingsState.CPUID").doc == "Product main cpu id" + + +def test_enum_values_are_positions_and_names_stay_verbatim(): + band = xml_index.get("ardrone3.Network.WifiScan").args[0] + assert band.is_enum + assert [(m.name, m.value) for m in band.members] == [("2_4ghz", 0), ("5ghz", 1), ("all", 2)] + # Member names that are not valid identifiers must survive as strings. + framerate = xml_index.get("ardrone3.PictureSettings.VideoFramerate").args[0] + assert [m.name for m in framerate.members][:3] == ["24_FPS", "25_FPS", "30_FPS"] + + +def test_only_name_and_type_on_args(): + # Nothing in the parse may depend on an arg attribute Parrot does not set. + for command in xml_index.all_commands(): + for arg in command.args: + assert arg.name and arg.type + + +def test_search_ranks_the_command_above_the_prose(): + results = xml_index.search("takeoff") + assert results + assert results[0].name == "TakeOff" + assert any(c.full_name == "ardrone3.PilotingState.FlyingStateChanged" for c in xml_index.search("flying")) + assert xml_index.search(" ") == [] + + +def test_myclass_spelling_also_loads(tmp_path, monkeypatch): + """pyparrot renamed to ; a copy from it must still parse.""" + source = xml_index.xml_dir() + for name in xml_index.XML_FILES: + text = (source / name).read_text() + text = text.replace("", "") + (tmp_path / name).write_text(text) + + monkeypatch.setenv(xml_index.XML_DIR_ENV, str(tmp_path)) + xml_index.index.cache_clear() + try: + assert len(xml_index.all_commands()) == TOTAL_WITH_FLAT_TRIM + finally: + monkeypatch.delenv(xml_index.XML_DIR_ENV) + xml_index.index.cache_clear() + + +def test_missing_xml_dir_says_what_to_set(tmp_path, monkeypatch): + real = xml_index.xml_dir() + shutil.copy(real / "common.xml", tmp_path / "common.xml") # one of the two, not both + monkeypatch.setenv(xml_index.XML_DIR_ENV, str(tmp_path)) + xml_index.index.cache_clear() + try: + with pytest.raises(FileNotFoundError, match=xml_index.XML_DIR_ENV): + xml_index.all_commands() + finally: + monkeypatch.delenv(xml_index.XML_DIR_ENV) + xml_index.index.cache_clear() + + +@pytest.mark.parametrize( + ("support", "expected"), + [ + ("drones", True), + ("none", False), + ("0901;090c;090e", True), + ("0901;090e", False), + ("090c", True), + ("090c:4.3.0", True), # firmware 4.7.1 clears a 4.3.0 minimum + ("090c:4.8.0", False), + ("0901:2.0.29;090c;090e", True), + ("0914:1.6.3;0919:1.6.3", False), + ("090e;090c:4.3.0", True), + (None, None), + ("", None), + (" ", None), + ], +) +def test_supports_bebop2(support, expected): + assert xml_index.supports_bebop2(support) is expected + + +def test_support_versions_compare_as_integers_not_strings(): + # Lexically "2.0.29" < "2.0.3", which would make this unsupported. + assert xml_index.supports_bebop2("090c:2.0.29", firmware=(2, 0, 3)) is False + assert xml_index.supports_bebop2("090c:2.0.3", firmware=(2, 0, 29)) is True + assert xml_index.supports_bebop2("090c:10.0.0", firmware=(9, 0, 0)) is False + + +def test_unknown_support_is_not_unsupported(): + unknown = [c for c in xml_index.all_commands() if xml_index.supports_bebop2(c.support) is None] + # 30 commands carry a with no support attribute, 2 carry an empty + # one, 6 have no at all, and FlatTrim is hand-added. + assert len(unknown) == 39 + assert xml_index.get("ardrone3.Piloting.FlatTrim") in unknown diff --git a/tests/test_safety.py b/tests/test_safety.py new file mode 100644 index 0000000..e5148b8 --- /dev/null +++ b/tests/test_safety.py @@ -0,0 +1,131 @@ +"""Tiering decides what needs arming, so a misfiled command is a safety bug.""" + +import collections + +from mcbebop.protocol import xml_index +from mcbebop.protocol.safety import ALWAYS_ALLOWED, tier_for +from mcbebop.protocol.types import Direction, Tier + +MOTION = { + # The aircraft itself. + "ardrone3.Piloting.FlatTrim", + "ardrone3.Piloting.TakeOff", + "ardrone3.Piloting.PCMD", + "ardrone3.Piloting.Landing", + "ardrone3.Piloting.Emergency", + "ardrone3.Piloting.NavigateHome", + "ardrone3.Piloting.AutoTakeOffMode", + "ardrone3.Piloting.moveBy", + "ardrone3.Piloting.UserTakeOff", + "ardrone3.Piloting.Circle", + "ardrone3.Piloting.moveTo", + "ardrone3.Piloting.CancelMoveTo", + "ardrone3.Piloting.StartPilotedPOI", + "ardrone3.Piloting.StopPilotedPOI", + "ardrone3.Piloting.CancelMoveBy", + "ardrone3.Piloting.StartPilotedPOIV2", + "ardrone3.Animations.Flip", + # Running a stored flight plan moves the whole mission. + "common.Mavlink.Start", + "common.Mavlink.Pause", + "common.Mavlink.Stop", + # Calibration asks a powered aircraft to rotate. + "common.Calibration.MagnetoCalibration", + "common.Calibration.PitotCalibration", + "common.Animations.StartAnimation", + "common.Animations.StopAnimation", + "common.Animations.StopAllAnimations", +} + + +def test_motion_set_is_exactly_these(): + tiered = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.MOTION} + assert tiered == MOTION + + +def test_every_motion_command_is_sendable(): + for name in MOTION: + command = xml_index.get(name) + assert command is not None, name + assert command.direction is Direction.TO_DRONE, name + + +def test_observe_is_the_two_dump_everything_commands(): + sendable_observe = { + c.full_name + for c in xml_index.all_commands() + if c.tier is Tier.OBSERVE and c.direction is Direction.TO_DRONE + } + assert sendable_observe == {"common.Common.AllStates", "common.Settings.AllSettings"} + + +def test_envelope_covers_the_limit_and_radio_classes(): + envelope = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.ENVELOPE} + for name in ( + "ardrone3.PilotingSettings.MaxAltitude", + "ardrone3.PilotingSettings.NoFlyOverMaxDistance", + "ardrone3.SpeedSettings.MaxVerticalSpeed", + "ardrone3.GPSSettings.SetHome", + "ardrone3.Network.WifiScan", + "ardrone3.NetworkSettings.WifiSelection", + "common.Network.Disconnect", + "common.WifiSettings.OutdoorSetting", + "common.FlightPlanSettings.ReturnHomeOnDisconnect", + "common.Settings.Country", + "common.Settings.AutoCountry", + "common.Settings.Reset", + "common.Common.Reboot", + "common.Factory.Reset", + "common.OverHeat.SwitchOff", + "common.Charger.SetMaxChargeRate", + "common.Controller.isPiloting", + ): + assert name in envelope, name + assert len(envelope) == 44 + + +def test_config_is_the_harmless_remainder(): + config = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.CONFIG} + assert len(config) == 31 + for name in ( + "ardrone3.MediaRecord.Picture", + "ardrone3.MediaStreaming.VideoEnable", + "ardrone3.PictureSettings.VideoFramerate", + "ardrone3.Camera.Orientation", + "common.Common.CurrentDate", + "common.Settings.ProductName", + "common.Headlights.intensity", + ): + assert name in config, name + assert not any( + c.direction is Direction.FROM_DRONE for c in xml_index.all_commands() if c.tier is Tier.CONFIG + ) + + +def test_sendable_tiers_account_for_every_to_drone_command(): + counts = collections.Counter( + c.tier for c in xml_index.all_commands() if c.direction is Direction.TO_DRONE + ) + assert counts[Tier.MOTION] == 25 + assert counts[Tier.ENVELOPE] == 44 + assert counts[Tier.OBSERVE] == 2 + assert counts[Tier.CONFIG] == 31 + assert sum(counts.values()) == 102 # 101 in the XML plus the injected FlatTrim + + +def test_events_are_observe(): + for command in xml_index.all_commands(): + if command.direction is Direction.FROM_DRONE: + assert command.tier is Tier.OBSERVE, command.full_name + + +def test_always_allowed_are_the_two_that_end_a_flight(): + assert frozenset({"ardrone3.Piloting.Landing", "ardrone3.Piloting.Emergency"}) == ALWAYS_ALLOWED + for name in ALWAYS_ALLOWED: + command = xml_index.get(name) + assert command.tier is Tier.MOTION # locked by tier, exempted by name + + +def test_tier_for_is_what_the_index_stamped(): + for command in xml_index.all_commands(): + assert tier_for(command) is command.tier