497 lines
No EOL
18 KiB
Python
497 lines
No EOL
18 KiB
Python
#V0.1b @PC
|
||
import requests
|
||
import json
|
||
from pathlib import Path
|
||
from datetime import datetime, timezone
|
||
from openai import OpenAI
|
||
import os
|
||
import sys
|
||
import random
|
||
import shutil
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
|
||
|
||
def setup_logging(app_dir, level=logging.INFO, console=True):
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
log_file = app_dir / "Simbriefimport.log"
|
||
|
||
logger = logging.getLogger()
|
||
logger.setLevel(level)
|
||
logger.handlers.clear()
|
||
|
||
fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||
|
||
# Datei: rotierend, damit sie nicht unendlich wächst
|
||
fh = RotatingFileHandler(log_file, maxBytes=2_000_000, backupCount=5, encoding="utf-8")
|
||
fh.setFormatter(fmt)
|
||
logger.addHandler(fh)
|
||
|
||
if console:
|
||
ch = logging.StreamHandler()
|
||
ch.setFormatter(fmt)
|
||
logger.addHandler(ch)
|
||
|
||
logging.info("Logging gestartet: %s", log_file)
|
||
def conf():
|
||
appname = os.path.basename(sys.argv[0])
|
||
appname = appname.replace(".py", ".conf")
|
||
|
||
absFilePath = os.getcwd()
|
||
absFilePath = absFilePath.replace(".exe", ".conf")
|
||
print(appname)
|
||
print(absFilePath)
|
||
|
||
if os.path.isfile(absFilePath) is False:
|
||
print(appname + ' scheint es nicht zu geben.')
|
||
print('Ich lege eine neue Datei ' + appname + ' an.')
|
||
passw = {'Prompt':
|
||
{
|
||
"PurserPrompt": "",
|
||
"CptPrompt": "",
|
||
"CptName":""
|
||
|
||
},
|
||
'AI':
|
||
{
|
||
"Token": "",
|
||
"Model": ""
|
||
}
|
||
}
|
||
|
||
print(str(appname) + ' bitte entsprechend befüllen.')
|
||
with open(absFilePath, 'w') as fp:
|
||
json.dump(passw, fp, sort_keys=True, indent=4)
|
||
quit()
|
||
else:
|
||
with open(absFilePath) as file:
|
||
passw = json.load(file)
|
||
return passw
|
||
def get_app_dir():
|
||
# Fall: als .exe (z. B. mit PyInstaller gebaut)
|
||
if getattr(sys, 'frozen', False):
|
||
return Path(sys.executable).parent
|
||
# Fall: normales Python-Skript
|
||
return Path(__file__).resolve().parent
|
||
def resource_path(filename: str) -> Path:
|
||
if hasattr(sys, "_MEIPASS"):
|
||
return Path(sys._MEIPASS) / filename
|
||
return Path(__file__).resolve().parent / filename
|
||
def minuten_zu_zeit(minuten) -> str:
|
||
minuten = int(minuten)
|
||
stunden = minuten // 60 # volle Stunden
|
||
rest_min = minuten % 60 # verbleibende Minuten
|
||
return f"{stunden}:{rest_min:02d}" # zweistellige Minutenanzeige
|
||
def old_aircr_name(aircraft_icao):
|
||
with open("aircraft_full.json", "r", encoding="utf-8") as f:
|
||
aircraft_data = json.load(f)
|
||
if aircraft_icao in aircraft_data:
|
||
antwort = (aircraft_data[aircraft_icao]["name"])
|
||
else:
|
||
print("Code nicht vorhanden")
|
||
antwort = "unbekanntes Luftfahrzeug"
|
||
return (antwort)
|
||
def aircr_name(aircraft_icao: str) -> str:
|
||
json_path = resource_path("aircraft_full.json")
|
||
|
||
with json_path.open("r", encoding="utf-8") as f:
|
||
aircraft_data = json.load(f)
|
||
|
||
if aircraft_icao in aircraft_data:
|
||
return aircraft_data[aircraft_icao].get("name", "unbekanntes Luftfahrzeug")
|
||
else:
|
||
return "unbekanntes Luftfahrzeug"
|
||
def old_start_name(ori_icao):
|
||
|
||
with open("airports_full.json", "r", encoding="utf-8") as f:
|
||
ori_data = json.load(f)
|
||
if ori_icao in ori_data:
|
||
antwort = (ori_data[ori_icao]["name"])
|
||
else:
|
||
print("Code nicht vorhanden")
|
||
antwort = "unbekannter Startflughafen"
|
||
return (antwort)
|
||
def start_name(ori_icao: str) -> str:
|
||
global _airports_cache
|
||
|
||
if _airports_cache is None:
|
||
with resource_path("airports_full.json").open(encoding="utf-8") as f:
|
||
_airports_cache = json.load(f)
|
||
|
||
return _airports_cache.get(
|
||
ori_icao, {}
|
||
).get("name", "unbekannter Startflughafen")
|
||
def old_airlinename(name):
|
||
with open("airlines_full.json", "r", encoding="utf-8") as f:
|
||
airline_data = json.load(f)
|
||
if name in airline_data:
|
||
antwort = (airline_data[name]["name"])
|
||
else:
|
||
print("Code nicht vorhanden")
|
||
antwort = "unbekanntes Luftfahrzeug"
|
||
return (antwort)
|
||
def airlinename(airline_icao: str) -> str:
|
||
global _airlines_cache
|
||
|
||
if _airlines_cache is None:
|
||
with resource_path("airlines_full.json").open(encoding="utf-8") as f:
|
||
_airlines_cache = json.load(f)
|
||
|
||
return _airlines_cache.get(
|
||
airline_icao, {}
|
||
).get(
|
||
"name",
|
||
"unbekannte Fluggesellschaft"
|
||
)
|
||
def replacedynamic(message):
|
||
message = message.replace('{AIRCRAFT_NAME}', aircraft_name)
|
||
message = message.replace('{DESTINATION_NAME}', dest_name)
|
||
message = message.replace('{AIRLINE_NAME}', airline)
|
||
message = message.replace('{DESTINATION_CITY}', dest_name)
|
||
message = message.replace('{ORIGIN_CITY}', origin_name)
|
||
message = message.replace('{xml_number<general><initial_altitude>}', flightlevel)
|
||
return (message)
|
||
def unix_to_datetime(timestamp: int) -> str:
|
||
return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%d.%m.%Y %H:%M:%S")
|
||
def ai(prompt):
|
||
completion = client.chat.completions.create(
|
||
model="gpt-4.1-mini",
|
||
messages=[
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
)
|
||
return completion
|
||
def stimmung():
|
||
tone = (passw['Prompt']['tone'])
|
||
tone = random.choice(list(tone))
|
||
|
||
#print(tone)
|
||
return tone
|
||
def WelcomePilot():
|
||
prompt_org = passw['Prompt']['CptPrompt']
|
||
info = {"Zielflughafen: " + dest.get("icao_code") + dest_name, "Reisedauer:" + block,
|
||
"Entfernung: " + distance + "nm",
|
||
#"Menge Sprit: " + fuel + "kg",
|
||
"Passagieranzahl" + pax,
|
||
"Flugzeugtyp:" + aircraft_name,
|
||
"Du freust dich auf den Flug und bedankst dich, das man sich für " + airline + " entschieden hat.",
|
||
"Dein Name:" + passw['Prompt']['CptName'],
|
||
"Startflughafen: " + origin.get("icao_code") + origin_name,
|
||
"Du bist von deiner Art her:" + Stimmung,
|
||
"Wetter am Ziel" + metar_dest,
|
||
"Verwende nie die ICAO Abkürzungen, übersetze diese.",
|
||
#"Erwähne 2 bis 3 Sehenswürdigkeiten, die auf dem Flug von Start zum Ziel überflogen werden. Die Route lautet:" + route,
|
||
"Flugnummer" + general.get("flight_number"),
|
||
"Wetter am Start" + metar_origin}
|
||
|
||
for i in info:
|
||
prompt_org = prompt_org + i
|
||
antwort_AI = ai(prompt_org)
|
||
|
||
return antwort_AI
|
||
def CruiseElapsed10Percent():
|
||
prompt_org = passw['Prompt']['Cruise10']
|
||
info = ["Zielflughafen: " + dest.get("icao_code") + dest_name, "Reisedauer:" + block,
|
||
"Menge Sprit: " + fuel + "kg",
|
||
"Reiseflughöhe: ",
|
||
"Erwähne die typische Reiseflughöhe für diesen Flug",
|
||
"Dein Name:" + passw['Prompt']['CptName'],
|
||
"Startflughafen: " + origin.get("icao_code") + origin_name,
|
||
"Du bist von deiner Art her:" + Stimmung,
|
||
"Erwähne 2 bis 3 Sehenswürdigkeiten, die auf dem Flug von Start zum Ziel überflogen werden. Die Route lautet:" + route]
|
||
for i in info:
|
||
prompt_org = prompt_org + i
|
||
antwort_AI = ai(prompt_org)
|
||
|
||
return antwort_AI
|
||
def CruiseElapsed50Percent():
|
||
prompt_org = passw['Prompt']['Cruise50']
|
||
info = {"Zielflughafen: " + dest.get("icao_code") + dest_name, "Gesamtreisedauer:" + block,
|
||
"Du bist von deiner Art her:" + Stimmung,
|
||
"Erwähne 3 bis 6 Städte, welche entlang der Route liegen. Nenne nicht die Wegpunkte. Route: " + route,
|
||
"Informiere über die bevorstehenden, typischen Landevorbereitungen"}
|
||
for i in info:
|
||
prompt_org = prompt_org + i
|
||
antwort_AI = ai(prompt_org)
|
||
return antwort_AI
|
||
def CruiseElapsed80Percent():
|
||
prompt_org = passw['Prompt']['Cruise80']
|
||
info = {"Zielflughafen: " + dest.get("icao_code") + dest_name,
|
||
"Du bist von deiner Art her:" + Stimmung,
|
||
"Die abgeflogene Route lautet:" + route}
|
||
for i in info:
|
||
prompt_org = prompt_org + i
|
||
antwort_AI = ai(prompt_org)
|
||
|
||
return antwort_AI
|
||
def SafetyBriefing():
|
||
prompt_org = passw['Prompt']['SafetyBriefing']
|
||
info = {"Zielflughafen: " + dest.get("icao_code") + dest_name, "Reisedauer:" + block,
|
||
|
||
"Passagieranzahl" + pax,
|
||
"Flugzeugtyp:" + aircraft_name,
|
||
"Du freust dich auf den Flug und bedankst dich, das man sich für " + airline + " entschieden hat.",
|
||
"Dein Name:" + passw['Prompt']['PurserName'],
|
||
"Startflughafen: " + origin.get("icao_code") + origin_name,
|
||
"Du bist von deiner Art her:" + str(stimmung()),
|
||
|
||
"Flugnummer" + general.get("flight_number"),
|
||
"Wetter am Start" + metar_origin}
|
||
|
||
for i in info:
|
||
prompt_org = prompt_org + i
|
||
antwort_AI = ai(prompt_org)
|
||
|
||
return antwort_AI
|
||
def METAR():
|
||
prompt_org = passw['Prompt']['PurserPrompt_METAR']
|
||
prompt2 = prompt_org + 'Die METAR lautet ' + metar_dest
|
||
antwort_AI = ai(prompt2)
|
||
return antwort_AI
|
||
def replaceFiles():
|
||
files = []
|
||
for datei in Path(Pfad_wd).glob("*.txt"):
|
||
with open(datei, "r", encoding="utf-8") as f:
|
||
files.append(f"{datei.name}")
|
||
inhalt = f.read()
|
||
inhalt_neu = replacedynamic(inhalt)
|
||
dateineu = str(ordnerneu) + '/' + (f"{datei.name}")
|
||
with open(dateineu, 'w', encoding="utf-8") as e:
|
||
e.write(inhalt_neu)
|
||
return files
|
||
def Simbriefimport():
|
||
USERNAME = "hubobel"
|
||
URL = "https://www.simbrief.com/api/xml.fetcher.php"
|
||
params = {
|
||
"username": USERNAME,
|
||
"json": 1, # JSON statt XML
|
||
}
|
||
resp = requests.get(URL, params=params, timeout=15)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
general = data.get("general", {})
|
||
origin = data.get("origin", {})
|
||
origin_name = start_name(origin["icao_code"])
|
||
origin_icao = origin["icao_code"]
|
||
dest = data.get("destination", {})
|
||
dest_name = start_name(dest["icao_code"])
|
||
dest_icao = dest["icao_code"]
|
||
aircraft = data.get("aircraft", {})
|
||
time = data.get("times", {})
|
||
block = minuten_zu_zeit(int(time.get("sched_block")) / 60)
|
||
start_time = unix_to_datetime(int(time.get("sched_out")))
|
||
land_time = unix_to_datetime(int(time.get("sched_in")))
|
||
aircraft_icao = aircraft["icaocode"]
|
||
aircraft_name = aircr_name((aircraft_icao))
|
||
airline = airlinename(general.get("icao_airline"))
|
||
flightnumber = general.get("flight_number")
|
||
flightlevel = data['general']['initial_altitude']
|
||
fl = 'FL' + str(int(flightlevel) // 100)
|
||
distance = data['general']['route_distance']
|
||
fuel = data['fuel']['plan_ramp']
|
||
metar_origin = data['weather']['orig_metar']
|
||
metar_dest = data['weather']['dest_metar']
|
||
pax = data['weights']['pax_count']
|
||
payload = data['weights']['payload']
|
||
tow = data['weights']['est_tow']
|
||
zfw = data['weights']['est_zfw']
|
||
airline_icao = general.get("icao_airline")
|
||
route = data['general']['route']
|
||
print(data.keys())
|
||
print(data['general'])
|
||
return (zfw, tow, payload, pax, metar_dest, metar_origin, fuel, distance, fl, flightlevel, flightnumber, airline,
|
||
aircraft_name, aircraft_icao, land_time, start_time, block, time, aircraft, dest_icao, dest_name, dest,
|
||
origin_icao, origin_name, origin, general, airline_icao, route)
|
||
def txtSave(Datei, Inhalt):
|
||
voice = None
|
||
Inhalt = Inhalt.replace("„", "").replace("“", "").replace('"', "")
|
||
try:
|
||
|
||
with open(Datei, "r", encoding="utf-8") as f:
|
||
erste_zeile = f.readline().strip()
|
||
if erste_zeile.startswith("##Voice:"):
|
||
voice = erste_zeile + "\n"
|
||
Inhalt = voice + Inhalt
|
||
except:
|
||
None
|
||
with open(Datei, 'w', encoding="utf-8") as e:
|
||
e.write(Inhalt)
|
||
return None
|
||
def BACKUP(Pfad):
|
||
ryr_dir = Path(Pfad)
|
||
backup_dir = ryr_dir / "BACKUP"
|
||
|
||
# Prüfen, ob BACKUP bereits existiert
|
||
if backup_dir.exists():
|
||
logging.info("BACKUP existiert bereits – keine Dateien wurden kopiert.")
|
||
#print("BACKUP existiert bereits – keine Dateien wurden kopiert.")
|
||
else:
|
||
# BACKUP anlegen
|
||
backup_dir.mkdir(parents=True)
|
||
|
||
# Alle .txt Dateien kopieren
|
||
txt_files = list(ryr_dir.glob("*.txt"))
|
||
|
||
for src in txt_files:
|
||
dst = backup_dir / src.name
|
||
shutil.copy2(src, dst)
|
||
print(f"Kopiert: {src.name}")
|
||
|
||
logging.info(
|
||
"Backup erstellt. %d Datei(en) gesichert.",
|
||
len(txt_files)
|
||
)
|
||
return None
|
||
|
||
|
||
host = os.getcwd()
|
||
Pfad = os.getcwd() + '/Announcements/'
|
||
_airports_cache = None
|
||
_airlines_cache = None
|
||
|
||
|
||
|
||
app_dir = get_app_dir()
|
||
setup_logging(app_dir, level=logging.INFO, console=True)
|
||
conf_file = app_dir / "Simbriefimport.conf"
|
||
|
||
if conf_file.exists():
|
||
#print("Konfigurationsdatei gefunden:", conf_file)
|
||
logging.info(
|
||
"Konfigurationsdatei gefunden: %s",
|
||
conf_file
|
||
)
|
||
with open(conf_file) as file:
|
||
passw = json.load(file)
|
||
|
||
else:
|
||
#print("Keine Konfigurationsdatei vorhanden:", conf_file)
|
||
logging.info(
|
||
"keine Konfigurationsdatei gefunden: %s",
|
||
conf_file
|
||
)
|
||
|
||
#passw = conf()
|
||
client = OpenAI(
|
||
api_key=passw['AI']['Token'])
|
||
|
||
# dict
|
||
|
||
Stimmung = str(stimmung())
|
||
|
||
|
||
(zfw, tow, payload, pax, metar_dest, metar_origin, fuel, distance, fl, flightlevel, flightnumber, airline,
|
||
aircraft_name, aircraft_icao, land_time, start_time, block, time, aircraft, dest_icao, dest_name, dest, origin_icao,
|
||
origin_name, origin, general, airline_icao, route) = Simbriefimport()
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("SimbriefUsername: %s",passw['Prompt']['PurserName'])
|
||
logging.info("Simbriefdate: %s",passw['Prompt']['CptName'])
|
||
logging.info("Simbrief Route: %s", Stimmung)
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("PurserName: %s",passw['Prompt']['PurserName'])
|
||
logging.info("Cpt.Name: %s",passw['Prompt']['CptName'])
|
||
logging.info("Stimmung: %s", Stimmung)
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("FLUGINFORMATIONEN")
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("Airline (ICAO): %s", general.get("icao_airline"))
|
||
logging.info("Airlinename: %s", airline)
|
||
logging.info("Flugnummer: %s", general.get("flight_number"))
|
||
|
||
logging.info("Abflug ICAO: %s", origin.get("icao_code"))
|
||
logging.info("Abflugort: %s", origin_name)
|
||
|
||
logging.info("Ziel ICAO: %s", dest.get("icao_code"))
|
||
logging.info("Zielort: %s", dest_name)
|
||
logging.info("Route: %s", route)
|
||
|
||
logging.info("Geplante Blockzeit: %s", block)
|
||
logging.info("Geplanter Start: %s UTC", start_time)
|
||
logging.info("Geplante Landung: %s UTC", land_time)
|
||
|
||
logging.info("Fluggerät: %s (%s)", aircraft_icao, aircraft_name)
|
||
logging.info("Cruiselevel: FL%s", fl)
|
||
logging.info("Entfernung: %s nm", distance)
|
||
|
||
logging.info("Fuel: %s kg", fuel)
|
||
logging.info("Passagiere: %s", pax)
|
||
logging.info("ZFW: %s kg", zfw)
|
||
logging.info("TOW: %s kg", tow)
|
||
|
||
logging.info(
|
||
"Wetter Abflug %s: %s",
|
||
origin.get("icao_code"),
|
||
metar_origin
|
||
)
|
||
|
||
logging.info(
|
||
"Wetter Ziel %s: %s",
|
||
dest.get("icao_code"),
|
||
metar_dest
|
||
)
|
||
|
||
logging.info("--------------------------------------------------")
|
||
|
||
|
||
Pfad_wd = Pfad + airline_icao + '/'
|
||
|
||
ordnerneu = Path(Pfad_wd + '/neu/')
|
||
logging.info(Pfad_wd)
|
||
BACKUP(Pfad_wd)
|
||
#print(replaceFiles())
|
||
|
||
ordnerneu.mkdir(exist_ok=True)
|
||
logging.info("--------------------------------------------------")
|
||
logging.info("START .txt Erzeugung ")
|
||
logging.info("--------------------------------------------------")
|
||
metarCabin = METAR()
|
||
metar = metarCabin.choices[0].message.content
|
||
metar = "\n".join(filter(None, map(str.strip, metar.splitlines())))
|
||
logging.info("METAR erzeugt")
|
||
welcomePilot = WelcomePilot()
|
||
Inhalt = welcomePilot.choices[0].message.content
|
||
Inhalt = "\n".join(filter(None, map(str.strip, Inhalt.splitlines())))
|
||
logging.info("WelcomePilot erzeugt")
|
||
|
||
InhaltCruise10 = CruiseElapsed10Percent().choices[0].message.content
|
||
InhaltCruise10 = "\n".join(filter(None, map(str.strip, InhaltCruise10.splitlines())))
|
||
logging.info("Cruise10 erzeugt")
|
||
|
||
InhaltCruise50 = CruiseElapsed50Percent().choices[0].message.content
|
||
InhaltCruise50 = "\n".join(filter(None, map(str.strip, InhaltCruise50.splitlines())))
|
||
logging.info("Cruise50 erzeugt")
|
||
|
||
InhaltCruise80 = CruiseElapsed80Percent().choices[0].message.content
|
||
InhaltCruise80 = "\n".join(filter(None, map(str.strip, InhaltCruise80.splitlines())))
|
||
|
||
logging.info("Cruise80 erzeugt")
|
||
|
||
Safety = SafetyBriefing().choices[0].message.content
|
||
Safety = "\n".join(filter(None, map(str.strip, Safety.splitlines())))
|
||
logging.info("SafetyBriefing erzeugt")
|
||
|
||
Pfad = Pfad_wd + 'BoardingWelcomePilot.txt'
|
||
txtSave(Pfad, Inhalt)
|
||
|
||
Pfad = Pfad_wd + 'FastenSeatbelt.txt'
|
||
txtSave(Pfad, metar)
|
||
|
||
Pfad = Pfad_wd + 'CruiseElapsed10Percent.txt'
|
||
txtSave(Pfad, InhaltCruise10)
|
||
|
||
Pfad = Pfad_wd + 'CruiseElapsed50Percent.txt'
|
||
txtSave(Pfad, InhaltCruise50)
|
||
|
||
Pfad = Pfad_wd + 'CruiseElapsed80Percent.txt'
|
||
txtSave(Pfad, InhaltCruise80)
|
||
|
||
Pfad = Pfad_wd + 'CruiseElapsed80Percent.txt'
|
||
txtSave(Pfad, InhaltCruise80)
|
||
|
||
Pfad = Pfad_wd + 'SafetyBriefing.txt'
|
||
txtSave(Pfad, Safety)
|
||
|
||
logging.info("--------------------------------------------------")
|
||
logging.info(" F E R T S C H ")
|
||
logging.info("--------------------------------------------------") |