275 lines
9.4 KiB
Python
275 lines
9.4 KiB
Python
#V0.1 @Mac
|
||
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
|
||
|
||
def conf():
|
||
appname = os.path.basename(sys.argv[0])
|
||
appname = appname.replace(".py", ".conf")
|
||
absFilePath = os.path.abspath(__file__)
|
||
absFilePath = absFilePath.replace(".py", ".conf")
|
||
|
||
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 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 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 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 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 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"), "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"),
|
||
"Du bist von deiner Art her:" + str(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",
|
||
"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")
|
||
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)
|
||
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():
|
||
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}")
|
||
|
||
print(f"Backup erstellt. {len(txt_files)} Datei(en) gesichert.")
|
||
return None
|
||
|
||
host = os.getcwd()
|
||
Pfad = os.getcwd() + '/Announcements/'
|
||
|
||
passw = conf()
|
||
client = OpenAI(
|
||
api_key=passw['AI']['Token'])
|
||
|
||
# dict
|
||
|
||
#print(data.keys())
|
||
#print(data['weights'])
|
||
|
||
(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) = Simbriefimport()
|
||
|
||
print('')
|
||
print("Airline:", general.get("icao_airline"))
|
||
print("Airlinename:", airline)
|
||
print("Flugnummer:", general.get("flight_number"))
|
||
print("Abflug ICAO:", origin.get("icao_code"))
|
||
print("von:", origin_name)
|
||
print("Ziel ICAO:", dest.get("icao_code"))
|
||
print("nach:", dest_name)
|
||
print("Geplante Blockzeit:", block)
|
||
print("geplanter Start:", start_time,"UTC")
|
||
print("geplante Landung:", land_time, "UTC")
|
||
print("Fluggerät:", aircraft_icao, aircraft_name)
|
||
print("Cruiselevel:", fl)
|
||
print("Entfernung:", distance, 'nm')
|
||
print("Fuel:", fuel, 'kg')
|
||
print("Passagiere:", pax)
|
||
print("ZFW:", zfw, 'kg')
|
||
print("TOW:", tow, 'kg')
|
||
print("Wetter:", origin.get("icao_code"), metar_origin)
|
||
print("Wetter:", dest["icao_code"], metar_dest)
|
||
|
||
Pfad_wd = Pfad + airline_icao + '/'
|
||
|
||
ordnerneu = Path(Pfad_wd + '/neu/')
|
||
print(Pfad_wd)
|
||
BACKUP(Pfad_wd)
|
||
print(replaceFiles())
|
||
|
||
ordnerneu.mkdir(exist_ok=True)
|
||
|
||
metarCabin = METAR()
|
||
print(metarCabin.choices[0].message.content)
|
||
metar = metarCabin.choices[0].message.content
|
||
metar = "\n".join(filter(None, map(str.strip, metar.splitlines())))
|
||
|
||
welcomePilot = WelcomePilot()
|
||
|
||
Inhalt = welcomePilot.choices[0].message.content
|
||
Inhalt = "\n".join(filter(None, map(str.strip, Inhalt.splitlines())))
|
||
print(Inhalt)
|
||
print(os.getcwd())
|
||
|
||
Pfad = Pfad_wd + 'BoardingWelcomePilot.txt'
|
||
txtSave(Pfad, Inhalt)
|
||
|
||
Pfad = Pfad_wd + 'FastenSeatbelt.txt'
|
||
txtSave(Pfad, metar)
|