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/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/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_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 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 == []