937 lines
36 KiB
Python
937 lines
36 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. "
|
|
"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": "",
|
|
}
|
|
|
|
# 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.001, "MHz"),
|
|
("sim/cockpit/radios/com2_freq_hz", "COM2", 0.001, "MHz"),
|
|
("sim/cockpit/radios/nav1_freq_hz", "NAV1", 0.001, "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"
|
|
|
|
# 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
|
|
|
|
|
|
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
|
|
|
|
# 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
|
|
|
|
# VR
|
|
self._last_vr = None
|
|
|
|
# Glos (STT / TTS)
|
|
self._recording = False
|
|
self._record_proc = None
|
|
self._ptt_toggled = False # dla trybu "toggle"
|
|
|
|
# ---------------------------- 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)
|
|
# 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)
|
|
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 _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 = []
|
|
for name, label, mult, unit in CONTEXT_DATAREFS:
|
|
v = self._get_dataref(name, is_float=(name != "sim/cockpit/radios/transponder_code"))
|
|
if v is None:
|
|
continue
|
|
if name == "sim/cockpit/radios/transponder_code":
|
|
ctx[label] = int(v)
|
|
lines.append(f"{label}: {int(v)}")
|
|
elif name in ("sim/cockpit/radios/com1_freq_hz",
|
|
"sim/cockpit/radios/com2_freq_hz",
|
|
"sim/cockpit/radios/nav1_freq_hz"):
|
|
val = v * mult
|
|
ctx[label] = val
|
|
lines.append(f"{label}: {val:.3f} {unit}".strip())
|
|
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)")
|
|
|
|
# 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 _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):
|
|
"""Buduje liste messages. flight_ctx_text musi byc przekazany (bez odczytu datarefow)."""
|
|
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})
|
|
|
|
for m in self.messages[-12:]:
|
|
msgs.append({"role": m["role"], "content": m["text"]})
|
|
|
|
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):
|
|
"""Watek roboczy: TYLKO siec i aktualizacja self.messages. Bez xp.getData*."""
|
|
self.streaming = True
|
|
self.status = "ATC mysli..."
|
|
|
|
# Historia do UI
|
|
self.messages.append({"role": "user", "text": user_text})
|
|
self.messages.append({"role": "assistant", "text": ""})
|
|
|
|
try:
|
|
messages = self._build_messages(user_text, flight_ctx_text)
|
|
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
|
|
|
|
# Zbierz kontekst lotu NA GLOWNYM WATKU
|
|
flight_ctx_text = 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
|
|
|
|
self._stop_event.clear()
|
|
self._worker = threading.Thread(
|
|
target=self._run_stream, args=(text, flight_ctx_text), 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 _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_")
|
|
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)
|
|
_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)
|
|
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}")
|
|
|
|
# ---------------------------- VR ----------------------------
|
|
def _vr_flight_loop(self, lastCall, elapsed, counter, refCon):
|
|
"""Co 1s sprawdza stan VR i ustawia tryb pozycjonowania okien (glowny watek)."""
|
|
vr = self._get_vr_enabled()
|
|
if vr != self._last_vr:
|
|
self._last_vr = vr
|
|
self._apply_vr_mode(vr)
|
|
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()
|
|
if imgui.button("Konfiguracja"):
|
|
self._open_config_window()
|
|
|
|
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("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()
|
|
|
|
# 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", None)
|
|
|
|
# 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 == self._menu_cfg_idx:
|
|
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)
|