From 09126893015b807455709c41eba5e55015b6cd63 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 2 Oct 2026 12:07:26 -0600 Subject: [PATCH 1/2] Add an RTP/H.264 packetiser and a capture/replay pair The packetiser is RFC 6184: single-NAL packets, FU-A fragmentation, and STAP-A aggregation for parameter sets and SEI. Pure stdlib, because the simulator ships in the package and cannot drag a media library behind it. A VCL NAL is never aggregated and interleaved mode is absent; the docstring says so rather than leaving it to be discovered. Parameter sets repeat rather than appearing once at the head of the stream, which is what the aircraft does and the only reason a viewer joining a flight already in progress can recover. A start offset, random with a controllable seed, exists to hand a decoder a stream that begins mid-GOP on purpose. Capture and replay is the other half, and the two are different instruments. The packetised path paces at a nominal frame rate, which proves a decoder and a renderer work. A replay reproduces the recorded inter-packet gaps packet for packet, which is what a latency or jitter measurement needs. The aircraft's packet-type mix cannot be derived from first principles: the measurement we have read only the outer header byte of each packet, so what its 740 FU-A packets carried is still unknown, and only a real capture settles it. The loop point is the one gap a file cannot describe. It is estimated from the capture's mean frame period rather than its median packet gap, which was the first thing tried and is wrong: with a dozen packets to a frame most gaps are intra-frame, so the stream looped a frame early every time. --- src/mcbebop/media/capture.py | 280 +++++++++++++++++++++ src/mcbebop/media/rtp.py | 397 +++++++++++++++++++++++++++++ tests/test_rtp.py | 475 +++++++++++++++++++++++++++++++++++ 3 files changed, 1152 insertions(+) create mode 100644 src/mcbebop/media/capture.py create mode 100644 src/mcbebop/media/rtp.py create mode 100644 tests/test_rtp.py diff --git a/src/mcbebop/media/capture.py b/src/mcbebop/media/capture.py new file mode 100644 index 0000000..5e14af4 --- /dev/null +++ b/src/mcbebop/media/capture.py @@ -0,0 +1,280 @@ +"""Record the aircraft's RTP stream, and replay it packet for packet. + +Two halves of one idea. `record` binds the port we name in the ARSDK +handshake and writes every datagram the drone sends us, with the time it +arrived. `ReplaySource` feeds that file back to a client with the gaps it was +recorded with, which is the only way to put a viewer in front of the +aircraft's real pacing without the aircraft. + +That distinction is the point of having two paths at all. +`rtp.PacketisedSource` builds a stream from any video file and sends it at a +steady frame rate, which proves a decoder and a renderer work. A replay +reproduces bursts, reordering and jitter as they happened, which is what a +latency measurement needs. They are different instruments and a caller should +know which one it has. + +The file is deliberately dull: a magic string, a version, then a record per +packet. Nothing is parsed, nothing is dropped, and the payload bytes are kept +whole, so a later pass can answer questions the recorder never asked. Counting +the NAL types carried inside FU-A packets is the obvious one: the only +measurement we have of the drone's stream read the outer header byte of each +packet and stopped there, so what the fragments carried is still unknown. + +Recording needs the drone. Reading, writing and replaying a file do not, which +is what makes the format testable here. + + python -m mcbebop.media.capture out.rtpcap --seconds 10 +""" + +from __future__ import annotations + +import logging +import random +import socket +import statistics +import struct +import time +from collections.abc import Iterable, Iterator +from dataclasses import dataclass, field +from itertools import pairwise +from pathlib import Path + +from mcbebop.media.rtp import CLOCK_RATE, DEFAULT_FPS, parse_packet, resolve_start_offset, rewrite_header + +log = logging.getLogger(__name__) + +MAGIC = b"BEBOPRTP" +VERSION = 1 +_FILE_HEADER = struct.Struct("<8sHH") +_RECORD = struct.Struct(" int: + """Write `(arrival seconds, datagram)` pairs. Returns the packet count. + + Times are relative to the first packet, so a file is comparable with + itself regardless of when it was taken. + """ + out = Path(path) + out.parent.mkdir(parents=True, exist_ok=True) + count = 0 + with out.open("wb") as fh: + fh.write(_FILE_HEADER.pack(MAGIC, VERSION, 0)) + for at, data in packets: + fh.write(_RECORD.pack(at, len(data)) + data) + count += 1 + return count + + +def read_capture(path: str | Path) -> Iterator[tuple[float, bytes]]: + """Stream a capture back as `(arrival seconds, datagram)` pairs. + + A truncated tail stops the read rather than raising: a capture cut short + by Ctrl-C is still worth replaying, and the alternative is losing the + whole file to its last partial record. + """ + with Path(path).open("rb") as fh: + head = fh.read(_FILE_HEADER.size) + if len(head) < _FILE_HEADER.size: + raise CaptureFormatError(f"{path} is too short to be a capture") + magic, version, _flags = _FILE_HEADER.unpack(head) + if magic != MAGIC: + raise CaptureFormatError(f"{path} does not start with {MAGIC!r}") + if version != VERSION: + raise CaptureFormatError(f"{path} is capture version {version}, this reads {VERSION}") + while True: + raw = fh.read(_RECORD.size) + if len(raw) < _RECORD.size: + if raw: + log.warning("%s ends mid-record; stopping there", path) + return + at, length = _RECORD.unpack(raw) + data = fh.read(length) + if len(data) < length: + log.warning("%s ends mid-packet; stopping there", path) + return + yield at, data + + +def record( + path: str | Path, + *, + port: int = DEFAULT_CAPTURE_PORT, + host: str = "", + seconds: float | None = None, + max_packets: int | None = None, +) -> int: + """Bind `port` and write every datagram that arrives. Returns the count. + + Read-only with respect to the aircraft: it sends nothing, so it cannot be + the reason a stream stops. Something else has to hold the ARSDK session + and send VideoEnable, and it has to do it after this is listening, because + RTP is connectionless and whatever arrives before the bind is gone. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((host, port)) + sock.settimeout(_POLL) + deadline = None if seconds is None else time.monotonic() + seconds + first: float | None = None + packets: list[tuple[float, bytes]] = [] + log.info("recording RTP on %s:%d to %s", host or "0.0.0.0", port, path) + try: + while deadline is None or time.monotonic() < deadline: + try: + data, _addr = sock.recvfrom(_MAX_DATAGRAM) + except TimeoutError: + continue + now = time.monotonic() + first = now if first is None else first + packets.append((now - first, data)) + if max_packets is not None and len(packets) >= max_packets: + break + except KeyboardInterrupt: + log.info("interrupted; writing what arrived") + finally: + sock.close() + count = write_capture(path, packets) + span = packets[-1][0] if packets else 0.0 + log.info("wrote %d packets over %.2fs to %s", count, span, path) + return count + + +def _median_positive(values: Iterable[int | float], fallback: float) -> float: + positive = [v for v in values if v > 0] + return statistics.median(positive) if positive else fallback + + +@dataclass +class ReplaySource: + """A recorded RTP stream, sent again with its original inter-packet gaps. + + Only the SSRC and the sequence numbers are rewritten, so the client sees a + stream from us rather than one that looks like a replay of someone else's. + Timestamps keep their recorded deltas; what is added is a running offset, + so a loop carries the clock forward instead of jumping back to the start + of the capture and stalling every decoder downstream. + + The loop point is the one gap in the file that does not exist in it: the + interval between the last packet and a next one that was never recorded. + It is estimated as what would carry the stream to the start of the frame + after the last, from the capture's own mean frame period. The median gap + is the wrong estimator here and was the first thing tried: with a dozen + packets to a frame, most gaps are the ones *inside* a frame, so the median + is an intra-frame gap and the stream would loop a frame early every time. + """ + + packets_in: tuple[tuple[float, bytes], ...] + start_offset: float | str | None = None + seed: int | None = None + ssrc: int = field(default_factory=lambda: random.getrandbits(32)) + + @classmethod + def from_path( + cls, + path: str | Path, + *, + start_offset: float | str | None = None, + seed: int | None = None, + ) -> ReplaySource: + packets = tuple(read_capture(path)) + if not packets: + raise CaptureFormatError(f"{path} holds no packets") + return cls(packets_in=packets, start_offset=start_offset, seed=seed) + + def __post_init__(self) -> None: + # Both counters live on the instance, not in `packets()`, so a client + # that disables and re-enables video sees the stream continue rather + # than a clock that jumps backwards under an unchanged SSRC. + counters = random.Random(self.seed) + self._seq = counters.getrandbits(16) + self._ticks = counters.getrandbits(32) + # A separate generator, so the chosen offset does not move when the + # number of counter draws above changes. + offset = resolve_start_offset(self.start_offset, self.duration, random.Random(self.seed)) + self.start_index = next( + (i for i, (at, _) in enumerate(self.packets_in) if at >= offset), + 0, + ) + stamps = [parse_packet(data)[2] for _at, data in self.packets_in] + # Ticks step only between frames, so the median of the positive steps + # is one frame's worth however many packets a frame took. + ticks = [(b - a) % (1 << 32) for a, b in pairwise(stamps)] + self._loop_ticks = int(_median_positive(ticks, CLOCK_RATE / DEFAULT_FPS)) + self._loop_gap = self._gap_to_the_next_frame(stamps) + + def _gap_to_the_next_frame(self, stamps: list[int]) -> float: + """How long after the last recorded packet the next frame would start.""" + starts = [at for i, (at, _) in enumerate(self.packets_in) if i == 0 or stamps[i] != stamps[i - 1]] + if len(starts) < 2: + return 1.0 / DEFAULT_FPS + period = (starts[-1] - starts[0]) / (len(starts) - 1) + return max(0.0, period - (self.duration - starts[-1])) + + @property + def duration(self) -> float: + return self.packets_in[-1][0] if self.packets_in else 0.0 + + @property + def describe(self) -> str: + return ( + f"{len(self.packets_in)} recorded packets over {self.duration:.2f}s, " + f"replayed from index {self.start_index}" + ) + + def packets(self) -> Iterator[tuple[float, bytes]]: + total = len(self.packets_in) + index = self.start_index + prev_at, prev_ticks = None, None + while True: + at, data = self.packets_in[index] + stamp = parse_packet(data)[2] + if prev_at is None: + delay, step = 0.0, 0 + elif index == 0: + # The wrap: the file's own gaps say nothing about this one. + delay, step = self._loop_gap, self._loop_ticks + else: + delay, step = at - prev_at, (stamp - prev_ticks) % (1 << 32) + self._ticks = (self._ticks + step) % (1 << 32) + yield ( + max(0.0, delay), + rewrite_header(data, ssrc=self.ssrc, seq=self._seq, timestamp=self._ticks), + ) + self._seq = (self._seq + 1) & 0xFFFF + prev_at, prev_ticks = at, stamp + index = (index + 1) % total + + +def main(argv: list[str] | None = None) -> int: + import argparse + import sys + + parser = argparse.ArgumentParser( + prog="python -m mcbebop.media.capture", + description="Record the Bebop 2's RTP video stream for later replay.", + ) + parser.add_argument("out", help="where to write the capture") + parser.add_argument("--port", type=int, default=DEFAULT_CAPTURE_PORT) + parser.add_argument("--seconds", type=float, default=None, help="stop after this long") + parser.add_argument("--packets", type=int, default=None, help="stop after this many packets") + args = parser.parse_args(argv) + + # stderr, because this module is importable from the MCP server and + # stdout there is the JSON-RPC transport. + logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stderr) + count = record(args.out, port=args.port, seconds=args.seconds, max_packets=args.packets) + return 0 if count else 1 + + +if __name__ == "__main__": # pragma: no cover - a hand-run tool + raise SystemExit(main()) diff --git a/src/mcbebop/media/rtp.py b/src/mcbebop/media/rtp.py new file mode 100644 index 0000000..d6981df --- /dev/null +++ b/src/mcbebop/media/rtp.py @@ -0,0 +1,397 @@ +"""RTP/H.264 packetisation, so the simulator can emit a video stream. + +The Bebop 2 does not serve RTSP. It pushes plain RTP to the port the +controller named in its handshake, which means a simulator can imitate the +whole video path with a UDP socket and nothing else. This module is the +packetiser half of that: Annex-B H.264 in, RFC 6184 datagrams out. Pure +stdlib, because the simulator ships in the package and must not drag a media +library in behind it. + +What is implemented, and what is not: + +Single-NAL packets when a NAL fits the MTU, FU-A fragmentation when it does +not, and STAP-A aggregation for runs of parameter sets and SEI. Nothing else. +In particular a VCL NAL is never aggregated, so an access unit that fits in +one datagram still goes out as one single-NAL packet rather than a STAP-A. +Interleaved mode (types 25 to 29 other than 28), MTAP and the RTCP side are +absent; a viewer that needs RTCP from the simulator needs a capture replay. + +The live aircraft's packet-type mix is not reproduced and cannot be from +first principles. Six seconds off the drone carried 1461 packets whose outer +NAL byte was STAP-A, 740 FU-A and 200 non-IDR, which is around eight STAP-A +per frame at 30 fps and far more aggregation than parameter sets alone can +explain. Those are *outer* header bytes: for a FU-A packet the type lives in +the next byte and was never read, so what the carried NALs were is unknown. +Closing that gap needs a real capture, which is what `media/capture.py` and +`ReplaySource` exist for. + +Repeating the parameter sets is the one behaviour here that is not optional. +The aircraft's SPS and PPS recur throughout the stream rather than appearing +once at the start, which is why a viewer can join a flight already in +progress and recover in a second or two. `PacketisedSource` does the same. +""" + +from __future__ import annotations + +import logging +import random +import struct +from collections.abc import Iterator, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Protocol + +log = logging.getLogger(__name__) + +PAYLOAD_TYPE = 96 # dynamic, bound to H264/90000 by the SDP +CLOCK_RATE = 90_000 +DEFAULT_MTU = 1400 # payload bytes, so a 1500-byte link does not fragment +DEFAULT_FPS = 30.0 + +# version 2, no padding, no extension, no CSRC. +_V2 = 0x80 +_MARKER = 0x80 +_RTP_HEADER = struct.Struct(">BBHII") +_STAP_A_LENGTH = struct.Struct(">H") + +NAL_STAP_A = 24 +NAL_FU_A = 28 + +_VCL = frozenset(range(1, 6)) # 1 non-IDR .. 5 IDR +_SPS, _PPS = 7, 8 +# What is worth aggregating: parameter sets, SEI and the access unit +# delimiter. All small, all repeated, none of them a picture. +_AGGREGATABLE = frozenset({6, _SPS, _PPS, 9}) + +_START_CODE = b"\x00\x00\x01" + + +def nal_type(nal: bytes) -> int: + return nal[0] & 0x1F + + +def _nri(nal: bytes) -> int: + return nal[0] & 0x60 + + +def iter_nals(data: bytes) -> Iterator[bytes]: + """Split an Annex-B elementary stream into NAL units. + + Both start codes are accepted. The three-byte form is the one actually + searched for, and a fourth leading zero is simply a trailing zero of the + previous NAL, which is why trailing zeros are stripped: a decoder ignores + them but they would otherwise inflate every NAL's length and so change + where the MTU falls. + """ + start = data.find(_START_CODE) + if start < 0: + return + pos = start + len(_START_CODE) + while True: + nxt = data.find(_START_CODE, pos) + end = len(data) if nxt < 0 else nxt + nal = data[pos:end].rstrip(b"\x00") + if nal: + yield nal + if nxt < 0: + return + pos = nxt + len(_START_CODE) + + +def _opens_picture(nal: bytes) -> bool: + """Is this a VCL NAL whose slice starts at macroblock zero? + + `first_mb_in_slice` is the first ue(v) in the slice header, and ue(v) is + zero exactly when the first bit is set. That makes the boundary between + access units readable without a bitstream parser, and it keeps a + multi-slice picture in one access unit instead of splitting it into one + per slice, which would put the marker bit in the wrong places. + """ + return len(nal) > 1 and nal_type(nal) in _VCL and bool(nal[1] & 0x80) + + +@dataclass(frozen=True) +class AccessUnit: + """One decodable picture, with whatever headers precede it.""" + + nals: tuple[bytes, ...] + + @property + def carries_parameter_sets(self) -> bool: + return any(nal_type(n) in (_SPS, _PPS) for n in self.nals) + + @property + def is_idr(self) -> bool: + return any(nal_type(n) == 5 for n in self.nals) + + +def access_units(nals: Sequence[bytes]) -> list[AccessUnit]: + """Group NALs into access units. + + A picture-opening VCL NAL closes the previous unit, so parameter sets and + SEI attach to the picture they precede rather than to the one before. + """ + units: list[AccessUnit] = [] + current: list[bytes] = [] + have_picture = False + for nal in nals: + if _opens_picture(nal) and have_picture: + units.append(AccessUnit(tuple(current))) + current, have_picture = [], False + current.append(nal) + have_picture = have_picture or nal_type(nal) in _VCL + if current: + units.append(AccessUnit(tuple(current))) + return units + + +@dataclass(frozen=True) +class AnnexBStream: + """A parsed Annex-B file, ready to be looped.""" + + units: tuple[AccessUnit, ...] + parameter_sets: tuple[bytes, ...] + + @classmethod + def from_bytes(cls, data: bytes) -> AnnexBStream: + nals = list(iter_nals(data)) + if not nals: + raise ValueError("no NAL units found; is this an Annex-B elementary stream?") + # Keep the last of each kind: a stream whose resolution changes + # mid-file should be repeated with the sets that are in force, and + # x264 writes the same pair every time anyway. + latest: dict[int, bytes] = {} + for nal in nals: + kind = nal_type(nal) + if kind in (_SPS, _PPS): + latest[kind] = nal + sets = tuple(latest[k] for k in (_SPS, _PPS) if k in latest) + if not sets: + log.warning("the stream carries no SPS/PPS, so a viewer joining late cannot start") + return cls(units=tuple(access_units(nals)), parameter_sets=sets) + + @classmethod + def from_path(cls, path: str | Path) -> AnnexBStream: + return cls.from_bytes(Path(path).read_bytes()) + + +@dataclass +class Packetiser: + """Annex-B access units in, RTP datagrams out. + + Stateful on purpose. The sequence number, the timestamp and the SSRC + belong to a stream rather than to a frame, and a looping source must not + reset any of them: a decoder handed a timestamp that jumps backwards + treats the whole stream as corrupt. + """ + + mtu: int = DEFAULT_MTU + payload_type: int = PAYLOAD_TYPE + ssrc: int = field(default_factory=lambda: random.getrandbits(32)) + seq: int = field(default_factory=lambda: random.getrandbits(16)) + timestamp: int = field(default_factory=lambda: random.getrandbits(32)) + aggregate: bool = True + + def __post_init__(self) -> None: + # Two bytes of FU-A header, so an MTU below three could never carry a + # payload byte and would loop forever. + if self.mtu < 3: + raise ValueError(f"mtu must be at least 3 payload bytes, got {self.mtu}") + + def packetise(self, nals: Sequence[bytes], *, advance: int = 0) -> list[bytes]: + """Datagrams for one access unit, then move the clock on by `advance`. + + The marker bit lands on the last packet of the unit, which is how a + receiver knows the picture is complete without parsing it. + """ + payloads: list[bytes] = [] + for group in self._groups(nals): + if len(group) > 1: + payloads.append(self._stap_a(group)) + elif len(group[0]) <= self.mtu: + payloads.append(group[0]) + else: + payloads.extend(self._fu_a(group[0])) + + packets = [self._frame(p, marker=i == len(payloads) - 1) for i, p in enumerate(payloads)] + self.timestamp = (self.timestamp + advance) % (1 << 32) + return packets + + # -- grouping -------------------------------------------------------- + def _groups(self, nals: Sequence[bytes]) -> list[list[bytes]]: + """Runs of NALs that travel together. Singletons unless aggregating. + + A STAP-A of one NAL is legal but pointless, so a run of length one is + emitted as a single-NAL packet instead. + """ + if not self.aggregate: + return [[nal] for nal in nals] + groups: list[list[bytes]] = [] + run: list[bytes] = [] + used = 1 # the STAP-A header byte + for nal in nals: + cost = _STAP_A_LENGTH.size + len(nal) + if nal_type(nal) in _AGGREGATABLE and used + cost <= self.mtu: + run.append(nal) + used += cost + continue + if run: + groups.append(run) + run, used = [], 1 + if nal_type(nal) in _AGGREGATABLE and 1 + cost <= self.mtu: + run, used = [nal], 1 + cost + else: + groups.append([nal]) + if run: + groups.append(run) + return groups + + # -- packet shapes --------------------------------------------------- + def _frame(self, payload: bytes, *, marker: bool) -> bytes: + seq = self.seq + self.seq = (seq + 1) & 0xFFFF + second = (_MARKER if marker else 0) | self.payload_type + return _RTP_HEADER.pack(_V2, second, seq, self.timestamp, self.ssrc) + payload + + def _stap_a(self, nals: Sequence[bytes]) -> bytes: + # The aggregate's NRI is the highest of what it carries, so dropping + # it costs a receiver no more than dropping the most important NAL in + # it would have. + out = bytearray([max(_nri(n) for n in nals) | NAL_STAP_A]) + for nal in nals: + out += _STAP_A_LENGTH.pack(len(nal)) + nal + return bytes(out) + + def _fu_a(self, nal: bytes) -> list[bytes]: + indicator = _nri(nal) | NAL_FU_A + kind = nal_type(nal) + body = nal[1:] # the original header is rebuilt by the receiver + budget = self.mtu - 2 + chunks = [body[i : i + budget] for i in range(0, len(body), budget)] + out = [] + for i, chunk in enumerate(chunks): + flags = (0x80 if i == 0 else 0) | (0x40 if i == len(chunks) - 1 else 0) + out.append(bytes([indicator, flags | kind]) + chunk) + return out + + +def parse_packet(packet: bytes) -> tuple[int, int, int, int, bool, bytes]: + """Read an RTP packet back: payload type, seq, timestamp, ssrc, marker, payload. + + Here rather than in the tests because the capture tools and the replay + path need it too, and a second implementation would be a second chance to + get the field order wrong. + """ + if len(packet) < _RTP_HEADER.size: + raise ValueError(f"an RTP packet is at least {_RTP_HEADER.size} bytes, got {len(packet)}") + first, second, seq, timestamp, ssrc = _RTP_HEADER.unpack_from(packet) + if first >> 6 != 2: + raise ValueError(f"not RTP version 2: first byte {first:#04x}") + csrc = first & 0x0F + offset = _RTP_HEADER.size + 4 * csrc + return second & 0x7F, seq, timestamp, ssrc, bool(second & _MARKER), packet[offset:] + + +def rewrite_header(packet: bytes, *, ssrc: int, seq: int, timestamp: int) -> bytes: + """Restamp a recorded packet with our SSRC, sequence number and clock. + + The caller supplies the timestamp rather than an offset because a replay + that loops has to carry the recorded gaps forward past the loop point + instead of jumping back to where the capture started. `ReplaySource` + accumulates the recorded deltas to do that. + """ + first, second, _seq, _timestamp, _ssrc = _RTP_HEADER.unpack_from(packet) + head = _RTP_HEADER.pack(first, second, seq & 0xFFFF, timestamp & 0xFFFFFFFF, ssrc & 0xFFFFFFFF) + return head + packet[_RTP_HEADER.size :] + + +class VideoSource(Protocol): + """Where the simulator gets datagrams and when to send them. + + One pair per packet: how long to wait before sending it, and the bytes. + The generator is expected to be endless, so a viewer can be left running. + """ + + def packets(self) -> Iterator[tuple[float, bytes]]: ... + + @property + def describe(self) -> str: ... + + +def resolve_start_offset(offset: float | str | None, duration: float, rng: random.Random) -> float: + """Where in the stream to begin, in seconds. + + `"random"` exists to reproduce a viewer switched on while the drone is + already flying, which hands a stateless decoder a stream that begins + mid-GOP. The seed is the caller's so a failure can be run again. + """ + if offset is None: + return 0.0 + if isinstance(offset, str): + if offset != "random": + raise ValueError(f"start_offset must be a number or 'random', got {offset!r}") + return rng.uniform(0.0, duration) if duration > 0 else 0.0 + if offset < 0: + raise ValueError(f"start_offset must not be negative, got {offset}") + return offset % duration if duration > 0 else 0.0 + + +@dataclass +class PacketisedSource: + """An Annex-B file, packetised and paced at a nominal frame rate. + + Steady pacing: every frame goes out one period after the last and the + packets within a frame go out back to back. That is a synthetic + instrument. It is the right one for checking that a decoder and a + renderer work, and the wrong one for measuring latency or jitter against + the aircraft, which sends in bursts this does not imitate. Use + `ReplaySource` over a real capture for that. + """ + + stream: AnnexBStream + fps: float = DEFAULT_FPS + mtu: int = DEFAULT_MTU + parameter_set_period: int = 30 + start_offset: float | str | None = None + seed: int | None = None + + def __post_init__(self) -> None: + if self.fps <= 0: + raise ValueError(f"fps must be positive, got {self.fps}") + if not self.stream.units: + raise ValueError("the stream has no access units") + self.packetiser = Packetiser(mtu=self.mtu) + self._rng = random.Random(self.seed) + self.start_index = round(self._offset_seconds() * self.fps) % len(self.stream.units) + + def _offset_seconds(self) -> float: + return resolve_start_offset(self.start_offset, len(self.stream.units) / self.fps, self._rng) + + @property + def describe(self) -> str: + return ( + f"{len(self.stream.units)} access units paced at {self.fps:g} fps from index {self.start_index}" + ) + + def packets(self) -> Iterator[tuple[float, bytes]]: + units = self.stream.units + ticks = round(CLOCK_RATE / self.fps) + period = 1.0 / self.fps + index = self.start_index + frame = 0 + while True: + unit = units[index] + nals = list(unit.nals) + # Nothing else here lets a viewer join a stream in progress: it + # cannot decode a picture whose parameter sets it never saw. + if ( + self.parameter_set_period > 0 + and frame % self.parameter_set_period == 0 + and not unit.carries_parameter_sets + ): + nals = [*self.stream.parameter_sets, *nals] + datagrams = self.packetiser.packetise(nals, advance=ticks) + for i, datagram in enumerate(datagrams): + yield (period if i == 0 else 0.0), datagram + index = (index + 1) % len(units) + frame += 1 diff --git a/tests/test_rtp.py b/tests/test_rtp.py new file mode 100644 index 0000000..79727e8 --- /dev/null +++ b/tests/test_rtp.py @@ -0,0 +1,475 @@ +"""The RTP packetiser and the capture format, against bytes. + +Synthetic NALs throughout: a NAL is a header byte and a payload as far as +RFC 6184 is concerned, so real H.264 would only make the assertions harder to +read. The one test that needs real video is the ffmpeg decode in +`test_sim_video.py`, which is the only one that can tell whether any of this +is actually correct. +""" + +import random +import struct +from itertools import pairwise + +import pytest + +from mcbebop.media import capture, rtp + +# header bytes: NRI 3 is "most important", which parameter sets carry. +SPS = bytes([0x67]) + b"\x42\xc0\x1e" +PPS = bytes([0x68]) + b"\xce\x3c\x80" +SEI = bytes([0x06]) + b"\x05\x02\x00\x00\x80" +# A slice whose first bit is set, so first_mb_in_slice is 0 and it opens a +# picture. 0x65 is an IDR at NRI 3, 0x41 a non-IDR at NRI 2. +IDR = bytes([0x65, 0x88]) + b"\xaa" * 40 +SLICE = bytes([0x41, 0x9A]) + b"\xbb" * 40 +# first bit clear, so first_mb_in_slice is not 0: a continuation slice. +SLICE_2 = bytes([0x41, 0x1A]) + b"\xcc" * 40 + + +def annex_b(*nals: bytes, four_byte: bool = False) -> bytes: + code = b"\x00\x00\x00\x01" if four_byte else b"\x00\x00\x01" + return b"".join(code + nal for nal in nals) + + +# -- Annex-B parsing ----------------------------------------------------- +def test_three_and_four_byte_start_codes_both_split(): + assert list(rtp.iter_nals(annex_b(SPS, PPS))) == [SPS, PPS] + assert list(rtp.iter_nals(annex_b(SPS, PPS, four_byte=True))) == [SPS, PPS] + + +def test_trailing_zeros_belong_to_the_start_code_not_the_nal(): + # A fourth zero before 00 00 01 is a trailing byte of the previous NAL. + # Counting it would change the length and so change where the MTU falls. + data = b"\x00\x00\x01" + SPS + b"\x00\x00\x00\x00\x01" + PPS + assert list(rtp.iter_nals(data)) == [SPS, PPS] + + +def test_a_stream_with_no_start_code_yields_nothing(): + assert list(rtp.iter_nals(b"\xde\xad\xbe\xef")) == [] + with pytest.raises(ValueError, match="Annex-B"): + rtp.AnnexBStream.from_bytes(b"\xde\xad\xbe\xef") + + +def test_leading_garbage_before_the_first_start_code_is_skipped(): + assert list(rtp.iter_nals(b"junk" + annex_b(SPS))) == [SPS] + + +def test_nal_type_reads_the_low_five_bits(): + assert rtp.nal_type(SPS) == 7 + assert rtp.nal_type(PPS) == 8 + assert rtp.nal_type(IDR) == 5 + assert rtp.nal_type(SLICE) == 1 + + +# -- access units -------------------------------------------------------- +def test_parameter_sets_attach_to_the_picture_that_follows_them(): + units = rtp.access_units([SPS, PPS, IDR, SLICE]) + assert [u.nals for u in units] == [(SPS, PPS, IDR), (SLICE,)] + assert units[0].is_idr and units[0].carries_parameter_sets + assert not units[1].is_idr and not units[1].carries_parameter_sets + + +def test_a_continuation_slice_stays_in_the_same_access_unit(): + # Splitting a multi-slice picture per slice would put the marker bit and + # the timestamp step in the wrong places. + units = rtp.access_units([IDR, SLICE_2, SLICE]) + assert [u.nals for u in units] == [(IDR, SLICE_2), (SLICE,)] + + +def test_the_stream_keeps_the_parameter_sets_for_reuse(): + stream = rtp.AnnexBStream.from_bytes(annex_b(SPS, PPS, IDR, SLICE, SLICE)) + assert stream.parameter_sets == (SPS, PPS) + assert len(stream.units) == 3 + + +def test_a_stream_without_parameter_sets_warns(caplog): + rtp.AnnexBStream.from_bytes(annex_b(IDR)) + assert "joining late" in caplog.text + + +# -- RTP headers --------------------------------------------------------- +def unpack(packet): + return rtp.parse_packet(packet) + + +def test_a_small_nal_becomes_exactly_one_single_nal_packet(): + p = rtp.Packetiser(mtu=1400, ssrc=0x11223344, seq=7, timestamp=9000) + packets = p.packetise([IDR]) + assert len(packets) == 1 + pt, seq, ts, ssrc, marker, payload = unpack(packets[0]) + assert (pt, seq, ts, ssrc, marker) == (96, 7, 9000, 0x11223344, True) + assert payload == IDR # the NAL travels whole, header byte included + + +def test_the_header_is_twelve_bytes_of_version_two(): + packets = rtp.Packetiser().packetise([IDR]) + assert packets[0][0] == 0x80 # V=2, no padding, no extension, no CSRC + assert len(packets[0]) == 12 + len(IDR) + + +def test_parse_packet_rejects_anything_that_is_not_version_two(): + with pytest.raises(ValueError, match="version 2"): + rtp.parse_packet(b"\x40" + b"\x00" * 15) + with pytest.raises(ValueError, match="at least"): + rtp.parse_packet(b"\x80\x60") + + +# -- FU-A ---------------------------------------------------------------- +def big_nal(size: int, header: int = 0x65) -> bytes: + return bytes([header]) + bytes((i % 251) + 1 for i in range(size - 1)) + + +def test_an_oversized_nal_fragments_into_well_formed_fu_a(): + nal = big_nal(4000) + p = rtp.Packetiser(mtu=1400) + packets = p.packetise([nal]) + payloads = [unpack(pkt)[5] for pkt in packets] + assert len(payloads) == 3 # 1398 payload bytes per fragment, 3999 to carry + + for payload in payloads: + assert payload[0] & 0x1F == rtp.NAL_FU_A + assert payload[0] & 0x60 == nal[0] & 0x60 # NRI is preserved + assert payload[1] & 0x1F == rtp.nal_type(nal) # the carried type + assert payload[1] & 0x20 == 0 # the reserved bit must be zero + + starts = [bool(p[1] & 0x80) for p in payloads] + ends = [bool(p[1] & 0x40) for p in payloads] + assert starts == [True, False, False] + assert ends == [False, False, True] + + +def test_fu_a_fragments_reassemble_into_the_original_nal(): + nal = big_nal(4000) + payloads = [unpack(p)[5] for p in rtp.Packetiser(mtu=1400).packetise([nal])] + rebuilt = bytes([(payloads[0][0] & 0xE0) | (payloads[0][1] & 0x1F)]) + rebuilt += b"".join(p[2:] for p in payloads) + assert rebuilt == nal + + +def test_no_fragment_exceeds_the_mtu(): + for mtu in (3, 50, 1400): + packets = rtp.Packetiser(mtu=mtu).packetise([big_nal(4000)]) + assert max(len(unpack(p)[5]) for p in packets) <= mtu + + +def test_a_nal_exactly_at_the_mtu_is_not_fragmented(): + nal = big_nal(100) + assert len(rtp.Packetiser(mtu=100, aggregate=False).packetise([nal])) == 1 + assert len(rtp.Packetiser(mtu=99, aggregate=False).packetise([nal])) == 2 + + +def test_an_mtu_too_small_to_carry_a_payload_byte_is_refused(): + with pytest.raises(ValueError, match="at least 3"): + rtp.Packetiser(mtu=2) + + +# -- STAP-A -------------------------------------------------------------- +def test_parameter_sets_aggregate_into_one_stap_a(): + packets = rtp.Packetiser(mtu=1400).packetise([SPS, PPS, SEI, IDR]) + assert len(packets) == 2 # one STAP-A for the headers, one for the picture + payload = unpack(packets[0])[5] + assert payload[0] & 0x1F == rtp.NAL_STAP_A + assert payload[0] & 0x60 == 0x60 # the highest NRI of what it carries + + carried, pos = [], 1 + while pos < len(payload): + (length,) = struct.unpack_from(">H", payload, pos) + carried.append(payload[pos + 2 : pos + 2 + length]) + pos += 2 + length + assert carried == [SPS, PPS, SEI] + + +def test_a_picture_is_never_aggregated(): + # A STAP-A carrying a slice would be legal, but it is not what this does, + # and the docstring says so. Two small slices stay two packets. + packets = rtp.Packetiser(mtu=1400).packetise([IDR, SLICE_2]) + assert len(packets) == 2 + assert all(unpack(p)[5][0] & 0x1F != rtp.NAL_STAP_A for p in packets) + + +def test_a_lone_parameter_set_goes_out_as_a_single_nal_not_a_stap_a(): + (packet,) = rtp.Packetiser(mtu=1400).packetise([SPS]) + assert unpack(packet)[5] == SPS + + +def test_aggregation_can_be_switched_off(): + packets = rtp.Packetiser(mtu=1400, aggregate=False).packetise([SPS, PPS, SEI, IDR]) + assert len(packets) == 4 + assert [unpack(p)[5] for p in packets] == [SPS, PPS, SEI, IDR] + + +def test_aggregation_stops_at_the_mtu_rather_than_overflowing(): + sets = [big_nal(300, header=0x68) for _ in range(5)] + packets = rtp.Packetiser(mtu=700).packetise(sets) + assert len(packets) == 3 # 2 + 2 + 1 at 302 bytes of cost each + assert all(len(unpack(p)[5]) <= 700 for p in packets) + + +# -- sequence, marker, clock --------------------------------------------- +def test_sequence_numbers_advance_by_one_per_packet(): + p = rtp.Packetiser(mtu=1400, seq=100) + packets = p.packetise([SPS, PPS, IDR]) + p.packetise([SLICE]) + assert [unpack(pkt)[1] for pkt in packets] == [100, 101, 102] + assert p.seq == 103 + + +def test_sequence_numbers_wrap_at_sixteen_bits(): + p = rtp.Packetiser(mtu=1400, seq=0xFFFE) + packets = p.packetise([IDR]) + p.packetise([IDR]) + p.packetise([IDR]) + assert [unpack(pkt)[1] for pkt in packets] == [0xFFFE, 0xFFFF, 0] + + +def test_the_marker_bit_lands_only_on_the_last_packet_of_an_access_unit(): + p = rtp.Packetiser(mtu=1400) + packets = p.packetise([SPS, PPS, big_nal(4000)]) + markers = [unpack(pkt)[4] for pkt in packets] + assert markers == [False, False, False, True] + + +def test_the_timestamp_is_one_per_access_unit_and_advances_between_them(): + p = rtp.Packetiser(mtu=1400, timestamp=1000) + first = p.packetise([SPS, PPS, big_nal(4000)], advance=3000) + second = p.packetise([SLICE], advance=3000) + assert {unpack(pkt)[2] for pkt in first} == {1000} + assert {unpack(pkt)[2] for pkt in second} == {4000} + + +def test_the_timestamp_wraps_at_thirty_two_bits(): + p = rtp.Packetiser(timestamp=(1 << 32) - 1000) + p.packetise([IDR], advance=3000) + assert p.timestamp == 2000 + + +def test_the_ssrc_is_the_same_on_every_packet(): + p = rtp.Packetiser(mtu=1400) + packets = p.packetise([SPS, PPS, big_nal(4000)]) + p.packetise([SLICE]) + assert len({unpack(pkt)[3] for pkt in packets}) == 1 + + +# -- start offsets ------------------------------------------------------- +def test_no_offset_starts_at_the_beginning(): + assert rtp.resolve_start_offset(None, 10.0, random.Random(0)) == 0.0 + + +def test_a_random_offset_lands_inside_the_stream_and_repeats_with_a_seed(): + first = rtp.resolve_start_offset("random", 10.0, random.Random(5)) + second = rtp.resolve_start_offset("random", 10.0, random.Random(5)) + assert first == second + assert 0.0 <= first < 10.0 + + +def test_an_offset_past_the_end_wraps_rather_than_falling_off(): + assert rtp.resolve_start_offset(12.5, 10.0, random.Random(0)) == 2.5 + + +def test_a_nonsense_offset_is_an_error(): + with pytest.raises(ValueError, match="'random'"): + rtp.resolve_start_offset("middle", 10.0, random.Random(0)) + with pytest.raises(ValueError, match="negative"): + rtp.resolve_start_offset(-1.0, 10.0, random.Random(0)) + + +# -- the packetised source ---------------------------------------------- +@pytest.fixture +def stream(): + # Ten pictures, parameter sets only at the front, which is the case that + # makes repetition matter. + nals = [SPS, PPS, IDR] + [SLICE] * 9 + return rtp.AnnexBStream.from_bytes(annex_b(*nals)) + + +def take(source, n): + out = [] + for item in source.packets(): + out.append(item) + if len(out) >= n: + break + return out + + +def test_the_source_paces_one_period_per_frame_and_nothing_between_packets(stream): + source = rtp.PacketisedSource(stream, fps=30.0) + delays = [delay for delay, _ in take(source, 6)] + assert delays[0] == pytest.approx(1 / 30) + assert delays[1] == 0.0 # the second packet of the same frame + assert sum(1 for d in delays if d > 0) == len([d for d in delays if d > 0]) + + +def test_the_source_loops_without_the_clock_going_backwards(stream): + source = rtp.PacketisedSource(stream, fps=30.0, parameter_set_period=0) + packets = [data for _delay, data in take(source, 40)] + stamps = [unpack(p)[2] for p in packets] + assert stamps == sorted(stamps), "a timestamp that goes back makes a decoder give up" + seqs = [unpack(p)[1] for p in packets] + assert seqs == list(range(seqs[0], seqs[0] + len(seqs))) + assert len(packets) > len(stream.units), "it must have passed the loop point" + + +def test_parameter_sets_recur_so_a_viewer_can_join_late(stream): + source = rtp.PacketisedSource(stream, fps=30.0, parameter_set_period=4) + payloads = [unpack(data)[5] for _delay, data in take(source, 30)] + stap_a = [p for p in payloads if p[0] & 0x1F == rtp.NAL_STAP_A] + # One at the file's own head plus one every fourth frame after it. + assert len(stap_a) >= 5 + assert all(SPS in p and PPS in p for p in stap_a) + + +def test_turning_repetition_off_sends_the_parameter_sets_once_per_pass(stream): + source = rtp.PacketisedSource(stream, fps=30.0, parameter_set_period=0) + payloads = [unpack(data)[5] for _delay, data in take(source, 25)] + assert sum(1 for p in payloads if p[0] & 0x1F == rtp.NAL_STAP_A) == 3 + + +def test_a_random_start_offset_begins_mid_stream_and_is_reproducible(stream): + a = rtp.PacketisedSource(stream, fps=30.0, start_offset="random", seed=11) + b = rtp.PacketisedSource(stream, fps=30.0, start_offset="random", seed=11) + assert a.start_index == b.start_index + assert 0 <= a.start_index < len(stream.units) + assert rtp.PacketisedSource(stream, fps=30.0, start_offset=0.1).start_index == 3 + + +def test_a_source_needs_a_positive_frame_rate(stream): + with pytest.raises(ValueError, match="fps"): + rtp.PacketisedSource(stream, fps=0) + + +# -- the capture format -------------------------------------------------- +def rtp_packet(seq, timestamp, ssrc=0xAABBCCDD, payload=b"\x41\x9a\xff"): + return struct.pack(">BBHII", 0x80, 96, seq, timestamp, ssrc) + payload + + +def test_a_capture_round_trips(tmp_path): + path = tmp_path / "c.rtpcap" + packets = [(0.0, rtp_packet(1, 9000)), (0.033, rtp_packet(2, 12000))] + assert capture.write_capture(path, packets) == 2 + assert list(capture.read_capture(path)) == packets + + +def test_a_capture_creates_its_parent_directory(tmp_path): + path = tmp_path / "deep" / "c.rtpcap" + capture.write_capture(path, [(0.0, rtp_packet(1, 0))]) + assert path.is_file() + + +def test_something_that_is_not_a_capture_is_rejected(tmp_path): + path = tmp_path / "nope.rtpcap" + path.write_bytes(b"not a capture at all") + with pytest.raises(capture.CaptureFormatError, match="does not start"): + list(capture.read_capture(path)) + short = tmp_path / "short.rtpcap" + short.write_bytes(b"BEBO") + with pytest.raises(capture.CaptureFormatError, match="too short"): + list(capture.read_capture(short)) + + +def test_a_future_capture_version_is_refused_rather_than_misread(tmp_path): + path = tmp_path / "future.rtpcap" + path.write_bytes(struct.pack("<8sHH", capture.MAGIC, 99, 0)) + with pytest.raises(capture.CaptureFormatError, match="version 99"): + list(capture.read_capture(path)) + + +def test_a_capture_cut_short_still_reads_up_to_the_break(tmp_path, caplog): + path = tmp_path / "cut.rtpcap" + capture.write_capture(path, [(0.0, rtp_packet(1, 0)), (0.033, rtp_packet(2, 3000))]) + path.write_bytes(path.read_bytes()[:-5]) # a Ctrl-C mid-write + got = list(capture.read_capture(path)) + assert len(got) == 1 + assert "mid-packet" in caplog.text + + +# -- replay -------------------------------------------------------------- +@pytest.fixture +def capture_file(tmp_path): + """Three frames of two packets each, with a gap only between frames.""" + packets = [] + for frame in range(3): + at = frame * 0.04 + packets.append((at, rtp_packet(frame * 2, 9000 + frame * 3000))) + packets.append((at + 0.001, rtp_packet(frame * 2 + 1, 9000 + frame * 3000))) + path = tmp_path / "flight.rtpcap" + capture.write_capture(path, packets) + return path + + +def test_a_replay_reproduces_the_recorded_gaps(capture_file): + source = capture.ReplaySource.from_path(capture_file) + delays = [delay for delay, _ in take(source, 6)] + assert delays[0] == 0.0 + assert delays[1] == pytest.approx(0.001) + assert delays[2] == pytest.approx(0.039) + assert delays[3] == pytest.approx(0.001) + + +def test_a_replay_restamps_the_ssrc_and_renumbers_continuously(capture_file): + source = capture.ReplaySource.from_path(capture_file, seed=3) + packets = [data for _delay, data in take(source, 10)] + assert {unpack(p)[3] for p in packets} == {source.ssrc} + assert source.ssrc != 0xAABBCCDD + seqs = [unpack(p)[1] for p in packets] + assert seqs == list(range(seqs[0], seqs[0] + len(seqs))) + + +def test_a_replay_keeps_the_recorded_timestamp_deltas(capture_file): + source = capture.ReplaySource.from_path(capture_file) + stamps = [unpack(data)[2] for _delay, data in take(source, 6)] + # Two packets per frame share a timestamp; frames are 3000 ticks apart. + assert [b - a for a, b in pairwise(stamps)] == [0, 3000, 0, 3000, 0] + + +def test_a_replay_carries_the_clock_past_the_loop_point(capture_file): + source = capture.ReplaySource.from_path(capture_file) + items = take(source, 14) # six recorded packets, then round again + stamps = [unpack(data)[2] for _delay, data in items] + assert stamps == sorted(stamps), "looping must not rewind the clock" + # The gap the file cannot know: filled with its own median, one frame. + assert items[6][0] == pytest.approx(0.039) + assert stamps[6] - stamps[5] == 3000 + + +def test_a_replay_can_start_part_way_in(capture_file): + source = capture.ReplaySource.from_path(capture_file, start_offset=0.04) + assert source.start_index == 2 + assert capture.ReplaySource.from_path(capture_file, start_offset="random", seed=2).start_index >= 0 + + +def test_an_empty_capture_is_an_error_not_a_silent_stall(tmp_path): + path = tmp_path / "empty.rtpcap" + capture.write_capture(path, []) + with pytest.raises(capture.CaptureFormatError, match="no packets"): + capture.ReplaySource.from_path(path) + + +def test_the_recorder_writes_what_arrives_on_the_port_it_binds(tmp_path): + import socket + import threading + + path = tmp_path / "live.rtpcap" + port = 0 + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + + done = threading.Event() + count = [] + + def run(): + count.append(capture.record(path, port=port, host="127.0.0.1", max_packets=2)) + done.set() + + thread = threading.Thread(target=run, daemon=True) + thread.start() + sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + for i in range(2): + for _attempt in range(20): + sender.sendto(rtp_packet(i, i * 3000), ("127.0.0.1", port)) + if done.wait(0.05) or len(count) or i == 0: + break + done.wait(5) + sender.close() + assert count == [2] + got = list(capture.read_capture(path)) + assert len(got) == 2 + assert got[0][0] == 0.0 # times are relative to the first packet From ae2cf54e68b434e02a3e992d5ac5b59db5196bac Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 2 Oct 2026 12:07:26 -0600 Subject: [PATCH 2/2] Stream video from the simulator, gated on VideoEnable as the drone gates it The sim takes an optional video source. Nothing flows until ardrone3.MediaStreaming.VideoEnable arrives with 1; RTP then goes from the sim's own 5004 to whatever arstream2_client_stream_port the controller named in its handshake, and stops on a 0, on a link loss, or at shutdown. The MediaStreamingState.VideoEnableChanged reply goes out either way, because that reports the aircraft's state and not its bitrate, so a client waiting on the confirmation must not hang for want of a file to stream. Video stays optional: a FakeBebop() with no source gains no thread and sends no packets. One source object for the life of the sim, so sequence numbers and timestamps keep advancing across both the loop point in the file and a disable/enable cycle. A decoder handed a timestamp that went backwards treats the stream as corrupt and stays that way. The test that proves any of this works is the ffmpeg decode: frames come out at 856x480, including when ffmpeg joins a stream that has already been running for two seconds, which is the case a goggle viewer actually faces. The clip is generated at setup and never committed. Also fixed stop() on a sim that was never started: the sockets bind in __post_init__, so it had ports to release, and joining an unstarted thread raised and left them held. --- README.md | 47 +++ src/mcbebop/config.py | 7 + src/mcbebop/sim.py | 268 ++++++++++++++++- src/mcbebop/tools/connection.py | 2 +- tests/test_sim_video.py | 490 ++++++++++++++++++++++++++++++++ 5 files changed, 801 insertions(+), 13 deletions(-) create mode 100644 tests/test_sim_video.py diff --git a/README.md b/README.md index 5eafd46..d3e88bb 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,53 @@ land an airborne aircraft is the more dangerous answer. `connect(target="sim")` runs everything against a protocol-accurate simulator, which is where anything involving motion should be rehearsed. +## The simulator streams video + +The simulator can push RTP/H.264 exactly as the aircraft does, so a viewer's +whole video path can be developed and measured without a drone. It answers the +handshake with `arstream2_server_stream_port: 5004`, sends nothing until +`ardrone3.MediaStreaming.VideoEnable` arrives with 1, then streams from its own +5004 to whatever `arstream2_client_stream_port` the client named, and stops on a +0, on a link loss, or at shutdown. + +```bash +python -m mcbebop.sim --video clip.h264 # steady 30 fps +python -m mcbebop.sim --video clip.h264 --start-offset random --seed 7 +python -m mcbebop.sim --video flight.rtpcap # a real capture, replayed +``` + +`MCBEBOP_SIM_VIDEO_SOURCE=clip.h264` does the same for `connect(target="sim")`. + +Two kinds of source, and they are **different instruments**: + +| Source | Pacing | Use it for | +|---|---|---| +| `.h264` Annex-B elementary stream | packetised here, steady frame rate | does the decoder work, does the renderer work | +| `.rtpcap` capture off the aircraft | the recorded inter-packet gaps, packet for packet | latency and jitter, bursts, loss behaviour | + +Make the first from any video, at the resolution the aircraft streams: + +```bash +ffmpeg -i anything.mp4 -t 10 -vf scale=856:480 -r 30 \ + -c:v libx264 -preset ultrafast -pix_fmt yuv420p -g 30 -f h264 clip.h264 +``` + +`-f h264` already writes Annex-B, so no bitstream filter is wanted; +`h264_mp4toannexb` converts the other direction and ffmpeg rejects it here. + +Make the second from a real drone. Start the recorder first, because RTP is +connectionless and anything sent before the bind is gone, then enable video +from a session that holds the ARSDK link: + +```bash +python -m mcbebop.media.capture flight.rtpcap --seconds 30 # binds 55004 +``` + +`--start-offset random` is worth knowing about. It begins mid-GOP, which is +what a viewer switched on while the drone is already flying is handed, and +`--seed` makes a failure repeatable. Parameter sets repeat about once a second +on the packetised path, which is what lets a late joiner recover at all. + ## Install ```bash diff --git a/src/mcbebop/config.py b/src/mcbebop/config.py index d3fb890..9030425 100644 --- a/src/mcbebop/config.py +++ b/src/mcbebop/config.py @@ -20,6 +20,13 @@ class Settings(BaseSettings): capture_dir: Path = Field( default=Path("captures"), description="Where recordings and snapshots are written." ) + sim_video_source: Path | None = Field( + default=None, + description=( + "An Annex-B .h264 file or a .rtpcap capture for connect(target='sim') to stream. " + "Unset means the simulator answers VideoEnable but sends no RTP." + ), + ) transport: str = Field(default="stdio", description="stdio or http.") host: str = Field(default="127.0.0.1", description="Bind address when transport is http.") port: int = Field(default=8440, description="Port when transport is http.") diff --git a/src/mcbebop/sim.py b/src/mcbebop/sim.py index 2e76eeb..5ca8899 100644 --- a/src/mcbebop/sim.py +++ b/src/mcbebop/sim.py @@ -19,6 +19,14 @@ would agree with it about a shared mistake. One deliberate fault is baked in: the magnetometer self-test reports failure, because "all six sensors fine" is the one answer that never exercises the code that reads them. + +Video is optional and off unless a source is handed to `FakeBebop`. With one, +the sim imitates the aircraft's ARStream2 behaviour: nothing flows until +`ardrone3.MediaStreaming.VideoEnable` arrives with 1, RTP then goes from the +sim's own port 5004 to the `arstream2_client_stream_port` the controller named +in its handshake, and it stops on a 0, on a disconnect, or at shutdown. The +`MediaStreamingState.VideoEnableChanged` reply is sent either way, because +that is a protocol fact rather than a property of having video to send. """ from __future__ import annotations @@ -36,6 +44,7 @@ from pathlib import Path from typing import Any from mcbebop.arsdk.types import COMMAND_HEADER, FRAME_HEADER, BufferId, DataType +from mcbebop.media.rtp import DEFAULT_FPS, DEFAULT_MTU, AnnexBStream, PacketisedSource, VideoSource from mcbebop.protocol.types import ArgSpec, Buffer, CommandSpec, EnumSpec log = logging.getLogger(__name__) @@ -53,6 +62,16 @@ _C2D_BUFFERS = frozenset( _ALL_STATES = (0, 4, 0) _ALL_SETTINGS = (0, 2, 0) +_VIDEO_ENABLE = (1, 21, 0) + +# What the drone's handshake reply names, and where it sends from. The client +# half (55004/55005) is the controller's to choose and arrives in its request. +VIDEO_SERVER_STREAM_PORT = 5004 +VIDEO_SERVER_CONTROL_PORT = 5005 +VIDEO_CLIENT_STREAM_PORT = 55004 + +#: Suffixes read as a previously recorded raw RTP stream rather than H.264. +_REPLAY_SUFFIXES = frozenset({".rtpcap"}) # Sensor self-test order as the aircraft reports it, with the one that fails. _SENSORS = ("IMU", "barometer", "ultrasound", "GPS", "magnetometer", "vertical_camera") @@ -173,6 +192,29 @@ def _arg(spec: CommandSpec, name: str) -> ArgSpec: raise KeyError(f"{spec.full_name} has no argument {name!r}") +def video_source_for( + path: str | Path, + *, + fps: float = DEFAULT_FPS, + mtu: int = DEFAULT_MTU, + start_offset: float | str | None = None, + seed: int | None = None, +) -> VideoSource: + """Turn a file into something the sim can stream, choosing by suffix. + + A `.rtpcap` is a capture taken off the aircraft and is replayed with its + own inter-packet timing. Anything else is read as an Annex-B H.264 + elementary stream and packetised here at a steady `fps`. See + `media/capture.py` for why the difference matters. + """ + if Path(path).suffix.lower() in _REPLAY_SUFFIXES: + from mcbebop.media.capture import ReplaySource + + return ReplaySource.from_path(path, start_offset=start_offset, seed=seed) + stream = AnnexBStream.from_path(path) + return PacketisedSource(stream, fps=fps, mtu=mtu, start_offset=start_offset, seed=seed) + + @dataclass class FakeBebop: """A drone-shaped thing on a socket. @@ -191,9 +233,27 @@ class FakeBebop: stream_hz: float = 5.0 battery_start: int = 87 + # Video is off unless a source is given, so an existing FakeBebop() is + # byte for byte the drone it was before this existed. A path is resolved + # here rather than at VideoEnable time, so a typo fails at construction + # instead of silently producing a drone that never streams. + video_source: VideoSource | str | Path | None = None + video_fps: float = DEFAULT_FPS + video_mtu: int = DEFAULT_MTU + #: Seconds into the stream to start, or "random" to begin mid-GOP, which + #: is what a viewer switched on mid-flight is handed. + video_start_offset: float | str | None = None + video_seed: int | None = None + #: The drone streams from its own 5004. If that port is taken, which it is + #: whenever a second sim is already streaming, the OS picks one instead: + #: an RTP receiver binds rather than connects, so it does not care. + video_source_port: int = VIDEO_SERVER_STREAM_PORT + received: list[Received] = field(default_factory=list) handshakes: list[dict[str, Any]] = field(default_factory=list) pongs: int = 0 + video_packets_sent: int = 0 + video_bytes_sent: int = 0 def __post_init__(self) -> None: self.specs = load_specs() @@ -203,6 +263,21 @@ class FakeBebop: self._lock = threading.Lock() self._battery = self.battery_start + self._video = ( + video_source_for( + self.video_source, + fps=self.video_fps, + mtu=self.video_mtu, + start_offset=self.video_start_offset, + seed=self.video_seed, + ) + if isinstance(self.video_source, str | Path) + else self.video_source + ) + self._video_wanted = threading.Event() + self._video_target: tuple[str, int] | None = None + self._video_udp: socket.socket | None = None + self._udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self._udp.bind((self.host, self.c2d_port)) self.c2d_port = self._udp.getsockname()[1] @@ -212,13 +287,15 @@ class FakeBebop: self.discovery_port = self._tcp.getsockname()[1] self._tcp.settimeout(0.2) + targets = [ + ("discovery", self._discovery_loop), + ("commands", self._command_loop), + ("stream", self._stream_loop), + ] + if self._video is not None: + targets.append(("video", self._video_loop)) self._threads = [ - threading.Thread(target=target, name=f"sim-{name}", daemon=True) - for name, target in ( - ("discovery", self._discovery_loop), - ("commands", self._command_loop), - ("stream", self._stream_loop), - ) + threading.Thread(target=target, name=f"sim-{name}", daemon=True) for name, target in targets ] # -- lifecycle ------------------------------------------------------- @@ -237,18 +314,34 @@ class FakeBebop: def stop(self) -> None: self._stop.set() + self._video_wanted.clear() for thread in self._threads: - thread.join(timeout=2) + # The sockets bind in __post_init__ but the threads only start in + # start(), so a sim that was built and never run still has ports + # to release. Joining an unstarted thread raises, which would + # leave those ports held for the rest of the process. + if thread.ident is not None: + thread.join(timeout=2) self._udp.close() self._tcp.close() + video, self._video_udp = self._video_udp, None + if video is not None: + video.close() @property def occupied(self) -> bool: return self._d2c is not None + @property + def video_streaming(self) -> bool: + return self._video_wanted.is_set() + def release(self) -> None: """Forget the current controller, as a real drone does on link loss.""" self._d2c = None + # The aircraft's stream dies with the controlling link, which is the + # behaviour a viewer has to survive. + self._video_wanted.clear() def wait_for_controller(self, timeout: float = 5.0) -> bool: deadline = time.monotonic() + timeout @@ -307,14 +400,21 @@ class FakeBebop: conn.sendall(json.dumps({"status": 1}).encode() + b"\x00") continue self._d2c = (addr[0], int(request["d2c_port"])) + # The controller names where video should go; the drone does + # not choose it. A controller that names nothing gets the + # usual port, which is what libARController would have sent. + self._video_target = ( + addr[0], + int(request.get("arstream2_client_stream_port", VIDEO_CLIENT_STREAM_PORT)), + ) reply = { "status": 0, "c2d_port": self.c2d_port, "arstream_fragment_size": 65000, "arstream_fragment_maximum_number": 128, "arstream_max_ack_interval": -1, - "arstream2_server_stream_port": 5004, - "arstream2_server_control_port": 5005, + "arstream2_server_stream_port": VIDEO_SERVER_STREAM_PORT, + "arstream2_server_control_port": VIDEO_SERVER_CONTROL_PORT, } conn.sendall(json.dumps(reply).encode() + b"\x00") # The burst goes out per controller attach, not once per @@ -344,7 +444,10 @@ class FakeBebop: if buffer_id not in _C2D_BUFFERS or len(body) < COMMAND_HEADER.size: return ids = COMMAND_HEADER.unpack_from(body) - self.received.append(Received(ids, body[COMMAND_HEADER.size :], buffer_id, int(data_type), int(seq))) + args = body[COMMAND_HEADER.size :] + self.received.append(Received(ids, args, buffer_id, int(data_type), int(seq))) + if ids == _VIDEO_ENABLE: + self._set_video(bool(args[0]) if args else False) if ids in (_ALL_STATES, _ALL_SETTINGS): # The real drone answers these with a burst of its current state, # which is what makes request_full_state worth calling. @@ -392,6 +495,81 @@ class FakeBebop: self.emit("common.CommonState.SensorsStatesListChanged", acked=True, sensorName=sensor, sensorState=0 if sensor == _FAULTY_SENSOR else 1) # fmt: skip + # -- video ----------------------------------------------------------- + def _set_video(self, wanted: bool) -> None: + """Answer VideoEnable, and start or stop the stream. + + The event goes out whether or not there is anything to stream: a + client checking that its VideoEnable took effect is reading the + aircraft's state, not its bitrate. + """ + self.emit( + "ardrone3.MediaStreamingState.VideoEnableChanged", + acked=True, + enabled="enabled" if wanted else "disabled", + ) + if self._video is None: + if wanted: + log.debug("VideoEnable(1) with no video source; nothing to stream") + return + if wanted: + self._video_wanted.set() + else: + self._video_wanted.clear() + + def _video_socket(self) -> socket.socket: + if self._video_udp is not None: + return self._video_udp + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind((self.host, self.video_source_port)) + except OSError as exc: + log.warning( + "cannot send video from port %d (%s); using an ephemeral one instead", + self.video_source_port, + exc, + ) + sock.bind((self.host, 0)) + self._video_udp = sock + return sock + + def _video_loop(self) -> None: + """Push RTP for as long as the controller wants it. + + One source object for the life of the sim, so its sequence numbers + and timestamps keep advancing across both the loop point in the file + and a disable/enable cycle. A decoder handed a timestamp that went + backwards treats the stream as corrupt and stays that way. + """ + assert self._video is not None + while not self._stop.is_set(): + if not self._video_wanted.wait(0.1): + continue + target = self._video_target + if target is None: + self._video_wanted.clear() + continue + sock = self._video_socket() + log.info("streaming video to %s:%d: %s", *target, self._video.describe) + deadline = time.monotonic() + for delay, datagram in self._video.packets(): + if self._stop.is_set() or not self._video_wanted.is_set(): + break + if delay > 0: + deadline = max(deadline + delay, time.monotonic()) + now = time.monotonic() + if deadline > now: + time.sleep(deadline - now) + try: + sock.sendto(datagram, target) + except OSError as exc: # the socket can close under us at teardown + log.debug("video send failed: %s", exc) + break + self.video_packets_sent += 1 + self.video_bytes_sent += len(datagram) + log.info("video stopped after %d packets", self.video_packets_sent) + def _stream_loop(self) -> None: while self._d2c is None and not self._stop.is_set(): time.sleep(0.02) @@ -432,13 +610,29 @@ class FakeBebop: self.emit("common.CommonState.BatteryStateChanged", acked=True, percent=self._battery) -def serve(seconds: float = 0.0, *, host: str = "127.0.0.1", discovery_port: int = 44444) -> None: +def serve( + seconds: float = 0.0, + *, + host: str = "127.0.0.1", + discovery_port: int = 44444, + video_source: VideoSource | str | Path | None = None, + video_fps: float = DEFAULT_FPS, + video_start_offset: float | str | None = None, + video_seed: int | None = None, +) -> None: """Run a sim until interrupted. Logs; nothing goes to stdout. stdout is the MCP server's JSON-RPC transport, and this module is importable from it. """ - with FakeBebop(host=host, discovery_port=discovery_port) as sim: + with FakeBebop( + host=host, + discovery_port=discovery_port, + video_source=video_source, + video_fps=video_fps, + video_start_offset=video_start_offset, + video_seed=video_seed, + ) as sim: log.info("fake Bebop 2 on %s:%d", sim.host, sim.discovery_port) end = time.monotonic() + seconds if seconds else None try: @@ -446,3 +640,53 @@ def serve(seconds: float = 0.0, *, host: str = "127.0.0.1", discovery_port: int time.sleep(0.25) except KeyboardInterrupt: pass + + +def main(argv: list[str] | None = None) -> int: + """Run the simulator from a shell, which is how a viewer gets developed. + + python -m mcbebop.sim --video clip.h264 + + The client then handshakes on 44444 as it would with the aircraft, names + its own stream port, and sends VideoEnable to start the RTP. + """ + import argparse + import sys + + parser = argparse.ArgumentParser( + prog="python -m mcbebop.sim", description="A fake Bebop 2 on localhost, optionally with video." + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--discovery-port", type=int, default=44444) + parser.add_argument("--seconds", type=float, default=0.0, help="0 runs until interrupted") + parser.add_argument("--video", default=None, help="an Annex-B .h264 file, or a .rtpcap capture") + parser.add_argument("--fps", type=float, default=DEFAULT_FPS, help="ignored for a .rtpcap replay") + parser.add_argument( + "--start-offset", + default=None, + help="seconds into the stream, or 'random' to begin mid-GOP", + ) + parser.add_argument("--seed", type=int, default=None, help="makes --start-offset=random repeatable") + args = parser.parse_args(argv) + + offset: float | str | None = args.start_offset + if isinstance(offset, str) and offset != "random": + offset = float(offset) + + # stderr: stdout is the MCP server's transport and this module is + # importable from it. + logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stderr) + serve( + args.seconds, + host=args.host, + discovery_port=args.discovery_port, + video_source=args.video, + video_fps=args.fps, + video_start_offset=offset, + video_seed=args.seed, + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover - a hand-run tool + raise SystemExit(main()) diff --git a/src/mcbebop/tools/connection.py b/src/mcbebop/tools/connection.py index 2c4616e..b67ee4e 100644 --- a/src/mcbebop/tools/connection.py +++ b/src/mcbebop/tools/connection.py @@ -65,7 +65,7 @@ def register(mcp: FastMCP, settings: Settings) -> None: if target == SIM_TARGET: from mcbebop.sim import FakeBebop - sim = FakeBebop() + sim = FakeBebop(video_source=settings.sim_video_source) sim.__enter__() state.sim = sim session = DroneSession(ip=sim.host, discovery_port=sim.discovery_port) diff --git a/tests/test_sim_video.py b/tests/test_sim_video.py new file mode 100644 index 0000000..e68f4a9 --- /dev/null +++ b/tests/test_sim_video.py @@ -0,0 +1,490 @@ +"""The simulator's video path, end to end over real sockets. + +The assertion that counts is at the bottom: ffmpeg is pointed at the +simulator and has to produce frames at 856x480. Everything above it checks +behaviour the decode test cannot distinguish, such as whether the stream +actually stops when told to. + +Synthetic NALs for everything except the decode, so the suite still runs on a +machine without ffmpeg. The clip for the decode is generated at setup and +never committed: a video file in the repository would be a 2 MB answer to a +question ffmpeg answers in three seconds. +""" + +import json +import shutil +import socket +import struct +import subprocess +import time +from pathlib import Path + +import pytest +from PIL import Image + +from mcbebop.arsdk.types import COMMAND_HEADER, BufferId, DataType, Frame +from mcbebop.media import rtp, video +from mcbebop.sim import FakeBebop, load_specs, video_source_for + +SPECS = load_specs() +VIDEO_ENABLE_CHANGED = SPECS["ardrone3.MediaStreamingState.VideoEnableChanged"].ids + +WIDTH, HEIGHT, FPS = 856, 480, 30 + + +def free_port() -> int: + """A port nothing holds, for something else to bind in a moment. + + Racy in principle. The alternative is the real 55004, which is worse: + one left-behind ffmpeg and every run of this file fails. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +class Controller: + """A hand-rolled controller that also binds its own video port.""" + + def __init__(self, sim: FakeBebop, *, stream_port: int | None = None) -> None: + self.udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.udp.bind(("127.0.0.1", 0)) + self.udp.settimeout(0.3) + self.rtp: socket.socket | None = None + if stream_port is None: + self.rtp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.rtp.bind(("127.0.0.1", 0)) + self.rtp.settimeout(0.3) + stream_port = self.rtp.getsockname()[1] + self.stream_port = stream_port + request = { + "d2c_port": self.udp.getsockname()[1], + "controller_type": "computer", + "controller_name": "video-test", + "arstream2_client_stream_port": stream_port, + "arstream2_client_control_port": stream_port + 1, + } + with socket.create_connection(("127.0.0.1", sim.discovery_port), timeout=2) as tcp: + tcp.sendall(json.dumps(request).encode()) + raw = b"" + while b"\x00" not in raw: + raw += tcp.recv(4096) + self.reply = json.loads(raw.split(b"\x00")[0].decode()) + self.c2d = ("127.0.0.1", self.reply["c2d_port"]) + self._seq = 0 + + def video_enable(self, on: bool) -> None: + self._seq += 1 + payload = COMMAND_HEADER.pack(1, 21, 0) + bytes([1 if on else 0]) + frame = Frame(DataType.DATA_WITH_ACK, BufferId.C2D_ACK, self._seq, payload) + self.udp.sendto(frame.encode(), self.c2d) + + def rtp_packets(self, seconds: float = 1.0) -> list[bytes]: + assert self.rtp is not None + out: list[bytes] = [] + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + try: + out.append(self.rtp.recv(65535)) + except TimeoutError: + continue + return out + + def wait_for_rtp(self, seconds: float = 3.0) -> list[bytes]: + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + got = self.rtp_packets(0.3) + if got: + return got + pytest.fail("no RTP arrived on the port the handshake named") + + def events(self, seconds: float = 1.0) -> list[tuple[tuple[int, int, int], bytes]]: + out = [] + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + try: + data = self.udp.recv(65535) + except TimeoutError: + continue + for frame in Frame.decode_all(data): + if ( + frame.buffer_id in (BufferId.D2C_ACK, BufferId.D2C_NON_ACK) + and len(frame.payload) >= COMMAND_HEADER.size + ): + ids = COMMAND_HEADER.unpack_from(frame.payload) + out.append((ids, frame.payload[COMMAND_HEADER.size :])) + return out + + def close(self) -> None: + self.udp.close() + if self.rtp is not None: + self.rtp.close() + + +def decode_argv(sdp: Path, pattern: Path, frames: int = 10) -> list[str]: + """ffmpeg reading our SDP. rtp and udp have to be whitelisted explicitly + or ffmpeg refuses the file with an error that reads like a bad path.""" + return [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-protocol_whitelist", "file,rtp,udp", "-i", str(sdp), + "-frames:v", str(frames), "-fps_mode", "passthrough", str(pattern), + ] # fmt: skip + + +# -- sources ------------------------------------------------------------- +SPS = bytes([0x67]) + b"\x42\xc0\x1e" +PPS = bytes([0x68]) + b"\xce\x3c\x80" +IDR = bytes([0x65, 0x88]) + b"\xaa" * 2000 # big enough to need FU-A +SLICE = bytes([0x41, 0x9A]) + b"\xbb" * 400 + + +def synthetic_annex_b(frames: int = 10) -> bytes: + nals = [SPS, PPS, IDR] + [SLICE] * (frames - 1) + return b"".join(b"\x00\x00\x00\x01" + nal for nal in nals) + + +@pytest.fixture +def fake_source(): + stream = rtp.AnnexBStream.from_bytes(synthetic_annex_b()) + return rtp.PacketisedSource(stream, fps=FPS, parameter_set_period=5) + + +@pytest.fixture +def sim(fake_source): + # Port 0 rather than the aircraft's 5004: a receiver binds, so it never + # looks at where a packet came from, and a fixed port makes two sims in + # one test session fight. + with FakeBebop(video_source=fake_source, video_source_port=0) as fake: + yield fake + + +@pytest.fixture +def controller(sim): + client = Controller(sim) + yield client + client.close() + + +# -- no video source: unchanged ------------------------------------------ +def test_a_sim_without_video_has_no_video_thread(): + with FakeBebop() as plain: + assert not [t for t in plain._threads if t.name == "sim-video"] + assert plain.video_streaming is False + + +def test_video_enable_is_answered_even_with_nothing_to_stream(): + # The event reports the aircraft's state, not its bitrate. A client that + # waits for the confirmation must not hang because the sim has no file. + with FakeBebop() as plain: + client = Controller(plain) + try: + client.events(0.5) # drain the identity burst + client.video_enable(True) + enabled = [args for ids, args in client.events(1.0) if ids == VIDEO_ENABLE_CHANGED] + assert enabled == [struct.pack("BBHII", 0x80, 96, 1, 9000, 0x1234) + b"\x41\x9a" + capture.write_capture(cap, [(0.0, packet)]) + assert isinstance(video_source_for(cap), capture.ReplaySource) + + +# -- with a source ------------------------------------------------------- +def test_nothing_streams_until_video_enable_arrives(sim, controller): + assert controller.rtp_packets(0.5) == [] + assert sim.video_packets_sent == 0 + + +def test_video_enable_starts_rtp_on_the_port_the_handshake_named(sim, controller): + assert controller.reply["arstream2_server_stream_port"] == 5004 + controller.video_enable(True) + packets = controller.wait_for_rtp() + for packet in packets: + payload_type, _seq, _ts, _ssrc, _marker, body = rtp.parse_packet(packet) + assert payload_type == 96 + assert body + assert any(rtp.parse_packet(p)[4] for p in packets), "no frame was ever marked complete" + assert sim.video_streaming is True + + +def test_the_stream_reports_itself_enabled_then_disabled(sim, controller): + controller.events(0.4) + controller.video_enable(True) + controller.wait_for_rtp() + controller.video_enable(False) + reported = [args for ids, args in controller.events(1.0) if ids == VIDEO_ENABLE_CHANGED] + assert struct.pack(" 0 + finally: + first.stop() + second.stop() + + +# -- the test that proves it --------------------------------------------- +@pytest.fixture(scope="session") +def clip(tmp_path_factory): + """Three seconds of H.264 at the resolution the aircraft streams. + + Generated, not committed. `-f h264` already writes Annex-B, so no + bitstream filter is needed: `h264_mp4toannexb` is for the other + direction and ffmpeg rejects it on an input that is already Annex-B. + """ + if shutil.which("ffmpeg") is None: + pytest.skip("no ffmpeg") + out = tmp_path_factory.mktemp("clip") / "testsrc.h264" + argv = [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-f", "lavfi", "-i", f"testsrc=size={WIDTH}x{HEIGHT}:rate={FPS}", + "-t", "3", "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", + "-g", "15", "-f", "h264", str(out), + ] # fmt: skip + subprocess.run( + argv, + check=True, + capture_output=True, + timeout=120, + ) + return out + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="needs ffmpeg to decode") +def test_ffmpeg_decodes_the_simulated_stream_at_the_right_size(clip, tmp_path): + """If this fails the feature does not work, whatever the unit tests say.""" + port = free_port() + sdp = tmp_path / "sim.sdp" + sdp.write_text(video.sdp_text(port=port)) + pattern = tmp_path / "frame%03d.png" + + with FakeBebop(video_source=clip, video_fps=FPS, video_source_port=0) as sim: + client = Controller(sim, stream_port=port) + # ffmpeg binds before the stream starts: RTP is connectionless, so + # anything sent before it is listening is simply gone. + proc = subprocess.Popen( + decode_argv(sdp, pattern), + stderr=subprocess.PIPE, + ) + try: + time.sleep(1.0) # let it bind and read the SDP + client.video_enable(True) + _out, err = proc.communicate(timeout=60) + except subprocess.TimeoutExpired: + proc.kill() + _out, err = proc.communicate() + finally: + client.close() + + frames = sorted(tmp_path.glob("frame*.png")) + assert len(frames) >= 5, f"ffmpeg decoded {len(frames)} frames; stderr was {err.decode()[-2000:]}" + for frame in frames: + with Image.open(frame) as im: + assert im.size == (WIDTH, HEIGHT) + assert sim.video_packets_sent > len(frames) + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="needs ffmpeg to decode") +def test_ffmpeg_can_join_a_stream_already_in_progress(clip, tmp_path): + """The customer's actual case: goggles switched on mid-flight. + + Nothing but the repeated parameter sets makes this work. A stream that + sent its SPS and PPS once at the start would leave a decoder that joined + later with no way to size a frame, and it would never recover. + """ + port = free_port() + sdp = tmp_path / "late.sdp" + sdp.write_text(video.sdp_text(port=port)) + pattern = tmp_path / "late%03d.png" + + with FakeBebop( + video_source=clip, + video_fps=FPS, + video_source_port=0, + video_start_offset="random", + video_seed=19, + ) as sim: + client = Controller(sim, stream_port=port) + client.video_enable(True) + time.sleep(2.0) # the drone has been flying a while + proc = subprocess.Popen( + decode_argv(sdp, pattern), + stderr=subprocess.PIPE, + ) + try: + _out, err = proc.communicate(timeout=60) + except subprocess.TimeoutExpired: + proc.kill() + _out, err = proc.communicate() + finally: + client.close() + + frames = sorted(tmp_path.glob("late*.png")) + assert len(frames) >= 5, f"joining late decoded {len(frames)}; stderr was {err.decode()[-2000:]}" + with Image.open(frames[0]) as im: + assert im.size == (WIDTH, HEIGHT) + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="needs ffmpeg to decode") +def test_a_capture_of_our_own_stream_replays_and_still_decodes(clip, tmp_path): + """Proves the replay path on a capture we can actually make. + + A capture off the aircraft would be better and we have none, so this + records the simulator's own output instead. It exercises the file format, + the restamping and the pacing; what it cannot exercise is the aircraft's + bursts, which is the whole reason the replay path exists. + """ + from mcbebop.media import capture as cap + + record_port = free_port() + recorded: list[tuple[float, bytes]] = [] + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(("127.0.0.1", record_port)) + sock.settimeout(0.3) + + with FakeBebop(video_source=clip, video_fps=FPS, video_source_port=0) as sim: + client = Controller(sim, stream_port=record_port) + client.video_enable(True) + first = None + deadline = time.monotonic() + 6.0 + while time.monotonic() < deadline and len(recorded) < 400: + try: + data = sock.recv(65535) + except TimeoutError: + continue + now = time.monotonic() + first = now if first is None else first + recorded.append((now - first, data)) + client.close() + sock.close() + assert len(recorded) > 50, "nothing to replay" + + path = tmp_path / "own.rtpcap" + cap.write_capture(path, recorded) + + port = free_port() + sdp = tmp_path / "replay.sdp" + sdp.write_text(video.sdp_text(port=port)) + pattern = tmp_path / "replay%03d.png" + + with FakeBebop(video_source=path, video_source_port=0) as replay: + client = Controller(replay, stream_port=port) + proc = subprocess.Popen( + decode_argv(sdp, pattern), + stderr=subprocess.PIPE, + ) + try: + time.sleep(1.0) + client.video_enable(True) + _out, err = proc.communicate(timeout=60) + except subprocess.TimeoutExpired: + proc.kill() + _out, err = proc.communicate() + finally: + client.close() + + frames = sorted(tmp_path.glob("replay*.png")) + assert len(frames) >= 5, f"the replay decoded {len(frames)}; stderr was {err.decode()[-2000:]}" + with Image.open(frames[0]) as im: + assert im.size == (WIDTH, HEIGHT) + + +def test_nothing_in_the_repository_is_a_video_file(): + # The clip above is generated at setup for exactly this reason. + root = Path(__file__).resolve().parents[1] + tracked = subprocess.run( + ["git", "-C", str(root), "ls-files"], capture_output=True, text=True, check=True + ).stdout.split() + bad = [f for f in tracked if f.endswith((".h264", ".264", ".mp4", ".rtpcap", ".ts"))] + assert bad == []