pali112
00a771c7e0
- monitor w flight loop (co 1s) sprawdza wyzwalacze - wyzwalacze: zmiana fazy (start/lądowanie), bliski ruch <5nm, nowe lotnisko w przylocie - cooldown konfigurowalny (domyślnie 30s) zapobiega spamowaniu - ATC inicjuje wiadomość w aktualnej roli (bez wiadomości pilota) - checkbox + slider w oknie konfiguracji
1470 lines
59 KiB
Python
1470 lines
59 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
XSayATC - lokalny asystent ATC dla X-Plane 12 (llama.cpp)
|
|
|
|
Podlacza sie do lokalnego serwera llama.cpp (OpenAI-compat /v1/chat/completions),
|
|
pobiera kontekst lotu z datarefow, wysyla Twoja wiadomosc (na razie tekstowa),
|
|
i streamuje odpowiedz ATC do okna widocznego rowniez w VR.
|
|
|
|
Wymaga: XPPython3 (dostarcza wlasnego Pythona 3.12 + imgui + requests).
|
|
|
|
Konfiguracja zapisywana jest w config.json obok tego pliku.
|
|
|
|
WAZNE (bezpieczenstwo watkow):
|
|
- Odczyty datarefow (xp.getData*) MUSZA dzialac na glownym watku X-Plane.
|
|
- Watek roboczy robi WYLACZNIE siec (requests) i aktualizuje self.messages.
|
|
- Kontekst lotu zbierany jest na glownym watku i przekazywany do watku.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
import subprocess
|
|
import time
|
|
|
|
from XPPython3 import xp
|
|
from XPPython3.xp_imgui import Window as ImguiWindow
|
|
|
|
import imgui
|
|
import requests
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Konfiguracja i domyslne wartosci
|
|
# --------------------------------------------------------------------------
|
|
|
|
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
|
|
|
DEFAULT_CONFIG = {
|
|
"base_url": "http://localhost:8081", # adres serwera llama.cpp
|
|
"model": "", # puste = pobierze pierwszy z /v1/models
|
|
"temperature": 0.7,
|
|
"max_tokens": 512,
|
|
"system_prompt": (
|
|
"You are a concise, professional air traffic controller in X-Plane 12. "
|
|
"Give short, clear, realistic ATC instructions. Use standard phraseology. "
|
|
"You have access to the pilot's current flight context (position, altitude, "
|
|
"speed, heading, transponder, radio frequencies, nearest airport, weather) "
|
|
"and to nearby traffic. "
|
|
"IMPORTANT: adopt the role of the facility the pilot is currently tuned to "
|
|
"(the 'Tuned facility' field): Delivery/Clearance gives IFR/VFR clearances and "
|
|
"squawk codes; Ground issues taxi instructions; Tower controls takeoff, landing "
|
|
"and the active runway; Tracon/Approach handles arrivals, departures and "
|
|
"vectors; Center handles enroute. If 'Tuned facility' is 'None', act as a "
|
|
"generic ATC using the active COM frequency. "
|
|
"HANDOFF RULE (very important): when you finish with the pilot, you MUST hand "
|
|
"them off to the next facility exactly like real ATC, using standard phraseology "
|
|
"such as 'contact Ground on 121.900', 'contact Tower on 118.300', or 'frequency "
|
|
"change approved'. Use the 'Airport frequencies' list to give the correct "
|
|
"frequency for the 'Next expected facility'. If the next facility has several "
|
|
"frequencies, pick the most plausible one and state it clearly. "
|
|
"DIRECTION: use the 'Flight phase' field. On DEPARTURE the flow is Delivery->"
|
|
"Ground->Tower->Departure->Center (so Tower hands off to Departure). On ARRIVAL "
|
|
"the flow is Center->Approach->Tower->Ground (so Tower hands off to Ground). "
|
|
"Never hand off to yourself (e.g. Tower must NOT say 'contact Tower'). "
|
|
"Use the nearby traffic list when issuing traffic advisories "
|
|
"(e.g. bearing in clock positions and distance). "
|
|
"Respond with ATC instructions only, no extra commentary."
|
|
),
|
|
"include_flight_context": True,
|
|
"no_think": True, # dodaje /no_think do promptu (modele Qwen3 z reasoningiem)
|
|
"window_always_on_top": True,
|
|
# --- Glos (whisper / piper) ---
|
|
"voice_enabled": True,
|
|
"voice_ptt_mode": "hold", # "hold" = przytrzymaj PTT, mowisz, puszczasz; "toggle" = wcisnij aby zaczac/konczyc
|
|
"whisper_cli": "/usr/bin/whisper-cli",
|
|
"whisper_model": "/home/pali112/models/ggml-small.bin",
|
|
"whisper_language": "auto",
|
|
"record_cmd": ["pw-record", "--format=s16", "--rate=16000", "--channels=1"],
|
|
"piper": "/opt/piper-tts/piper",
|
|
"piper_model": "/home/pali112/.local/share/piper-voices/en_US-lessac-medium.onnx",
|
|
"play_cmd": ["pw-play"],
|
|
"tts_enabled": True,
|
|
# --- Plan lotu ---
|
|
# Fallback dla samolotow z wlasnym FMC (Toliss/Zibo), ktore nie zapisuja
|
|
# planu do natywnego FMS X-Plane. Wskaz plik tekstowy; kazda linia to jeden
|
|
# punkt: "NAZWA [WYSOKOSC_stop]" (wysokosc opcjonalna), '#' = komentarz.
|
|
"flightplan_file": "",
|
|
# --- Procedury ATC ---
|
|
# Katalog z plikami procedur. Puste = <katalog_wtyczki>/procedures.
|
|
# _global.txt = zawsze dolaczane; <ICAO>.txt = per-lotnisko.
|
|
"procedures_dir": "",
|
|
"use_procedures": True,
|
|
# --- Proaktywny ATC ---
|
|
# Gdy wlaczone, ATC sam inicjuje kontakt (np. traffic advisory, zmiana fazy
|
|
# lotu), zamiast tylko odpowiadac na Twoje wiadomosci.
|
|
"proactive_enabled": True,
|
|
# Minimalny odstęp (sekundy) miedzy dwoma proaktywnymi komunikatami.
|
|
"proactive_cooldown": 30,
|
|
}
|
|
|
|
# Dataref -> (nazwa kontekstu, mnoznik, jednostka)
|
|
CONTEXT_DATAREFS = [
|
|
("sim/flightmodel/position/latitude", "Latitude", 1.0, "deg"),
|
|
("sim/flightmodel/position/longitude", "Longitude", 1.0, "deg"),
|
|
("sim/flightmodel/misc/h_ind", "Alt (baro)", 1.0, "ft"),
|
|
("sim/flightmodel/position/indicated_airspeed", "IAS", 1.0, "kt"),
|
|
("sim/flightmodel/position/groundspeed", "GS", 1.943844, "kt"),
|
|
("sim/flightmodel/position/mag_psi", "HDG (mag)", 1.0, "deg"),
|
|
("sim/cockpit/radios/transponder_code", "Squawk", 1.0, ""),
|
|
("sim/cockpit/radios/com1_freq_hz", "COM1", 0.01, "MHz"),
|
|
("sim/cockpit/radios/com2_freq_hz", "COM2", 0.01, "MHz"),
|
|
("sim/cockpit/radios/nav1_freq_hz", "NAV1", 0.01, "MHz"),
|
|
]
|
|
|
|
TAILNUM_DATAREF = "sim/aircraft/view/acf_tailnum"
|
|
VR_ENABLED_DATAREF = "sim/graphics/VR/enabled"
|
|
TUNED1_DATAREF = "sim/atc/com1_tuned_facility"
|
|
TUNED2_DATAREF = "sim/atc/com2_tuned_facility"
|
|
ELEVATION_DATAREF = "sim/flightmodel/position/elevation"
|
|
|
|
# Nowoczesne datarefy radiowe (cockpit2/radios/actuators). Samoloty z wlasnym
|
|
# systemem radiowym (Garmin / REP / Toliss / Zibo) czesto zapisuja czestotliwosc
|
|
# TU, a legacy sim/cockpit/radios/*_freq_hz zostaje 0. Uzywamy jako fallback.
|
|
RADIO_MODERN_DATAREFS = {
|
|
"sim/cockpit/radios/com1_freq_hz": "sim/cockpit2/radios/actuators/com1_frequency_hz",
|
|
"sim/cockpit/radios/com2_freq_hz": "sim/cockpit2/radios/actuators/com2_frequency_hz",
|
|
"sim/cockpit/radios/nav1_freq_hz": "sim/cockpit2/radios/actuators/nav1_frequency_hz",
|
|
}
|
|
|
|
# Mapa typu sluzby ATC (sim/atc/*_tuned_facility)
|
|
FACILITY_NAMES = {
|
|
0: "None",
|
|
1: "Delivery/Clearance",
|
|
2: "Ground",
|
|
3: "Tower",
|
|
4: "Tracon/Approach-Departure",
|
|
5: "Center",
|
|
}
|
|
|
|
# Datarefy TCAS (ruch z LiveTraffic / AI / multiplayer)
|
|
# TCAS daje gotowe: namiar wzgledny, dystans, wysokosc wzgledna.
|
|
TCAS_BEARING_DATAREF = "sim/cockpit2/tcas/indicators/relative_bearing_degs"
|
|
TCAS_DISTANCE_DATAREF = "sim/cockpit2/tcas/indicators/relative_distance_mtrs"
|
|
TCAS_ALTITUDE_DATAREF = "sim/cockpit2/tcas/indicators/relative_altitude_mtrs"
|
|
TCAS_NUM_ACF_DATAREF = "sim/cockpit2/tcas/indicators/tcas_num_acf"
|
|
TCAS_FLIGHT_ID_DATAREF = "sim/cockpit2/tcas/targets/flight_id"
|
|
TCAS_ICAO_TYPE_DATAREF = "sim/cockpit2/tcas/targets/icao_type"
|
|
|
|
# Ile slotow TCAS sprawdzac (X-Plane ma tablice float[64])
|
|
NUM_TCAS_SLOTS = 64
|
|
|
|
# Typy sluzb ATC w plikach scenery apt.dat (kolumna 1, wiersze 1050-1056)
|
|
# 1050=ATIS/AWOS/ASOS, 1051=UNICOM/CTAF, 1052=Delivery/Clearance,
|
|
# 1053=Ground, 1054=Tower, 1055=Approach, 1056=Departure
|
|
APT_FREQ_CODES = {
|
|
"1050": "ATIS",
|
|
"1051": "UNICOM/CTAF",
|
|
"1052": "Delivery/Clearance",
|
|
"1053": "Ground",
|
|
"1054": "Tower",
|
|
"1055": "Approach",
|
|
"1056": "Departure",
|
|
}
|
|
|
|
# Sluzby ATC, ktore powinny przy przelaczeniu podac nastepna czestotliwosc.
|
|
# Dwie sciezki zalezne od kierunku lotu (odlot / przylot).
|
|
# Odlot: Delivery -> Ground -> Tower -> Departure -> Center
|
|
# Przylot: Center -> Approach -> Tower -> Ground
|
|
NEXT_FACILITY_DEPARTURE = {
|
|
"Delivery/Clearance": "Ground",
|
|
"Ground": "Tower",
|
|
"Tower": "Departure",
|
|
"Tracon/Approach-Departure": "Center",
|
|
"Center": None,
|
|
}
|
|
|
|
NEXT_FACILITY_ARRIVAL = {
|
|
"Delivery/Clearance": "Ground",
|
|
"Ground": None,
|
|
"Tower": "Ground",
|
|
"Tracon/Approach-Departure": "Tower",
|
|
"Center": "Approach",
|
|
}
|
|
|
|
ELEVATION_AGL_DATAREF = "sim/flightmodel/position/y_agl"
|
|
|
|
|
|
class XSayATC:
|
|
def __init__(self):
|
|
self.config = self._load_config()
|
|
self.datarefs = {} # name -> dataref object
|
|
self._resolved_datarefs = False
|
|
self._tailnum_ref = None
|
|
self._vr_ref = None
|
|
self._modern_radio_refs = {}
|
|
|
|
# Stan rozmowy
|
|
self.messages = [] # lista slownikow {"role": "user"/"assistant"/"system", "text": str}
|
|
self.pending_text = "" # tekst w polu wprowadzania
|
|
self.status = "Gotowy."
|
|
self.streaming = False
|
|
|
|
self._model_cache = None
|
|
self._stop_event = threading.Event()
|
|
self._worker = None
|
|
|
|
# Okna ImGui
|
|
self._imgui_window = None
|
|
self._config_window = None
|
|
|
|
# Menu
|
|
self._menu = None
|
|
self._menu_open_idx = None
|
|
self._menu_cfg_idx = None
|
|
|
|
# Cache wyswietlania czestotliwosci w oknie glownym
|
|
self._freq_display_cache = None # (icao, [linie]) lub None
|
|
self._freq_display_ts = 0.0
|
|
|
|
# VR
|
|
self._last_vr = None
|
|
|
|
# Glos (STT / TTS)
|
|
self._recording = False
|
|
self._record_proc = None
|
|
self._ptt_toggled = False # dla trybu "toggle"
|
|
|
|
# Sluzba ATC, na ktora bylismy ostatnio nastrojeni (do resetu rozmowy)
|
|
self._last_facility = None
|
|
|
|
# Kierunek lotu (odlot / przylot) — wybierany recznie przez pilota.
|
|
# Determinuje handoff Tower/Ground/Approach/Departure.
|
|
self.flight_phase = "departure" # "departure" | "arrival"
|
|
|
|
# Proaktywny ATC (sam inicjuje kontakt)
|
|
self._proactive_last_ts = 0.0 # timestamp ostatniego proaktywnego komunikatu
|
|
self._proactive_last_phase = None # ostatnia znana faza (ground/airborne)
|
|
self._proactive_announced_airports = set() # lotniska, na ktorych juz powitalismy/przegadalismy
|
|
|
|
# Czestotliwosci lotnisk z apt.dat (leniwie wczytywane w tle)
|
|
self._apt_freqs = {} # ICAO -> {kategoria: [freq_khz, ...]}
|
|
self._apt_loading = False
|
|
self._apt_loaded = False
|
|
self._elevation_agl_ref = None
|
|
# Cache najblizszego lotniska (czasowy, zeby nie iterowac navaidow co klatke)
|
|
self._nearest_icao_cache = None # (icao, timestamp)
|
|
|
|
# ---------------------------- Config ----------------------------
|
|
def _load_config(self):
|
|
cfg = dict(DEFAULT_CONFIG)
|
|
try:
|
|
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if isinstance(data, dict):
|
|
cfg.update(data)
|
|
except Exception:
|
|
pass
|
|
return cfg
|
|
|
|
def _save_config(self):
|
|
try:
|
|
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(self.config, f, ensure_ascii=False, indent=2)
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad zapisu config.json: {e}")
|
|
|
|
# ---------------------------- Datarefs ----------------------------
|
|
def _ensure_datarefs(self):
|
|
if self._resolved_datarefs:
|
|
return
|
|
for name, _label, _mult, _unit in CONTEXT_DATAREFS:
|
|
self.datarefs[name] = xp.findDataRef(name)
|
|
self._tailnum_ref = xp.findDataRef(TAILNUM_DATAREF)
|
|
self._vr_ref = xp.findDataRef(VR_ENABLED_DATAREF)
|
|
self._tuned1_ref = xp.findDataRef(TUNED1_DATAREF)
|
|
self._tuned2_ref = xp.findDataRef(TUNED2_DATAREF)
|
|
self._elevation_ref = xp.findDataRef(ELEVATION_DATAREF)
|
|
self._elevation_agl_ref = xp.findDataRef(ELEVATION_AGL_DATAREF)
|
|
# Datarefy TCAS (ruch z LiveTraffic / AI / multiplayer)
|
|
self._tcas_bearing_ref = xp.findDataRef(TCAS_BEARING_DATAREF)
|
|
self._tcas_distance_ref = xp.findDataRef(TCAS_DISTANCE_DATAREF)
|
|
self._tcas_altitude_ref = xp.findDataRef(TCAS_ALTITUDE_DATAREF)
|
|
self._tcas_num_ref = xp.findDataRef(TCAS_NUM_ACF_DATAREF)
|
|
self._tcas_flightid_ref = xp.findDataRef(TCAS_FLIGHT_ID_DATAREF)
|
|
self._tcas_icao_ref = xp.findDataRef(TCAS_ICAO_TYPE_DATAREF)
|
|
# Nowoczesne datarefy radiowe jako fallback
|
|
self._modern_radio_refs = {
|
|
name: xp.findDataRef(modern)
|
|
for name, modern in RADIO_MODERN_DATAREFS.items()
|
|
}
|
|
self._resolved_datarefs = True
|
|
|
|
def _get_dataref(self, name, is_float=True):
|
|
d = self.datarefs.get(name)
|
|
if d is None:
|
|
return None
|
|
try:
|
|
if is_float:
|
|
return xp.getDataf(d)
|
|
return xp.getDatai(d)
|
|
except Exception:
|
|
return None
|
|
|
|
def _get_radio_hz(self, name):
|
|
"""Czyta czestotliwosc radia (int, 10 kHz). Najpierw legacy
|
|
sim/cockpit/radios/*_freq_hz, a jesli 0 -> nowoczesny cockpit2."""
|
|
legacy = self.datarefs.get(name)
|
|
legacy_v = 0
|
|
try:
|
|
if legacy is not None:
|
|
legacy_v = xp.getDatai(legacy)
|
|
except Exception:
|
|
legacy_v = 0
|
|
modern_ref = self._modern_radio_refs.get(name)
|
|
modern_v = 0
|
|
if modern_ref is not None:
|
|
try:
|
|
modern_v = xp.getDatai(modern_ref)
|
|
except Exception:
|
|
modern_v = 0
|
|
if not legacy_v and not modern_v:
|
|
xp.log(f"[XSayATC] radio {name}: legacy i modern = 0 (radio wylaczone?)")
|
|
return legacy_v or modern_v
|
|
|
|
def _read_string(self, dref):
|
|
try:
|
|
return xp.getDatas(dref)
|
|
except Exception:
|
|
return None
|
|
|
|
# ---------------------------- Kontekst lotu (TYLKO glowny watek) ----------------------------
|
|
def _collect_context(self):
|
|
"""Zbiera kontekst lotu. WYWOLYWAC TYLKO na glownym watku."""
|
|
ctx = {}
|
|
lines = []
|
|
INT_RADIO_DATAREFS = (
|
|
"sim/cockpit/radios/transponder_code",
|
|
"sim/cockpit/radios/com1_freq_hz",
|
|
"sim/cockpit/radios/com2_freq_hz",
|
|
"sim/cockpit/radios/nav1_freq_hz",
|
|
)
|
|
RADIO_FREQ_DATAREFS = (
|
|
"sim/cockpit/radios/com1_freq_hz",
|
|
"sim/cockpit/radios/com2_freq_hz",
|
|
"sim/cockpit/radios/nav1_freq_hz",
|
|
)
|
|
for name, label, mult, unit in CONTEXT_DATAREFS:
|
|
if name in RADIO_FREQ_DATAREFS:
|
|
# Czestotliwosc radia czytana z fallbackiem na cockpit2
|
|
v = self._get_radio_hz(name)
|
|
val = int(v) * mult
|
|
ctx[label] = val
|
|
lines.append(f"{label}: {val:.3f} {unit}".strip())
|
|
continue
|
|
|
|
v = self._get_dataref(name, is_float=(name not in INT_RADIO_DATAREFS))
|
|
if v is None:
|
|
continue
|
|
if name == "sim/cockpit/radios/transponder_code":
|
|
ctx[label] = int(v)
|
|
lines.append(f"{label}: {int(v)}")
|
|
else:
|
|
val = v * mult
|
|
ctx[label] = val
|
|
unit_s = (" " + unit) if unit else ""
|
|
lines.append(f"{label}: {val:.1f}{unit_s}")
|
|
|
|
tail = self._read_string(self._tailnum_ref)
|
|
if tail:
|
|
tail = tail.strip()
|
|
ctx["Callsign"] = tail
|
|
lines.insert(0, f"Callsign: {tail}")
|
|
|
|
# Sluzba ATC, na ktora jestes nastrojony
|
|
fac1 = self._get_tuned_facility(self._tuned1_ref)
|
|
fac2 = self._get_tuned_facility(self._tuned2_ref)
|
|
com1 = ctx.get("COM1")
|
|
com2 = ctx.get("COM2")
|
|
tuned = None
|
|
if fac1 and fac1 != "None":
|
|
tuned = (fac1, com1, 1)
|
|
elif fac2 and fac2 != "None":
|
|
tuned = (fac2, com2, 2)
|
|
if tuned:
|
|
name, freq, num = tuned
|
|
freq_s = f"{freq:.3f} MHz" if freq is not None else "? MHz"
|
|
ctx["TunedFacility"] = name
|
|
lines.append(f"Tuned facility: {name} ({freq_s} on COM{num})")
|
|
else:
|
|
ctx["TunedFacility"] = "None"
|
|
lines.append("Tuned facility: None (no active ATC service on COM1/COM2)")
|
|
|
|
# Faza lotu (na ziemi / w powietrzu) z AGL
|
|
agl_m = self._get_agl()
|
|
on_ground = (agl_m is not None and agl_m < 1.0)
|
|
ctx["Phase"] = "ground" if on_ground else "airborne"
|
|
lines.append("Phase: " + ("on ground (taxi/ramp)" if on_ground else "airborne"))
|
|
|
|
# Kierunek lotu (odlot / przylot) — wybrany recznie przez pilota
|
|
ctx["FlightPhase"] = self.flight_phase
|
|
lines.append("Flight phase (pilot selection): " +
|
|
("DEPARTURE" if self.flight_phase == "departure" else "ARRIVAL"))
|
|
|
|
# Czestotliwosci najblizszego lotniska (do podpowiedzi nastepnej czestotliwosci)
|
|
icao = self._nearest_airport_icao()
|
|
if icao:
|
|
ctx["NearestAirport"] = icao
|
|
lines.append(f"Nearest airport: {icao}")
|
|
freq_lines = self._format_airport_freqs(icao)
|
|
if freq_lines:
|
|
lines.append("Airport frequencies:")
|
|
lines.extend(freq_lines)
|
|
ctx["AirportFreqs"] = freq_lines
|
|
else:
|
|
lines.append("Airport frequencies: (not available)")
|
|
|
|
# Podpowiedz nastepna sluzbe, na ktora pilot ma sie przelaczyc
|
|
if tuned:
|
|
nxt = self._next_facility_hint(tuned[0])
|
|
if nxt:
|
|
ctx["NextFacility"] = nxt
|
|
lines.append(f"Next expected facility (per standard flow): {nxt}")
|
|
|
|
# Ruch lotniczy w poblizu
|
|
traffic_lines = self._collect_traffic()
|
|
ctx["Traffic"] = traffic_lines
|
|
if traffic_lines:
|
|
lines.append("Nearby traffic:")
|
|
lines.extend(traffic_lines)
|
|
else:
|
|
lines.append("Nearby traffic: none detected")
|
|
|
|
# Plan lotu z FMS/GPS
|
|
fp_lines = self._collect_flight_plan()
|
|
ctx["FlightPlan"] = fp_lines
|
|
if fp_lines:
|
|
lines.append("Flight plan (FMS/GPS):")
|
|
lines.extend(fp_lines)
|
|
|
|
return ctx, "\n".join(lines)
|
|
|
|
def _get_tuned_facility(self, ref):
|
|
if ref is None:
|
|
return None
|
|
try:
|
|
code = xp.getDatai(ref)
|
|
return FACILITY_NAMES.get(code, f"Unknown({code})")
|
|
except Exception:
|
|
return None
|
|
|
|
def _get_agl(self):
|
|
"""Zwraca wysokosc AGL w metrach (lub None)."""
|
|
if self._elevation_agl_ref is None:
|
|
return None
|
|
try:
|
|
return xp.getDataf(self._elevation_agl_ref)
|
|
except Exception:
|
|
return None
|
|
|
|
def _procedures_dir(self):
|
|
"""Zwraca sciezke katalogu procedur."""
|
|
if self.config.get("procedures_dir"):
|
|
return self.config["procedures_dir"]
|
|
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "procedures")
|
|
|
|
def _nearest_airport_icao(self, force=False):
|
|
"""Zwraca ICAO najblizszego lotniska (z natywnej bazy navaidow).
|
|
|
|
Skan calej bazy navaidow jest ciezki, wiec normalnie wynik jest
|
|
cache'owany (10 s). force=True wymusza ponowny skan."""
|
|
import time as _time
|
|
now = _time.time()
|
|
if not force and self._nearest_icao_cache is not None:
|
|
c_icao, c_ts = self._nearest_icao_cache
|
|
if now - c_ts < 10.0:
|
|
return c_icao
|
|
lat = self._get_dataref("sim/flightmodel/position/latitude", is_float=True)
|
|
lon = self._get_dataref("sim/flightmodel/position/longitude", is_float=True)
|
|
if lat is None or lon is None:
|
|
return None
|
|
try:
|
|
import math
|
|
best_id = None
|
|
best_dist = None
|
|
nav = xp.getFirstNavAid()
|
|
while nav is not None and nav != xp.NAV_NOT_FOUND:
|
|
info = xp.getNavAidInfo(nav)
|
|
if info is not None and getattr(info, "type", None) == xp.Nav_Airport:
|
|
dlat = info.latitude - lat
|
|
dlon = info.longitude - lon
|
|
dist = dlat * dlat + dlon * dlon
|
|
if best_dist is None or dist < best_dist:
|
|
best_dist = dist
|
|
best_id = getattr(info, "navAidID", None)
|
|
nav = xp.getNextNavAid(nav)
|
|
self._nearest_icao_cache = (best_id, now)
|
|
return best_id
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad szukania lotniska: {e}")
|
|
return None
|
|
|
|
def _read_procedure_file(self, path):
|
|
"""Czyta plik procedury, zwraca tekst (bez linii komentarza '#')."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
lines = [ln.rstrip() for ln in f if ln.strip() and not ln.lstrip().startswith("#")]
|
|
return "\n".join(lines)
|
|
except Exception:
|
|
return ""
|
|
|
|
def _collect_procedures(self):
|
|
"""Zbiera tekst procedur: _global.txt + <ICAO>.txt najblizszego lotniska."""
|
|
if not self.config.get("use_procedures", True):
|
|
return ""
|
|
d = self._procedures_dir()
|
|
if not os.path.isdir(d):
|
|
return ""
|
|
parts = []
|
|
g = os.path.join(d, "_global.txt")
|
|
if os.path.exists(g):
|
|
t = self._read_procedure_file(g)
|
|
if t:
|
|
parts.append(t)
|
|
icao = self._nearest_airport_icao()
|
|
if icao:
|
|
p = os.path.join(d, icao.upper() + ".txt")
|
|
if os.path.exists(p):
|
|
t = self._read_procedure_file(p)
|
|
if t:
|
|
parts.append(f"[Procedures for {icao.upper()}]\n" + t)
|
|
return "\n\n".join(parts)
|
|
|
|
# ---------------------------- Czestotliwosci lotnisk (apt.dat) ----------------------------
|
|
def _airport_freq_path(self):
|
|
"""Zwraca sciezke do apt.dat Global Airports (najwieksza baza lotnisk)."""
|
|
try:
|
|
xp_dir = xp.getSystemPath()
|
|
except Exception:
|
|
xp_dir = None
|
|
if xp_dir:
|
|
p = os.path.join(xp_dir, "Global Scenery", "Global Airports", "Earth nav data", "apt.dat")
|
|
if os.path.exists(p):
|
|
return p
|
|
# Fallback: zgadnij wzgledem tego pliku (Resources/plugins/PythonPlugins)
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
p = os.path.normpath(os.path.join(here, "..", "..", "..", "Global Scenery",
|
|
"Global Airports", "Earth nav data", "apt.dat"))
|
|
if os.path.exists(p):
|
|
return p
|
|
return None
|
|
|
|
def _ensure_apt_freqs(self):
|
|
"""Wczytuje czestotliwosci z apt.dat w tle (raz, leniwie)."""
|
|
if self._apt_loaded or self._apt_loading:
|
|
return
|
|
path = self._airport_freq_path()
|
|
if not path:
|
|
self._apt_loaded = True
|
|
xp.log("[XSayATC] Nie znaleziono apt.dat (Global Airports)")
|
|
return
|
|
self._apt_loading = True
|
|
threading.Thread(target=self._parse_apt_freqs, args=(path,), daemon=True).start()
|
|
|
|
def _parse_apt_freqs(self, path):
|
|
"""Watek roboczy: parsuje apt.dat i zapisuje self._apt_freqs."""
|
|
freqs = {}
|
|
try:
|
|
cur_icao = None
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
|
for line in f:
|
|
parts = line.split()
|
|
if not parts:
|
|
continue
|
|
code = parts[0]
|
|
# Wiersz naglowka lotniska: 1|16|17 ... ICAO Nazwa
|
|
if code in ("1", "16", "17") and len(parts) >= 5:
|
|
cur_icao = parts[4]
|
|
freqs.setdefault(cur_icao, {})
|
|
continue
|
|
# Wiersz czestotliwosci: 1050-1056 FREQ NAZWA
|
|
if code in APT_FREQ_CODES and len(parts) >= 2:
|
|
cat = APT_FREQ_CODES[code]
|
|
try:
|
|
freq_khz = int(parts[1])
|
|
except ValueError:
|
|
continue
|
|
if cur_icao:
|
|
freqs.setdefault(cur_icao, {}).setdefault(cat, []).append(freq_khz)
|
|
self._apt_freqs = freqs
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad parsowania apt.dat: {e}")
|
|
self._apt_freqs = {}
|
|
finally:
|
|
self._apt_loaded = True
|
|
self._apt_loading = False
|
|
|
|
def _airport_freqs(self, icao):
|
|
"""Zwraca slownik {kategoria: [freq_khz, ...]} dla lotniska (albo {})."""
|
|
self._ensure_apt_freqs()
|
|
if not icao:
|
|
return {}
|
|
return self._apt_freqs.get(icao.upper(), self._apt_freqs.get(icao, {}))
|
|
|
|
def _format_airport_freqs(self, icao):
|
|
"""Zwraca liste linii tekstowych z czestotliwosciami lotniska."""
|
|
freqs = self._airport_freqs(icao)
|
|
if not freqs:
|
|
return []
|
|
lines = []
|
|
order = ["ATIS", "Delivery/Clearance", "Ground", "Tower", "Departure", "Approach", "UNICOM/CTAF"]
|
|
for cat in order:
|
|
vals = freqs.get(cat)
|
|
if not vals:
|
|
continue
|
|
for f in vals:
|
|
lines.append(f" {cat}: {f / 1000.0:.3f} MHz")
|
|
# Ewentualne kategorie spoza listy (np. inne nazwy)
|
|
for cat, vals in freqs.items():
|
|
if cat in order:
|
|
continue
|
|
for f in vals:
|
|
lines.append(f" {cat}: {f / 1000.0:.3f} MHz")
|
|
return lines
|
|
|
|
def _next_facility_hint(self, name):
|
|
"""Zwraca nastepna sluzbe ATC wg kierunku lotu (odlot / przylot)."""
|
|
if not name:
|
|
return None
|
|
if self.flight_phase == "arrival":
|
|
return NEXT_FACILITY_ARRIVAL.get(name)
|
|
return NEXT_FACILITY_DEPARTURE.get(name)
|
|
|
|
def _freq_display_lines(self):
|
|
"""Zwraca (icao, [linie]) czestotliwosci z cache. NIE liczy nic — tylko
|
|
zwraca to, co bylo ostatnio policzone (przez _refresh_freq_display)."""
|
|
if self._freq_display_cache is None:
|
|
return (None, [])
|
|
return self._freq_display_cache
|
|
|
|
def _refresh_freq_display(self):
|
|
"""Recznie (przycisk) przelicza najblizsze lotnisko i jego czestotliwosci."""
|
|
import time as _time
|
|
icao = self._nearest_airport_icao(force=True)
|
|
lines = self._format_airport_freqs(icao) if icao else []
|
|
self._freq_display_cache = (icao, lines)
|
|
self._freq_display_ts = _time.time()
|
|
return self._freq_display_cache
|
|
|
|
def _read_float_array(self, ref, count):
|
|
"""Czyta tablice float z datarefu. Zwraca liste lub []."""
|
|
if ref is None:
|
|
return []
|
|
try:
|
|
values = []
|
|
xp.getDatavf(ref, values, 0, count)
|
|
return values
|
|
except Exception:
|
|
return []
|
|
|
|
def _read_packed_strings(self, ref, count, strlen=8):
|
|
"""Czyta byte[512] jako 64 stringow po strlen bajtow (7 znakow + NUL)."""
|
|
if ref is None:
|
|
return []
|
|
try:
|
|
raw = xp.getDatas(ref)
|
|
if not raw:
|
|
return []
|
|
out = []
|
|
for i in range(count):
|
|
chunk = raw[i * strlen:(i + 1) * strlen]
|
|
s = chunk.split("\x00", 1)[0].strip()
|
|
out.append(s)
|
|
return out
|
|
except Exception:
|
|
return []
|
|
|
|
def _collect_traffic(self):
|
|
"""Zbiera liste pobliskiego ruchu z TCAS (LiveTraffic / AI / multiplayer)."""
|
|
out = []
|
|
bearing = self._read_float_array(self._tcas_bearing_ref, NUM_TCAS_SLOTS)
|
|
distance = self._read_float_array(self._tcas_distance_ref, NUM_TCAS_SLOTS)
|
|
altitude = self._read_float_array(self._tcas_altitude_ref, NUM_TCAS_SLOTS)
|
|
flight_ids = self._read_packed_strings(self._tcas_flightid_ref, NUM_TCAS_SLOTS)
|
|
icao_types = self._read_packed_strings(self._tcas_icao_ref, NUM_TCAS_SLOTS)
|
|
|
|
if not distance:
|
|
return out
|
|
|
|
for i in range(NUM_TCAS_SLOTS):
|
|
if i >= len(distance):
|
|
break
|
|
dist_m = distance[i]
|
|
# 0 lub blisko 0 = pusty slot
|
|
if not dist_m or dist_m <= 0:
|
|
continue
|
|
|
|
dist_nm = dist_m / 1852.0
|
|
rel_alt_m = altitude[i] if i < len(altitude) else 0.0
|
|
rel_alt_ft = rel_alt_m * 3.28084
|
|
|
|
brg_rel = bearing[i] if i < len(bearing) else 0.0
|
|
brg_rel = brg_rel % 360.0
|
|
|
|
# pozycja zegarowa (12-godzinna) wzgledem namiaru wzglednego
|
|
clock = int(round((brg_rel / 30.0) % 12)) or 12
|
|
|
|
callsign = flight_ids[i] if i < len(flight_ids) else ""
|
|
icao = icao_types[i] if i < len(icao_types) else ""
|
|
|
|
tag = ""
|
|
if callsign:
|
|
tag += callsign
|
|
if icao:
|
|
tag += ((" " + icao) if tag else icao)
|
|
if tag:
|
|
tag = " (" + tag + ")"
|
|
|
|
out.append(
|
|
f" {dist_nm:.1f} nm, {clock}:00, alt {rel_alt_ft:+.0f} ft (brg {brg_rel:.0f}){tag}"
|
|
)
|
|
|
|
return out
|
|
|
|
def _collect_flight_plan(self):
|
|
"""Zbiera plan lotu z FMS/GPS lub z pliku fallback. TYLKO glowny watek."""
|
|
lines = []
|
|
# 1) Najpierw natywny FMS/GPS X-Plane
|
|
try:
|
|
n = xp.countFMSEntries()
|
|
except Exception:
|
|
n = 0
|
|
if n and n > 0:
|
|
try:
|
|
dest_idx = xp.getDestinationFMSEntry()
|
|
except Exception:
|
|
dest_idx = -1
|
|
for i in range(n):
|
|
try:
|
|
info = xp.getFMSEntryInfo(i)
|
|
except Exception:
|
|
continue
|
|
if info is None:
|
|
continue
|
|
navid = getattr(info, "navAidID", "") or ""
|
|
alt = getattr(info, "altitude", 0) or 0
|
|
marker = " [DEST]" if i == dest_idx else ""
|
|
alt_s = f", alt {int(alt)} ft" if alt else ""
|
|
if navid:
|
|
lines.append(f" {i}: {navid}{alt_s}{marker}")
|
|
else:
|
|
lat = getattr(info, "lat", None)
|
|
lon = getattr(info, "lon", None)
|
|
if lat is not None and lon is not None:
|
|
lines.append(f" {i}: lat/lon {lat:.4f}/{lon:.4f}{alt_s}{marker}")
|
|
return lines
|
|
|
|
# 2) Fallback: plik wskazany w konfiguracji (samoloty z wlasnym FMC)
|
|
fp_file = self.config.get("flightplan_file", "")
|
|
if not fp_file or not os.path.exists(fp_file):
|
|
return lines
|
|
try:
|
|
with open(fp_file, "r", encoding="utf-8") as f:
|
|
for i, raw in enumerate(f):
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
# "NAZWA [WYSOKOSC]" — oddziel nazwe od opcjonalnej wysokosci
|
|
parts = line.split()
|
|
alt = None
|
|
if len(parts) >= 2:
|
|
# ostatni token moze byc liczba (wysokosc w stopach)
|
|
try:
|
|
alt = int(parts[-1])
|
|
name = " ".join(parts[:-1])
|
|
except ValueError:
|
|
alt = None
|
|
name = " ".join(parts)
|
|
else:
|
|
name = " ".join(parts)
|
|
alt_s = f", alt {alt} ft" if alt else ""
|
|
lines.append(f" {i}: {name}{alt_s}")
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad odczytu pliku planu lotu: {e}")
|
|
return lines
|
|
|
|
# ---------------------------- Komunikacja z LLM ----------------------------
|
|
def _resolve_model(self):
|
|
if self._model_cache:
|
|
return self._model_cache
|
|
if self.config.get("model"):
|
|
self._model_cache = self.config["model"]
|
|
return self._model_cache
|
|
try:
|
|
r = requests.get(self.config["base_url"].rstrip("/") + "/v1/models", timeout=5)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
models = data.get("data") or []
|
|
if models:
|
|
self._model_cache = models[0].get("id")
|
|
return self._model_cache
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Nie mozna pobrac modelu: {e}")
|
|
return None
|
|
|
|
def _build_messages(self, user_text, flight_ctx_text, procedures_text, proactive_reason=None):
|
|
"""Buduje liste messages. flight_ctx_text i procedures_text musza byc
|
|
przekazane (bez odczytu datarefow / plikow na watku roboczym).
|
|
proactive_reason != None -> ATC ma sie odezwac sam (bez wiadomosci pilota)."""
|
|
msgs = []
|
|
sp = self.config.get("system_prompt", "")
|
|
if self.config.get("no_think", True) and sp and "/no_think" not in sp:
|
|
sp = sp + "\n/no_think"
|
|
msgs.append({"role": "system", "content": sp})
|
|
|
|
if self.config.get("include_flight_context", True) and flight_ctx_text:
|
|
msgs.append({"role": "system", "content": "Pilot's current flight context:\n" + flight_ctx_text})
|
|
|
|
if procedures_text:
|
|
msgs.append({"role": "system", "content": "Local ATC procedures:\n" + procedures_text})
|
|
|
|
for m in self.messages[-12:]:
|
|
msgs.append({"role": m["role"], "content": m["text"]})
|
|
|
|
if proactive_reason:
|
|
# ATC inicjuje kontakt — nie ma wiadomosci pilota, jest wyzwalacz.
|
|
msgs.append({"role": "system", "content":
|
|
"ATC PROACTIVE trigger: " + proactive_reason +
|
|
". You must now initiate contact with the pilot in your "
|
|
"current facility role, using standard phraseology."})
|
|
else:
|
|
msgs.append({"role": "user", "content": user_text})
|
|
return msgs
|
|
|
|
def _request(self, messages):
|
|
url = self.config["base_url"].rstrip("/") + "/v1/chat/completions"
|
|
model = self._resolve_model() or self.config.get("model") or "model"
|
|
payload = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"temperature": self.config.get("temperature", 0.7),
|
|
"max_tokens": self.config.get("max_tokens", 512),
|
|
"stream": True,
|
|
}
|
|
return requests.post(url, json=payload, stream=True, timeout=(10, 120))
|
|
|
|
def _run_stream(self, user_text, flight_ctx_text, procedures_text, proactive_reason=None):
|
|
"""Watek roboczy: TYLKO siec i aktualizacja self.messages. Bez xp.getData*."""
|
|
self.streaming = True
|
|
self.status = "ATC mysli..."
|
|
|
|
# Historia do UI
|
|
if proactive_reason:
|
|
# Brak wiadomosci pilota — ATC mowi sam. Zapisujemy tylko odpowiedz.
|
|
self.messages.append({"role": "assistant", "text": ""})
|
|
else:
|
|
self.messages.append({"role": "user", "text": user_text})
|
|
self.messages.append({"role": "assistant", "text": ""})
|
|
|
|
try:
|
|
messages = self._build_messages(user_text, flight_ctx_text, procedures_text, proactive_reason)
|
|
resp = self._request(messages)
|
|
resp.raise_for_status()
|
|
|
|
buf = ""
|
|
for raw in resp.iter_lines(decode_unicode=True):
|
|
if self._stop_event.is_set():
|
|
break
|
|
if not raw or not raw.startswith("data:"):
|
|
continue
|
|
data = raw[len("data:"):].strip()
|
|
if data == "[DONE]":
|
|
break
|
|
try:
|
|
chunk = json.loads(data)
|
|
except Exception:
|
|
continue
|
|
choices = chunk.get("choices") or []
|
|
if not choices:
|
|
continue
|
|
delta = choices[0].get("delta") or {}
|
|
content = delta.get("content")
|
|
if content:
|
|
buf += content
|
|
self.messages[-1]["text"] = buf
|
|
|
|
if self.messages[-1]["text"] == "":
|
|
self.messages[-1]["text"] = "(brak odpowiedzi)"
|
|
self.status = "Gotowy."
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad zapytania: {e}")
|
|
self.status = f"Blad: {e}"
|
|
if self.messages and self.messages[-1]["role"] == "assistant":
|
|
if not self.messages[-1]["text"]:
|
|
self.messages[-1]["text"] = f"(blad: {e})"
|
|
finally:
|
|
self.streaming = False
|
|
# Odczytaj odpowiedz glosem (TTS) w osobnym watku
|
|
if self.config.get("tts_enabled", True) and self.messages:
|
|
last_text = self.messages[-1].get("text", "")
|
|
if last_text and last_text not in ("(brak odpowiedzi)",):
|
|
threading.Thread(target=self._speak, args=(last_text,), daemon=True).start()
|
|
|
|
def send(self, text):
|
|
"""Wywolywane na glownym watku (przycisk / PTT). Zbiera kontekst tu, bo tu wolno."""
|
|
text = (text or "").strip()
|
|
if not text or self.streaming:
|
|
return
|
|
self._dispatch(text, None)
|
|
|
|
def _dispatch(self, user_text, proactive_reason):
|
|
"""Wspolna sciezka wysylania (tekst pilota lub wyzwalacz proaktywny).
|
|
Zbiera kontekst i procedury na glownym watku, potem startuje watek."""
|
|
if self.streaming:
|
|
return
|
|
|
|
# Zbierz kontekst lotu NA GLOWNYM WATKU
|
|
flight_ctx_text = None
|
|
_ctx = None
|
|
if self.config.get("include_flight_context", True):
|
|
try:
|
|
_ctx, flight_ctx_text = self._collect_context()
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad zbierania kontekstu: {e}")
|
|
flight_ctx_text = None
|
|
|
|
# Reset rozmowy przy zmianie sluzby ATC (Ground->Tower itp.),
|
|
# zeby model nie mieszal frazeologii poprzednich rol.
|
|
facility = _ctx.get("TunedFacility") if _ctx else None
|
|
if facility is not None and facility != self._last_facility:
|
|
self.messages = []
|
|
self.status = f"Przelaczono na: {facility}"
|
|
self._last_facility = facility
|
|
|
|
# Zbierz procedury ATC NA GLOWNYM WATKU (czytanie plikow/navaidow)
|
|
procedures_text = ""
|
|
try:
|
|
procedures_text = self._collect_procedures()
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad zbierania procedur: {e}")
|
|
procedures_text = ""
|
|
|
|
self._stop_event.clear()
|
|
self._worker = threading.Thread(
|
|
target=self._run_stream,
|
|
args=(user_text, flight_ctx_text, procedures_text, proactive_reason),
|
|
daemon=True)
|
|
self._worker.start()
|
|
|
|
def stop(self):
|
|
self._stop_event.set()
|
|
|
|
# ---------------------------- Glos (STT / TTS) ----------------------------
|
|
def _voice_enabled(self):
|
|
return bool(self.config.get("voice_enabled", True))
|
|
|
|
def _start_recording(self):
|
|
"""Rozpoczyna nagrywanie mikrofonu do pliku tymczasowego."""
|
|
if self._recording:
|
|
return
|
|
import tempfile
|
|
self._record_file = tempfile.mktemp(suffix=".wav", prefix="xsayatc_")
|
|
cmd = list(self.config.get("record_cmd", ["pw-record", "--format=s16", "--rate=16000", "--channels=1"]))
|
|
cmd.append(self._record_file)
|
|
try:
|
|
self._record_proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
self._recording = True
|
|
self.status = "Nagrywanie... (pusc PTT aby wyslac)"
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad nagrywania: {e}")
|
|
self._record_proc = None
|
|
|
|
def _stop_recording_and_send(self):
|
|
"""Konczy nagrywanie, transkrybuje whisperem i wysyla tekst."""
|
|
if not self._recording:
|
|
return
|
|
self._recording = False
|
|
if self._record_proc is not None:
|
|
try:
|
|
# SIGINT kaze pw-record poprawnie domknac plik WAV
|
|
self._record_proc.send_signal(2) # 2 = SIGINT
|
|
try:
|
|
self._record_proc.wait(timeout=3)
|
|
except Exception:
|
|
self._record_proc.kill()
|
|
self._record_proc.wait(timeout=2)
|
|
except Exception:
|
|
pass
|
|
self._record_proc = None
|
|
|
|
wav = getattr(self, "_record_file", None)
|
|
if not wav or not os.path.exists(wav):
|
|
self.status = "Blad: brak nagrania"
|
|
return
|
|
|
|
self.status = "Transkrybuje (whisper)..."
|
|
text = self._transcribe(wav)
|
|
try:
|
|
os.remove(wav)
|
|
except Exception:
|
|
pass
|
|
|
|
if text:
|
|
self.send(text)
|
|
else:
|
|
self.status = "Nie rozpoznano mowy."
|
|
|
|
def _transcribe(self, wav_path):
|
|
"""Uruchamia whisper-cli i zwraca rozpoznany tekst."""
|
|
import tempfile
|
|
out_file = tempfile.mktemp(suffix="", prefix="xsayatc_stt_")
|
|
cli = self.config.get("whisper_cli", "/usr/bin/whisper-cli")
|
|
model = self.config.get("whisper_model", "")
|
|
lang = self.config.get("whisper_language", "auto")
|
|
cmd = [cli, "-m", model, "-l", lang, "-otxt", "-of", out_file, "-f", wav_path, "-np"]
|
|
try:
|
|
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60)
|
|
txt_file = out_file + ".txt"
|
|
if os.path.exists(txt_file):
|
|
with open(txt_file, "r", encoding="utf-8") as f:
|
|
text = f.read().strip()
|
|
try:
|
|
os.remove(txt_file)
|
|
except Exception:
|
|
pass
|
|
return text
|
|
return ""
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad whisper: {e}")
|
|
return ""
|
|
|
|
def _clean_env(self):
|
|
"""Zwraca srodowisko bez LD_LIBRARY_PATH/LD_PRELOAD ze Steam, zeby
|
|
piper/pw-play uzywaly wlasnych bibliotek (inaczej Steam Runtime podklada
|
|
stara libespeak-ng i piper zwraca pusty plik)."""
|
|
env = dict(os.environ)
|
|
env.pop("LD_PRELOAD", None)
|
|
# Zostawiamy tylko sciezki systemowe, nie Steam Runtime.
|
|
ld = env.pop("LD_LIBRARY_PATH", "")
|
|
# Dopisujemy /opt/piper-tts na poczatek, by piper znalazl swoje libs.
|
|
env["LD_LIBRARY_PATH"] = "/opt/piper-tts"
|
|
return env
|
|
|
|
def _speak(self, text):
|
|
"""Synteza mowy (piper) i odtworzenie. Watek roboczy."""
|
|
if not self.config.get("tts_enabled", True):
|
|
xp.log("[XSayATC] TTS wylaczone.")
|
|
return
|
|
text = (text or "").strip()
|
|
if not text:
|
|
return
|
|
piper = self.config.get("piper", "/opt/piper-tts/piper")
|
|
model = self.config.get("piper_model", "")
|
|
import tempfile
|
|
wav = tempfile.mktemp(suffix=".wav", prefix="xsayatc_tts_")
|
|
env = self._clean_env()
|
|
try:
|
|
xp.log(f"[XSayATC] TTS: piper -> {wav}")
|
|
p = subprocess.Popen(
|
|
[piper, "-m", model, "-f", wav],
|
|
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
|
|
env=env)
|
|
_out, err = p.communicate(input=text.encode("utf-8"), timeout=60)
|
|
if err:
|
|
xp.log(f"[XSayATC] TTS piper stderr: {err.decode('utf-8', 'ignore')[:200]}")
|
|
if os.path.exists(wav) and os.path.getsize(wav) > 0:
|
|
play_cmd = list(self.config.get("play_cmd", ["pw-play"]))
|
|
xp.log(f"[XSayATC] TTS: odtwarzam: {play_cmd} {wav}")
|
|
r = subprocess.run(play_cmd + [wav], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=60, env=env)
|
|
if r.returncode != 0:
|
|
xp.log(f"[XSayATC] TTS odtwarzanie kod={r.returncode} stderr={r.stderr.decode('utf-8','ignore')[:200]}")
|
|
try:
|
|
os.remove(wav)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
xp.log(f"[XSayATC] TTS: brak pliku WAV (rozmiar 0 lub nie istnieje)")
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad TTS: {e}")
|
|
|
|
# ---------------------------- Proaktywny ATC ----------------------------
|
|
def _proactive_check(self):
|
|
"""Sprawdza warunki i ewentualnie kaze ATC samodzielnie sie odezwac.
|
|
|
|
Wywolywane na glownym watku (z flight loop), co ~1 s. Samo wyslanie
|
|
do LLM idzie do watku roboczego przez _dispatch()."""
|
|
if not self.config.get("proactive_enabled", True):
|
|
return
|
|
if self.streaming:
|
|
return
|
|
|
|
import time as _time
|
|
now = _time.time()
|
|
cooldown = self.config.get("proactive_cooldown", 30)
|
|
if now - self._proactive_last_ts < cooldown:
|
|
return
|
|
|
|
reason = self._detect_proactive_reason()
|
|
if not reason:
|
|
return
|
|
|
|
# Zaznacz czas, zeby nie spamowac, i wyslij.
|
|
self._proactive_last_ts = now
|
|
self._dispatch("", reason)
|
|
|
|
def _detect_proactive_reason(self):
|
|
"""Wykrywa prosty wyzwalacz proaktywny. Zwraca opis powodu lub None."""
|
|
# 1) Zmiana fazy: ziemia -> powietrze (start) lub powietrze -> ziemia (ladowanie)
|
|
agl_m = self._get_agl()
|
|
if agl_m is not None:
|
|
phase = "ground" if agl_m < 1.0 else "airborne"
|
|
if self._proactive_last_phase is not None and phase != self._proactive_last_phase:
|
|
if phase == "airborne":
|
|
return "aircraft just became airborne (takeoff)"
|
|
else:
|
|
return "aircraft just touched down (landing)"
|
|
self._proactive_last_phase = phase
|
|
|
|
# 2) Nowe lotnisko w poblizu (podejscie/zbliżenie) — witamy raz.
|
|
icao = self._nearest_airport_icao()
|
|
if icao and icao not in self._proactive_announced_airports:
|
|
# Tylko gdy jestesmy w powietrzu (przylot) — nie przy starcie na ziemi.
|
|
if agl_m is not None and agl_m >= 1.0:
|
|
self._proactive_announced_airports.add(icao)
|
|
return f"pilot is approaching airport {icao}"
|
|
|
|
# 3) Bliski ruch (traffic advisory) — tylko jesli cos jest < 5 nm.
|
|
close = self._close_traffic(5.0)
|
|
if close:
|
|
return "nearby traffic within 5 nm (issue traffic advisory)"
|
|
|
|
return None
|
|
|
|
def _close_traffic(self, max_nm):
|
|
"""Zwraca liste bliskiego ruchu (dist_nm < max_nm). TYLKO glowny watek."""
|
|
bearing = self._read_float_array(self._tcas_bearing_ref, NUM_TCAS_SLOTS)
|
|
distance = self._read_float_array(self._tcas_distance_ref, NUM_TCAS_SLOTS)
|
|
if not distance:
|
|
return []
|
|
out = []
|
|
for i in range(NUM_TCAS_SLOTS):
|
|
if i >= len(distance):
|
|
break
|
|
dist_m = distance[i]
|
|
if not dist_m or dist_m <= 0:
|
|
continue
|
|
dist_nm = dist_m / 1852.0
|
|
if dist_nm < max_nm:
|
|
brg_rel = (bearing[i] % 360.0) if i < len(bearing) else 0.0
|
|
clock = int(round((brg_rel / 30.0) % 12)) or 12
|
|
out.append(f"{dist_nm:.1f} nm at {clock}:00")
|
|
return out
|
|
|
|
# ---------------------------- VR ----------------------------
|
|
def _vr_flight_loop(self, lastCall, elapsed, counter, refCon):
|
|
"""Co 1s sprawdza stan VR i wywoluje monitor proaktywny (glowny watek)."""
|
|
vr = self._get_vr_enabled()
|
|
if vr != self._last_vr:
|
|
self._last_vr = vr
|
|
self._apply_vr_mode(vr)
|
|
# Proaktywny ATC (bezpiecznie w try/except, zeby nie wywalic pętli)
|
|
try:
|
|
self._proactive_check()
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad monitora proaktywnego: {e}")
|
|
return 1.0
|
|
|
|
def _get_vr_enabled(self):
|
|
if self._vr_ref is None:
|
|
return 0
|
|
try:
|
|
return xp.getDatai(self._vr_ref)
|
|
except Exception:
|
|
return 0
|
|
|
|
def _apply_vr_mode(self, vr):
|
|
for win in (self._imgui_window, self._config_window):
|
|
if win is None or win.windowID is None:
|
|
continue
|
|
try:
|
|
if vr:
|
|
# Okno w przestrzeni VR (3D), widoczne i interaktywne w headsetcie
|
|
xp.setWindowPositioningMode(win.windowID, xp.WindowVR, -1)
|
|
xp.setWindowGeometryVR(win.windowID, 800, 900)
|
|
else:
|
|
xp.setWindowPositioningMode(win.windowID, xp.WindowPositionFree, -1)
|
|
except Exception as e:
|
|
xp.log(f"[XSayATC] Blad ustawiania VR: {e}")
|
|
|
|
# ---------------------------- Okno ImGui ----------------------------
|
|
def _draw(self, windowID, refCon):
|
|
imgui.text("XSayATC - lokalny ATC")
|
|
imgui.separator()
|
|
|
|
# Obszar czatu (scrollowalny)
|
|
avail_h = imgui.get_content_region_available()[1]
|
|
chat_h = max(60.0, avail_h - 130.0)
|
|
with imgui.begin_child("chat", 0, chat_h, border=True):
|
|
for m in self.messages:
|
|
if m["role"] == "user":
|
|
imgui.text_colored("Ty: " + m["text"], 0.5, 0.8, 1.0, 1.0)
|
|
elif m["role"] == "assistant":
|
|
imgui.text_wrapped("ATC: " + m["text"])
|
|
if self.streaming:
|
|
imgui.text_colored("...", 0.9, 0.9, 0.3, 1.0)
|
|
imgui.set_scroll_here_y(1.0)
|
|
|
|
# Pole tekstowe + przycisk PTT
|
|
changed, self.pending_text = imgui.input_text(
|
|
"##msg", self.pending_text, 512,
|
|
flags=imgui.INPUT_TEXT_ENTER_RETURNS_TRUE
|
|
)
|
|
|
|
imgui.same_line()
|
|
if imgui.button("Wyslij (PTT)") or (changed and self.pending_text.strip()):
|
|
self.send(self.pending_text)
|
|
self.pending_text = ""
|
|
|
|
imgui.same_line()
|
|
if imgui.button("Stop"):
|
|
self.stop()
|
|
|
|
imgui.text(self.status)
|
|
|
|
imgui.separator()
|
|
# Wybor kierunku lotu (determinuje handoff ATC)
|
|
imgui.text("Kierunek lotu:")
|
|
if imgui.radio_button("Odlot", self.flight_phase == "departure"):
|
|
self.flight_phase = "departure"
|
|
imgui.same_line()
|
|
if imgui.radio_button("Przylot", self.flight_phase == "arrival"):
|
|
self.flight_phase = "arrival"
|
|
|
|
imgui.separator()
|
|
# Czestotliwosci najblizszego lotniska (odswiezane recznie przyciskiem)
|
|
if imgui.button("Odswiez czestotliwosci"):
|
|
self._refresh_freq_display()
|
|
icao, freq_lines = self._freq_display_lines()
|
|
if icao:
|
|
imgui.text(f"Czestotliwosci - {icao}:")
|
|
else:
|
|
imgui.text_colored("Czestotliwosci: (kliknij Odswiez)", 0.7, 0.7, 0.7, 1.0)
|
|
# Wlasny przewijany obszar, zeby czestotliwosci mialy zawsze suwak
|
|
with imgui.begin_child("freqs", 0, 120.0, border=True):
|
|
if freq_lines:
|
|
for ln in freq_lines:
|
|
imgui.text(ln)
|
|
else:
|
|
imgui.text_colored("(brak danych)", 0.7, 0.7, 0.7, 1.0)
|
|
|
|
def _draw_config(self, windowID, refCon):
|
|
imgui.text("Konfiguracja XSayATC")
|
|
imgui.separator()
|
|
|
|
_, self.config["base_url"] = imgui.input_text(
|
|
"Base URL", self.config["base_url"], 256)
|
|
_, self.config["model"] = imgui.input_text(
|
|
"Model (puste=auto)", self.config["model"], 512)
|
|
_, self.config["temperature"] = imgui.slider_float(
|
|
"Temperature", self.config["temperature"], 0.0, 2.0, "%.2f")
|
|
_, self.config["max_tokens"] = imgui.slider_int(
|
|
"Max tokens", self.config["max_tokens"], 64, 4096)
|
|
_, self.config["no_think"] = imgui.checkbox(
|
|
"Tryb /no_think (Qwen3)", self.config["no_think"])
|
|
_, self.config["include_flight_context"] = imgui.checkbox(
|
|
"Dolacz kontekst lotu", self.config["include_flight_context"])
|
|
|
|
imgui.separator()
|
|
imgui.text("Glos (whisper / piper):")
|
|
_, self.config["voice_enabled"] = imgui.checkbox(
|
|
"Glos wlaczony (STT)", self.config["voice_enabled"])
|
|
_, self.config["tts_enabled"] = imgui.checkbox(
|
|
"Odczytywanie odpowiedzi (TTS)", self.config["tts_enabled"])
|
|
_, self.config["whisper_model"] = imgui.input_text(
|
|
"Model whisper", self.config["whisper_model"], 512)
|
|
_, self.config["piper_model"] = imgui.input_text(
|
|
"Model piper", self.config["piper_model"], 512)
|
|
|
|
imgui.separator()
|
|
imgui.text("Plan lotu (fallback dla samolotow z wlasnym FMC):")
|
|
_, self.config["flightplan_file"] = imgui.input_text(
|
|
"Plik planu lotu", self.config["flightplan_file"], 512)
|
|
|
|
imgui.separator()
|
|
imgui.text("Procedury ATC:")
|
|
_, self.config["use_procedures"] = imgui.checkbox(
|
|
"Wlacz procedury", self.config["use_procedures"])
|
|
_, self.config["procedures_dir"] = imgui.input_text(
|
|
"Katalog procedur", self.config["procedures_dir"], 512)
|
|
|
|
imgui.separator()
|
|
imgui.text("Proaktywny ATC:")
|
|
_, self.config["proactive_enabled"] = imgui.checkbox(
|
|
"ATC sam inicjuje kontakt (ruch, faza lotu)", self.config["proactive_enabled"])
|
|
_, self.config["proactive_cooldown"] = imgui.slider_int(
|
|
"Odstep (s)", self.config["proactive_cooldown"], 5, 300)
|
|
|
|
imgui.separator()
|
|
imgui.text("System prompt (instrukcje ATC):")
|
|
_, self.config["system_prompt"] = imgui.input_text_multiline(
|
|
"##sysprompt", self.config["system_prompt"], 4096,
|
|
width=0, height=120)
|
|
|
|
imgui.separator()
|
|
if imgui.button("Zapisz"):
|
|
self._save_config()
|
|
self._model_cache = None
|
|
self.status = "Konfiguracja zapisana."
|
|
|
|
# pokaz kontekst lotu na zywo
|
|
imgui.separator()
|
|
imgui.text("Kontekst lotu (podglad):")
|
|
try:
|
|
_, ctx_text = self._collect_context()
|
|
imgui.text_wrapped(ctx_text)
|
|
except Exception as e:
|
|
imgui.text(f"(blad: {e})")
|
|
|
|
# ---------------------------- Cykl zycia pluginu ----------------------------
|
|
def XPluginStart(self):
|
|
self._ensure_datarefs()
|
|
|
|
# Rozpocznij wczytywanie czestotliwosci z apt.dat w tle od razu,
|
|
# zeby pierwsze klikniecie "Odswiez czestotliwosci" nie musialo czekac.
|
|
self._ensure_apt_freqs()
|
|
|
|
# Komenda PTT (przypisz klawisz w Settings > Keyboard)
|
|
self._cmd = xp.createCommand("xsayatc/ptt", "XSayATC push-to-talk")
|
|
xp.registerCommandHandler(self._cmd, self._ptt_handler, 1, None)
|
|
|
|
# Komenda toggle okna (przypisz klawisz, zeby wlaczac/wylaczac okno)
|
|
self._cmd_toggle = xp.createCommand("xsayatc/toggle_window", "XSayATC toggle window")
|
|
xp.registerCommandHandler(self._cmd_toggle, self._toggle_handler, 1, None)
|
|
|
|
# Menu
|
|
self._menu = xp.createMenu("XSayATC", None, 0, self._menu_handler, None)
|
|
self._menu_open_idx = xp.appendMenuItemWithCommand(
|
|
self._menu, "Pokaz/ukryj okno", self._cmd_toggle)
|
|
self._menu_cfg_idx = xp.appendMenuItem(self._menu, "Konfiguracja", "xsayatc_config")
|
|
|
|
# Okno glowne (ukryte domyslnie — pokaz klawiszem/menu "Pokaz/ukryj okno")
|
|
self._imgui_window = ImguiWindow(
|
|
left=100, top=400, right=600, bottom=100,
|
|
visible=0, draw=self._draw, refCon=None,
|
|
decoration=1, layer=1)
|
|
self._imgui_window.setTitle("XSayATC")
|
|
|
|
# Okno konfiguracji (ukryte domyslnie)
|
|
self._config_window = ImguiWindow(
|
|
left=150, top=500, right=700, bottom=120,
|
|
visible=0, draw=self._draw_config, refCon=None,
|
|
decoration=1, layer=1)
|
|
self._config_window.setTitle("XSayATC - Konfiguracja")
|
|
|
|
# Ustaw limity zmiany rozmiaru
|
|
for win in (self._imgui_window, self._config_window):
|
|
try:
|
|
xp.setWindowResizingLimits(win.windowID, 200, 100, 2000, 2000)
|
|
except Exception:
|
|
pass
|
|
|
|
# Flight loop do detekcji VR
|
|
xp.registerFlightLoopCallback(self._vr_flight_loop, 1.0, None)
|
|
self._last_vr = self._get_vr_enabled()
|
|
self._apply_vr_mode(self._last_vr)
|
|
|
|
return "XSayATC", "pali112.xsayatc", "Lokalny asystent ATC (llama.cpp)"
|
|
|
|
def XPluginStop(self):
|
|
self.stop()
|
|
try:
|
|
xp.unregisterFlightLoopCallback(self._vr_flight_loop, None)
|
|
except Exception:
|
|
pass
|
|
for win in (self._config_window, self._imgui_window):
|
|
if win is not None:
|
|
try:
|
|
win.delete()
|
|
except Exception:
|
|
pass
|
|
self._config_window = None
|
|
self._imgui_window = None
|
|
|
|
def XPluginEnable(self):
|
|
return 1
|
|
|
|
def XPluginDisable(self):
|
|
pass
|
|
|
|
def XPluginReceiveMessage(self, inFromWho, inMessage, inParam):
|
|
pass
|
|
|
|
# ---------------------------- Handlery ----------------------------
|
|
def _ptt_handler(self, cmdRef, phase, refCon):
|
|
# Glos: PTT steruje nagrywaniem
|
|
if self._voice_enabled():
|
|
mode = self.config.get("voice_ptt_mode", "hold")
|
|
if mode == "toggle":
|
|
if phase == xp.CommandBegin:
|
|
if self._recording:
|
|
self._stop_recording_and_send()
|
|
else:
|
|
self._start_recording()
|
|
return 0
|
|
else: # "hold"
|
|
if phase == xp.CommandBegin:
|
|
self._start_recording()
|
|
return 0
|
|
elif phase == xp.CommandEnd:
|
|
self._stop_recording_and_send()
|
|
return 0
|
|
return 1
|
|
|
|
# Tryb tekstowy (bez glosu)
|
|
if phase == xp.CommandBegin:
|
|
text = self.pending_text.strip()
|
|
if text:
|
|
self.send(text)
|
|
self.pending_text = ""
|
|
return 0
|
|
return 1
|
|
|
|
def _toggle_handler(self, cmdRef, phase, refCon):
|
|
if phase == xp.CommandBegin:
|
|
self._toggle_main_window()
|
|
return 0
|
|
return 1
|
|
|
|
def _menu_handler(self, menuRef, itemRef):
|
|
if itemRef == "xsayatc_config":
|
|
self._toggle_config_window()
|
|
|
|
def _toggle_main_window(self):
|
|
if self._imgui_window is not None and self._imgui_window.windowID is not None:
|
|
cur = xp.getWindowIsVisible(self._imgui_window.windowID)
|
|
xp.setWindowIsVisible(self._imgui_window.windowID, 0 if cur else 1)
|
|
|
|
def _open_config_window(self):
|
|
if self._config_window is not None and self._config_window.windowID is not None:
|
|
xp.setWindowIsVisible(self._config_window.windowID, 1)
|
|
xp.bringWindowToFront(self._config_window.windowID)
|
|
|
|
def _toggle_config_window(self):
|
|
if self._config_window is not None and self._config_window.windowID is not None:
|
|
cur = xp.getWindowIsVisible(self._config_window.windowID)
|
|
xp.setWindowIsVisible(self._config_window.windowID, 0 if cur else 1)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Punkt wejscia XPPython3
|
|
# --------------------------------------------------------------------------
|
|
|
|
_plugin = XSayATC()
|
|
|
|
|
|
class PythonInterface:
|
|
def XPluginStart(self):
|
|
return _plugin.XPluginStart()
|
|
|
|
def XPluginStop(self):
|
|
_plugin.XPluginStop()
|
|
|
|
def XPluginEnable(self):
|
|
return _plugin.XPluginEnable()
|
|
|
|
def XPluginDisable(self):
|
|
_plugin.XPluginDisable()
|
|
|
|
def XPluginReceiveMessage(self, inFromWho, inMessage, inParam):
|
|
_plugin.XPluginReceiveMessage(inFromWho, inMessage, inParam)
|