#!/usr/bin/env python3 """ Three Flights agent — always connected. Starts with Windows, waits for Microsoft Flight Simulator, records every flight and sends each one to your Three Flights account on its own. No codes per flight. One-time setup (Command Prompt, in the folder where this file lives): pip install SimConnect requests python three_flights_agent.py --link ABC123 (code from Profile → Your sim PC → Link this PC) python three_flights_agent.py --install (starts with Windows from now on) Run by hand: python three_flights_agent.py Nothing about your account is stored on this PC — only a device token you can revoke from the site. Copyright © 2026 UK Centre of Excellence Ltd. This source is published so anyone can read exactly what runs on their PC before running it. That is a promise of transparency, not a licence: all rights reserved, and it may not be reused, modified or redistributed outside of running Three Flights. See https://three-flights.vercel.app/terms. """ import json import os import subprocess import sys import time from datetime import datetime, timezone try: import requests except ImportError: # pragma: no cover print("Missing dependency: run pip install SimConnect requests") sys.exit(1) API = "https://three-flights.vercel.app/api/agent" AGENT_VERSION = "agent-0.31.0" # When PyInstaller freezes this into ThreeFlights.exe, __file__ points inside a temporary # extraction folder that Windows deletes on exit — a device token written there would be lost # every time. Frozen builds must sit beside the .exe instead. FROZEN = getattr(sys, "frozen", False) HERE = os.path.dirname(sys.executable) if FROZEN else os.path.dirname(os.path.abspath(__file__)) TOKEN_FILE = os.path.join(HERE, "device.json") SIM_PROCESSES = ("FlightSimulator.exe", "FlightSimulator2024.exe") SAMPLE_SECONDS = 1.0 # Near the ground we sample twenty times a second. At 1 Hz the last airborne reading could be most of a # second before the wheels touched, and in the last second of a flare the vertical speed is still decaying # fast — so the "measured" touchdown was whichever value the tick happened to land on. A real landing read # -185 fpm on the sim's own report and -159 here, and the error was not even consistent: the same landing # could have read -120 or -300 depending on the timing of a tick. That is not a measurement. FAST_SAMPLE_SECONDS = 0.05 FAST_BELOW_FT = 200 FT_PER_SEC_TO_KT = 0.5924838 # DESIGN SPEED VS0 and friends are published in feet per second KT_TO_FT_PER_SEC = 1.6878099 # knots to feet per second, for how fast the ground can move underneath FLARE_ROLLOUT_SECONDS = 4 # keep sampling fast this long after touchdown, for the rollout FLARE_MAX_SAMPLES = 1600 # ~80 s of 20 Hz — bounds memory and the payload TRACK_EVERY_SECONDS = 10 ROUTE_EVERY_SECONDS = 60 # coarse whole-flight route ROUTE_MAX_POINTS = 500 # thinned as the flight grows, so a 14-hour leg still fits LOG_FILE = os.path.join(HERE, "agent.log") LOG_MAX_BYTES = 2_000_000 # about a fortnight of flying; one old copy is kept beside it def log(msg): """Say it on the console and keep it on disk. Shannon flew Heathrow to Athens with the top-of-descent pause set and it did not fire. The one record of what the recorder had actually been told was the console window, and by the morning the agent had restarted and wiped it. A recorder that cannot say afterwards what it did is not much of a recorder, so every line now also goes to agent.log beside this program.""" line = f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {msg}" print(line[11:], flush=True) try: if os.path.exists(LOG_FILE) and os.path.getsize(LOG_FILE) > LOG_MAX_BYTES: old = LOG_FILE + ".1" if os.path.exists(old): os.remove(old) os.replace(LOG_FILE, old) with open(LOG_FILE, "a", encoding="utf-8") as fh: fh.write(line + "\n") except Exception: pass # a log that breaks the flight is worse than no log def post(kind, **fields): try: r = requests.post(API, json={"kind": kind, **fields}, timeout=20) return r.json() except Exception as exc: return {"ok": False, "error": str(exc)} def load_token(): try: with open(TOKEN_FILE, "r", encoding="utf-8") as f: return json.load(f).get("token") except Exception: return None def save_token(token): with open(TOKEN_FILE, "w", encoding="utf-8") as f: json.dump({"token": token, "linkedAt": datetime.now(timezone.utc).isoformat()}, f) def link(code): r = post("claim", code=code.strip().upper()) if not r.get("ok"): log(f"Link failed: {r.get('error')}. Get a fresh code from Profile → Your sim PC.") if FROZEN: try: input("\nPress Enter to close this window.") except Exception: pass sys.exit(1) save_token(r["token"]) if FROZEN: log("Linked. This PC now records your flights automatically.") else: log("Linked. This PC now records your flights automatically. Run with --install to start with Windows.") def install(): """Put a shortcut in the Startup folder so the agent runs (hidden) when Windows starts.""" startup = os.path.join(os.environ.get("APPDATA", ""), r"Microsoft\Windows\Start Menu\Programs\Startup") if not os.path.isdir(startup): log("Could not find the Windows Startup folder; run the agent by hand instead.") sys.exit(1) lnk = os.path.join(startup, "Three Flights agent.lnk") if FROZEN: target, arguments = sys.executable, "" else: pythonw = os.path.join(os.path.dirname(sys.executable), "pythonw.exe") if not os.path.exists(pythonw): pythonw = sys.executable target, arguments = pythonw, '"%s"' % os.path.abspath(__file__) ps = ( "$s=(New-Object -ComObject WScript.Shell).CreateShortcut('%s');" "$s.TargetPath='%s';$s.Arguments='%s';$s.WorkingDirectory='%s';$s.WindowStyle=7;$s.Save()" % (lnk.replace("'", "''"), target, arguments, HERE) ) subprocess.run(["powershell", "-NoProfile", "-Command", ps], check=False) log(f"Installed: {lnk}. The agent will start with Windows and sit quietly until MSFS runs.") def uninstall(): lnk = os.path.join(os.environ.get("APPDATA", ""), r"Microsoft\Windows\Start Menu\Programs\Startup", "Three Flights agent.lnk") if os.path.exists(lnk): os.remove(lnk) log("Removed from Windows startup.") else: log("Nothing to remove.") _sim_seen_at = 0.0 _sim_last_answer = True SIM_CHECK_SECONDS = 10 def sim_running(force=False): """Is MSFS still there? Answered from a ten-second-old cache unless `force`. This used to spawn `tasklist` on every pass of the recording loop — a whole Windows process, hundreds of milliseconds, four times a second by design and in practice the thing that set the loop's speed. The flare was meant to be sampled twenty times a second and was in fact sampled about once, on every aircraft, because of this one line. A simulator that closes stays closed for ten seconds; nothing is lost by asking less often.""" global _sim_seen_at, _sim_last_answer now = time.time() if not force and now - _sim_seen_at < SIM_CHECK_SECONDS: return _sim_last_answer try: out = subprocess.run(["tasklist", "/FO", "CSV", "/NH"], capture_output=True, text=True, timeout=10).stdout _sim_last_answer = any(p.lower() in out.lower() for p in SIM_PROCESSES) except Exception: _sim_last_answer = False _sim_seen_at = now return _sim_last_answer # Variables python-SimConnect has never heard of. # # Its catalogue is a fixed list, and `aq.get()` returns None for anything outside it — which is why # the mixture and propeller levers were unreachable. But its Request class takes a raw (name, unit) # pair straight through to SimConnect, so anything the simulator publishes can be registered by hand. # Registered lazily, once, and only asked for after that. _EXTRA_REQUESTS = {} # The fuel cutoff switches, read directly. See the shutdown logic for why. _FUEL_VALVES = [f"GENERAL_ENG_FUEL_VALVE:{n}" for n in (1, 2, 3, 4)] def fuel_cut(aq): """True when every engine that answers reports its fuel valve shut; None when none answer. "Engines off" is measured from the engines actually having stopped turning, and on a big turbofan that is thirty to sixty seconds after the pilot has finished — Shannon moved the A350's switches and then sat watching a spinning-down N1 decide when her flight was over. The switch itself is a better answer to "has the pilot shut down", because it is the thing the pilot did. It is used only to shorten the wait, never to end a flight: it is read after the wheels are down and the aeroplane is stationary, so an aeroplane that reports its valves oddly in the air cannot cut a flight short. If nothing answers, nothing changes.""" answered = False for name in _FUEL_VALVES: v = read(aq, name, None) if v is None: continue answered = True try: if float(v) > 0.5: return False # at least one engine is still being fed except (TypeError, ValueError): return None return True if answered else None _EXTRA_SPEC = { # name simvar unit "MIXTURE_LEVER:1": (b"GENERAL ENG MIXTURE LEVER POSITION:1", b"Percent"), "MIXTURE_LEVER:2": (b"GENERAL ENG MIXTURE LEVER POSITION:2", b"Percent"), "PROP_LEVER:1": (b"GENERAL ENG PROPELLER LEVER POSITION:1", b"Percent"), "PROP_LEVER:2": (b"GENERAL ENG PROPELLER LEVER POSITION:2", b"Percent"), # Carburettor heat. MSFS drives it from the engine anti-ice position on piston types, not from # anything with "carb" in the name — which is why looking for one never found it. "CARB_HEAT:1": (b"GENERAL ENG ANTI ICE POSITION:1", b"Bool"), "CARB_HEAT:2": (b"GENERAL ENG ANTI ICE POSITION:2", b"Bool"), } def read_extra(sm, name): """One of the hand-registered variables, or None when this aeroplane does not model it. None is a real and useful answer: an aeroplane with no mixture lever is not a pilot who forgot to set one, and the review must never mark them for a control they have not got.""" if sm is None: return None try: req = _EXTRA_REQUESTS.get(name) if req is None: from SimConnect.RequestList import Request spec = _EXTRA_SPEC.get(name) if spec is None: return None req = Request(spec, sm, _time=0) _EXTRA_REQUESTS[name] = req return req.value except Exception: return None def reset_extra_requests(): _EXTRA_REQUESTS.clear() def speed_up_reads(sm): """Stop waiting ten milliseconds at a time for every single variable. python-SimConnect asks the simulator for one variable per round trip and then waits for the answer like this: while _Request.outData is None and attemps < _Request.attemps: time.sleep(.01) Ten milliseconds is an age. The nine variables the flare needs therefore cost at least ninety milliseconds of pure sleeping per sample, and the first real circuit flown on 0.19.0 came back with a fastest gap of exactly 0.200 s — never once quicker, on any sample, which is the shape of a fixed floor rather than a slow simulator. The answer usually arrives in well under a millisecond. So poll at one millisecond instead of ten, and raise the attempt count to keep exactly the same patience in wall-clock terms — a hundred tries of 1 ms is the same 100 ms of waiting the library always allowed, just checked more often. Nothing is given up: a variable that is genuinely slow still gets its full hundred milliseconds. If a future version of the library changes shape underneath this, the patch quietly does nothing and the recorder runs at the old speed rather than breaking. """ try: import time as _t def get_data(_request): sm.request_data(_request) waited = 0.0 while _request.outData is None and waited < 0.1: _t.sleep(0.001) waited += 0.001 return _request.outData is not None sm.get_data = get_data log("Fast variable reads enabled.") except Exception: pass # old speed is not broken, only slow def read(aq, name, default=None, optional=False): """One simulator variable, or the default. `optional` is accepted and ignored: it used to mean "give up on this variable after five silences", which was a mistake serious enough to be worth remembering. While MSFS loads, every variable is silent — so the engine variables were all blacklisted within the first five samples of every session, permanently, before the aeroplane had even appeared. engines_state() then never got an answer, "engines off" could never be detected, and the flight only sent when the simulator was closed. Discovery, below, is the right shape.""" try: v = aq.get(name) except Exception: return default return default if v is None else v # The engine variables this aeroplane actually answers, discovered once and then reused. # # Aircraft disagree about which of these exist: MSFS 2024 rejects ENG N1 RPM outright, some # study-level add-ons drive TURB ENG N1 and not ENG COMBUSTION, and asking all sixteen every second # is what filled ten hours of log with SIMCONNECT_EXCEPTION_UNRECOGNIZED_ID. So we probe until # something answers, remember exactly which ones did, and ask only those from then on. Nothing is # ever permanently given up on: a fresh flight starts a fresh discovery. _ENGINE_CANDIDATES = [(f"{name}:{n}", threshold) for n in (1, 2, 3, 4) # ENG_N1_RPM was here and is gone: MSFS rejects the name outright on every # engine — four guaranteed rejections per probe, for nothing that # GENERAL_ENG_RPM does not already answer. for name, threshold in (("ENG_COMBUSTION", 0.5), ("TURB_ENG_N1", 5.0), ("GENERAL_ENG_RPM", 50.0))] _engine_working = None def reset_engine_discovery(): """Called at the start of each flight: the next aeroplane may not be this one.""" global _engine_working _engine_working = None def engines_state(aq): """True, False, or None when the simulator will not say. Never guess "off": an unknown answer means the flight is not over, because ending a recording early loses it.""" global _engine_working probe = _engine_working if _engine_working else _ENGINE_CANDIDATES answered = [] running = False for name, threshold in probe: v = read(aq, name, None) if v is None: continue answered.append((name, threshold)) try: if float(v) > threshold: running = True except (TypeError, ValueError): continue if _engine_working is None and answered: _engine_working = answered # lock to what this aeroplane actually reports if not answered: return None return running def nearest_icao(lat, lon): """Best-effort airport lookup for the logbook via the site's own database endpoint; empty when offline.""" try: r = requests.get("https://three-flights.vercel.app/api/nearest", params={"lat": lat, "lon": lon}, timeout=8) return (r.json() or {}).get("icao", "") or "" except Exception: return "" def trim_flare(buf, touchdown_t): """The landing, kept instead of thrown away. The recorder has always watched the last 200 ft twenty times a second — and used exactly one of those samples (the touchdown) before discarding the rest. This packages them: time is re-zeroed on the touchdown so the replay can count down to it, and everything is rounded to what a replay can actually show. ~1,200 rows of eight small numbers; about the weight of a single photo thumbnail. Groundspeed rides along with airspeed because on a windy approach they are different stories: 144 kt indicated into a strong headwind is far less across the ground, and it is the groundspeed the pilot actually felt in the flare.""" if touchdown_t is None or not buf: return None rows = [] for t, agl, vs, ias, gs, lat, lon, bank in buf[-FLARE_MAX_SAMPLES:]: rows.append([round(t - touchdown_t, 2), round(agl), round(vs), round(ias), round(gs), round(lat, 5), round(lon, 5), round(bank, 1)]) return rows or None def procedure_snapshot(aq, sm): """Everything the procedure review reads, at one moment, as it is. Deliberately raw. The recorder's job is to write down what the switches and levers were doing; deciding whether that was right for this aeroplane is the site's job, where the rules can change without anyone rebuilding an .exe. A value that comes back None means the simulator did not answer for this aircraft, and the review will simply not mention it — silence, never a failure. """ def light(name): v = read(aq, name, None) return None if v is None else bool(v) def pct(v): return None if v is None else round(float(v)) return { "navLight": light("LIGHT_NAV"), "beaconLight": light("LIGHT_BEACON"), "taxiLight": light("LIGHT_TAXI"), "landingLight": light("LIGHT_LANDING"), "strobeLight": light("LIGHT_STROBE"), "parkingBrake": light("BRAKE_PARKING_POSITION"), "spoilersArmed": light("SPOILERS_ARMED"), "autopilot": light("AUTOPILOT_MASTER"), "stallWarning": light("STALL_WARNING"), "overspeed": light("OVERSPEED_WARNING"), "mixturePct": pct(read_extra(sm, "MIXTURE_LEVER:1")), "propPct": pct(read_extra(sm, "PROP_LEVER:1")), "pitchDeg": (lambda v: None if v is None else round(__import__("math").degrees(float(v)), 1))(read(aq, "PLANE_PITCH_DEGREES", None)), "flapsPct": pct(read(aq, "TRAILING_EDGE_FLAPS_LEFT_PERCENT", None)), # Carburettor heat, and whether this aeroplane even has one. A fuel-injected engine has no # carburettor to ice up, so the review must not ask about a lever that does not exist — # hence recording the availability alongside the position rather than inferring it. "carbHeatAvailable": light("CARB_HEAT_AVAILABLE"), "carbHeat": (lambda v: None if v is None else bool(v))(read_extra(sm, "CARB_HEAT:1")), # The aeroplane's own stall speed in the landing configuration, so approach speed can be # judged against what this type actually needs rather than against a number I picked. # # DESIGN SPEED VS0 is published in FEET PER SECOND. Reading it as knots told Shannon her # Bonanza stalls at 91 kt and wanted 118 on final — it stalls at 54 and wants about 70, and # the 85 kt she actually flew was fifteen knots fast rather than thirty knots slow. Confidently # wrong flying advice is the worst thing this product can produce, so the unit is converted # here, once, and named in the key. "vs0Kt": (lambda v: None if v is None else round(float(v) * FT_PER_SEC_TO_KT))(read(aq, "DESIGN_SPEED_VS0", None)), "iasKt": pct(read(aq, "AIRSPEED_INDICATED", None)), } def near_the_ground(airborne, on_ground, agl, already_down): """Is this the part of the flight worth measuring twenty times a second? Airborne, below FAST_BELOW_FT, wheels not down yet. Everything above that is cruise, where one sample a second is plenty and 20 Hz for two and a half hours would be waste. Once the wheels are down the touchdown is already captured, so we drop back to the slow rate for the taxi in.""" if not airborne or on_ground or already_down: return False if agl is None: return False return agl <= FAST_BELOW_FT def is_touchdown(on_ground, prev_on_ground, agl, ias): """A real touchdown happens low and slow, on the transition from airborne to on-ground. Anything else claiming to be a landing is a SimConnect glitch. This rule exists because one did: a phantom -1031 fpm 'landing' logged at top of descent when the simulator stalled during a pause and the ground flag was guessed rather than read.""" if not on_ground or prev_on_ground: return False if agl is None or ias is None: return False return agl < 150 and ias < 200 def is_takeoff(airborne, on_ground, agl, ias): """Wheels up is: not already airborne, off the ground, above it, AND moving. The speed test is the one a loading simulator cannot fake. While MSFS loads, SimConnect can report not-on-ground with a stale altitude, and a stationary aeroplane at the gate was being logged as a departure because of it — which showed on the site as "paused, still recording" before the pilot had moved, and inflated the airborne time by however long the load took. Same lesson as the phantom touchdown: never trust one reading from a simulator that is busy.""" if airborne or on_ground: return False if agl is None or ias is None: return False return agl > 50 and ias > 30 def sane_agl(agl, prev_agl, vs_fpm, seconds, gs_kt=0.0): """Is this height above the ground a real one, or is it the terrain loader lagging? PLANE ALT ABOVE GROUND is the aeroplane's altitude minus the elevation of the terrain tile underneath it. When that tile has not streamed in yet the elevation is wrong, and the height goes wrong with it — by thousands of feet, for as long as the tile takes to arrive. This is not theoretical. An A380 descending into Heathrow reported minus nine hundred and thirty-three feet over the Channel, at 284 knots, forty miles out. Minus nine hundred is below a thousand and below five hundred, so both approach gates fired on that one sample: the review then said the approach was flown at 284 kt with the gear up and the spoilers unarmed, when the gear had in fact come down at 3,232 ft and the real touchdown was 133 kt. One bad number, four wrong statements, and every one of them accused the pilot of something they had not done. A real height changes by what the aeroplane is doing vertically, plus what the ground is doing underneath it. The vertical part is known — that is the vertical speed. The ground part was first allowed a flat 150 ft a second, and that number was too small, because it was guessed. Descending into Bogota — a city at 8,300 ft in the Andes — five perfectly real heights were thrown away as glitches: 8,228 ft after 8,602, 1,562 after 2,115. Mountains really do climb that fast underneath an aeroplane. So the ground is now allowed to rise or fall as fast as the aeroplane is travelling over it: at 250 knots that is 422 ft a second, a slope of one in one, a cliff face. Nothing real is steeper, and it is measured rather than chosen. The A380's minus nine hundred feet — eleven thousand feet of change in one sample — is still rejected by a factor of twenty.""" if agl is None: return False try: agl = float(agl) except (TypeError, ValueError): return False if agl < -20: # a parked A350 reads +16; nothing real reads below the ground return False if prev_agl is None or seconds <= 0: return True try: moved = abs(float(vs_fpm)) / 60.0 * seconds except (TypeError, ValueError): moved = 0.0 try: terrain = abs(float(gs_kt)) * KT_TO_FT_PER_SEC * seconds # a one-in-one slope, the whole way except (TypeError, ValueError): terrain = 0.0 return abs(agl - prev_agl) <= moved + terrain + 100.0 def crossed_down_through(level, agl, prev_agl, ceiling): """Did the aeroplane fly down through this height, rather than appear below it? A gate is a moment on an approach, so it has to be reached from just above. Requiring the previous good sample to be above the level and below a sane ceiling is what stops a single wrong height a long way out from being mistaken for short final — and it also makes the thousand-foot gate and the five-hundred-foot gate impossible to capture in the same pass, which is what produced two identical readings of 284 kt and -2,177 fpm. The five-hundred-foot gate is given a ceiling of a thousand, so its crossing window and the thousand-foot gate's cannot overlap at all: whatever the sampling does, the two gates are different moments.""" if prev_agl is None: return False return agl <= level < prev_agl <= ceiling def nm_between(lat1, lon1, lat2, lon2): """Great-circle distance in nautical miles.""" import math p1, p2 = math.radians(lat1), math.radians(lat2) dlat = p2 - p1 dlon = math.radians(lon2 - lon1) h = math.sin(dlat / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlon / 2) ** 2 return 2 * 3440.065 * math.asin(min(1.0, math.sqrt(h))) def descent_distance_nm(alt_ft, field_elev_ft): """How far out the top of descent is, by the rule every pilot uses: three miles per thousand feet to lose, and ten more to slow down. FL350 into a sea-level field is a hundred and five plus ten. Not the FMS's number — that depends on the wind and the aeroplane and we cannot see it — but the right order of magnitude, and the pause sits a further margin before it.""" return max(0.0, (float(alt_ft) - float(field_elev_ft)) / 1000.0 * 3.0) + 10.0 def tod_pause_due(intent, lat, lon, alt_ft, vs_fpm, on_ground, airborne, already): """Should the simulator be paused now, ahead of the top of descent? Shannon flies the long ones overnight — Heathrow to San Francisco with a pause at the top of descent, so she can sleep through the cruise and wake up to fly the landing. And on a shorter flight people need to eat. So: once per flight, when the pilot has said where they are going, the aeroplane is airborne and still in the cruise (not already descending), and the distance to go has come down to the estimated descent distance plus the pilot's chosen margin.""" if already or not airborne or on_ground or not intent: return False if intent.get("lat") is None or intent.get("lon") is None: return False if float(alt_ft) < 10000: # a low cruise is not a flight anyone sleeps through return False if float(vs_fpm) < -500: # already going down: too late, and do not fight the pilot return False to_go = nm_between(lat, lon, float(intent["lat"]), float(intent["lon"])) trigger = descent_distance_nm(alt_ft, intent.get("elevFt") or 0) + float(intent.get("pauseNm") or 30) return to_go <= trigger def intent_sentence(intent): """One plain line saying what the recorder has been told about the pause, and why it will or will not fire. Said whenever the answer changes, so the log shows the truth at every moment of the flight rather than a single claim made at take-off.""" if not intent: return ("TOD pause OFF: no destination has been set on the site for this flight, or it was set " "more than a day ago. Pick the flight on the briefing page before you depart to arm it.") dest = intent.get("destination") nm = intent.get("pauseNm") or 0 if intent.get("lat") is None: return f"TOD pause: {dest} is not an airfield this recorder knows, so it cannot work out the descent. The sim will not pause." if nm <= 0: return f"TOD pause OFF for {dest} — the switch on the briefing is off. The sim will not pause." return f"TOD pause ARMED: the sim will pause {nm} nm before the top of descent for {dest}." def intent_key(intent): if not intent: return None return (intent.get("destination"), intent.get("pauseNm") or 0, intent.get("lat") is not None) _tod_said_key = False # the last thing the log said about the pause; False means nothing yet def report_intent(reply): """Say what the recorder has been told about the pause, whenever that answer changes — waiting on the ramp, sitting in the cruise, or with the simulator not even open. Returns the intent.""" global _tod_said_key new_intent = reply.get("intent") if isinstance(reply, dict) and reply.get("ok") else None if intent_key(new_intent) != _tod_said_key: _tod_said_key = intent_key(new_intent) log(intent_sentence(new_intent)) return new_intent def pause_sim(sm): """Ask the simulator to pause, through the same SimConnect link the readings come over. Returns True when the event was accepted. The pilot resumes from inside the sim, as usual.""" try: from SimConnect import AircraftEvents AircraftEvents(sm).find("PAUSE_ON")() return True except Exception as exc: log(f"Could not pause the simulator: {exc}") return False def sane_position(lat, lon, prev, seconds, max_kt=800): """Reject what a loading simulator reports: null island, the poles, and teleports. A real aeroplane cannot move faster than max_kt between two samples.""" try: lat = float(lat); lon = float(lon) except (TypeError, ValueError): return False if abs(lat) < 0.02 and abs(lon) < 0.02: return False if abs(lat) > 89.5: return False if prev is not None and seconds > 0: import math dlat = math.radians(lat - prev[0]); dlon = math.radians(lon - prev[1]) h = math.sin(dlat / 2) ** 2 + math.cos(math.radians(prev[0])) * math.cos(math.radians(lat)) * math.sin(dlon / 2) ** 2 nm = 2 * 3440.065 * math.asin(min(1.0, math.sqrt(h))) if nm > (max_kt * seconds / 3600.0) + 5: return False return True def record_one_flight(token, aq, sm=None): """Wait for wheels-up, record to shutdown, return the summary (or None if the sim went away).""" title = read(aq, "TITLE", b"") title = title.decode("utf-8", "ignore") if isinstance(title, bytes) else str(title or "") post("heartbeat", token=token, state="waiting", aircraft=title) log(f"Sim connected · {title or 'aircraft'} · waiting for take-off.") airborne = False joined_in_flight = False # we connected after the take-off: say so rather than invent an origin departed_at = landed_at = None origin = destination = "" max_alt = 0.0 track = [] last_track = 0.0 route = [] # the whole flight, coarse last_route = 0.0 route_every = ROUTE_EVERY_SECONDS last_heartbeat = time.time() prev_on_ground = True ground_events = go_arounds = 0 was_below_500 = False approach = [] gate1000 = gate500 = touchdown = last_sample = None prev_agl = None # the last height we believed prev_agl_t = None agl_rejected = 0 # consecutive heights thrown away stopped_since = None # not moving at all, as opposed to taxiing slowly engines_logged = False # the engine-variable diagnosis, said once fuel_off_since = None # the switches, which the pilot moved themselves intent = None # where the pilot said they are going, from the site tod_paused = False # the one pause before descent, done last_tod_report = 0.0 # a progress line every ten minutes while armed ground_stable_since = None engines_off_since = None gear_retractable = bool(read(aq, "IS_GEAR_RETRACTABLE", 1)) gear_up_agl = gear_down_agl = None prev_gear_down = True sim_misses = 0 last_origin_try = 0.0 last_fix = None # Pause awareness: simmers pause for hours (dinner, sleep, a phone call). Paused time never counts # as flying, never adds track points, and never times out a recording. A pause is detected when the # sim clock stops advancing or the aircraft is frozen in the air (identical position, altitude and speed). paused = False paused_since = None paused_total = 0.0 prev_sim_time = None prev_pos = None frozen_samples = 0 paused_at_departure = paused_at_landing = 0.0 settle = 0 sim_time_departed = sim_time_landed = None # the simulator's own clock: it stops when the sim does g_watch = 0 # keep watching g for a few seconds after touchdown fast = False # sampling twenty times a second, close to the ground flare_buf = [] # the 20 Hz samples of the landing itself reset_engine_discovery() # this flight's aeroplane, not the last one's engines_reported = False # say once, in the log, that shutdown can be seen timing_reported = False # the cost of a sample, measured rather than assumed reset_extra_requests() # the hand-registered variables belong to this session proc = {} # what the switches and levers were doing, by moment moved_off = False slew_used = False stall_seen = False spoilers_ever_armed = False # did SPOILERS ARMED ever read true on this aeroplane? mixture_ever_set = False # did a mixture lever ever read above zero? see below prop_ever_set = False # the same question for the propeller lever stall_lowest_agl = None # the lowest it sounded at stall_highest_agl = None # and the highest — the two tell very different stories overspeed_seen = False max_pitch = 0.0 last_slow = 0.0 # when the once-a-second variables were last read # Carried between fast samples: read at the slow rate, used at every rate. alt = alt_ind = flaps = wind = wind_dir = heading = sim_time = 0.0 gear = True engines_running = True took_off_clear = False # the climb-out has been dropped from the flare buffer touchdown_t = None # flight_t at the moment the wheels touched last_pause_check = 0.0 # pause detection stays on its one-second cadence while True: t = time.time() try: og_raw = read(aq, "SIM_ON_GROUND", None) if og_raw is None: raise ValueError("no ground state") # counts as a missed sample below; never guess on-ground on_ground = bool(og_raw) # --- the fast nine ---------------------------------------------------------- # Every SimConnect read is a blocking round trip, and there are two dozen of them below. # Read all of them twenty times a second and the loop takes about a second per pass, which # is exactly what was happening: four aircraft, four flights, an average gap of 0.8 to 1.2 # seconds through a "20 Hz" flare. These nine are the ones that change inside a flare and # are the only ones the replay and the touchdown measurement use. agl = float(read(aq, "PLANE_ALT_ABOVE_GROUND", 0.0) or 0.0) ias = float(read(aq, "AIRSPEED_INDICATED", 0.0) or 0.0) vs = float(read(aq, "VERTICAL_SPEED", 0.0) or 0.0) bank = abs(__import__("math").degrees(float(read(aq, "PLANE_BANK_DEGREES", 0.0) or 0.0))) g = float(read(aq, "G_FORCE", 1.0) or 1.0) lat = float(read(aq, "PLANE_LATITUDE", 0.0) or 0.0) lon = float(read(aq, "PLANE_LONGITUDE", 0.0) or 0.0) gs = float(read(aq, "GROUND_VELOCITY", 0.0) or 0.0) # Is that height believable? A rejected one is not used for anything that ends up in the # review — no gate, no flare sample, no gear height. If the simulator keeps insisting on # it for five seconds then the terrain has genuinely changed and we take the new number # as the datum, but that first re-synced sample still triggers nothing. agl_ok = sane_agl(agl, prev_agl, vs, (t - prev_agl_t) if prev_agl_t else 0, gs) if agl_ok: agl_rejected = 0 else: agl_rejected += 1 if agl_rejected == 1: log(f"Height above ground reported as {agl:.0f} ft" + (f" after {prev_agl:.0f} ft" if prev_agl is not None else "") + " — terrain not loaded. Ignoring it.") if agl_rejected * SAMPLE_SECONDS >= 5: prev_agl = agl # re-sync, but this sample still measures nothing prev_agl_t = t agl_rejected = 0 prev_good_agl = prev_agl if agl_ok: prev_agl = agl prev_agl_t = t # --- everything else, once a second ----------------------------------------- # Wind, heading, gear, flaps, the altimeter, the engines and the aeroplane's name do not # change meaningfully inside fifty milliseconds. Reading them at the slow rate is what buys # the flare its real twenty samples a second. slow_due = (not fast) or (t - last_slow >= SAMPLE_SECONDS) if slow_due: last_slow = t alt = float(read(aq, "PLANE_ALTITUDE", 0.0) or 0.0) # PLANE_ALTITUDE is true altitude above sea level. INDICATED_ALTITUDE is what the # altimeter reads, which is what the pilot set, flew and remembers: FL350 is 35,000 on # the dial even when the aeroplane is physically 1,700 ft higher in warm air. Report # the dial. alt_ind = float(read(aq, "INDICATED_ALTITUDE", alt) or alt) gear_pct = read(aq, "GEAR_TOTAL_PCT_EXTENDED", None) gear = (float(gear_pct) >= 0.95) if gear_pct is not None else (float(read(aq, "GEAR_HANDLE_POSITION", 1.0) or 1.0) >= 0.99) flaps = float(read(aq, "TRAILING_EDGE_FLAPS_LEFT_PERCENT", 0.0) or 0.0) wind = float(read(aq, "AMBIENT_WIND_VELOCITY", 0.0) or 0.0) # Direction the wind is FROM, and where the nose is pointing. Wind speed alone cannot # say whether 39 kt was on the nose or across the runway, and those are entirely # different landings — one holds you up, the other pushes you sideways. wind_dir = float(__import__("math").degrees(float(read(aq, "AMBIENT_WIND_DIRECTION", 0.0) or 0.0))) heading = float(__import__("math").degrees(float(read(aq, "PLANE_HEADING_DEGREES_TRUE", 0.0) or 0.0))) _eng = engines_state(aq) engines_running = True if _eng is None else _eng if _eng is not None and not engines_reported: engines_reported = True log(f"Engine state readable from {', '.join(n for n, _ in (_engine_working or []))} — shutdown will be detected.") sim_time = float(read(aq, "SIMULATION_TIME", 0.0) or 0.0) # The aircraft, re-read until the wheels leave the ground. `if not title` latched the # first aeroplane the agent ever saw and never looked again, so changing aircraft on the # ramp left the site reporting the old one — a 737-800 sitting at the gate while the # strip said 777-300ER. After take-off it is fixed: the aeroplane cannot change in # flight, and a bad read at altitude must not be allowed to rewrite what is being # recorded. if not airborne or not title: t_raw = read(aq, "TITLE", b"") fresh = t_raw.decode("utf-8", "ignore") if isinstance(t_raw, bytes) else str(t_raw or "") if fresh and fresh != title: if title: log(f"Aircraft changed: {title} → {fresh}") # The lock belongs to the aeroplane, not to the session. Changing from an # A340 to a Bonanza on the ramp left the recorder still asking the A340's # variables; it happened to work because both answer GENERAL_ENG_RPM, but # the next pair might not be so forgiving. reset_engine_discovery() engines_reported = False post("heartbeat", token=token, state="waiting", aircraft=fresh) last_heartbeat = t title = fresh sim_misses = 0 except Exception: sim_misses += 1 frozen_samples = max(frozen_samples, 1) # a stalled SimConnect usually means the sim is paused if sim_misses > 10: if touchdown is not None: log("Lost contact with the simulator after landing — sending the flight rather than losing it.") break return None time.sleep(SAMPLE_SECONDS) continue if not sim_running(): # A flight that has already landed is finished work. It used to be discarded here: a pilot # who taxied in and quit to the menu — or whose sim crashed on the runway — lost everything, # ten hours of it, with the touchdown already measured and sitting in memory. The wait for # engines off is how we prefer to end a flight, not a condition of keeping one. if touchdown is not None: log("Simulator closed after landing — sending the flight rather than losing it.") break return None # --- pause detection ------------------------------------------------------------- # Only ever once a second. Between two samples 50 ms apart a perfectly healthy sim clock advances # about 0.05 s, which is exactly what "stopped" looks like at this threshold — so at 20 Hz this # test would call every landing a pause. Compare second to second whatever rate we are sampling at. # Only when the simulator clock was actually re-read this pass. During the flare the slow # variables are a second old by design, and a stale clock looks exactly like a stopped one — # which would call every landing a pause and hold the recording at the worst possible moment. if slow_due and t - last_pause_check >= SAMPLE_SECONDS: last_pause_check = t pos = (round(lat, 6), round(lon, 6), round(alt, 1), round(ias, 1)) clock_stopped = prev_sim_time is not None and sim_time > 0 and abs(sim_time - prev_sim_time) < 0.05 frozen_in_air = prev_pos == pos and not on_ground frozen_samples = frozen_samples + 1 if (clock_stopped or frozen_in_air) else 0 prev_sim_time = sim_time prev_pos = pos now_paused = frozen_samples >= 2 if now_paused and not paused: paused = True paused_since = t if airborne: post("heartbeat", token=token, state="paused_tod" if tod_paused else "paused", aircraft=title) if not tod_paused: log("Sim paused — recording on hold, take as long as you like.") elif paused and not now_paused: paused = False paused_total += t - (paused_since or t) paused_since = None settle = 2 # first samples after a resume can be garbage; watch, do not judge if airborne: post("heartbeat", token=token, state="recording", aircraft=title) log(f"Sim resumed · {int(paused_total // 60)} min paused so far, not counted as flying.") # A pause after the landing must not strand a finished flight. The pilot who shuts down and # then hits Escape has done everything the recording needs; holding it until they un-pause is # how a completed flight sits in memory instead of arriving. if paused and touchdown is not None and not engines_running: log("Paused after shutdown — the flight is complete, sending it now.") break if paused: if t - last_heartbeat > 30: post("heartbeat", token=token, state=("paused_tod" if tod_paused else "paused") if airborne else "waiting", aircraft=title) last_heartbeat = t time.sleep(SAMPLE_SECONDS) continue flight_t = t - paused_total # wall clock with paused time removed; all timers below use it if settle > 0: settle -= 1 last_sample = {"vs": vs, "g": g, "bank": bank, "ias": ias, "gs": gs} prev_on_ground = on_ground time.sleep(SAMPLE_SECONDS) continue # Rolling for the first time: the taxi-out state, caught once. if slow_due and not airborne and not moved_off and on_ground and gs > 3: moved_off = True proc["taxiOut"] = procedure_snapshot(aq, sm) # Warnings and the slew key are latched wherever they happen — a stall warning for one second # over the threshold still happened, and a sample a second later would miss it. if slow_due: if read(aq, "STALL_WARNING", None): stall_seen = True # A chirp at ten feet in a light aeroplane is a well-flown landing; the same chirp at # eight hundred feet on the turn to final is a different aeroplane entirely. Recording # only that it happened threw away the part that decides which one it was. if airborne and agl_ok: _a = round(agl) stall_lowest_agl = _a if stall_lowest_agl is None else min(stall_lowest_agl, _a) stall_highest_agl = _a if stall_highest_agl is None else max(stall_highest_agl, _a) if read(aq, "OVERSPEED_WARNING", None): overspeed_seen = True # A study-level airliner runs its own systems, and some of them never touch the standard # SPOILERS ARMED variable at all. Shannon armed the A350's spoilers, left them armed # through the taxi in, and the review told her she had not armed them. Before that check # is allowed to say anything, the variable has to prove it works: if it never reads true # once in a whole flight — not on approach, not on the landing roll, not at the stand — # then the aeroplane is not driving it and we know nothing, which is a different answer # from "you did not arm them" and must not be confused with it. if read(aq, "SPOILERS_ARMED", None): spoilers_ever_armed = True # The same question for the mixture and propeller levers, and for the same reason. Shannon # flew the Black Square B36TP — a Bonanza airframe with a turbine in the nose. It has a # condition lever, not a mixture, so MIXTURE LEVER POSITION reads zero for the whole # flight; the review read that zero and told her she had taken off leaned and landed # leaned. A lever that never moves off zero from start-up to shutdown is a lever the # aeroplane does not have, and an aeroplane cannot be marked down for a control that is # not in the cockpit. for _n in ("MIXTURE_LEVER:1", "MIXTURE_LEVER:2"): _v = read_extra(sm, _n) if _v is not None and float(_v) > 1: mixture_ever_set = True break for _n in ("PROP_LEVER:1", "PROP_LEVER:2"): _v = read_extra(sm, _n) if _v is not None and float(_v) > 1: prop_ever_set = True break if read(aq, "IS_SLEW_ACTIVE", None): slew_used = True _p = read(aq, "PLANE_PITCH_DEGREES", None) if _p is not None and airborne: max_pitch = max(max_pitch, abs(__import__("math").degrees(float(_p)))) if is_takeoff(airborne, on_ground, agl, ias): airborne = True joined_in_flight = agl > 5000 # already high when we arrived: this is not a departure departed_at = datetime.now(timezone.utc).isoformat() paused_at_departure = paused_total sim_time_departed = sim_time origin = nearest_icao(lat, lon) post("heartbeat", token=token, state="recording", aircraft=title) proc["takeOff"] = procedure_snapshot(aq, sm) log(f"Wheels up{f' from {origin}' if origin else ''}.") if airborne and not origin and not joined_in_flight and last_origin_try + 30 < t: last_origin_try = t origin = nearest_icao(lat, lon) if airborne: if slow_due and "cruise" not in proc and agl > 3000 and abs(vs) < 300: proc["cruise"] = procedure_snapshot(aq, sm) max_alt = max(max_alt, alt_ind) if gear_retractable and agl_ok: if prev_gear_down and not gear and gear_up_agl is None: gear_up_agl = round(agl) if not prev_gear_down and gear and touchdown is None: gear_down_agl = round(agl) prev_gear_down = gear if flight_t - last_track >= TRACK_EVERY_SECONDS and sane_position(lat, lon, last_fix, flight_t - last_track): track.append([round(lat, 4), round(lon, 4), round(alt_ind)]) last_fix = (lat, lon) last_track = flight_t if flight_t - last_route >= route_every and sane_position(lat, lon, last_fix, flight_t - last_route): route.append([round(lat, 4), round(lon, 4), round(alt_ind)]) last_fix = (lat, lon) last_route = flight_t if len(route) > ROUTE_MAX_POINTS: # halve the resolution, keep the whole shape route = route[::2] route_every *= 2 if not on_ground and vs < -100 and agl_ok: # Each gate is a crossing, not a height. See crossed_down_through. if gate1000 is None and crossed_down_through(1000, agl, prev_good_agl, 2500): gate1000 = {"ias": ias, "vs": vs, "gear": gear, "flaps": flaps} proc["gate1000"] = procedure_snapshot(aq, sm) if gate500 is None and gate1000 is not None and crossed_down_through(500, agl, prev_good_agl, 1000): gate500 = {"ias": ias, "vs": vs} # Mixture rich and prop fine belong here, on short final, not on the ramp. proc["gate500"] = procedure_snapshot(aq, sm) if agl <= 1000: approach.append((agl, ias, vs, bank)) was_below_500 = was_below_500 or agl <= 500 if was_below_500 and not on_ground and vs > 500 and agl > 300: go_arounds += 1 was_below_500 = False gate1000 = gate500 = None approach = [] log("Go-around detected.") if is_touchdown(on_ground, prev_on_ground, agl, ias): ground_events += 1 if touchdown is None: ls = last_sample or {} # True airspeed, read once, here. The headwind used to be worked out from the # wind direction against the aeroplane's heading, and it did not survive contact # with Shannon's own groundspeed: 32 kt of wind reported as 16 kt of headwind on # a landing where 63 kt indicated became 32 kt over the ground. Whatever the # direction convention is, true airspeed minus groundspeed is the headwind, it # needs no convention at all, and both numbers are measured. _tas = read(aq, "AIRSPEED_TRUE", None) touchdown = {"fpm": ls.get("vs", vs), "g": max(g, ls.get("g", 1.0)), "bank": ls.get("bank", bank), "ias": ls.get("ias", ias), "gs": ls.get("gs", gs), "lat": lat, "lon": lon, "tas": round(float(_tas)) if _tas is not None else None, "windKt": round(wind), "windDir": round(wind_dir) % 360, "heading": round(heading) % 360} touchdown_t = flight_t landed_at = datetime.now(timezone.utc).isoformat() paused_at_landing = paused_total sim_time_landed = sim_time g_watch = 5 # the load peaks a moment after the wheels touch, not at the instant destination = nearest_icao(lat, lon) post("heartbeat", token=token, state="landed", aircraft=title) log(f"Touchdown {int(abs(touchdown['fpm']))} fpm{f' at {destination}' if destination else ''}. Taxi in and shut down — nothing is sent until the engines are off.") if touchdown is not None and on_ground and gs < 5: ground_stable_since = ground_stable_since or flight_t elif touchdown is not None and (not on_ground or gs >= 5): ground_stable_since = None if touchdown is not None and on_ground and gs < 1: stopped_since = stopped_since or flight_t else: stopped_since = None if touchdown is not None and ground_stable_since: if not engines_running: if engines_off_since is None: engines_off_since = flight_t log("Engines off. Sending the flight in 20 seconds.") else: if engines_off_since is not None: log("Engines running again — holding the flight.") engines_off_since = None # Which variable is holding the flight open, and what does it say? Add-on # aeroplanes do not agree about what "running" means, and a residual reading # that never falls is indistinguishable from a running engine unless we say so. if _engine_working and flight_t - ground_stable_since > 120 and not engines_logged: engines_logged = True log("Parked with the engines still reading as running: " + ", ".join(f"{n}={read(aq, n, None)}" for n, _ in _engine_working)) # The fuel is cut and the aeroplane is stopped: the pilot has finished, whatever the # spinning-down fans still read. Ten seconds rather than twenty, because there is # nothing left to wait for. if stopped_since and fuel_cut(aq): if fuel_off_since is None: fuel_off_since = flight_t log("Fuel cut off. Sending the flight in 10 seconds.") else: fuel_off_since = None # Twenty seconds of genuine silence; or three minutes not moving an inch, which is # parked whatever the engine variables claim — an A380 whose engine reading never # fell held a finished flight open until the simulator itself was closed; or a # quarter of an hour on the stand as the outer bound. if ((fuel_off_since and flight_t - fuel_off_since > 10) or (engines_off_since and flight_t - engines_off_since > 20) or (stopped_since and flight_t - stopped_since > 180) or flight_t - ground_stable_since > 900): proc["shutdown"] = procedure_snapshot(aq, sm) break if g_watch > 0 and touchdown is not None: touchdown["g"] = max(touchdown["g"], g) g_watch -= 1 last_sample = {"vs": vs, "g": g, "bank": bank, "ias": ias} prev_on_ground = on_ground if t - last_heartbeat > 30: reply = post("heartbeat", token=token, state="landed" if touchdown is not None else "recording" if airborne else "waiting", aircraft=title) last_heartbeat = t # The site tells the sim PC where the pilot is going. Read it every time: they may set it # after take-off, or change their mind. if isinstance(reply, dict): # Said once per flight, either way. Shannon left an eleven-hour flight running with # the pause showing ON in the browser and the sim never paused, and there was no way # to tell from the console whether the recorder had been told anything at all. A log # that only speaks when things are right is no use on the night they are not. # Said whenever it changes, on the ground as well as in the air. Setting the # destination after you have taken off still works; setting it after you have landed # does nothing, and the log now shows which of those happened. intent = report_intent(reply) # The pause before the top of descent, once, when it is due. if airborne and slow_due and not tod_paused and intent and (intent.get("pauseNm") or 0) > 0: if intent.get("lat") is not None and t - last_tod_report >= 600: last_tod_report = t _to_go = nm_between(lat, lon, float(intent["lat"]), float(intent["lon"])) _trigger = descent_distance_nm(alt, intent.get("elevFt") or 0) + float(intent.get("pauseNm") or 30) log(f"TOD pause: {_to_go:.0f} nm to {intent.get('destination')}, will pause at {_trigger:.0f} nm" + ("" if alt >= 10000 else " — but only above 10,000 ft, and you are lower than that")) if tod_pause_due(intent, lat, lon, alt, vs, on_ground, airborne, tod_paused): tod_paused = True if pause_sim(sm): to_go = nm_between(lat, lon, float(intent["lat"]), float(intent["lon"])) log(f"Paused {to_go:.0f} nm from {intent.get('destination')} — about {intent.get('pauseNm')} nm before the top of descent.") log("To fly again: press Esc in the simulator, then Resume. The message on screen says to use " "the toolbar; it is wrong about this pause.") post("heartbeat", token=token, state="paused_tod", aircraft=title) # Speed up for the part that is actually being measured: airborne and low — and now also the # first seconds of rollout, so the replay does not freeze at the moment of touchdown. # The take-off is also "airborne and below 200 ft", so the climb-out lands in the flare buffer # too. On an eleven-hour flight that meant the replay data began at the departure runway, half # a world and forty thousand seconds before the landing it was supposed to show. Once clear of # the ground with no touchdown yet, throw the climb-out away. if airborne and not took_off_clear and touchdown is None and agl_ok and agl > FAST_BELOW_FT * 2: took_off_clear = True flare_buf.clear() # What a sample actually costs, measured once on the first fast pass rather than reasoned about. # Three separate causes of slow sampling have been found and fixed by argument; the fourth is # being found by measurement. If one read is 20 ms then the nine are the whole story and the # answer is fewer reads; if one read is 2 ms then the cost is somewhere else entirely. if fast and airborne and not timing_reported: timing_reported = True t_one = time.time() read(aq, "PLANE_ALT_ABOVE_GROUND", 0.0) one_ms = (time.time() - t_one) * 1000 t_nine = time.time() for _n in ("PLANE_ALT_ABOVE_GROUND", "AIRSPEED_INDICATED", "VERTICAL_SPEED", "PLANE_BANK_DEGREES", "G_FORCE", "PLANE_LATITUDE", "PLANE_LONGITUDE", "GROUND_VELOCITY", "SIM_ON_GROUND"): read(aq, _n, 0.0) nine_ms = (time.time() - t_nine) * 1000 log(f"Sample cost: one variable {one_ms:.1f} ms, nine {nine_ms:.1f} ms " f"(target under 50 ms for twenty samples a second).") rollout = touchdown_t is not None and flight_t - touchdown_t <= FLARE_ROLLOUT_SECONDS fast = (agl_ok and near_the_ground(airborne, on_ground, agl, touchdown is not None)) or rollout if fast and airborne and agl_ok: flare_buf.append((flight_t, agl, vs, ias, gs, lat, lon, bank)) if len(flare_buf) > FLARE_MAX_SAMPLES + 200: del flare_buf[:200] # a long low approach: keep the newest, cheaply time.sleep(FAST_SAMPLE_SECONDS if fast else SAMPLE_SECONDS) import statistics below = [s for s in approach if s[0] <= 1000] ias_std = statistics.pstdev([s[1] for s in below]) if len(below) > 3 else None max_bank_500 = max([s[3] for s in approach if s[0] <= 500], default=None) airborne_min = None if departed_at and landed_at: if sim_time_departed is not None and sim_time_landed is not None and sim_time_landed > sim_time_departed: airborne_min = max(0, round((sim_time_landed - sim_time_departed) / 60)) # the simulator's own clock else: wall = (datetime.fromisoformat(landed_at) - datetime.fromisoformat(departed_at)).total_seconds() airborne_min = max(0, round((wall - (paused_at_landing - paused_at_departure)) / 60)) summary = { "aircraft": title, "departedAt": departed_at, "landedAt": landed_at, "airborneMin": airborne_min, "maxAltitudeFt": round(max_alt / 100) * 100, "pausedMin": round((paused_at_landing - paused_at_departure) / 60) if paused_at_landing - paused_at_departure >= 30 else None, "touchdownFpm": round(touchdown["fpm"]) if touchdown else None, "touchdownG": round(touchdown["g"], 2) if touchdown else None, "touchdownBankDeg": round(touchdown["bank"], 1) if touchdown else None, "gsAtTouchdown": round(touchdown["gs"]) if touchdown else None, "tasAtTouchdown": touchdown["tas"] if touchdown else None, "windKtAtTouchdown": touchdown["windKt"] if touchdown else None, "windDirAtTouchdown": touchdown["windDir"] if touchdown else None, "headingAtTouchdown": touchdown["heading"] if touchdown else None, "touchdownLat": round(touchdown["lat"], 5) if touchdown else None, "touchdownLon": round(touchdown["lon"], 5) if touchdown else None, "flare": trim_flare(flare_buf, touchdown_t), "maxBankBelow500": round(max_bank_500, 1) if max_bank_500 is not None else None, "iasAt1000": round(gate1000["ias"]) if gate1000 else None, "iasAt500": round(gate500["ias"]) if gate500 else None, "iasAtTouchdown": round(touchdown["ias"]) if touchdown else None, "vsAt1000": round(gate1000["vs"]) if gate1000 else None, "vsAt500": round(gate500["vs"]) if gate500 else None, "gearRetractable": gear_retractable, "gearDownBy1000": (gate1000["gear"] if gate1000 else None) if gear_retractable else None, "gearUpAgl": gear_up_agl, "gearDownAgl": gear_down_agl, "flapsLandingBy1000": (gate1000["flaps"] >= 60) if gate1000 else None, "iasStdBelow1000": round(ias_std, 1) if ias_std is not None else None, "joinedInFlight": joined_in_flight or None, "goArounds": go_arounds, "bounces": max(0, ground_events - 1), "windKt": round(wind), "procedure": ({**proc, "flags": {"spoilersReadable": spoilers_ever_armed, "mixtureReadable": mixture_ever_set, "propReadable": prop_ever_set, "slewUsed": slew_used, "stallWarning": stall_seen, "overspeed": overspeed_seen, "stallLowestAglFt": stall_lowest_agl, "stallHighestAglFt": stall_highest_agl, "maxPitchDeg": round(max_pitch, 1) if max_pitch else None}} if proc else None), "track": track[-600:], "route": route or None, "clientVersion": AGENT_VERSION, } summary = {k: v for k, v in summary.items() if v is not None} return {"aircraft": title, "origin": origin, "destination": destination, "summary": summary} def hold_window(): """A double-clicked .exe closes the instant it finishes. Give the reader time to read.""" if FROZEN: try: input("\nPress Enter to close this window.") except Exception: pass def first_run_setup(): """No token yet and someone has double-clicked the exe. Ask, rather than exit with advice.""" print("") print(" THREE FLIGHTS - sim recorder") print(" " + "-" * 46) print(" This PC is not linked to an account yet.") print("") print(" On three-flights.vercel.app/download - the page you downloaded this from -") print(" sign in and press 'Show my link code'. Six characters appear right there.") print("") try: code = input(" Type the code and press Enter: ").strip() except Exception: code = "" if not code: log("No code entered. Run this again when you have one.") return False link(code) try: answer = input("\n Start automatically with Windows from now on? [Y/n] ").strip().lower() except Exception: answer = "n" if answer in ("", "y", "yes"): install() return True def _stop_now(signum, frame): """Ctrl+C, made unconditional. The polite route — KeyboardInterrupt caught at the bottom of the file — only works when the interrupt reaches the main thread, and the SimConnect library's background thread often keeps it from ever arriving. So Ctrl+C looked dead: the one instruction every pilot knows, doing nothing. A signal handler fires regardless, says the same calm sentence, and leaves immediately. os._exit because a wave of daemon threads mid-shutdown is exactly what hangs a normal sys.exit here.""" log("Stopped. Nothing was lost — run it again whenever you like.") os._exit(0) def main(): import signal signal.signal(signal.SIGINT, _stop_now) if hasattr(signal, "SIGBREAK"): signal.signal(signal.SIGBREAK, _stop_now) # Ctrl+Break, the Windows sibling args = sys.argv[1:] if args[:1] == ["--version"]: print(AGENT_VERSION); return if args[:1] == ["--link"] and len(args) > 1: link(args[1]); hold_window(); return if args[:1] == ["--install"]: install(); hold_window(); return if args[:1] == ["--uninstall"]: uninstall(); hold_window(); return token = load_token() if not token and FROZEN: if not first_run_setup(): hold_window(); return token = load_token() if not token: log("Not linked yet. On the site: Profile → Your sim PC → Link this PC, then run: python three_flights_agent.py --link CODE") sys.exit(1) log(f"Three Flights agent {AGENT_VERSION} running. Waiting for MSFS…") log(f"Keeping a log at {LOG_FILE} — send me that file if a flight does not do what you expected.") report_intent(post("heartbeat", token=token, state="idle", aircraft="")) last_idle_beat = time.time() announced_starting = False while True: if not sim_running(): announced_starting = False if time.time() - last_idle_beat > 30: # "the agent is alive, the sim just isn't open" # Also the moment to say whether the pause is armed. Waiting until MSFS is open meant # the answer arrived after the pilot had already committed to the flight. report_intent(post("heartbeat", token=token, state="idle", aircraft="")) last_idle_beat = time.time() time.sleep(3) continue if not announced_starting: post("heartbeat", token=token, state="starting", aircraft="") # sim seen, connecting announced_starting = True log("MSFS is running. Connecting…") try: # python-SimConnect logs a warning every time the simulator rejects a variable name, and # with no handler configured Python prints those straight to the console. Probing for the # engine variables therefore filled the window with # SIMCONNECT_EXCEPTION_NAME_UNRECOGNIZED before every flight, burying "Wheels up" and # "Touchdown 185 fpm" in it. A rejected name is an expected answer here — it means this # aeroplane does not have that variable — so it is not warning-worthy. Errors still print. import logging logging.getLogger("SimConnect").setLevel(logging.ERROR) from SimConnect import SimConnect, AircraftRequests sm = SimConnect() aq = AircraftRequests(sm, _time=0) speed_up_reads(sm) except Exception: time.sleep(5) # sim is loading; try again shortly continue while True: result = record_one_flight(token, aq, sm) if result is None: log("Sim closed. Waiting for MSFS…") try: sm.exit() except Exception: pass post("heartbeat", token=token, state="idle", aircraft="") announced_starting = False break post("heartbeat", token=token, state="sending", aircraft=result["aircraft"]) r = post("complete", token=token, **result) if r.get("ok"): log("Flight sent. Open Three Flights — the observed review is waiting.") else: log(f"Could not send ({r.get('error')}); saved to three-flights-last-flight.json.") with open(os.path.join(HERE, "three-flights-last-flight.json"), "w", encoding="utf-8") as f: json.dump(result, f, indent=2) time.sleep(5) if __name__ == "__main__": try: main() except KeyboardInterrupt: # Ctrl+C is how a pilot stops the recorder. Without this, PyInstaller reports the # interrupt as "Failed to execute script ... unhandled exception", which reads like a # crash to anyone who did exactly the right thing. log("Stopped. Nothing was lost — run it again whenever you like.") sys.exit(0)