diff --git a/.gitignore b/.gitignore index 5c37fe7..7ab9cf1 100644 --- a/.gitignore +++ b/.gitignore @@ -55,6 +55,11 @@ inference/data/pilot_clusters.csv inference/data/group_centres.csv # Collected telemetry (one file per collection session, written by data_logger.py) inference/data/session_flight_data.csv +# Written by live_groups.py for one flight, and the files the README's replay demo makes +inference/data/live_flight_data.csv +inference/data/live_flight_data_instructions.csv +inference/data/replay_pilots.csv +inference/briefs/replay.csv # Local build archives /xplane_plugin.zip diff --git a/GLOSSARY.md b/GLOSSARY.md index a1dd2ac..f16c904 100644 --- a/GLOSSARY.md +++ b/GLOSSARY.md @@ -64,6 +64,10 @@ _Avoid_: Core metrics, KPIs A set of pilots whose **Deviation features** resemble each other. Groups are not ranked. _Avoid_: Tier (a **Synthetic pilot**'s known level is its **Hidden tier**, a different thing), skill level, class +**Reference cohort**: +The pilots whose **Deviation features** form the **Performance groups**, held fixed so a pilot outside them can be placed in a group. +_Avoid_: Training set, baseline, reference group + **Segment**: One of the seven phases of a **Flight**, from takeoff through landing. _Avoid_: Phase, section, leg (the five-value **Flight phase** in the out-of-scope section and a **Leg** are different things) @@ -73,9 +77,17 @@ A stretch of the route flown on one heading, such as the 190° climb or the 220 _Avoid_: Segment (the client's seven phases), phase **Instruction**: -The timestamped moment a pilot is told a new **Target**, such as "change heading to 20". Instructions are standardized across pilots. +A directive to the pilot that sets new **Targets** from the moment it is given, such as "change heading to 20". Targets it does not mention carry over from the one before. The client's instructions are standardized across pilots. _Avoid_: Command, prompt, target change +**Brief**: +The ordered list of **Instructions** for a **Flight**, each with the **Targets** it sets. The client's scenario briefing is one; so is one we write for a flight we fly ourselves. +_Avoid_: Script, scenario, flight plan, briefing + +**Settle window**: +The time after an **Instruction** during which the **Targets** it changes are not scored, so the turn or level-off it causes does not count as **Deviation**. Targets it leaves alone, or restates at the same value, are scored throughout. +_Avoid_: Grace period, transition, buffer + ### Pilots and flights **Pilot**: @@ -93,7 +105,7 @@ _Avoid_: Pilot profile, demographics ### Synthetic pilots **Synthetic pilot**: -A generated pilot with generated flights, built to deviate by a known amount, used to check that **Performance groups** find real differences before real pilots are available. +A generated pilot with generated flights, built to deviate by a known amount, used to check that **Performance groups** find real differences, and as the **Reference cohort**, before real pilots are available. _Avoid_: Test pilot, fake pilot, mock pilot **Hidden tier**: @@ -120,14 +132,16 @@ _Avoid_: Fake data, dummy data, mock data - A **Pilot** flies many **Flights**; a **Flight** produces many **Samples**. - A **Sample** has one value per telemetry field, and one **Deviation** per **Flight dynamics factor** once a **Target** exists. -- A **Flight** is divided into **Segments**, and is also flown as a sequence of **Legs**. An **Instruction** sets a new **Target** from that moment on. -- A **Pilot** falls into exactly one **Performance group**, decided by their **Deviation features** alone. +- A **Flight** is divided into **Segments**, and is also flown as a sequence of **Legs**. A **Brief** lists its **Instructions** in order. +- An **Instruction** sets new **Targets** from the moment it is given. The targets it changes are not scored during its **Settle window**. +- A **Pilot** falls into exactly one **Performance group**, decided by their **Deviation features** alone. A pilot outside the **Reference cohort** is placed in the nearest group without changing the groups. - **Pilot metadata** is compared against **Performance groups** after they are formed; it does not form them. - Out of scope: a **Sample** receives exactly one **Flight event label**, and a **Flight phase** is a kind of **Flight event label**. ## Flagged ambiguities - **Pilot rating** is the type of certification, not a performance score (client, see `docs/client_meeting_notes.md`). Whether it means the same as certification level, or an extra rating such as instrument, is still unclear. -- **Target** source is the **Instruction**: the client gives standardized, timestamped instructions during the flight, and iMotions holds the timestamps. None have been shared and no targets are captured yet, and the heading reference (true vs. magnetic) is unresolved. +- **Target** source is the **Instruction**: the client gives standardized, timestamped instructions during the flight, and iMotions holds the timestamps. Her **Brief** and timestamps have not been shared, so targets exist only for flights we brief ourselves. The heading reference for her routes (true vs. magnetic) is unresolved. +- Climbs are not settled: whether altitude **Deviation** counts while the pilot climbs to a new altitude **Target**, or only after level-off. A climb outlasts a **Settle window**. Our level-hold **Briefs** avoid the question; the client's scenario does not. - **Leg** and **Segment** may or may not line up. The client's seven **Segments** are unknown until she shares them, so how they map to our **Legs** is open. - **Segment** names and boundaries are unknown until the client shares her seven-phase segmentation, which she marks by hand from the iMotions audio and video. Detecting **Instructions** automatically, for example from the eye tracker's audio, is later work; today the client segments by hand. diff --git a/docs/adr/0004-live-clustering-runs-in-a-python-sidecar.md b/docs/adr/0004-live-clustering-runs-in-a-python-sidecar.md new file mode 100644 index 0000000..efaf6fe --- /dev/null +++ b/docs/adr/0004-live-clustering-runs-in-a-python-sidecar.md @@ -0,0 +1,17 @@ +# Live performance groups run in a Python sidecar, not in the relay + +To update **Performance groups** while a **Flight** is under way, the relay forwards each **Sample** over TCP to a separate Python program, the sidecar. The sidecar keeps the live pilot's running **Deviation features** and assigns the pilot to **Performance groups** fitted once on a reference cohort, using the existing functions in `inference/pilot_inference/performance/features.py` and `clustering.py`. The relay embeds no Python or ML and only shows the result. We chose this so there is one implementation of the clustering, the one the evidence in docs/adr/0002 was measured on. + +## Considered Options + +- **Port clustering to Rust inside the relay.** Rejected. It would duplicate KMeans, the log/floor scaling, the standardization over the cohort and the k-sweep stability check. Two copies could drift apart, and the ADR 0002 evidence would only describe one of them. The relay would also need an ML dependency for a 10 to 20 pilot cohort. +- **Embed Python in the relay** (for example through PyO3). Rejected. It couples the relay's build and release to a Python runtime and scikit-learn, and it puts a slow refit on the process that must keep forwarding to iMotions. +- **Python sidecar (chosen).** The sidecar is a separate process started like the other inference tools. A slow update, or a sidecar that is down, cannot block the relay's path to iMotions. + +## Consequences + +- The sidecar reads the relay's **Mirror** (docs/adr/0003), so the relay already forwards to iMotions unchanged when the sidecar is absent or slow. The copy is best effort, so the sidecar must tolerate dropped samples and must keep reading its socket on a thread separate from the slower work, or it loses the mirror. +- The return channel (how the sidecar tells the relay a pilot's group) is not decided here. +- The groups are fitted once and the live pilot is assigned to the nearest one, rather than refitting with the live pilot included. A refit would renumber the groups between updates, and one live pilot cannot form groups alone (`candidate_ks` needs 6). Until real pilots have **Targets**, the reference cohort is **Synthetic pilots**, so the live group is a demonstration rather than a measurement. +- The client's **Targets** are not available (`docs/telemetry_schema.md` section 3). For our own X-Plane flights we are the instructor: we write a **Brief** of **Instructions**, and the operator marks each one live as it is spoken, so the sidecar knows when every target starts. The targets an instruction changes are not scored during its **Settle window**, so the turn or level-off is not scored, since **Synthetic pilots** fly only steady holds. Targets it leaves alone are scored throughout. +- If the sidecar is later judged too heavy to run beside the relay, this decision should be revisited, since porting to Rust then means re-validating the clustering against ADR 0002's evidence. diff --git a/docs/telemetry_schema.md b/docs/telemetry_schema.md index 2c5249f..50d3cfa 100644 --- a/docs/telemetry_schema.md +++ b/docs/telemetry_schema.md @@ -65,8 +65,11 @@ outliers, but they are not part of the deviation metric set. - The plugin only reads and sends **actual** instrument values. There is no DataRef read, UI control, or CSV column for a *designated/target* value for altitude, heading, vertical speed, or airspeed. -- Nothing in `relay/`, `data_logger.py`, or the `inference/` scripts defines - or stores a target value either. +- Nothing in `relay/` or `data_logger.py` defines or stores a target value + either. The one exception is a flight we fly ourselves: there we are the + instructor, and `live_groups.py` takes the targets from a **Brief** we wrote + and writes them into its CSV (docs/adr/0004, `inference/README.md`). The + client's own targets are still missing. To satisfy the client's requirement, the pipeline needs a source of target values per flight (or per flight segment), one of: @@ -78,9 +81,9 @@ values per flight (or per flight segment), one of: some parameters (e.g. autopilot altitude/heading bugs) that could stand in for "designated" values when the autopilot is being used as the reference. -Until one of these is implemented, `target_*` columns in section 4 will be -unpopulated (`NaN`) and deviation cannot be computed from real data — only -from synthetic data where targets are generated. **Flag this to the client +Until one of these is implemented for the client's flights, `target_*` +columns in section 4 will be unpopulated (`NaN`) for her data, and deviation +can only be computed from synthetic data or from flights we brief ourselves. **Flag this to the client directly:** ask whether "designated" means an instructor-assigned briefing target, an autopilot bug setting, or something else, since the schema and ingestion path differ depending on the answer. @@ -91,20 +94,20 @@ ingestion path differ depending on the answer. | Field | Type | Units | Source | Status | |---|---|---|---|---| -| `timestamp` | ISO-8601 string, UTC | — | data_logger.py | exists | -| `pilot_id` | string | — | `data_logger.py` | exists, see capture-point note below | -| `flight_id` | string | — | `data_logger.py` | exists, see capture-point note below | +| `timestamp` | ISO-8601 string, UTC | — | `data_logger.py` or `live_groups.py` | exists | +| `pilot_id` | string | — | `data_logger.py` or `live_groups.py` | exists, see capture-point note below | +| `flight_id` | string | — | `data_logger.py` or `live_groups.py` | exists, see capture-point note below | | `altitude` | float | ft MSL | plugin | exists | -| `target_altitude` | float | ft MSL | scenario/autopilot (TBD, see §3) | **missing** | +| `target_altitude` | float | ft MSL | our **Brief**, via `live_groups.py`; client's source TBD (§3) | ours only; client's **missing** | | `altitude_deviation` | float | ft | computed | derived, see §5 | | `heading` | float | deg magnetic | plugin | exists | -| `target_heading` | float | deg magnetic | scenario/autopilot (TBD) | **missing** | +| `target_heading` | float | deg magnetic | our **Brief**, via `live_groups.py`; client's source TBD | ours only; client's **missing** | | `heading_deviation` | float | deg | computed | derived, see §5 | | `vertical_speed` | float | ft/min | plugin | exists | -| `target_vertical_speed` | float | ft/min | scenario/autopilot (TBD) | **missing** | +| `target_vertical_speed` | float | ft/min | our **Brief**, via `live_groups.py`; client's source TBD | ours only; client's **missing** | | `vertical_speed_deviation` | float | ft/min | computed | derived, see §5 | | `velocity` (airspeed) | float | kts (indicated) | plugin | exists | -| `target_airspeed` | float | kts | scenario/autopilot (TBD) | **missing** | +| `target_airspeed` | float | kts | our **Brief**, via `live_groups.py`; client's source TBD | ours only; client's **missing** | | `airspeed_deviation` | float | kts | computed | derived, see §5 | | `roll` | float | deg | plugin | exists (extra, not part of deviation set) | | `pitch` | float | deg | plugin | exists (extra) | @@ -119,7 +122,11 @@ anyway. `data_logger.py` takes `--pilot-id` and `--flight-id` and writes them into every row, which leaves the plugin, baton, and relay untouched. When `--flight-id` is omitted, a UTC timestamp is generated for each relay connection. When `--pilot-id` is omitted, rows are written as `unknown` and -cannot be grouped per pilot. +cannot be grouped per pilot. `live_groups.py`, which replaces the logger for a +flight we brief, takes the same two arguments and also writes the `target_*` +columns, blank where a target is not being scored (its **Settle window**). Its +generated flight id is one per run rather than per connection, so a mirror +reconnect continues the same flight. ### 4b. Pilot metadata table (one row per pilot, joined on `pilot_id`) diff --git a/inference/README.md b/inference/README.md index f37f8a4..d3d8a5b 100644 --- a/inference/README.md +++ b/inference/README.md @@ -11,14 +11,15 @@ Forest) is kept in its own folder, **out of scope** for the client's request; se ``` inference/ pyproject.toml, uv.lock + briefs/ instructions with targets, for flights we fly ourselves data/ input and output CSVs (not code) prepared/ train/test splits written by event_classification models/ trained models (not tracked; train_model.py writes them) pilot_inference/ the package angles.py wrapped angular difference, shared paths.py where data/ and models/ are - capture/ data_logger.py - performance/ features, clustering, synthetic pilots and their CLIs + capture/ data_logger.py, replay.py + performance/ features, clustering, synthetic pilots, live groups and their CLIs event_classification/ OUT OF SCOPE: labeler, Random Forest, evaluation tests/ mirrors pilot_inference/ ``` @@ -73,6 +74,9 @@ uv run python -m pilot_inference.performance.generate_synthetic_pilots uv run python -m pilot_inference.performance.compute_features --input data/synthetic_pilot_telemetry.csv uv run python -m pilot_inference.performance.cluster_pilots uv run python -m pilot_inference.performance.cluster_pilots --k 3 --truth data/synthetic_pilot_tiers.csv + +# Live groups during a flight (replaces the capture logger for that flight) +uv run python -m pilot_inference.performance.live_groups --brief briefs/example_level_holds.csv --pilot-id pilot-07 ``` `--pilot-id` labels every row so samples can be grouped per pilot and joined to pilot metadata. @@ -138,6 +142,65 @@ the adjusted Rand index against the hidden tiers; the file must list each pilot synthetic tiers overlap, so this is below 1: the default seed gives 0.79, and generator seeds 0 to 11 range from 0.37 to 1.00. +### Live groups + +`pilot_inference.performance.live_groups` is the sidecar from `docs/adr/0004`. It listens where +the capture logger would (port 5001, the relay's mirror suggestion) and prints, every 5 seconds, +the pilot's deviation features so far and the nearest performance group. Use it instead of the +logger for that flight: it writes the same columns plus the `target_*` columns to +`data/live_flight_data.csv`, and each instruction with its time to +`data/live_flight_data_instructions.csv`. Both files are appended to, like the logger's session +file, so one file can hold several flights and pilots. Running `compute_features` on it gives each +flight the features the console showed for it in `flight_features.csv`; the per-pilot features +average a pilot's flights. + +The client's targets are not available yet, so live groups work on flights we fly ourselves, +where we are the instructor: + +1. Write a brief: a CSV with an `instruction` column and a target per factor (`heading`, + `altitude`, `airspeed`, `vertical_speed`), one row per instruction in order. + `briefs/example_level_holds.csv` is four level holds. The first instruction sets all four + targets; after that a blank cell keeps the target from the instruction before, as a real + "turn right heading 360" keeps the altitude. Headings are magnetic, as the X-Plane heading + indicator and the plugin show them. +2. Start `live_groups` with the brief and a `--pilot-id`, then press `Connect mirror` in the relay + (or `Connect TCP` with no iMotions). It replaces the capture logger for the flight, so stop + `data_logger` first; if the port is still taken, `live_groups` says so and exits. + On Windows and Linux keep the plugin's "Positional Flight Data" window open the whole flight: + the stream stops when it closes (see the root README), and the status line's scored time then + stops growing. +3. Press Enter at the moment each instruction is spoken. The targets it changes are not scored + for 30 seconds (`--settle`), so the turn or level-off does not count; targets it leaves alone, + or restates at the same value, are scored throughout, so altitude lost in a turn counts. + Type `q` and Enter to stop. The sidecar ends the relay connection, writes every sample it + already received, then closes its files. + +The sidecar is used on Windows, the lab machine's OS. It also stops cleanly on Linux and macOS, +where closing a listening socket alone does not wake a waiting `accept()`, so stopping before a +relay connects would otherwise hang. That case was checked under WSL; the tests run on Windows. + +No group is shown until every factor has 60 seconds scored (`--min-scored`). Once one is, it +stays shown while a later instruction's factors settle, since the others are still scored. The +groups are fitted once at start-up on a reference cohort of 15 synthetic pilots with k = 3, and +the live pilot is placed in the nearest one without refitting, so the group numbers do not change +during the flight. The cohort is +synthetic, so the group demonstrates the pipeline; it does not measure the pilot against real +pilots. + +To try it without X-Plane, replay a synthetic flight the cohort does not contain (another seed), +using its targets as a one-instruction brief: + +``` +uv run python -c "from pilot_inference.performance.generate_synthetic_pilots import generate_pilots; generate_pilots(3, seed=7)[0].to_csv('data/replay_pilots.csv', index=False)" +uv run python -m pilot_inference.capture.replay --input data/replay_pilots.csv --flight P03-F1 --brief-out briefs/replay.csv +uv run python -m pilot_inference.performance.live_groups --brief briefs/replay.csv --settle 0 +# in a second terminal, then press Enter in the first +uv run python -m pilot_inference.capture.replay --input data/replay_pilots.csv --flight P03-F1 +``` + +`replay` sends a telemetry CSV in the relay's packet format at 20 Hz (`--hz`), so either listener +reads it like a live flight. It needs a single flight, so pick one with `--flight`. + ## Tests ``` @@ -145,9 +208,9 @@ uv sync # installs pytest from the dev group uv run pytest ``` -`tests/` mirrors the package: `tests/capture/`, `tests/performance/` (deviation features, the -synthetic pilot generator and performance groups, including recovering the synthetic tiers end to -end), `tests/event_classification/` (the labeler's heading change), and `tests/test_angles.py` +`tests/` mirrors the package: `tests/capture/` (the logger and replay), `tests/performance/` +(deviation features, the synthetic pilot generator, performance groups including recovering the +synthetic tiers end to end, and live groups from replayed packets to a saved CSV), `tests/event_classification/` (the labeler's heading change), and `tests/test_angles.py` for the angle helper. ## Out of scope: event classification diff --git a/inference/briefs/example_level_holds.csv b/inference/briefs/example_level_holds.csv new file mode 100644 index 0000000..8a75e2f --- /dev/null +++ b/inference/briefs/example_level_holds.csv @@ -0,0 +1,5 @@ +instruction,heading,altitude,airspeed,vertical_speed +"Fly heading 270, maintain 3000 feet, 100 knots",270,3000,100,0 +"Turn right heading 360",360,,, +"Turn right heading 090",90,,, +"Turn right heading 180",180,,, diff --git a/inference/pilot_inference/capture/data_logger.py b/inference/pilot_inference/capture/data_logger.py index b810ce5..12163ef 100644 --- a/inference/pilot_inference/capture/data_logger.py +++ b/inference/pilot_inference/capture/data_logger.py @@ -87,20 +87,20 @@ def new_flight_id() -> str: return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") -def ensure_csv(csv_path: Path) -> None: +def ensure_csv(csv_path: Path, fields: list[str] = CSV_FIELDS) -> None: csv_path.parent.mkdir(parents=True, exist_ok=True) if csv_path.exists() and csv_path.stat().st_size > 0: with csv_path.open("r", newline="", encoding="utf-8") as handle: header = next(csv.reader(handle), []) - if header != CSV_FIELDS: + if header != fields: raise ValueError( - f"{csv_path} has columns {header}, but this logger writes {CSV_FIELDS}. " + f"{csv_path} has columns {header}, but these rows have {fields}. " "Appending would misalign the rows. Pass --csv with a new path, or move " "the old file aside." ) return with csv_path.open("a", newline="", encoding="utf-8") as handle: - writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS) + writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() handle.flush() os.fsync(handle.fileno()) diff --git a/inference/pilot_inference/capture/replay.py b/inference/pilot_inference/capture/replay.py new file mode 100644 index 0000000..2343df6 --- /dev/null +++ b/inference/pilot_inference/capture/replay.py @@ -0,0 +1,145 @@ +""" +Send a telemetry CSV to a listener the way the relay would, to test without X-Plane. + +Each row becomes the relay's eight packets in the relay's order +(packets_for_fields in relay/src/state.rs), so data_logger and live_groups read +it exactly like a live flight. A telemetry column the CSV lacks is sent as 0: +synthetic telemetry has no roll, pitch, yaw or g_force. Target columns are not +sent, because the relay carries none; --brief-out writes them as a brief to +start live_groups with, and exits without sending. + +Rows are sent as they are. The committed CSVs in data/ predate #196, so their +altitude and velocity are mis-scaled (docs/telemetry_schema.md section 1); +replaying one sends those wrong values as if the plugin had. + +Usage (from inference/, after uv sync): + uv run python -m pilot_inference.capture.replay --input data/synthetic_pilot_telemetry.csv --flight P01-F1 +""" + +from __future__ import annotations + +import argparse +import socket +import time +from pathlib import Path + +import pandas as pd + +from pilot_inference.performance.features import METRICS, TARGET_COLUMN + +# The relay's send order, not the CSV's column order. +RELAY_EVENTS = [ + ("AltitudeSync", "altitude"), + ("AirspeedSync", "velocity"), + ("VerticalVelocitySync", "vertical_speed"), + ("HeadingSync", "heading"), + ("RollSync", "roll"), + ("PitchSync", "pitch"), + ("YawSync", "yaw"), + ("GForceSync", "g_force"), +] + + +def packets_for_row(row: dict) -> list[str]: + """One sample as the relay's iMotions packets, with each value sent twice as the relay does.""" + packets = [] + for event, column in RELAY_EVENTS: + value = row.get(column, 0.0) + packets.append(f"E;1;PilotDataSync;;;;;{event};{value:.6f};{value:.6f}\r\n") + return packets + + +def select_flight(telemetry: pd.DataFrame, flight_id: str | None) -> pd.DataFrame: + """ + The rows of one flight. Without a flight id the CSV must hold only one, + so two flights are never sent back to back as if they were one. + """ + if "flight_id" not in telemetry.columns: + if flight_id is not None: + raise ValueError( + f"The CSV has no flight_id column, so --flight {flight_id} cannot pick a flight" + ) + return telemetry + flights = telemetry["flight_id"].astype(str).unique().tolist() + if flight_id is None: + if len(flights) > 1: + shown = ", ".join(flights[:5]) + (", ..." if len(flights) > 5 else "") + raise ValueError(f"The CSV has {len(flights)} flights ({shown}); pick one with --flight") + return telemetry + if flight_id not in flights: + raise ValueError(f"No flight {flight_id!r} in the CSV") + return telemetry[telemetry["flight_id"].astype(str) == flight_id] + + +def brief_for(rows: pd.DataFrame) -> pd.DataFrame: + """ + A one-instruction brief holding the flight's targets, for a flight whose + targets never change, as a synthetic flight's do. + """ + brief = {"instruction": "Hold the replayed flight's targets"} + for metric in METRICS: + column = TARGET_COLUMN[metric] + if column not in rows.columns: + raise ValueError(f"No {column} column to build a brief from") + values = rows[column].dropna().unique() + if len(values) != 1: + raise ValueError(f"{column} is not one constant value, so one instruction cannot hold it") + brief[metric] = values[0] + return pd.DataFrame([brief]) + + +def replay(rows: pd.DataFrame, host: str, port: int, hz: float) -> int: + """Send every row to host:port at hz samples a second (0 for as fast as possible).""" + period = 1.0 / hz if hz > 0 else 0.0 + sent = 0 + with socket.create_connection((host, port)) as sock: + start = time.monotonic() + for row in rows.to_dict("records"): + sock.sendall("".join(packets_for_row(row)).encode("utf-8")) + sent += 1 + if period: + # Pace against the start, so a slow send does not stretch the flight. + delay = start + sent * period - time.monotonic() + if delay > 0: + time.sleep(delay) + return sent + + +def main(argv=None) -> None: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--input", type=Path, required=True, help="Telemetry CSV to send") + parser.add_argument("--flight", default=None, help="flight_id to send, if the CSV has several") + parser.add_argument("--host", default="127.0.0.1", help="Listener host") + parser.add_argument("--port", type=int, default=5001, help="Listener port") + parser.add_argument( + "--hz", type=float, default=20.0, help="Samples a second; 0 sends as fast as possible" + ) + parser.add_argument( + "--brief-out", + type=Path, + default=None, + help="Write the flight's targets as a brief CSV and exit, without sending", + ) + args = parser.parse_args(argv) + + try: + rows = select_flight(pd.read_csv(args.input), args.flight) + brief = brief_for(rows) if args.brief_out else None + except ValueError as error: + raise SystemExit(str(error)) from None + if brief is not None: + brief.to_csv(args.brief_out, index=False) + print(f"Wrote a brief with this flight's targets to {args.brief_out}") + return + print(f"Sending {len(rows)} samples to {args.host}:{args.port} at {args.hz:g} Hz") + try: + sent = replay(rows, args.host, args.port, args.hz) + except ConnectionRefusedError: + raise SystemExit(f"Nothing is listening on {args.host}:{args.port}. Start the listener first.") + except ConnectionError: + raise SystemExit("The listener closed the connection before the flight was fully sent.") + print(f"Sent {sent} samples") + + +if __name__ == "__main__": + main() diff --git a/inference/pilot_inference/performance/clustering.py b/inference/pilot_inference/performance/clustering.py index 135f032..43132f8 100644 --- a/inference/pilot_inference/performance/clustering.py +++ b/inference/pilot_inference/performance/clustering.py @@ -6,6 +6,8 @@ DataFrame-in functions; cluster_pilots.py is the CLI wrapper. """ +from dataclasses import dataclass + import numpy as np import pandas as pd from sklearn.cluster import KMeans @@ -52,6 +54,13 @@ def clustering_features(pilots: pd.DataFrame) -> pd.DataFrame: return features[FEATURE_COLUMNS] +def _logged(features: pd.DataFrame) -> pd.DataFrame: + floors = pd.Series( + {f"{metric}_{part}": LOG_FLOOR[metric] for metric in METRICS for part in FEATURE_PARTS} + ) + return np.log(features + floors[features.columns]) + + def _scaled(features: pd.DataFrame) -> np.ndarray: """ Log, then standardize over the cohort being clustered (never per pilot). @@ -59,14 +68,16 @@ def _scaled(features: pd.DataFrame) -> np.ndarray: Deviation sizes are positive and pilots differ by multiples, so on a linear scale the worst pilots stretch the axis and the rest bunch up. """ - floors = pd.Series( - {f"{metric}_{part}": LOG_FLOOR[metric] for metric in METRICS for part in FEATURE_PARTS} - ) - return StandardScaler().fit_transform(np.log(features + floors[features.columns])) + return StandardScaler().fit_transform(_logged(features)) + + +def _kmeans(k: int, seed: int) -> KMeans: + """The one KMeans setup every fit uses, so the sweep and the groups cannot drift apart.""" + return KMeans(n_clusters=k, n_init=10, random_state=seed) def _kmeans_labels(scaled: np.ndarray, k: int, seed: int) -> np.ndarray: - return KMeans(n_clusters=k, n_init=10, random_state=seed).fit_predict(scaled) + return _kmeans(k, seed).fit_predict(scaled) def candidate_ks(n_pilots: int) -> range: @@ -129,22 +140,66 @@ def fit(pilots: pd.DataFrame, k: int, seed: int = 0): Raises if k is outside candidate_ks, the range the sweep offers. That needs at least 6 pilots. """ + reference = fit_reference(pilots, k, seed) + return reference.assignments, reference.centres + + +def _centres(features: pd.DataFrame, labels: np.ndarray) -> pd.DataFrame: + # The raw-unit mean of each group's members, not the KMeans centre, which + # lives in log-scaled space. + grouped = features.groupby(labels) + centres = grouped.mean() + centres.insert(0, "pilots", grouped.size()) + return centres.rename_axis("group").reset_index() + + +@dataclass +class CohortGroups: + """ + The performance groups of a reference cohort, fitted once and kept so that + pilots outside the cohort can be placed in a group without refitting + (docs/adr/0004). + + assignments and centres are what fit() returns for the same cohort, k and + seed. + """ + + scaler: StandardScaler + kmeans: KMeans + assignments: pd.DataFrame + centres: pd.DataFrame + + def assign(self, pilots: pd.DataFrame) -> pd.DataFrame: + """ + The nearest of the cohort's groups for each pilot, as pilot_id and group. + + Pilots are scaled with the reference cohort's scaling, not their own, + so a group means the same thing for them as for the cohort. They do + not move the groups. Raises on NaN features, like fit(). + """ + features = clustering_features(pilots) + scaled = self.scaler.transform(_logged(features)) + return pd.DataFrame({"pilot_id": features.index, "group": self.kmeans.predict(scaled)}) + + +def fit_reference(pilots: pd.DataFrame, k: int, seed: int = 0) -> CohortGroups: + """ + fit(), keeping the fitted scaling and KMeans so other pilots can be assigned. + + Same rules as fit(): k must be in candidate_ks, which needs at least 6 pilots. + """ features = clustering_features(pilots) allowed = candidate_ks(len(features)) if k not in allowed: raise ValueError( f"k must be {allowed.start} to {allowed.stop - 1} for {len(features)} pilots, got {k}" ) - labels = _kmeans_labels(_scaled(features), k, seed) + logged = _logged(features) + scaler = StandardScaler().fit(logged) + kmeans = _kmeans(k, seed).fit(scaler.transform(logged)) + labels = kmeans.labels_ assignments = pd.DataFrame({"pilot_id": features.index, "group": labels}) - - # The raw-unit mean of each group's members, not the KMeans centre, which - # lives in log-scaled space. - grouped = features.groupby(labels) - centres = grouped.mean() - centres.insert(0, "pilots", grouped.size()) - centres = centres.rename_axis("group").reset_index() - return assignments, centres + return CohortGroups(scaler, kmeans, assignments, _centres(features, labels)) def truth_agreement(assignments: pd.DataFrame, truth: pd.DataFrame) -> float: diff --git a/inference/pilot_inference/performance/live.py b/inference/pilot_inference/performance/live.py new file mode 100644 index 0000000..938434b --- /dev/null +++ b/inference/pilot_inference/performance/live.py @@ -0,0 +1,273 @@ +""" +Deviation features for one flight while it is being flown (docs/adr/0004). + +For flights we fly ourselves we are the instructor, so we know the targets: a +brief lists the instructions with the targets each one sets, and the operator +marks each one as it is spoken. Targets an instruction does not name carry +over. LiveFlight attaches the targets in force to every sample, leaving the +ones an instruction changed blank for a settle window so the turn or level-off +is not scored, and summarises the scored samples with the same functions the +offline pipeline uses. Pure, with +time passed in; live_groups.py is the CLI that owns the socket, threads and +console. +""" + +from __future__ import annotations + +import copy +import math +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path + +import pandas as pd + +from pilot_inference.angles import angular_diff +from pilot_inference.capture.data_logger import CSV_FIELDS, TELEMETRY_FIELDS +from pilot_inference.performance.features import METRICS, TARGET_COLUMN, compute_features + +# The telemetry the capture logger writes, plus the target in force for each +# sample, so a saved flight goes straight into compute_features. +LIVE_CSV_FIELDS = CSV_FIELDS + [TARGET_COLUMN[metric] for metric in METRICS] + +BRIEF_COLUMNS = ["instruction", *METRICS] + +# Synthetic pilots fly only steady holds, so the turn or level-off after an +# instruction is left out rather than scored against them. +DEFAULT_SETTLE = timedelta(seconds=30) + +# Longest gap between two scored samples that still counts as scored time. The +# stream is 20 Hz, so anything longer is a stall or dropped copies, not flying. +MAX_SAMPLE_GAP_SECONDS = 1.0 + + +@dataclass(frozen=True) +class Instruction: + """ + One line of a brief: what the pilot is told, and the targets it sets. + + targets holds only the factors the instruction names; the others carry + over from the instruction before. + """ + + text: str + targets: dict[str, float] + + +def load_brief(path: Path) -> list[Instruction]: + """ + Read a brief: a CSV with an instruction column and one target column per + flight dynamics factor, one row per instruction in the order given. A blank + target carries over from the instruction before. + + Raises on a missing column, an empty brief, or a blank target in the first + instruction, which has nothing to carry over from and would leave the + pilot with NaN features that cannot be placed in a group. + """ + brief = pd.read_csv(path) + missing = [c for c in BRIEF_COLUMNS if c not in brief.columns] + if missing: + raise ValueError(f"{path} is missing column(s): {', '.join(missing)}") + if brief.empty: + raise ValueError(f"{path} has no instructions") + + instructions = [] + for number, row in enumerate(brief.itertuples(index=False), start=1): + given = {metric: float(getattr(row, metric)) for metric in METRICS} + infinite = [metric for metric, value in given.items() if math.isinf(value)] + if infinite: + raise ValueError( + f"{path} instruction {number} has an infinite target for {', '.join(infinite)}" + ) + blank = [metric for metric, value in given.items() if math.isnan(value)] + if number == 1 and blank: + raise ValueError( + f"{path} instruction 1 has no target for {', '.join(blank)}; " + "the first instruction sets all four" + ) + targets = {metric: value for metric, value in given.items() if metric not in blank} + instructions.append(Instruction(str(row.instruction), targets)) + return instructions + + +def _changed(metric: str, before: float, after: float) -> bool: + """Whether a target moved. Headings compare wrapped, so 0 after 360 is the same north.""" + if metric == "heading": + return angular_diff(after, before) != 0 + return after != before + + +@dataclass(frozen=True) +class _Given: + """ + An instruction as given: when, every target in force from then on, and + when each target starts being scored, after its settle window. + """ + + at: datetime + instruction: Instruction + targets: dict[str, float] + scored_from: dict[str, datetime] + + +class LiveFlight: + """ + One flight in progress: the brief, which instruction is in force, and the + samples scored against it. + + Not thread-safe; the CLI holds a lock around every call, or reads a snapshot(). + """ + + def __init__( + self, + brief: list[Instruction], + pilot_id: str, + flight_id: str, + settle: timedelta = DEFAULT_SETTLE, + ): + if not brief: + raise ValueError("A brief needs at least one instruction") + self.brief = brief + self.pilot_id = pilot_id + self.flight_id = flight_id + self.settle = settle + # Instructions as given, in order. + self._history: list[_Given] = [] + self.scored_seconds_by_factor = {metric: 0.0 for metric in METRICS} + # Per factor: when it was last scored, and when that scored stretch began. + self._last_scored: dict[str, tuple[datetime, datetime]] = {} + self._scored_rows: list[dict] = [] + + @property + def scored_seconds(self) -> float: + """Scored time of the least-scored factor, which is what a group waits on.""" + return min(self.scored_seconds_by_factor.values()) + + @property + def given(self) -> int: + """How many instructions have been given.""" + return len(self._history) + + @property + def current(self) -> Instruction | None: + return self._history[-1].instruction if self._history else None + + @property + def current_targets(self) -> dict[str, float]: + """Every target the latest instruction left in force, settling or not.""" + return dict(self._history[-1].targets) if self._history else {} + + def settling(self, at: datetime) -> dict[str, float]: + """ + The factors still inside their settle window at `at`, with the seconds + each has left, in METRICS order. An earlier instruction's window can + still be running for a factor the latest one left alone. + """ + index = self._in_force(at) + if index is None: + return {} + given = self._history[index] + return { + metric: (given.scored_from[metric] - at).total_seconds() + for metric in METRICS + if at < given.scored_from[metric] + } + + def instruct(self, at: datetime) -> Instruction | None: + """ + Put the next instruction in force from `at`, and return it. + + Returns None, changing nothing, once the brief is used up, so the last + instruction stays in force until the flight ends. + """ + if self.given >= len(self.brief): + return None + instruction = self.brief[self.given] + before = self._history[-1] if self._history else None + targets = {**(before.targets if before else {}), **instruction.targets} + # Only a target whose value changes settles; one left alone or restated + # at the same value keeps being scored. + scored_from = { + metric: at + self.settle + if before is None or _changed(metric, before.targets[metric], targets[metric]) + else before.scored_from[metric] + for metric in METRICS + } + self._history.append(_Given(at, instruction, targets, scored_from)) + return instruction + + def _in_force(self, at: datetime) -> int | None: + """Index into the history of the instruction in force at `at`, if any.""" + for index in range(len(self._history) - 1, -1, -1): + if self._history[index].at <= at: + return index + return None + + def targets_at(self, at: datetime) -> dict[str, float]: + """ + The targets in force at `at`: NaN before the first instruction and while settling. + + Goes by `at`, not by which instruction is newest, so a sample that + arrived just before an instruction but is handled just after it is + still judged against the one before. + """ + index = self._in_force(at) + if index is None: + return {metric: math.nan for metric in METRICS} + given = self._history[index] + return { + metric: math.nan if at < given.scored_from[metric] else given.targets[metric] + for metric in METRICS + } + + def add_sample(self, values: dict[str, float], at: datetime) -> dict: + """ + Record one sample that arrived at `at`, and return it as a LIVE_CSV_FIELDS row. + + values has one entry per TELEMETRY_FIELDS column. Targets in the row + are NaN where nothing is being scored. + """ + targets = self.targets_at(at) + row = { + "timestamp": at.isoformat(timespec="milliseconds"), + "pilot_id": self.pilot_id, + "flight_id": self.flight_id, + **{name: values[name] for name in TELEMETRY_FIELDS}, + **{TARGET_COLUMN[metric]: targets[metric] for metric in METRICS}, + } + scored = [metric for metric in METRICS if not math.isnan(targets[metric])] + if not scored: + return row + given = self._history[self._in_force(at)] + for metric in scored: + stretch = given.scored_from[metric] + last = self._last_scored.get(metric) + # A settle window ends a stretch, so the time across it is not scored time. + if last is not None and last[1] == stretch: + gap = (at - last[0]).total_seconds() + self.scored_seconds_by_factor[metric] += min(max(gap, 0.0), MAX_SAMPLE_GAP_SECONDS) + self._last_scored[metric] = (at, stretch) + self._scored_rows.append(row) + return row + + def snapshot(self) -> LiveFlight: + """ + A copy that later instructions and samples do not change, safe to read + without the caller's lock. Extend it when __init__ gains a mutable field. + """ + frozen = copy.copy(self) + frozen._history = list(self._history) + frozen.scored_seconds_by_factor = dict(self.scored_seconds_by_factor) + frozen._last_scored = dict(self._last_scored) + frozen._scored_rows = list(self._scored_rows) + return frozen + + def pilot_features(self) -> pd.DataFrame | None: + """ + The pilot's deviation features so far, shaped like + features.pilot_features output, or None before anything is scored. + """ + if not self._scored_rows: + return None + _, pilots = compute_features(pd.DataFrame(self._scored_rows)) + return pilots diff --git a/inference/pilot_inference/performance/live_groups.py b/inference/pilot_inference/performance/live_groups.py new file mode 100644 index 0000000..0f914f4 --- /dev/null +++ b/inference/pilot_inference/performance/live_groups.py @@ -0,0 +1,450 @@ +""" +Show a pilot's performance group while they fly (docs/adr/0004). + +Listens for the relay's mirror (or Connect TCP), assembles samples the way the +capture logger does, attaches the brief's targets as the operator marks each +instruction, and writes every sample to CSV. Every few seconds it prints the +pilot's deviation features and the nearest group of a reference cohort. The +cohort is synthetic until real pilots have targets, so the group demonstrates +the pipeline rather than measures the pilot. + +Usage (from inference/, after uv sync): + uv run python -m pilot_inference.performance.live_groups --brief briefs/example_level_holds.csv --pilot-id pilot-07 + +Press Enter as each instruction is spoken, and type q then Enter to stop. +""" + +from __future__ import annotations + +import argparse +import csv +import math +import os +import queue +import socket +import threading +import warnings +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pandas as pd + +from pilot_inference.capture.data_logger import ( + DEFAULT_PILOT_ID, + SampleBuffer, + ensure_csv, + new_flight_id, + parse_packet, +) +from pilot_inference.paths import DATA_DIR +from pilot_inference.performance.clustering import ( + FEATURE_COLUMNS, + CohortGroups, + fit_reference, + truth_agreement, +) +from pilot_inference.performance.features import METRICS, TARGET_COLUMN, compute_features +from pilot_inference.performance.generate_synthetic_pilots import generate_pilots +from pilot_inference.performance.live import ( + DEFAULT_SETTLE, + LIVE_CSV_FIELDS, + Instruction, + LiveFlight, + load_brief, +) + +INSTRUCTION_FIELDS = [ + "timestamp", + "pilot_id", + "flight_id", + "number", + "instruction", + *[TARGET_COLUMN[metric] for metric in METRICS], +] + +UNITS = {"altitude": "ft", "heading": "deg", "vertical_speed": "ft/min", "airspeed": "kts"} + +# Port 5001 is what the relay's mirror box suggests. This replaces the capture +# logger for a live flight, since it writes the same columns and more. +DEFAULT_PORT = 5001 + + +def now() -> datetime: + return datetime.now(timezone.utc) + + +def listen(host: str, port: int) -> socket.socket: + """ + A listening socket on host:port that nothing else can share. + + On Windows SO_REUSEADDR would let this bind a port the capture logger is + already listening on, leaving the relay to reach either one, so there the + port is claimed exclusively instead. Elsewhere SO_REUSEADDR only allows a + quick restart. Exits with the likely cause if the port is taken. + """ + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): + server.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) + else: + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + server.bind((host, port)) + except OSError as error: + server.close() + raise SystemExit( + f"Port {port} on {host} is in use ({error.strerror}). Is data_logger or another " + "live_groups still running? Stop it, or pass --port and point the relay's mirror there." + ) from None + server.listen() + return server + + +def read_relay(conn: socket.socket, samples: queue.Queue) -> None: + """ + Turn one relay connection into (values, arrival time) pairs on `samples`. + + Does nothing slower than parsing, so it keeps up with the mirror: the + mirror drops copies, and then closes, if its reader stops reading + (docs/adr/0003). + """ + sample = SampleBuffer() + with conn, conn.makefile("r", encoding="utf-8", newline="") as reader: + for raw_line in reader: + parsed = parse_packet(raw_line) + if parsed is None: + continue + sample.update(*parsed) + if sample.is_complete(): + samples.put((dict(sample.values), now())) + sample.reset() + + +class RelayListener: + """ + Accepts relay connections one after another on `server`, reading each into + `samples` on a thread of its own, until closed. + + A reconnect continues the same flight, as a restarted recording does. + """ + + def __init__(self, server: socket.socket, samples: queue.Queue): + self._server = server + self._samples = samples + self._lock = threading.Lock() + self._conn: socket.socket | None = None + self._closed = False + self._thread = threading.Thread(target=self._serve, daemon=True) + + @property + def port(self) -> int: + return self._server.getsockname()[1] + + def start(self) -> None: + self._thread.start() + + def close(self) -> None: + """ + Stop accepting, end the connection in progress and wait for the reader, + so nothing is queued once this returns. + """ + with self._lock: + self._closed = True + conn = self._conn + # On Linux and macOS closing a listening socket does not wake a thread + # blocked in accept(); shutting it down does. Windows raises here, and + # close() alone is enough there. + try: + self._server.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self._server.close() + if conn is not None: + # Unblocks the reader's readline, which then ends the connection itself. + try: + conn.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self._thread.join() + + def _serve(self) -> None: + while True: + try: + conn, addr = self._server.accept() + except OSError: + return + with self._lock: + if self._closed: + conn.close() + return + self._conn = conn + print(f"Relay connected from {addr[0]}:{addr[1]}") + try: + read_relay(conn, self._samples) + except ConnectionError: + pass + except (OSError, ValueError) as error: + # Garbled bytes or a socket error end this connection, not the + # listener: the mirror never reconnects by itself (docs/adr/0003), + # so the next Connect mirror must still find someone listening. + if not self._closed: + print(f"Relay connection failed: {error}") + with self._lock: + self._conn = None + if self._closed: + return + print("Relay disconnected. Press Connect mirror again to resume.") + + +def _cell(value) -> str: + if isinstance(value, float): + return "" if math.isnan(value) else f"{value:.6f}" + return str(value) + + +class CsvLog: + """Rows appended to a CSV and flushed one at a time, like the capture logger.""" + + def __init__(self, path: Path, fields: list[str]): + ensure_csv(path, fields) + self.path = path + self._handle = path.open("a", newline="", encoding="utf-8") + self._writer = csv.DictWriter(self._handle, fieldnames=fields) + + def write(self, row: dict) -> None: + self._writer.writerow({name: _cell(value) for name, value in row.items()}) + self._handle.flush() + os.fsync(self._handle.fileno()) + + def close(self) -> None: + self._handle.close() + + +# Put on the sample queue to stop record_samples once everything before it is written. +STOP = None + + +def record_samples( + samples: queue.Queue, flight: LiveFlight, lock: threading.Lock, log: CsvLog +) -> None: + """Score and write each sample as it comes off the queue, until STOP.""" + while True: + item = samples.get() + if item is STOP: + return + values, at = item + with lock: + row = flight.add_sample(values, at) + log.write(row) + + +def reference_cohort(n_pilots: int, k: int, seed: int) -> tuple[CohortGroups, float]: + """Synthetic pilots with their groups, and how well the groups match their hidden tiers.""" + telemetry, truth = generate_pilots(n_pilots=n_pilots, seed=seed) + _, pilots = compute_features(telemetry) + reference = fit_reference(pilots, k, seed) + return reference, truth_agreement(reference.assignments, truth) + + +def describe_groups(reference: CohortGroups) -> str: + """The cohort's groups as raw-unit centres, one line per group.""" + lines = [] + for centre in reference.centres.itertuples(index=False): + parts = [ + f"{metric} std {getattr(centre, f'{metric}_std'):.1f} " + f"|bias| {getattr(centre, f'{metric}_abs_bias'):.1f} {UNITS[metric]}" + for metric in METRICS + ] + lines.append(f" group {centre.group} ({centre.pilots} pilots): " + ", ".join(parts)) + return "\n".join(lines) + + +def status_line( + flight: LiveFlight, reference: CohortGroups, min_scored: float, at: datetime +) -> str: + """One console line: where the brief is, how much is scored, and the group if there is one.""" + if flight.current is None: + return "Waiting for the first instruction. Press Enter as you give it." + progress = f"instruction {flight.given}/{len(flight.brief)}" + settling = flight.settling(at) + if len(settling) == len(METRICS): + progress += " (settling)" + elif settling: + progress += f" (settling {', '.join(settling)})" + scored = f"scored {flight.scored_seconds:.0f}s" + with warnings.catch_warnings(): + # A metric with nothing scored yet warns; the line already says so. + warnings.simplefilter("ignore", UserWarning) + pilots = flight.pilot_features() + if pilots is None: + return f"{progress}, nothing scored yet" + features = ", ".join( + f"{metric} std {pilots[f'{metric}_std'].iloc[0]:.1f} " + f"|bias| {pilots[f'{metric}_abs_bias'].iloc[0]:.1f}" + for metric in METRICS + ) + if flight.scored_seconds < min_scored or pilots[FEATURE_COLUMNS].isna().to_numpy().any(): + group = f"provisional, {min_scored:.0f}s scored needed for a group" + else: + group = f"group {reference.assign(pilots)['group'].iloc[0]}" + return f"{progress}, {scored}, {group} | {features}" + + +def print_status( + flight: LiveFlight, + reference: CohortGroups, + lock: threading.Lock, + min_scored: float, + interval: float, + stop: threading.Event, +) -> None: + while not stop.wait(interval): + # Copy under the lock, recompute outside it, so the recorder is never held up. + with lock: + snapshot = flight.snapshot() + line = status_line(snapshot, reference, min_scored, now()) + print(f"[{now().astimezone():%H:%M:%S}] {line}") + + +def instruction_row(flight: LiveFlight, instruction: Instruction, at: datetime) -> dict: + return { + "timestamp": at.isoformat(timespec="milliseconds"), + "pilot_id": flight.pilot_id, + "flight_id": flight.flight_id, + "number": flight.given, + "instruction": instruction.text, + # The targets in force after it, so a turn's row still shows the altitude it keeps. + **{TARGET_COLUMN[metric]: value for metric, value in flight.current_targets.items()}, + } + + +def instructions_path(csv_path: Path) -> Path: + return csv_path.with_name(f"{csv_path.stem}_instructions.csv") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument( + "--brief", type=Path, required=True, help="Brief CSV: instruction plus one target per factor" + ) + parser.add_argument("--host", default="127.0.0.1", help="Host interface to bind") + parser.add_argument( + "--port", type=int, default=DEFAULT_PORT, help="TCP port the relay connects to" + ) + parser.add_argument("--pilot-id", default=DEFAULT_PILOT_ID, help="Identifier for the pilot flying") + parser.add_argument( + "--flight-id", default=None, help="Identifier for this flight; defaults to the start time" + ) + parser.add_argument( + "--csv", + type=Path, + default=DATA_DIR / "live_flight_data.csv", + help="Samples with targets; instructions go to _instructions.csv beside it", + ) + parser.add_argument( + "--settle", + type=float, + default=DEFAULT_SETTLE.total_seconds(), + help="Seconds after each instruction that are not scored", + ) + parser.add_argument( + "--min-scored", type=float, default=60.0, help="Scored seconds before naming a group" + ) + parser.add_argument("--interval", type=float, default=5.0, help="Seconds between status lines") + parser.add_argument("--k", type=int, default=3, help="Number of performance groups") + parser.add_argument("--seed", type=int, default=0, help="Seed for the synthetic reference cohort") + parser.add_argument( + "--cohort-pilots", type=int, default=15, help="Synthetic pilots in the reference cohort" + ) + return parser + + +def main(argv=None) -> None: + args = build_parser().parse_args(argv) + brief = load_brief(args.brief) + # Before the cohort fit and the CSVs, so a taken port fails at once and leaves nothing behind. + server = listen(args.host, args.port) + flight = LiveFlight( + brief, + args.pilot_id, + args.flight_id or new_flight_id(), + settle=timedelta(seconds=args.settle), + ) + + print(f"Fitting {args.k} groups on {args.cohort_pilots} synthetic pilots (seed {args.seed})...") + reference, agreement = reference_cohort(args.cohort_pilots, args.k, args.seed) + print("Performance groups of the synthetic reference cohort (a demonstration, not a measurement):") + print(describe_groups(reference)) + print(f" agreement with the synthetic pilots' hidden tiers (adjusted Rand): {agreement:.2f}") + + samples_log = CsvLog(args.csv, LIVE_CSV_FIELDS) + instructions_log = CsvLog(instructions_path(args.csv), INSTRUCTION_FIELDS) + lock = threading.Lock() + stop = threading.Event() + samples: queue.Queue = queue.Queue() + + recorder = threading.Thread( + target=record_samples, args=(samples, flight, lock, samples_log), daemon=True + ) + listener = RelayListener(server, samples) + listener.start() + threads = [ + recorder, + threading.Thread( + target=print_status, + args=(flight, reference, lock, args.min_scored, args.interval, stop), + daemon=True, + ), + ] + for thread in threads: + thread.start() + + print(f"Listening for the relay on {args.host}:{args.port}") + print(f"Pilot {flight.pilot_id}, flight {flight.flight_id}") + print(f"Writing samples to {args.csv} and instructions to {instructions_log.path}") + if flight.pilot_id == DEFAULT_PILOT_ID: + print("Pass --pilot-id to label these rows, or they cannot be grouped per pilot.") + print(f"Brief: {len(brief)} instruction(s).") + print("Press Enter as you give each one, q then Enter to stop.") + print(f" next: {brief[0].text}") + + try: + while True: + if input().strip().lower() == "q": + break + at = now() + with lock: + instruction = flight.instruct(at) + if instruction is not None: + row = instruction_row(flight, instruction, at) + settling = flight.settling(at) + if instruction is None: + print("No instructions left in the brief. The last one stays in force; q to stop.") + continue + instructions_log.write(row) + print(f"Instruction {flight.given}/{len(brief)} given: {instruction.text}") + if not settling: + print(" no target changed, so nothing settles") + # Group by whole seconds left, since an earlier window may still be running. + left = {metric: math.ceil(seconds) for metric, seconds in settling.items()} + for seconds in sorted(set(left.values()), reverse=True): + names = ", ".join(m for m, value in left.items() if value == seconds) + print(f" {names} scored again in {seconds}s") + if flight.given < len(brief): + print(f" next: {brief[flight.given].text}") + except (EOFError, KeyboardInterrupt): + pass + finally: + stop.set() + # Stop the reader first, so nothing is queued after STOP; then write out + # everything it queued before closing the files. + listener.close() + samples.put(STOP) + recorder.join() + samples_log.close() + instructions_log.close() + with lock: + print(f"Stopped. {flight.scored_seconds:.0f}s scored over {flight.given} instruction(s).") + + +if __name__ == "__main__": + main() diff --git a/inference/tests/capture/test_replay.py b/inference/tests/capture/test_replay.py new file mode 100644 index 0000000..1c9a466 --- /dev/null +++ b/inference/tests/capture/test_replay.py @@ -0,0 +1,74 @@ +import pandas as pd +import pytest + +from pilot_inference.capture import data_logger +from pilot_inference.capture.replay import brief_for, packets_for_row, select_flight + +ROW = {"altitude": 3000.5, "heading": 271.25, "vertical_speed": -40.0, "velocity": 99.5} + + +def test_packets_follow_the_relay_order(): + events = [packet.split(";")[7] for packet in packets_for_row(ROW)] + assert events == [ + "AltitudeSync", + "AirspeedSync", + "VerticalVelocitySync", + "HeadingSync", + "RollSync", + "PitchSync", + "YawSync", + "GForceSync", + ] + + +def test_packets_read_back_as_one_complete_sample(): + sample = data_logger.SampleBuffer() + for packet in packets_for_row(ROW): + sample.update(*data_logger.parse_packet(packet)) + assert sample.is_complete() + assert {k: sample.values[k] for k in ROW} == ROW + assert sample.values["roll"] == 0.0 + + +def flights(): + return pd.DataFrame( + { + "flight_id": ["A", "A", "B"], + "altitude": [1.0, 2.0, 3.0], + "target_altitude": [3000.0, 3000.0, 2000.0], + "target_heading": [90.0, 90.0, 90.0], + "target_vertical_speed": [0.0, 0.0, 0.0], + "target_airspeed": [100.0, 100.0, 100.0], + } + ) + + +def test_several_flights_need_one_picked(): + with pytest.raises(ValueError, match="--flight"): + select_flight(flights(), None) + assert select_flight(flights(), "B")["altitude"].tolist() == [3.0] + + +def test_an_unknown_flight_is_refused(): + with pytest.raises(ValueError, match="'C'"): + select_flight(flights(), "C") + + +def test_brief_holds_a_flights_constant_targets(): + brief = brief_for(select_flight(flights(), "A")) + assert brief.iloc[0][["altitude", "heading", "vertical_speed", "airspeed"]].tolist() == [ + 3000.0, + 90.0, + 0.0, + 100.0, + ] + + +def test_brief_is_refused_when_targets_change(): + with pytest.raises(ValueError, match="target_altitude"): + brief_for(flights()) + + +def test_a_flight_cannot_be_picked_from_a_csv_without_flight_ids(): + with pytest.raises(ValueError, match="flight_id"): + select_flight(pd.DataFrame({"altitude": [1.0, 2.0]}), "A") diff --git a/inference/tests/performance/test_clustering.py b/inference/tests/performance/test_clustering.py index 0481c79..67a50e6 100644 --- a/inference/tests/performance/test_clustering.py +++ b/inference/tests/performance/test_clustering.py @@ -2,7 +2,7 @@ import pandas as pd import pytest -from pilot_inference.performance.clustering import fit, sweep, truth_agreement +from pilot_inference.performance.clustering import fit, fit_reference, sweep, truth_agreement from pilot_inference.performance.features import METRICS, STATISTICS, compute_features from pilot_inference.performance.generate_synthetic_pilots import generate_pilots @@ -230,3 +230,46 @@ def test_pilot_off_target_in_opposite_directions_on_different_flights_is_not_on_ assert len(alternating) == 1 assert len(on_target) == 1 assert alternating != on_target + + +def synthetic_cohort(seed=0): + telemetry, truth = generate_pilots(n_pilots=15, seed=seed, flight_seconds=30) + _, pilots = compute_features(telemetry) + return pilots + + +def test_reference_groups_are_what_fit_returns(): + pilots = synthetic_cohort() + assignments, centres = fit(pilots, k=3) + reference = fit_reference(pilots, k=3) + pd.testing.assert_frame_equal(reference.assignments, assignments) + pd.testing.assert_frame_equal(reference.centres, centres) + + +def test_assigning_the_cohort_reproduces_its_groups(): + pilots = synthetic_cohort() + reference = fit_reference(pilots, k=3) + pd.testing.assert_frame_equal(reference.assign(pilots), reference.assignments) + + +def test_a_new_pilot_flying_like_a_member_joins_that_members_group(): + pilots = synthetic_cohort() + reference = fit_reference(pilots, k=3) + twin = pilots[pilots["pilot_id"] == "P15"].assign(pilot_id="live") + assert reference.assign(twin)["group"].iloc[0] == group_of(reference.assignments, "P15") + + +def test_assigning_does_not_move_the_groups(): + pilots = synthetic_cohort() + reference = fit_reference(pilots, k=3) + before = reference.kmeans.cluster_centers_.copy() + reference.assign(synthetic_cohort(seed=1)) + np.testing.assert_array_equal(reference.kmeans.cluster_centers_, before) + + +def test_assigning_a_pilot_with_nan_features_is_refused(): + pilots = synthetic_cohort() + reference = fit_reference(pilots, k=3) + gap = pilots.head(1).assign(pilot_id="live", altitude_std=np.nan) + with pytest.raises(ValueError, match="live"): + reference.assign(gap) diff --git a/inference/tests/performance/test_live.py b/inference/tests/performance/test_live.py new file mode 100644 index 0000000..0d439ff --- /dev/null +++ b/inference/tests/performance/test_live.py @@ -0,0 +1,242 @@ +import math +from datetime import datetime, timedelta, timezone + +import pytest + +from pilot_inference.performance.features import METRICS +from pilot_inference.performance.live import ( + MAX_SAMPLE_GAP_SECONDS, + Instruction, + LiveFlight, + load_brief, +) + +T0 = datetime(2026, 10, 26, 15, 0, tzinfo=timezone.utc) +HOLD = {"altitude": 3000.0, "heading": 270.0, "vertical_speed": 0.0, "airspeed": 100.0} + + +def at(seconds: float) -> datetime: + return T0 + timedelta(seconds=seconds) + + +def sample(**overrides) -> dict: + values = { + "altitude": 3010.0, + "heading": 268.0, + "vertical_speed": 20.0, + "velocity": 98.0, + "roll": 0.0, + "pitch": 0.0, + "yaw": 268.0, + "g_force": 1.0, + } + values.update(overrides) + return values + + +def two_holds(settle: float = 10.0) -> LiveFlight: + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", dict(HOLD, heading=360.0))] + return LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=settle)) + + +def write_brief(tmp_path, text: str): + path = tmp_path / "brief.csv" + path.write_text(text, encoding="utf-8") + return path + + +def test_brief_keeps_order_and_targets(tmp_path): + path = write_brief( + tmp_path, + "instruction,heading,altitude,airspeed,vertical_speed\n" + '"Fly heading 270, 3000 ft",270,3000,100,0\n' + "Turn 360,360,3000,100,0\n", + ) + brief = load_brief(path) + assert [i.text for i in brief] == ["Fly heading 270, 3000 ft", "Turn 360"] + assert brief[1].targets == dict(HOLD, heading=360.0) + + +def test_brief_missing_a_factor_is_refused(tmp_path): + path = write_brief(tmp_path, "instruction,heading,altitude,airspeed\nhold,270,3000,100\n") + with pytest.raises(ValueError, match="vertical_speed"): + load_brief(path) + + +def test_a_blank_target_is_left_to_carry_over(tmp_path): + path = write_brief( + tmp_path, + "instruction,heading,altitude,airspeed,vertical_speed\nhold,270,3000,100,0\nturn,360,,,\n", + ) + assert load_brief(path)[1].targets == {"heading": 360.0} + + +def test_the_first_instruction_must_set_every_target(tmp_path): + path = write_brief( + tmp_path, + "instruction,heading,altitude,airspeed,vertical_speed\nhold,270,,100,0\nturn,360,,,\n", + ) + with pytest.raises(ValueError, match="instruction 1 has no target for altitude"): + load_brief(path) + + +def test_nothing_is_scored_before_the_first_instruction(): + flight = two_holds() + row = flight.add_sample(sample(), at(0)) + assert all(math.isnan(row[f"target_{m}"]) for m in METRICS) + assert flight.pilot_features() is None + + +def test_targets_are_blank_while_settling_then_set(): + flight = two_holds(settle=10) + flight.instruct(at(0)) + settling = flight.add_sample(sample(), at(9.9)) + scored = flight.add_sample(sample(), at(10)) + assert math.isnan(settling["target_heading"]) + assert scored["target_heading"] == 270.0 + assert scored["target_altitude"] == 3000.0 + + +def test_an_exhausted_brief_keeps_the_last_instruction(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + flight.instruct(at(5)) + assert flight.instruct(at(10)) is None + assert flight.given == 2 + assert flight.targets_at(at(11))["heading"] == 360.0 + + +def test_scored_time_follows_the_samples(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + for i in range(41): + flight.add_sample(sample(), at(i * 0.05)) + assert flight.scored_seconds == pytest.approx(2.0) + + +def test_a_stall_counts_as_at_most_one_gap(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + flight.add_sample(sample(), at(0)) + flight.add_sample(sample(), at(30)) + assert flight.scored_seconds == pytest.approx(MAX_SAMPLE_GAP_SECONDS) + + +def test_settling_after_a_new_instruction_is_not_scored_time(): + flight = two_holds(settle=10) + flight.instruct(at(0)) + flight.add_sample(sample(), at(10)) + flight.add_sample(sample(), at(10.5)) + flight.instruct(at(11)) + flight.add_sample(sample(), at(15)) # heading settling, altitude still scored + flight.add_sample(sample(), at(21)) # first scored heading of the new hold + flight.add_sample(sample(), at(21.5)) + assert flight.scored_seconds_by_factor["heading"] == pytest.approx(1.0) + # Altitude never settled: 0.5 s, two stalls of at most a second each, then 0.5 s. + assert flight.scored_seconds_by_factor["altitude"] == pytest.approx(3.0) + # A group waits on the least-scored factor. + assert flight.scored_seconds == pytest.approx(1.0) + + +def test_features_score_each_sample_against_its_own_instruction(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + flight.add_sample(sample(heading=272.0), at(0)) + flight.add_sample(sample(heading=272.0), at(0.05)) + flight.instruct(at(1)) + # 2 deg past north is 2 deg off 360, not 358. + flight.add_sample(sample(heading=2.0), at(1)) + flight.add_sample(sample(heading=2.0), at(1.05)) + pilots = flight.pilot_features() + assert pilots["pilot_id"].tolist() == ["p1"] + assert pilots["heading_bias"].iloc[0] == pytest.approx(2.0) + assert pilots["heading_std"].iloc[0] == pytest.approx(0.0) + assert pilots["altitude_abs_bias"].iloc[0] == pytest.approx(10.0) + + +def test_a_sample_handled_after_an_instruction_keeps_the_one_it_arrived_under(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + flight.add_sample(sample(), at(4.95)) + flight.instruct(at(10)) + # Arrived at 5 s, before the turn to 360, but handled after it. + row = flight.add_sample(sample(), at(5)) + assert row["target_heading"] == 270.0 + assert flight.scored_seconds == pytest.approx(0.05) + + +def hold_then_turn(settle: float = 10.0) -> LiveFlight: + """A level hold on 270, then a turn to 360 that names only the heading.""" + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", {"heading": 360.0})] + return LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=settle)) + + +def test_targets_an_instruction_does_not_name_carry_over(): + flight = hold_then_turn(settle=0) + flight.instruct(at(0)) + flight.instruct(at(5)) + assert flight.targets_at(at(6)) == dict(HOLD, heading=360.0) + + +def test_only_the_targets_an_instruction_changes_settle(): + flight = hold_then_turn(settle=10) + flight.instruct(at(0)) + flight.instruct(at(20)) + during_turn = flight.targets_at(at(25)) + assert math.isnan(during_turn["heading"]) + assert during_turn["altitude"] == 3000.0 + assert flight.targets_at(at(30))["heading"] == 360.0 + + +def test_a_target_restated_at_the_same_value_does_not_settle(): + flight = two_holds(settle=10) + flight.instruct(at(0)) + # The turn to 360 restates 3000 ft, 100 kts and level flight. + flight.instruct(at(20)) + during_turn = flight.targets_at(at(25)) + assert math.isnan(during_turn["heading"]) + assert during_turn["altitude"] == 3000.0 + + +def test_settling_says_how_long_each_factor_has_left(): + flight = hold_then_turn(settle=10) + flight.instruct(at(0)) + # The turn comes before the first settle window ends: only heading restarts. + flight.instruct(at(4)) + left = flight.settling(at(4)) + assert left["heading"] == pytest.approx(10.0) + assert left["altitude"] == pytest.approx(6.0) + assert flight.settling(at(12)) == {"heading": pytest.approx(2.0)} + + +def test_north_restated_as_0_after_360_does_not_settle(): + brief = [Instruction("hold 360", dict(HOLD, heading=360.0)), Instruction("heading 0", {"heading": 0.0})] + flight = LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=10)) + flight.instruct(at(0)) + flight.instruct(at(20)) + assert flight.settling(at(21)) == {} + + +def test_an_infinite_target_is_refused(tmp_path): + path = write_brief( + tmp_path, + "instruction,heading,altitude,airspeed,vertical_speed\nhold,270,3000,100,0\nclimb,,inf,,\n", + ) + with pytest.raises(ValueError, match="instruction 2 .*altitude"): + load_brief(path) + + +def test_a_snapshot_does_not_move_as_the_flight_goes_on(): + flight = two_holds(settle=0) + flight.instruct(at(0)) + flight.add_sample(sample(altitude=3010.0), at(0)) + flight.add_sample(sample(altitude=3010.0), at(0.05)) + snapshot = flight.snapshot() + + flight.instruct(at(1)) + flight.add_sample(sample(altitude=3100.0), at(1)) + flight.add_sample(sample(altitude=3100.0), at(1.05)) + + assert snapshot.given == 1 + assert snapshot.pilot_features()["altitude_bias"].iloc[0] == pytest.approx(10.0) + assert snapshot.scored_seconds_by_factor["altitude"] == pytest.approx(0.05) diff --git a/inference/tests/performance/test_live_groups.py b/inference/tests/performance/test_live_groups.py new file mode 100644 index 0000000..2fcf14b --- /dev/null +++ b/inference/tests/performance/test_live_groups.py @@ -0,0 +1,274 @@ +"""The live sidecar end to end: relay packets in, samples with targets and a group out.""" + +import queue +import socket +import threading +from datetime import datetime, timedelta, timezone + +import pandas as pd +import pytest + +from pilot_inference.capture.replay import replay +from pilot_inference.performance.features import METRICS, compute_features +from pilot_inference.performance.live import LIVE_CSV_FIELDS, Instruction, LiveFlight +from pilot_inference.performance.live_groups import ( + RelayListener, + STOP, + CsvLog, + instruction_row, + instructions_path, + main, + record_samples, + reference_cohort, + status_line, +) + +T0 = datetime(2026, 10, 26, 15, 0, tzinfo=timezone.utc) +HOLD = {"altitude": 3000.0, "heading": 270.0, "vertical_speed": 0.0, "airspeed": 100.0} + + +def at(seconds: float) -> datetime: + return T0 + timedelta(seconds=seconds) + + +def wobbly_flight(n: int) -> pd.DataFrame: + """n samples at 20 Hz wobbling round HOLD, with a heading that crosses north.""" + i = pd.Series(range(n), dtype=float) + return pd.DataFrame( + { + "altitude": 3000.0 + 20.0 * (i % 7 - 3), + "heading": (270.0 + 3.0 * (i % 5 - 2)) % 360.0, + "vertical_speed": 50.0 * (i % 3 - 1), + "velocity": 100.0 + (i % 4 - 1.5), + } + ) + + +@pytest.fixture(scope="module") +def reference(): + groups, _ = reference_cohort(n_pilots=15, k=3, seed=0) + return groups + + +def listening_socket() -> socket.socket: + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen() + return server + + +def test_replayed_packets_arrive_as_samples(): + samples: queue.Queue = queue.Queue() + listener = RelayListener(listening_socket(), samples) + listener.start() + flight = wobbly_flight(25) + + replay(flight, "127.0.0.1", listener.port, hz=0) + + received = [samples.get(timeout=5) for _ in range(len(flight))] + listener.close() + assert [values["altitude"] for values, _ in received] == pytest.approx(flight["altitude"].tolist()) + assert [values["velocity"] for values, _ in received] == pytest.approx(flight["velocity"].tolist()) + # Columns the CSV lacks are sent as zero, so every sample is complete. + assert all(values["g_force"] == 0.0 for values, _ in received) + + +def test_saved_flight_gives_the_live_features_offline(tmp_path): + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", dict(HOLD, heading=360.0))] + flight = LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=2)) + log = CsvLog(tmp_path / "live.csv", LIVE_CSV_FIELDS) + samples: queue.Queue = queue.Queue() + recorder = threading.Thread( + target=record_samples, args=(samples, flight, threading.Lock(), log), daemon=True + ) + recorder.start() + + telemetry = wobbly_flight(200) + telemetry[["roll", "pitch", "yaw", "g_force"]] = 0.0 + flight.instruct(at(0)) + for i, values in enumerate(telemetry.to_dict("records")): + if i == 100: + flight.instruct(at(i * 0.05)) + samples.put((values, at(i * 0.05))) + samples.put(STOP) + recorder.join(timeout=5) + log.close() + + saved = pd.read_csv(tmp_path / "live.csv") + assert list(saved.columns) == LIVE_CSV_FIELDS + assert len(saved) == 200 + # Settling rows are saved with blank targets, so offline scoring skips them too. + assert saved["target_heading"].isna().sum() == 2 * 40 + _, offline = compute_features(saved) + live = flight.pilot_features() + columns = [f"{m}_{s}" for m in METRICS for s in ("mad", "std", "bias", "abs_bias")] + pd.testing.assert_frame_equal(offline[columns], live[columns], check_exact=False) + + +def test_status_waits_for_the_first_instruction(reference): + flight = LiveFlight([Instruction("hold", dict(HOLD))], "p1", "f1") + assert "Waiting for the first instruction" in status_line(flight, reference, 60, at(0)) + + +def test_status_says_settling_until_scoring_starts(reference): + flight = LiveFlight([Instruction("hold", dict(HOLD))], "p1", "f1", settle=timedelta(seconds=30)) + flight.instruct(at(0)) + line = status_line(flight, reference, 60, at(10)) + assert "instruction 1/1 (settling)" in line + assert "nothing scored yet" in line + + +def test_status_names_a_group_only_after_enough_scored_time(reference): + flight = LiveFlight([Instruction("hold", dict(HOLD))], "p1", "f1", settle=timedelta(0)) + flight.instruct(at(0)) + telemetry = wobbly_flight(2000) + telemetry[["roll", "pitch", "yaw", "g_force"]] = 0.0 + rows = telemetry.to_dict("records") + for i, values in enumerate(rows[:400]): + flight.add_sample(values, at(i * 0.05)) + assert "provisional" in status_line(flight, reference, 60, at(20)) + + for i, values in enumerate(rows[400:], start=400): + flight.add_sample(values, at(i * 0.05)) + line = status_line(flight, reference, 60, at(100)) + expected = reference.assign(flight.pilot_features())["group"].iloc[0] + assert f"group {expected} |" in line + + +def test_instructions_are_logged_with_their_targets(tmp_path): + flight = LiveFlight([Instruction("hold 270", dict(HOLD))], "p1", "f1") + instruction = flight.instruct(at(0)) + row = instruction_row(flight, instruction, at(0)) + assert row["number"] == 1 + assert row["instruction"] == "hold 270" + assert row["target_heading"] == 270.0 + assert instructions_path(tmp_path / "live.csv").name == "live_instructions.csv" + + +def test_a_busy_port_stops_before_anything_is_written(tmp_path): + # Held the way the capture logger holds it. + occupant = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + occupant.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + occupant.bind(("127.0.0.1", 0)) + occupant.listen() + port = occupant.getsockname()[1] + brief = tmp_path / "brief.csv" + brief.write_text("instruction,heading,altitude,airspeed,vertical_speed\nhold,270,3000,100,0\n") + csv_path = tmp_path / "live.csv" + + with pytest.raises(SystemExit) as stopped: + main(["--brief", str(brief), "--port", str(port), "--csv", str(csv_path)]) + occupant.close() + + assert f"Port {port}" in str(stopped.value) + assert "data_logger" in str(stopped.value) + assert not csv_path.exists() + assert not instructions_path(csv_path).exists() + + +def test_a_garbled_connection_does_not_stop_the_next_one(): + samples: queue.Queue = queue.Queue() + listener = RelayListener(listening_socket(), samples) + listener.start() + port = listener.port + + with socket.create_connection(("127.0.0.1", port)) as garbled: + garbled.sendall(b"\xff\xfe not utf-8\r\n") + replay(wobbly_flight(3), "127.0.0.1", port, hz=0) + + received = [samples.get(timeout=5) for _ in range(3)] + listener.close() + assert len(received) == 3 + + +def test_status_names_the_factors_still_settling(reference): + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", {"heading": 360.0})] + flight = LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=10)) + flight.instruct(at(0)) + flight.instruct(at(20)) + line = status_line(flight, reference, 60, at(25)) + assert "instruction 2/2 (settling heading)" in line + + +def test_a_logged_instruction_carries_its_unchanged_targets(tmp_path): + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", {"heading": 360.0})] + flight = LiveFlight(brief, "p1", "f1") + flight.instruct(at(0)) + turn = flight.instruct(at(20)) + row = instruction_row(flight, turn, at(20)) + assert row["target_heading"] == 360.0 + assert row["target_altitude"] == 3000.0 + + +def log_flight(log: CsvLog, flight: LiveFlight, telemetry: pd.DataFrame) -> None: + flight.instruct(at(0)) + for i, values in enumerate(telemetry.to_dict("records")): + log.write(flight.add_sample(values, at(i * 0.05))) + + +def test_an_appended_csv_still_gives_each_flight_its_live_features(tmp_path): + telemetry = wobbly_flight(120) + telemetry[["roll", "pitch", "yaw", "g_force"]] = 0.0 + log = CsvLog(tmp_path / "live.csv", LIVE_CSV_FIELDS) + earlier = LiveFlight([Instruction("hold", dict(HOLD, altitude=2000.0))], "p1", "f1", timedelta(0)) + log_flight(log, earlier, telemetry) + later = LiveFlight([Instruction("hold", dict(HOLD))], "p1", "f2", timedelta(0)) + log_flight(log, later, telemetry) + log.close() + + offline, _ = compute_features(pd.read_csv(tmp_path / "live.csv")) + columns = [f"{m}_{s}" for m in METRICS for s in ("mad", "std", "bias", "abs_bias")] + # The pilot's offline features average both flights; the flight's own row is the live one. + pd.testing.assert_frame_equal( + offline.loc[offline["flight_id"] == "f2", columns].reset_index(drop=True), + later.pilot_features()[columns], + check_exact=False, + ) + + +def test_a_group_stays_named_while_one_factor_settles(reference): + brief = [Instruction("hold 270", dict(HOLD)), Instruction("turn 360", {"heading": 360.0})] + flight = LiveFlight(brief, "p1", "f1", settle=timedelta(seconds=30)) + flight.instruct(at(0)) + # 100 s of flying, 70 s of it scored once the first settle window ends. + telemetry = wobbly_flight(2000) + telemetry[["roll", "pitch", "yaw", "g_force"]] = 0.0 + for i, values in enumerate(telemetry.to_dict("records")): + flight.add_sample(values, at(i * 0.05)) + flight.instruct(at(100)) + line = status_line(flight, reference, 60, at(105)) + assert "(settling heading)" in line + assert "provisional" not in line + assert "group " in line + + +def test_closing_the_listener_stops_a_connected_relay(): + samples: queue.Queue = queue.Queue() + listener = RelayListener(listening_socket(), samples) + listener.start() + + def stream_until_cut_off(): + try: + replay(wobbly_flight(100_000), "127.0.0.1", listener.port, 200) + except ConnectionError: + pass # what the relay sees when the sidecar closes + + relay = threading.Thread(target=stream_until_cut_off, daemon=True) + relay.start() + samples.get(timeout=5) # the relay is connected and streaming + + listener.close() + queued = samples.qsize() + relay.join(timeout=5) + # Nothing arrives once close() returns, even though the relay was still sending. + assert samples.qsize() == queued + assert not relay.is_alive() + + +def test_closing_the_listener_with_no_relay_returns(): + listener = RelayListener(listening_socket(), queue.Queue()) + listener.start() + closer = threading.Thread(target=listener.close, daemon=True) + closer.start() + closer.join(timeout=5) + assert not closer.is_alive()