#!/usr/bin/env python3 """Munfalke 1: kRPC 0.6 flight to Mun orbit, landing, and return to Kerbin. Default invocation is read-only. See ../doku/MUNFALKE_FLUG.md before --execute. --goal land stops after touchdown on the Mun; --goal return additionally ascends, transfers back to Kerbin, and lands under parachute. Both landing paths remain experimental and less flight-tested than the orbit-only path. """ from __future__ import annotations import argparse from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone import importlib.metadata import json import math from pathlib import Path import signal import socket import time ROOT = Path(__file__).resolve().parents[1] G0 = 9.80665 class FlightError(RuntimeError): pass @dataclass(frozen=True) class State: body: str situation: str stage: int periapsis: float eccentricity: float mun_encounter: bool = False def mission_phase(state: State) -> str: """Infer from the actual vessel, never from an old local progress file. 26.09.2026: the return leg reuses Kerbin/Mun and low stage numbers, so the ordering here matters. Stage 3 only ever occurs on the outbound leg (it is dropped for good once land() starts), and stage 1/2 at Kerbin only ever occurs on the return leg (the outbound leg never gets below stage 3 until it is at the Mun) -- that stage split is what keeps outbound and return states from colliding without needing extra bookkeeping. """ if state.body == "Kerbin": if state.situation in ("landed", "splashed") and state.stage <= 1: return "recovered" if state.situation == "pre_launch" and state.stage == 6: return "launch" if state.mun_encounter and state.stage in (3, 4): return "coast" if state.periapsis > 70000 and state.eccentricity < 1 and state.stage in (3, 4): return "transfer" if (state.situation not in ("landed", "splashed") and state.stage in (1, 2) and state.eccentricity < 1): return "reentry" if 20000 < state.periapsis < 45000 else "reentry_prep" if state.body == "Mun": if state.situation == "landed": return "landed" if state.stage == 3: if state.eccentricity >= 1: return "capture" if state.periapsis > 15000: return "land" if state.stage == 2: if state.eccentricity >= 1: return "return_coast" if state.periapsis > 10000: return "return_transfer" return "descent" raise FlightError("Dieser Flugzustand ist nicht automatisch wiederaufnehmbar. " "Status prüfen oder ausdrücklich einen Sicherungspunkt laden.") def booster_separation_ready(solid_fuel: float, altitude: float, dynamic_pressure: float) -> bool: # Avoid the dense-air, high-load separation that destroyed the carrier. return solid_fuel < .1 and altitude >= 15000 and dynamic_pressure <= 8000 def liquid_propellant_mass(liquid_fuel: float, oxidizer: float) -> float: return max(0.0, min(liquid_fuel / .9, oxidizer / 1.1) * 10.0) def rocket_delta_v(mass: float, propellant: float, isp: float = 345) -> float: if not (0 < propellant < mass) or isp <= 0: return 0.0 return isp * G0 * math.log(mass / (mass - propellant)) def burn_duration(mass: float, thrust: float, isp: float, delta_v: float) -> float: if min(mass, thrust, isp) <= 0 or delta_v < 0: raise FlightError("Ungültige Triebwerksdaten für die Brenndauer.") ve = isp * G0 return mass * ve / thrust * (1 - math.exp(-delta_v / ve)) def capture_delta_v(mu: float, periapsis_radius: float, sma: float, target_apoapsis_radius: float) -> float: if target_apoapsis_radius <= periapsis_radius: raise FlightError("Ziel-Apoapsis muss über der Periapsis liegen.") incoming = math.sqrt(mu * (2 / periapsis_radius - 1 / sma)) target = math.sqrt(mu * (2 / periapsis_radius - 2 / (periapsis_radius + target_apoapsis_radius))) return target - incoming def circularize_delta_v(mu: float, radius: float, sma: float) -> float: """Prograde dv to circularize at the given point of an orbit with this sma.""" if radius <= 0 or sma <= 0: raise FlightError("Ungültige Bahndaten für die Zirkularisierung.") current = math.sqrt(mu * (2 / radius - 1 / sma)) target = math.sqrt(mu / radius) return target - current def descent_command(altitude: float, velocity: tuple[float, float, float], gravity: float, maximum_acceleration: float): """Velocity is BODY-relative, rotated into surface axes (up, north, east).""" if maximum_acceleration <= gravity: raise FlightError("Landeschub reicht nicht zum Abbremsen.") up, north, east = velocity target = -min(250.0, math.sqrt(4.4 * max(0.0, altitude - 15)) + 1) # Brake early enough to allow for the two-second controller response lag. # The old 30-metre transition contacted at over 5 m/s in the offline model. if altitude < 500: target = -min(35.0, math.sqrt(1.6 * max(0.0, altitude - 40)) + 1) if altitude < 80: target = -min(6.0, max(.7, (altitude - 5) * .08)) vertical = max(0.0, gravity + (target - up) / 2) limit = max(.3, vertical * .5) lateral = [max(-limit, min(limit, -x / 2)) for x in (north, east)] direction = (vertical, *lateral) acceleration = math.sqrt(sum(x * x for x in direction)) return ((direction if acceleration > .05 else (1, 0, 0)), min(1.0, acceleration / maximum_acceleration)) @contextmanager def flight_lock(path: Path): """Protect against two instances of THIS script; other autopilots must be off.""" import fcntl # This project is run on Linux/Pop!_OS. path.parent.mkdir(parents=True, exist_ok=True) with path.open("a+") as file: try: fcntl.flock(file, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise FlightError("Eine weitere Munfalke-Steuerung läuft bereits.") from exc try: yield finally: fcntl.flock(file, fcntl.LOCK_UN) class Mission: def __init__(self, conn, log_path: Path | None = None): self.conn = conn self.sc = conn.space_center self.v = self.sc.active_vessel self.log_path = log_path self.last_report = 0.0 self.last_stage = -math.inf self.session = datetime.now().strftime("%Y%m%d-%H%M%S") self.checkpoint_number = 0 if self.v.name != "Munfalke 1": raise FlightError(f"Erwartet Munfalke 1, aktiv ist {self.v.name!r}.") def emit(self, event, **fields): row = {"time": datetime.now(timezone.utc).isoformat(), "event": event, **fields} print(json.dumps(row, ensure_ascii=False), flush=True) if self.log_path: with self.log_path.open("a", encoding="utf-8") as file: file.write(json.dumps(row, ensure_ascii=False) + "\n") @staticmethod def encounter(orbit, body_name="Mun"): for _ in range(6): if orbit is None: break if orbit.body.name == body_name: return orbit orbit = orbit.next_orbit return None def state(self): o = self.v.orbit return State(o.body.name, self.v.situation.name, self.v.control.current_stage, o.periapsis_altitude, o.eccentricity, o.body.name == "Kerbin" and self.encounter(o) is not None) def snapshot(self): v = self.v f = v.flight(v.orbit.body.reference_frame) return dict(vessel=v.name, body=v.orbit.body.name, situation=v.situation.name, stage=v.control.current_stage, ut=self.sc.ut, altitude=f.mean_altitude, altitude_agl=f.surface_altitude, vertical_speed=f.vertical_speed, horizontal_speed=f.horizontal_speed, apoapsis=v.orbit.apoapsis_altitude, periapsis=v.orbit.periapsis_altitude, mass=v.mass, fuel=v.resources.amount("LiquidFuel"), electricity=v.resources.amount("ElectricCharge"), throttle=v.control.throttle) def report(self, force=False): if force or time.monotonic() - self.last_report >= 5: self.emit("telemetry", **self.snapshot()) self.last_report = time.monotonic() def guard(self, deadline=None): if self.sc.active_vessel != self.v: raise FlightError("Aktives Fahrzeug gewechselt; Steuerung angehalten.") if deadline is not None and time.monotonic() > deadline: raise FlightError("Zeitlimit der Flugphase überschritten.") if self.v.resources.amount("ElectricCharge") < 3: raise FlightError("Stromversorgung unter 3 Einheiten.") def save(self, label): self.checkpoint_number += 1 name = f"Munfalke-auto-{self.session}-{self.checkpoint_number:02d}-{label}" self.sc.save(name) self.emit("checkpoint", name=name, **self.snapshot()) def _activate_stage(self, expected): """One command, then acknowledge the new stage. No policy on which stage.""" self.guard() if self.v.control.current_stage != expected: raise FlightError(f"Stufentrennung abgelehnt: erwartet {expected}, " f"aktuell {self.v.control.current_stage}.") if time.monotonic() - self.last_stage < 2: raise FlightError("Stufentrennungen folgen zu schnell aufeinander.") self.v.control.activate_next_stage() self.last_stage = time.monotonic() deadline = time.monotonic() + 6 while self.v.control.current_stage == expected: self.guard(deadline) time.sleep(.1) time.sleep(1) if self.v.control.current_stage != expected - 1: raise FlightError("Unerwartete zusätzliche Stufentrennung; Autostage prüfen.") self.emit("stage", old=expected, new=expected - 1) def stage(self, expected): """Outbound/landing staging only: never drop stage 1 (capsule) or 0 (chute).""" if expected <= 2: raise FlightError(f"Stufentrennung abgelehnt: erwartet {expected}, " f"aktuell {self.v.control.current_stage}.") self._activate_stage(expected) def stage_resource(self, decouple_stage, resource): return sum(p.resources.amount(resource) for p in self.v.parts.all if p.decouple_stage == decouple_stage) def reserve(self): detach = {4: 3, 3: 2, 2: 1}.get(self.v.control.current_stage) if detach is None: return 0.0 prop = liquid_propellant_mass(self.stage_resource(detach, "LiquidFuel"), self.stage_resource(detach, "Oxidizer")) return rocket_delta_v(self.v.mass, prop, 320 if detach == 3 else 345) def align(self, reference_frame, direction, timeout=35): a = self.v.auto_pilot self.v.control.sas = False a.reference_frame = reference_frame a.target_direction = direction a.engaged = True # kRPC 0.6: NOT the old engage() method. deadline = time.monotonic() + timeout while a.error > 3: self.guard(deadline) time.sleep(.1) def warp(self, target_ut, expected_body, timeout=240): """Bounded on-rails warp; factors are indices, unlike warp_to's rates.""" self.v.control.throttle = 0 self.v.auto_pilot.engaged = False self.sc.physics_warp_factor = 0 deadline = time.monotonic() + timeout try: while self.sc.ut < target_ut - .5: self.guard(deadline) if self.v.orbit.body.name != expected_body: break # Stop and re-evaluate immediately on an SOI transition. body = self.v.orbit.body # 26.09.2026: this used to call flight() with no reference_frame, # which defaults to the vessel's own surface_reference_frame -- # a frame that moves WITH the vessel. mean_altitude read from # that frame is not reliably altitude above the body (see the # same class of bug fixed in the Astronautica horizontal_speed # case on 24.09.). It wrongly blocked the post-separation warp # home right after reentry()'s capsule separation, stranding # the capsule without ever reaching the parachute-arming loop. if (body.has_atmosphere and self.v.flight(body.reference_frame).mean_altitude < body.atmosphere_depth + 1000): # Re-check once before aborting: a single reading right after # a stage/vessel-changing event is not trusted on its own. time.sleep(.3) if (self.v.flight(body.reference_frame).mean_altitude < body.atmosphere_depth + 1000): raise FlightError("Kein Rails-Warp innerhalb der Atmosphäre.") remaining = target_ut - self.sc.ut factor = next((factor for threshold, factor in [(3000, 6), (500, 5), (100, 4), (30, 3), (10, 2), (3, 1)] if remaining > threshold), 0) if self.sc.rails_warp_factor != factor: self.sc.rails_warp_factor = factor self.report() time.sleep(.1) finally: self.sc.rails_warp_factor = 0 def verify_parts(self, expected): actual = set(self.v.parts.all) if actual != expected: self.v.control.throttle = 0 raise FlightError("Unerwarteter Teileverlust nach der Boostertrennung; Flug angehalten.") def launch(self): v = self.v if self.state().situation != "pre_launch" or v.control.current_stage != 6: raise FlightError("Start nur von der unveränderten Startrampe, Stufe 6.") if len(v.parts.all) != 39 or len(v.parts.engines) != 9 or v.crew_count != 1: raise FlightError("Munfalke-Aufbau/Besatzung weicht vom geprüften Entwurf ab.") if sum(p.name in ("radialDecoupler", "radialDecoupler2") for p in v.parts.all) != 2: raise FlightError("Erwartet zwei radiale Booster-Abtrenner.") sepratrons = [p for p in v.parts.all if p.name == "sepMotor1"] if (len(sepratrons) != 4 or any(p.stage != 4 or p.decouple_stage != 4 or p.resources.amount("SolidFuel") < 7.9 for p in sepratrons)): raise FlightError("Vier volle Sepratrons an den Boostern in Stufe 4 erwartet.") carrier_parts = set(p for p in v.parts.all if p.decouple_stage != 4) if (v.resources.amount("LiquidFuel") < 3500 or v.resources.amount("Oxidizer") < 4280 or v.resources.amount("SolidFuel") < 1630): raise FlightError("Starttanks nicht voll.") self.save("vor-Start") self.align(v.surface_reference_frame, (1, 0, 0)) v.control.rcs = False v.control.throttle = 1 self.stage(6) f = v.flight(v.orbit.body.reference_frame) deadline = time.monotonic() + 1100 booster_done = core_done = False while v.orbit.periapsis_altitude < 80000: self.guard(deadline) stage = v.control.current_stage expected_stage = 3 if core_done else (4 if booster_done else 5) if stage != expected_stage: raise FlightError("Externe Stufenschaltung erkannt; Flug angehalten.") if (not booster_done and stage == 5 and booster_separation_ready(sum(p.resources.amount("SolidFuel") for p in v.parts.all if p.name == "solidBooster1-1"), f.mean_altitude, f.dynamic_pressure)): booster_done = True # Set BEFORE issuing the RPC; avoid duplicate commands. self.stage(5) self.verify_parts(carrier_parts) if booster_done and not core_done and v.control.current_stage == 4: if self.stage_resource(3, "LiquidFuel") < .2: core_done = True v.control.throttle = 0 self.stage(4) if booster_done and not core_done: self.verify_parts(carrier_parts) if v.control.current_stage not in (3, 4, 5): raise FlightError("Unerwartete Stufe im Aufstieg.") alt = f.mean_altitude vs = f.vertical_speed apo, tta = v.orbit.apoapsis_altitude, v.orbit.time_to_apoapsis if alt < 1000: pitch = 90 elif alt < 10000: pitch = 90 - 45 * (alt - 1000) / 9000 elif alt < 40000: pitch = 45 - 35 * (alt - 10000) / 30000 else: pitch = max(-5, min(20, (45 - tta) * .7)) # 26.09.2026: time_to_apoapsis is only meaningful on the way UP. # Once the vessel has coasted past its actual apoapsis without the # periapsis-raising burn finishing, KSP reports the time to the # NEXT lap's apoapsis instead -- nearly a full orbital period away # -- which satisfied "tta > 45" again and left the throttle stuck # at 0 while the vessel fell straight back toward the atmosphere # with an unsafe periapsis. vertical_speed has no such ambiguity: # only coast while genuinely still climbing with time to spare. coasting = apo > 100000 and vs > 0 and tta > 45 throttle = 0 if coasting else 1 if apo > 100000 and not coasting: pitch = max(-10, min(15, (20 - min(tta, 45)) * .6)) v.auto_pilot.target_pitch_and_heading(pitch, 90) v.control.throttle = throttle if alt > 5000 and vs < -150: raise FlightError("Aufstieg gescheitert: zu starke Sinkgeschwindigkeit.") self.report() time.sleep(.15) v.control.throttle = 0 self.save("Kerbinorbit") def transfer(self): if self.v.orbit.periapsis_altitude <= 70000: raise FlightError("Transfer braucht einen sicheren Kerbinorbit.") if self.v.control.current_stage == 4: # Deliberate discard of the carrier IN ORBIT, not a flameout heuristic. self.v.control.throttle = 0 self.save("vor-Traegertrennung") self.stage(4) if self.v.control.current_stage != 3 or self.reserve() < 1300: raise FlightError("Zu wenig Transferreserve; Aufstieg prüfen.") self.save("vor-Transfer") self.sc.target_body = self.sc.bodies["Mun"] # 26.09.2026: a single node covering the whole ~900 m/s Mun transfer sweeps # enough orbital arc during the (necessarily long) burn that periapsis can # dip into the atmosphere even though burn() aborts safely before that # becomes dangerous (see burn()'s periapsis guard). Rather than just giving # up on a safe abort, replan and finish the transfer over several shorter # passes: each partial burn raises apoapsis, which makes the next pass's # remaining burn shorter and safer, converging quickly in practice. for attempt in range(4): node = self._plan_transfer_node() if self.burn(node): self.correct_course() return self.emit("transfer_retry", attempt=attempt + 1) raise FlightError("Transfer nach mehreren Teil-Burns nicht sicher abgeschlossen.") def _plan_transfer_node(self): o = self.v.orbit mu, radius = o.body.gravitational_parameter, o.semi_major_axis moon_radius = self.sc.bodies["Mun"].orbit.semi_major_axis seed = math.sqrt(mu / radius) * (math.sqrt(2 * moon_radius / (radius + moon_radius)) - 1) period = o.period base = self.sc.ut + 480 node = self.v.control.add_node(base, prograde=seed) best = None deadline = time.monotonic() + 180 try: def evaluate(ut, prograde): nonlocal best self.guard(deadline) node.ut, node.prograde = ut, prograde time.sleep(.02) encounter = self.encounter(node.orbit) if encounter: pe = encounter.periapsis_altitude score = abs(pe - 40000) + 2 * prograde if best is None or score < best[0]: best = (score, ut, prograde, pe) for i in range(73): for extra in (0, 20, 40, 70, 100): evaluate(base + period * i / 72, seed + extra) if best is None: raise FlightError("Keine Mün-Begegnung gefunden; kein Schub ausgeführt.") for dt, dv in [(period / 72, 15), (period / 300, 4), (period / 1200, 1)]: center = best for ti in range(-4, 5): for di in range(-4, 5): evaluate(center[1] + ti * dt / 4, center[2] + di * dv / 4) node.ut, node.prograde = best[1:3] if not 20000 < best[3] < 100000: raise FlightError("Geplante Mün-Passage nicht im sicheren Höhenband.") self.emit("transfer_plan", delta_v=node.delta_v, mun_periapsis=best[3]) except BaseException: node.remove() raise return node def burn(self, node): v = self.v v.control.throttle = 0 dv = node.delta_v if dv + 5 > self.reserve(): raise FlightError(f"Manöver {dv:.1f} m/s übersteigt die Stufenreserve.") duration = burn_duration(v.mass, v.available_thrust, v.specific_impulse, dv) start = node.ut - duration / 2 if start < self.sc.ut + 20: raise FlightError("Manöverzeitpunkt zu nah/vergangen; neu planen.") self.emit("burn_start", delta_v=dv, duration=duration) # 26.09.2026: a long warp can leave the vessel tumbling in whatever # orientation it happened to freeze in; 25 s wasn't always enough real # time left to turn all the way to the burn vector before the ignition # deadline below (observed: missed the Mun capture window entirely, # ending up on an uncaptured flyby). align()'s own timeout is 35 s, so # 90 s of lead time leaves real slack even for a near-180 deg turn. self.warp(start - 90, v.orbit.body.name) self.align(node.reference_frame, (0, 1, 0)) # 26.09.2026: a node can legitimately be close enough to "now" (e.g. a # Mun capture planned right at SOI entry, already near periapsis) that # the 90 s lead-time above never materialises -- warp() just returns # immediately once its target is in the past. align() only reaches # here once the vessel IS correctly pointed (it raises its own # timeout otherwise), so a late finish means a late-but-correct burn, # not a wrong one. Missing the capture window entirely once cost a # dangerous uncaptured Mun flyby; starting a few seconds late and # relying on capture()'s own post-burn safety check is far safer. if self.sc.ut > start + 2: self.emit("alignment_late", overshoot=self.sc.ut - start) start = self.sc.ut deadline = time.monotonic() + duration * 3 + 100 while self.sc.ut < start: self.guard(deadline) time.sleep(.1) # 26.09.2026: a long, high-dv burn split around the node still sweeps a real # orbital arc, which can pull periapsis down toward/into the atmosphere even # though the impulsive node math looked safe (observed: TMI burn from an # 85 km Kerbin parking orbit grazed down to ~56 km at Mach 10). Abort the # burn early rather than complete it once periapsis crosses into that danger # zone; skip the check if periapsis was already unsafe before ignition (e.g. # a deliberate deorbit), since this guard is only meant to catch a burn # dragging a previously-safe orbit down, not to block an intentional one. body = v.orbit.body safe_periapsis = body.atmosphere_depth + 5000 if body.has_atmosphere else None if safe_periapsis is not None and v.orbit.periapsis_altitude < safe_periapsis: safe_periapsis = None minimum = node.remaining_delta_v cut_short = False try: while node.remaining_delta_v > .5: self.guard(deadline) remaining = node.remaining_delta_v if remaining > minimum + 3 and minimum < 10: raise FlightError("Manöverrest steigt wieder; Schub beendet, Bahn prüfen.") minimum = min(minimum, remaining) if v.available_thrust < 1 or not any(e.active and e.has_fuel for e in v.parts.engines): raise FlightError("Triebwerk ohne Treibstoff; keine automatische Trennung.") if v.auto_pilot.error > 10: raise FlightError("Ausrichtung im Manöver verloren.") if safe_periapsis is not None and v.orbit.periapsis_altitude < safe_periapsis: self.emit("burn_cut_short", periapsis=v.orbit.periapsis_altitude, remaining_delta_v=remaining) cut_short = True break acceleration = v.available_thrust / v.mass v.control.throttle = min(1, max(.015, remaining / (acceleration * 2))) self.report() time.sleep(.08) finally: v.control.throttle = 0 node.remove() if cut_short: return False self.save("nach-Manoever") return True def correct_course(self): encounter = self.encounter(self.v.orbit) if encounter and 25000 < encounter.periapsis_altitude < 100000: self.save("sicherer-Muenkurs") return node = self.v.control.add_node(self.sc.ut + 360, prograde=0) best = None deadline = time.monotonic() + 120 try: for step in (20, 5, 1, .2): center = (0, 0) if best is None else best[1:3] for i in range(-4, 5): for j in range(-4, 5): self.guard(deadline) pro, radial = center[0] + i * step, center[1] + j * step node.prograde, node.radial = pro, radial time.sleep(.02) encounter = self.encounter(node.orbit) if encounter: pe = encounter.periapsis_altitude score = abs(pe - 40000) + 10 * math.hypot(pro, radial) if best is None or score < best[0]: best = (score, pro, radial, pe) if best is None or not 25000 < best[3] < 100000: raise FlightError("Keine sichere Kurskorrektur gefunden.") node.prograde, node.radial = best[1:3] self.emit("course_correction", delta_v=node.delta_v, periapsis=best[3]) except BaseException: node.remove() raise self.burn(node) encounter = self.encounter(self.v.orbit) if not encounter or not 20000 < encounter.periapsis_altitude < 120000: raise FlightError("Reale Bahn nach Korrektur nicht sicher; kein Weiter-Warp.") self.save("sicherer-Muenkurs") def coast(self): self.correct_course() transition = self.v.orbit.time_to_soi_change if not math.isfinite(transition) or transition <= 0: raise FlightError("Kein gültiger Mün-SOI-Wechsel.") self.warp(self.sc.ut + transition + 5, "Kerbin") if self.v.orbit.body.name != "Mun": raise FlightError("Mün-Einflussbereich nicht erreicht.") self.save("Muenankunft") def capture(self): o = self.v.orbit if o.body.name != "Mun" or o.periapsis_altitude < 20000: raise FlightError("Mün-Ankunft nicht auf sicherer Einbremsbahn.") target_apoapsis = max(400000, o.periapsis_altitude + 50000) dv = capture_delta_v(o.body.gravitational_parameter, o.periapsis, o.semi_major_axis, o.body.equatorial_radius + target_apoapsis) if dv >= 0: raise FlightError("Kein gültiges rückwärts gerichtetes Einbremsmanöver.") node = self.v.control.add_node(self.sc.ut + o.time_to_periapsis, prograde=dv) self.burn(node) if self.v.orbit.eccentricity >= 1 or self.v.orbit.periapsis_altitude < 15000: raise FlightError("Keine sichere Mün-Umlaufbahn erreicht.") self.save("Muenorbit") def land(self, resume_descent=False): v, b = self.v, self.v.orbit.body if b.name != "Mun" or v.control.current_stage not in (2, 3): raise FlightError("Landung braucht Munfalke-Lander am Mün.") if not resume_descent: self.save("vor-Landung") o = v.orbit if not 15000 < o.periapsis_altitude < 100000: raise FlightError("Landung nur aus einer Bahn mit 15–100 km Periapsis.") # The flown trajectory started braking just after periapsis near 40 km. if v.flight().mean_altitude > o.periapsis_altitude + 10000: self.warp(self.sc.ut + o.time_to_periapsis, "Mun") if v.control.current_stage == 3: v.control.throttle = 0 self.stage(3) if v.control.current_stage != 2 or len(v.parts.legs) != 4: raise FlightError("Landestufe oder Landebeine fehlen.") if not resume_descent and self.reserve() < 1500: raise FlightError("Landestufe hat weniger als 1500 m/s Reserve.") for leg in v.parts.legs: leg.deployed = True f = v.flight(b.reference_frame) if f.horizontal_speed > 50: if f.surface_altitude < 8000: raise FlightError("Zu niedriger Einstieg für das horizontale Bremsmanöver.") self.align(v.surface_velocity_reference_frame, (0, -1, 0)) deadline = time.monotonic() + 200 while f.speed > 45: self.guard(deadline) if v.available_thrust < 1: raise FlightError("Landetriebwerk liefert keinen Schub.") if v.auto_pilot.error > 10: raise FlightError("Bremsausrichtung verloren.") v.control.throttle = 1 self.report() time.sleep(.1) v.control.throttle = 0 a = v.auto_pilot v.control.sas = False a.reference_frame = v.surface_reference_frame a.target_direction = (1, 0, 0) a.engaged = True deadline = time.monotonic() + 1000 while v.situation != self.sc.VesselSituation.landed: self.guard(deadline) # The surface reference frame moves WITH the vessel. Direct velocity # in that frame is zero. Rotate body-relative velocity instead. velocity = self.sc.transform_direction(v.velocity(b.reference_frame), b.reference_frame, v.surface_reference_frame) gravity = b.gravitational_parameter / (b.equatorial_radius + f.mean_altitude) ** 2 direction, throttle = descent_command(f.surface_altitude, velocity, gravity, v.available_thrust / v.mass) a.target_direction = direction if a.error > 20 and f.surface_altitude > 1000: throttle = min(.1, throttle) v.control.throttle = throttle self.report() time.sleep(.1) v.control.throttle = 0 # Confirm contact, rather than guessing touchdown from centre-of-mass height. a.target_direction = (1, 0, 0) deadline = time.monotonic() + 15 stable_since = time.monotonic() while time.monotonic() - stable_since < 3: self.guard(deadline) if v.situation != self.sc.VesselSituation.landed or f.speed > 1: stable_since = time.monotonic() time.sleep(.1) self.save("Muenlandung") self.emit("landed", latitude=f.latitude, longitude=f.longitude, remaining_delta_v=self.reserve(), **self.snapshot()) def ascend(self): """Lift the lander (stage 2, same engine as the landing) back to a low, near-circular Mun orbit. Never stages further -- capsule separation only happens at Kerbin, in separate_capsule().""" v, b = self.v, self.v.orbit.body if b.name != "Mun" or v.situation != self.sc.VesselSituation.landed: raise FlightError("Aufstieg braucht eine gelandete Munfalke-Landestufe am Mün.") if v.control.current_stage != 2 or len(v.parts.legs) != 4: raise FlightError("Landestufe oder Landebeine fehlen für den Aufstieg.") if self.reserve() < 700: raise FlightError("Landestufe hat weniger als 700 m/s Reserve für den Aufstieg.") self.save("vor-Muenstart") v.control.sas = False v.control.rcs = False v.auto_pilot.engaged = True v.control.throttle = 1 f = v.flight(b.reference_frame) target_apoapsis = 20000.0 deadline = time.monotonic() + 600 while v.orbit.apoapsis_altitude < target_apoapsis or v.situation == self.sc.VesselSituation.landed: self.guard(deadline) if v.available_thrust < 1 or not any(e.active and e.has_fuel for e in v.parts.engines): raise FlightError("Triebwerk ohne Treibstoff beim Mün-Aufstieg.") alt = f.surface_altitude # Airless body: no gravity turn needed for drag, just a gentle pitchover # once clear of local terrain so the burn also builds orbital speed. pitch = 90 if alt < 3000 else max(5, 90 - (alt - 3000) / 150) v.auto_pilot.target_pitch_and_heading(pitch, 90) self.report() time.sleep(.15) v.control.throttle = 0 self.save("Muenaufstieg") o = v.orbit node = v.control.add_node(self.sc.ut + o.time_to_apoapsis, prograde=circularize_delta_v( o.body.gravitational_parameter, o.body.equatorial_radius + o.apoapsis_altitude, o.semi_major_axis)) if not self.burn(node): raise FlightError("Zirkularisierung nach dem Aufstieg abgebrochen.") if v.orbit.eccentricity >= 1 or v.orbit.periapsis_altitude < 10000: raise FlightError("Kein sicherer Münorbit nach dem Aufstieg erreicht.") self.save("Muenorbit-Rueckflug") def return_transfer(self): if self.v.orbit.body.name != "Mun" or self.v.orbit.eccentricity >= 1: raise FlightError("Rückkehr-Transfer braucht einen sicheren Münorbit.") if self.v.control.current_stage != 2 or self.reserve() < 500: raise FlightError("Zu wenig Rückkehrreserve; Aufstieg prüfen.") self.save("vor-Ruecktransfer") self.sc.target_body = self.sc.bodies["Kerbin"] for attempt in range(4): node = self._plan_return_node() if self.burn(node): self.save("Muenabflug") return self.emit("return_retry", attempt=attempt + 1) raise FlightError("Rückkehr-Transfer nach mehreren Teil-Burns nicht sicher abgeschlossen.") def _plan_return_node(self): """Aim only for SOME safe Kerbin encounter here; reentry_prep() refines the periapsis into the tight reentry band once closer to Kerbin, the same two-stage approach transfer()/correct_course() use outbound.""" o = self.v.orbit period = o.period base = self.sc.ut + 60 node = self.v.control.add_node(base, prograde=200) best = None deadline = time.monotonic() + 180 try: def evaluate(ut, prograde): nonlocal best self.guard(deadline) node.ut, node.prograde = ut, max(0.0, prograde) time.sleep(.02) encounter = self.encounter(node.orbit, "Kerbin") if encounter: pe = encounter.periapsis_altitude score = abs(pe - 35000) + 2 * prograde if best is None or score < best[0]: best = (score, ut, prograde, pe) for i in range(37): for extra in (0, 50, 100, 150, 250): evaluate(base + period * i / 36, 200 + extra) if best is None: raise FlightError("Keine Kerbin-Rückkehrbahn gefunden; kein Schub ausgeführt.") for dt, dv in [(period / 36, 30), (period / 150, 8), (period / 600, 2)]: center = best for ti in range(-4, 5): for di in range(-4, 5): evaluate(center[1] + ti * dt / 4, center[2] + di * dv / 4) node.ut, node.prograde = best[1:3] if not 20000 < best[3] < 500000: raise FlightError("Geplante Kerbin-Passage nicht im sicheren Höhenband.") self.emit("return_plan", delta_v=node.delta_v, kerbin_periapsis=best[3]) except BaseException: node.remove() raise return node def return_coast(self): transition = self.v.orbit.time_to_soi_change if not math.isfinite(transition) or transition <= 0: raise FlightError("Kein gültiger Kerbin-SOI-Wechsel.") self.warp(self.sc.ut + transition + 5, "Mun") if self.v.orbit.body.name != "Kerbin": raise FlightError("Kerbin-Einflussbereich nicht erreicht.") self.save("Kerbinankunft") def reentry_prep(self): encounter = self.encounter(self.v.orbit, "Kerbin") if encounter and 20000 < encounter.periapsis_altitude < 45000: self.save("sicherer-Kerbinkurs") return node = self.v.control.add_node(self.sc.ut + 360, prograde=0) best = None deadline = time.monotonic() + 120 try: for step in (20, 5, 1, .2): center = (0, 0) if best is None else best[1:3] for i in range(-4, 5): for j in range(-4, 5): self.guard(deadline) pro, radial = center[0] + i * step, center[1] + j * step node.prograde, node.radial = pro, radial time.sleep(.02) encounter = self.encounter(node.orbit, "Kerbin") if encounter: pe = encounter.periapsis_altitude score = abs(pe - 32000) + 10 * math.hypot(pro, radial) if best is None or score < best[0]: best = (score, pro, radial, pe) if best is None or not 20000 < best[3] < 45000: raise FlightError("Keine sichere Kerbin-Wiedereintrittsbahn gefunden.") node.prograde, node.radial = best[1:3] self.emit("reentry_correction", delta_v=node.delta_v, periapsis=best[3]) except BaseException: node.remove() raise self.burn(node) encounter = self.encounter(self.v.orbit, "Kerbin") if not encounter or not 15000 < encounter.periapsis_altitude < 60000: raise FlightError("Reale Bahn nach Korrektur nicht sicher; kein Weiter-Warp.") self.save("sicherer-Kerbinkurs") def separate_capsule(self): """Drop the spent Mun lander; only the capsule/heat shield returns. Deliberately bypasses stage()'s "never below 2" guard -- this is the one place in the mission where dropping stage 2 is correct.""" v = self.v if v.orbit.body.name != "Kerbin" or v.control.current_stage != 2: raise FlightError("Kapseltrennung nur bei Kerbin-Rückkehr aus Stufe 2 vorgesehen.") if not 15000 < v.orbit.periapsis_altitude < 100000: raise FlightError("Kapseltrennung nur mit sicherer Kerbin-Periapsis (15-100 km).") v.control.throttle = 0 v.auto_pilot.engaged = False self.save("vor-Kapseltrennung") self._activate_stage(2) def arm_parachute(self): """Deliberately bypasses stage()'s guard the same way; only valid low and slow enough in the atmosphere for a stock chute to survive opening.""" v = self.v if v.control.current_stage != 1: raise FlightError("Fallschirm-Scharfschaltung nur aus Stufe 1 vorgesehen.") f = v.flight(v.orbit.body.reference_frame) if f.surface_altitude > 9000 or f.mach > 1.5: raise FlightError("Zu hoch/schnell für die Fallschirm-Scharfschaltung.") self._activate_stage(1) def reentry(self): v, b = self.v, self.v.orbit.body if b.name != "Kerbin": raise FlightError("Wiedereintritt braucht Kerbin als Zielkörper.") o = v.orbit if not 20000 < o.periapsis_altitude < 45000: raise FlightError("Periapsis nicht im sicheren Wiedereintrittsfenster (20-45 km).") # 26.09.2026: hold retrograde from here on -- BEFORE separating the capsule # and BEFORE the long coast to periapsis, not just right before the final # descent loop. Two Kerbals were lost when a mid-sequence abort (a warp() # safety trip) left the freshly-separated capsule with no attitude hold at # all while it kept falling in real time, heat shield not forward, with no # parachute armed either. sas=True alone only freezes whatever attitude it # last had; sas_mode has to be set explicitly to actually turn it. v.auto_pilot.engaged = False v.control.rcs = False v.control.sas = True v.control.sas_mode = self.sc.SASMode.retrograde if v.control.current_stage == 2: self.separate_capsule() if v.control.current_stage != 1: raise FlightError("Erwartet Rückkehrkapsel in Stufe 1 vor dem Wiedereintritt.") # Re-assert: separation can reset SAS mode on the newly split-off vessel. v.control.sas = True v.control.sas_mode = self.sc.SASMode.retrograde ttp = v.orbit.time_to_periapsis if ttp > 90: # A refused/aborted warp here (already low, already close, a # transient reading) is not fatal for reentry() the way it is for # an outbound burn: we were heading into the atmosphere anyway, so # just fall through into real-time monitoring instead of losing # the capsule to an uncaught exception with SAS never re-checked. try: self.warp(self.sc.ut + ttp - 90, "Kerbin") except FlightError as exc: self.emit("reentry_warp_skipped", reason=str(exc)) v.control.sas = True v.control.sas_mode = self.sc.SASMode.retrograde # 26.09.2026: a shallow-enough entry does not always capture/land on the # first pass -- it can skip back out to a much smaller apoapsis (seen: # 238 km after the first atmospheric brush) and need another orbit or # two to bleed off the rest. The capsule is safe and passive while # coasting, so give this real patience instead of the tight budget a # single-pass, actively-thrusting phase would need. deadline = time.monotonic() + 10800 f = v.flight(b.reference_frame) landed_states = {self.sc.VesselSituation.landed, self.sc.VesselSituation.splashed} while v.situation not in landed_states: self.guard(deadline) # Climbed back out after a first, incomplete atmospheric pass: warp # the rest of that climb-and-fall-back away instead of monitoring it # in real time, same safe/non-fatal warp as the approach above. if (v.control.current_stage == 1 and b.has_atmosphere and f.mean_altitude > b.atmosphere_depth + 5000 and v.orbit.time_to_periapsis > 60): try: self.warp(self.sc.ut + v.orbit.time_to_periapsis - 60, "Kerbin") except FlightError as exc: self.emit("reentry_warp_skipped", reason=str(exc)) if v.control.current_stage == 1 and f.surface_altitude < 9000 and f.mach < 1.5: self.arm_parachute() self.report() time.sleep(.2) self.save("Kerbinlandung") self.emit("recovered", latitude=f.latitude, longitude=f.longitude, remaining_delta_v=self.reserve(), **self.snapshot()) def run(self, goal): if self.v.control.nodes: raise FlightError("Vorhandene Manöverknoten zuerst prüfen/entfernen. " "Sie werden nicht stillschweigend überschrieben.") self.sc.rails_warp_factor = self.sc.physics_warp_factor = 0 self.v.control.throttle = 0 for _ in range(14): phase = mission_phase(self.state()) self.emit("phase", phase=phase) if phase == "recovered": self.report(force=True) return if phase == "landed": if goal != "return": self.report(force=True) return self.ascend() elif phase == "launch": self.launch() elif phase == "transfer": self.transfer() elif phase == "coast": self.coast() elif phase == "capture": self.capture() elif phase == "land": if goal == "orbit": self.report(force=True) return self.land() if goal != "return": return elif phase == "descent": if goal == "orbit": raise FlightError("Bereits im Abstieg; Ziel Orbit ist hier nicht unterstützt.") self.land(resume_descent=True) if goal != "return": return elif phase == "return_transfer": self.return_transfer() elif phase == "return_coast": self.return_coast() elif phase == "reentry_prep": self.reentry_prep() elif phase == "reentry": self.reentry() return raise FlightError("Zu viele Phasenwechsel; Mission angehalten.") def stop(self): # Best effort only: an unavailable server cannot accept a throttle cutoff. # 26.09.2026: sas=True alone just freezes whatever attitude the vessel had # -- after an abort mid-return-leg that can mean tumbling, heat-shield-away, # toward Kerbin with nothing correcting it. Two Kerbals were lost this way # when a mission aborted after capsule separation and nothing held the # capsule retrograde while it kept falling in real time. Every abort, not # just the reentry ones, should leave the vessel holding retrograde. for action in (lambda: setattr(self.v.control, "throttle", 0), lambda: setattr(self.sc, "rails_warp_factor", 0), lambda: setattr(self.sc, "physics_warp_factor", 0), lambda: setattr(self.v.auto_pilot, "engaged", False), lambda: setattr(self.v.control, "sas", True), lambda: setattr(self.v.control, "sas_mode", self.sc.SASMode.retrograde)): try: action() except Exception as exc: print(f"Abschaltung nicht bestätigt: {exc}", flush=True) def parse_args(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--execute", action="store_true", help="Flug tatsächlich steuern") parser.add_argument("--goal", choices=("orbit", "land", "return"), default="orbit", help="orbit: Mün-Umlaufbahn (Standard); land: experimentelle Landung; " "return: Landung plus Rückflug zur Kerbin-Landung (experimentell)") parser.add_argument("--address", default="127.0.0.1") parser.add_argument("--rpc-port", type=int, default=50000) parser.add_argument("--stream-port", type=int, default=50001) parser.add_argument("--load-checkpoint", help="Diesen KSP-Spielstand ausdrücklich laden") args = parser.parse_args(argv) if args.load_checkpoint and not args.execute: parser.error("--load-checkpoint benötigt --execute; Standardmodus liest nur.") return args def main(argv=None): args = parse_args(argv) try: client_version = importlib.metadata.version("krpc") except importlib.metadata.PackageNotFoundError as exc: raise FlightError("kRPC fehlt in dieser Python-Umgebung. Auf Pop!_OS: " "source .venv-popos/bin/activate; " "python -m pip install -r requirements.txt") from exc if client_version != "0.6.0": raise FlightError("Python-Client 0.6.0 erforderlich: python -m pip install -r requirements.txt") with socket.create_connection((args.address, args.rpc_port), timeout=3): pass import krpc conn = krpc.connect(name="Munfalke Autoflight" if args.execute else "Munfalke Status", address=args.address, rpc_port=args.rpc_port, stream_port=args.stream_port) try: if conn.krpc.get_status().version != "0.6.0": raise FlightError("Das Skript benötigt kRPC-Server 0.6.0.") if not args.execute: mission = Mission(conn) mission.report(force=True) try: mission.emit("detected_phase", phase=mission_phase(mission.state())) except FlightError as exc: mission.emit("unsupported_state", reason=str(exc)) return 0 with flight_lock(ROOT / "logs/munfalke-autoflight.lock"): if args.load_checkpoint: conn.space_center.load(args.load_checkpoint) time.sleep(3) stamp = datetime.now().strftime("%Y%m%d-%H%M%S") mission = Mission(conn, ROOT / f"logs/munfalke-autoflight-{stamp}.jsonl") try: mission.run(args.goal) finally: mission.stop() finally: conn.close() return 0 if __name__ == "__main__": def interrupt(signum, frame): raise KeyboardInterrupt signal.signal(signal.SIGTERM, interrupt) try: raise SystemExit(main()) except KeyboardInterrupt: print("Flugsteuerung unterbrochen; Abschaltung wurde versucht.") raise SystemExit(130) except (FlightError, OSError) as exc: print(f"Nicht gestartet / angehalten: {exc}") raise SystemExit(1)