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 <BBH header (the command id is 16 bits, not 8) and reads enums as i32 and strings as NUL-terminated, both of which pyparrot gets wrong. safety assigns the arming tier by class, with Mavlink and Calibration counted as motion because they move the aircraft. expectations parses the undocumented #p-c-m grammar including | alternatives and this.<arg> echoes. 90 tests, counts asserted so a bad parse fails loudly.
This commit is contained in:
@@ -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 `<BBH`: project and class are single bytes but the
|
||||||
|
command id is a 16-bit little-endian word. pyparrot reads three bytes and
|
||||||
|
happens to work because every id in these files is below 256, which means a
|
||||||
|
hand-check against it will agree right up to the point it does not.
|
||||||
|
|
||||||
|
An `enum` argument is four bytes, not one. pyparrot's sensor parser reads a
|
||||||
|
single byte for enums, so any event with an enum followed by more arguments
|
||||||
|
decodes as garbage after the enum. Confirmed against live traffic.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import struct
|
||||||
|
from collections.abc import Iterable, Mapping
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from . import xml_index
|
||||||
|
from .types import ArgSpec, CommandSpec, ProtocolIndex
|
||||||
|
|
||||||
|
# project u8, class u8, command u16 little-endian.
|
||||||
|
HEADER = struct.Struct("<BBH")
|
||||||
|
|
||||||
|
_FORMATS: dict[str, str] = {
|
||||||
|
"u8": "<B",
|
||||||
|
"i8": "<b",
|
||||||
|
"u16": "<H",
|
||||||
|
"i16": "<h",
|
||||||
|
"u32": "<I",
|
||||||
|
"i32": "<i",
|
||||||
|
"u64": "<Q",
|
||||||
|
"i64": "<q",
|
||||||
|
"float": "<f",
|
||||||
|
"double": "<d",
|
||||||
|
}
|
||||||
|
|
||||||
|
# An enum travels as a signed 32-bit word carrying the member's position.
|
||||||
|
_ENUM_FORMAT = "<i"
|
||||||
|
|
||||||
|
|
||||||
|
def _struct_for(arg: ArgSpec) -> struct.Struct:
|
||||||
|
if arg.is_enum:
|
||||||
|
return struct.Struct(_ENUM_FORMAT)
|
||||||
|
if arg.is_bitfield:
|
||||||
|
# bitfield:<width>:<enumname>, 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 `<Command>_<arg>` 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)
|
||||||
@@ -0,0 +1,89 @@
|
|||||||
|
"""Parsing Parrot's `<expectations>` 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:
|
||||||
|
|
||||||
|
#<project>-<class>-<command> 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.<argname>`, meaning the event echoes the argument that was sent.
|
||||||
|
`<delayed>` 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<alt>\|\s*)?#(?P<project>\d+)-(?P<klass>\d+)-(?P<command>\d+)\s*(?:\((?P<fields>[^)]*)\))?"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
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 `<immediate>` or `<delayed>` 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 `<cmd>` 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)
|
||||||
@@ -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
|
||||||
@@ -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 <class>. pyparrot renamed it to <myclass> 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. `<comment result=...>` 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"):
|
||||||
|
# <arg> carries only name and type, and <enum> 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 <comment> 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 <comment>, 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()
|
||||||
@@ -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("<h", -70) + struct.pack("<i", 1) + b"\x0b"
|
||||||
|
assert codec.decode_args(spec, payload) == {
|
||||||
|
"WifiScanListChanged_ssid": "MyNet",
|
||||||
|
"WifiScanListChanged_rssi": -70,
|
||||||
|
"WifiScanListChanged_band": "5ghz",
|
||||||
|
"WifiScanListChanged_channel": 11,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_decoded_keys_match_the_existing_captures():
|
||||||
|
spec = xml_index.get("common.CommonState.BatteryStateChanged")
|
||||||
|
decoded = codec.decode_args(spec, b"\x4b")
|
||||||
|
assert decoded == {"BatteryStateChanged_percent": 75}
|
||||||
|
|
||||||
|
|
||||||
|
def test_decode_event_reads_the_header():
|
||||||
|
spec = xml_index.get("common.CommonState.BatteryStateChanged")
|
||||||
|
payload = codec.HEADER.pack(*spec.ids) + b"\x4b"
|
||||||
|
ids, values = codec.decode_event(payload)
|
||||||
|
assert ids == spec.ids
|
||||||
|
assert values == {"BatteryStateChanged_percent": 75}
|
||||||
|
|
||||||
|
|
||||||
|
def test_decode_event_rejects_a_truncated_header():
|
||||||
|
with pytest.raises(ValueError, match="too short"):
|
||||||
|
codec.decode_event(b"\x00\x04\x00")
|
||||||
|
|
||||||
|
|
||||||
|
def test_decode_event_names_an_unknown_command_rather_than_dropping_it():
|
||||||
|
with pytest.raises(ValueError, match=r"no command with ids \(7, 7, 7\)"):
|
||||||
|
codec.decode_event(codec.HEADER.pack(7, 7, 7))
|
||||||
|
|
||||||
|
|
||||||
|
def test_decode_rejects_a_short_payload():
|
||||||
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
||||||
|
with pytest.raises(ValueError, match="bytes but only"):
|
||||||
|
codec.decode_args(spec, b"\x01\x00")
|
||||||
|
|
||||||
|
|
||||||
|
def test_decode_rejects_an_unterminated_string():
|
||||||
|
spec = xml_index.get("common.Settings.ProductName")
|
||||||
|
with pytest.raises(ValueError, match="unterminated string"):
|
||||||
|
codec.decode_args(spec, b"Bebop")
|
||||||
|
|
||||||
|
|
||||||
|
def test_pcmd_round_trip():
|
||||||
|
# The stick packet: the one command sent dozens of times a second.
|
||||||
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
||||||
|
args = {"flag": 1, "roll": -20, "pitch": 30, "yaw": 0, "gaz": 10, "timestampAndSeqNum": 12345}
|
||||||
|
encoded = codec.encode_args(spec, args)
|
||||||
|
assert len(encoded) == 1 + 1 + 1 + 1 + 1 + 4
|
||||||
|
decoded = codec.decode_args(spec, encoded)
|
||||||
|
assert decoded["PCMD_roll"] == -20
|
||||||
|
assert decoded["PCMD_timestampAndSeqNum"] == 12345
|
||||||
|
|
||||||
|
|
||||||
|
def _plausible(arg: ArgSpec):
|
||||||
|
"""A value the spec allows, chosen so a round trip is exact."""
|
||||||
|
if arg.type == "string":
|
||||||
|
return "test"
|
||||||
|
if arg.is_enum:
|
||||||
|
return arg.members[0].name if arg.members else 0
|
||||||
|
if arg.is_bitfield:
|
||||||
|
return [arg.members[0].name] if arg.members else 0
|
||||||
|
if arg.type in ("float", "double"):
|
||||||
|
return 1.5 # exact in both widths, so no tolerance needed
|
||||||
|
if arg.type.startswith("i"):
|
||||||
|
return -3
|
||||||
|
return 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_round_trip_every_command_that_has_arguments():
|
||||||
|
checked = 0
|
||||||
|
for spec in xml_index.all_commands():
|
||||||
|
if not spec.args:
|
||||||
|
continue
|
||||||
|
sent = {arg.name: _plausible(arg) for arg in spec.args}
|
||||||
|
decoded = codec.decode_args(spec, codec.encode_args(spec, sent))
|
||||||
|
assert len(decoded) == len(spec.args), spec.full_name
|
||||||
|
for arg in spec.args:
|
||||||
|
got = decoded[f"{spec.name}_{arg.name}"]
|
||||||
|
expected = sent[arg.name]
|
||||||
|
if arg.is_enum and not arg.members:
|
||||||
|
expected = 0
|
||||||
|
assert got == expected, f"{spec.full_name}.{arg.name} ({arg.type}) gave {got!r}"
|
||||||
|
checked += 1
|
||||||
|
assert checked == 234 # every command that declares at least one argument
|
||||||
|
|
||||||
|
|
||||||
|
def test_encoded_length_matches_the_declared_types():
|
||||||
|
sizes = {"u8": 1, "i8": 1, "u16": 2, "i16": 2, "u32": 4, "i32": 4, "u64": 8, "i64": 8}
|
||||||
|
sizes |= {"float": 4, "double": 8}
|
||||||
|
for spec in xml_index.all_commands():
|
||||||
|
if any(a.type == "string" for a in spec.args):
|
||||||
|
continue
|
||||||
|
expected = sum(4 if a.is_enum else sizes[a.type] for a in spec.args)
|
||||||
|
sent = {arg.name: _plausible(arg) for arg in spec.args}
|
||||||
|
assert len(codec.encode_args(spec, sent)) == expected, spec.full_name
|
||||||
|
|
||||||
|
|
||||||
|
def test_missing_argument_is_named():
|
||||||
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
||||||
|
with pytest.raises(ValueError, match="requires argument 'flag'"):
|
||||||
|
codec.encode_args(spec, {})
|
||||||
|
|
||||||
|
|
||||||
|
def test_unknown_argument_is_named_and_the_real_ones_listed():
|
||||||
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
||||||
|
with pytest.raises(ValueError, match="has no argument nope; expected: enable"):
|
||||||
|
codec.encode_args(spec, {"enable": 1, "nope": 2})
|
||||||
|
|
||||||
|
|
||||||
|
def test_out_of_range_integer_is_named():
|
||||||
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
||||||
|
with pytest.raises(ValueError, match="'enable' \\(u8\\) cannot hold 999"):
|
||||||
|
codec.encode_args(spec, {"enable": 999})
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrong_type_is_named():
|
||||||
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
||||||
|
with pytest.raises(ValueError, match="'enable' is a u8, got str"):
|
||||||
|
codec.encode_args(spec, {"enable": "yes"})
|
||||||
|
with pytest.raises(ValueError, match="pass 1 or 0 rather than a bool"):
|
||||||
|
codec.encode_args(spec, {"enable": True})
|
||||||
|
|
||||||
|
|
||||||
|
def test_unknown_enum_member_lists_the_allowed_ones():
|
||||||
|
spec = xml_index.get("ardrone3.Network.WifiScan")
|
||||||
|
with pytest.raises(ValueError, match="no member named 'wifi6'; allowed: 2_4ghz, 5ghz, all"):
|
||||||
|
codec.encode_args(spec, {"band": "wifi6"})
|
||||||
|
with pytest.raises(ValueError, match="no member with value 9"):
|
||||||
|
codec.encode_args(spec, {"band": 9})
|
||||||
|
|
||||||
|
|
||||||
|
def test_embedded_nul_in_a_string_is_refused():
|
||||||
|
spec = xml_index.get("common.Settings.ProductName")
|
||||||
|
with pytest.raises(ValueError, match="contains a NUL byte"):
|
||||||
|
codec.encode_args(spec, {"name": "Beb\x00op"})
|
||||||
|
|
||||||
|
|
||||||
|
def _bitfield_spec(width: str) -> 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
|
||||||
@@ -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(
|
||||||
|
"<cmd name='X' id='0'><expectations><eventually>#0-0-0</eventually></expectations></cmd>"
|
||||||
|
)
|
||||||
|
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 == ()
|
||||||
@@ -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 <class> to <myclass>; 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("<class ", "<myclass ").replace("</class>", "</myclass>")
|
||||||
|
(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 <comment> with no support attribute, 2 carry an empty
|
||||||
|
# one, 6 have no <comment> at all, and FlatTrim is hand-added.
|
||||||
|
assert len(unknown) == 39
|
||||||
|
assert xml_index.get("ardrone3.Piloting.FlatTrim") in unknown
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user