mia/hafas-terminal-app
1
0
Fork 0
Code Issues 6 Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'main' of https://git.miaig.dev/mia/hafas-terminal-app

Browse source
This commit is contained in:
Sophia S 2026-06-03 12:30:46 +02:00
commit bf0d247840
2 changed files with 103 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

53
backend.py
View file

@ -49,9 +49,49 @@ class HafasClient:
res = self.stationRequest(stationString)
return [(station["name"], station["extId"]) for station in res["svcResL"][0]["res"]["match"]["locL"]]
def _format_time(self, timeValue):
if not timeValue:
return None
return datetime.strptime(timeValue, "%H%M%S").strftime("%H:%M")
def getArrDep(self, stationId, arrdep="DEP", count=1):
res = self.arrDepRequest(stationId, arrdep, count)
return res["svcResL"][0]["res"]["jnyL"]
departures = []
time_key = "aTimeS" if arrdep == "ARR" else "dTimeS"
time_key_real = "aTimeR" if arrdep == "ARR" else "dTimeR"
for departure in res["svcResL"][0]["res"]["jnyL"]:
prod_list = departure.get("prodL") or []
prod_index = departure.get("prodX")
if isinstance(prod_index, int) and 0 <= prod_index < len(prod_list):
product = prod_list[prod_index]
else:
product = prod_list[0] if prod_list else {}
stop = departure.get("stbStop", {})
# Raw JSON access pattern:
# - Name: departure["dirTxt"]
# - Line: departure["prodL"][0]["nameS"] or departure["prodL"][0]["prodCtx"]["line"]
# - Departure/arrival time: departure["stbStop"]["dTimeS"] or ["aTimeS"]
departures.append({
"name": departure.get("dirTxt"),
"line": (
product.get("nameS")
or product.get("number")
or product.get("name")
or product.get("prodCtx", {}).get("line")
or product.get("prodCtx", {}).get("name")
or product.get("prodCtx", {}).get("nameS")
),
"departure_time": self._format_time(stop.get(time_key)),
"departure_time_raw": stop.get(time_key),
"departure_time_real": self._format_time(stop.get(time_key_real)),
"departure_time_real_raw": stop.get(time_key_real),
"jid": departure.get("jid"),
"raw": departure,
})
return departures
def getTrip(self, departure):
res = self.tripRequest(departure)
@ -74,6 +114,14 @@ class HafasClient:
trip = self.getTrip(departures[0]["jid"])
__import__('pprint').pprint(trip)
def moniterTest(self):
stationInput = input("Enter Station Name: ")
station = self.getStationNames(stationInput)
print(station)
# departures = self.getArrDep(station[0][1], arrdep="DEP", count=1)
departures = self.arrDepRequest(station[0][1], arrdep="DEP", count=1)
print(departures)
# streq = self.stationRequest(station)["svcResL"][0]["res"]["match"]["locL"]
#
@ -91,4 +139,5 @@ class HafasClient:
if __name__ == '__main__':
client = HafasClient()
client.runTests()
# client.runTests()
client.moniterTest()

76
station_monitor.py
View file

@ -31,48 +31,76 @@ def run_station_monitor(hafas: backend.HafasClient | None = None):
hafas = backend.HafasClient()
keyB = KeyBindings()
inputBuffer = Buffer()
resultBuffer = Buffer()
prefixBuffer = Buffer()
prefixBuffer.text = "Haltestelle eingeben: "
prefixBuffer.read_only = True
userinputBuffer = Buffer()
arrivalBuffer = Buffer()
departureBuffer = Buffer()
stop_event = threading.Event()
app_state = "MONITOR"
inputBuffer.text = "Haltestelle eingeben: "
inputBuffer.cursor_position = len(inputBuffer.text)
resultBuffer.text = "Ausgabe: "
arrivalBuffer.text = "Ankünfte:\n"
departureBuffer.text = "Abfahrten:\n"
root_container = HSplit(children=[
VSplit(children=[
Window(content=BufferControl(buffer=inputBuffer, focusable=True)),
Window(content=BufferControl(buffer=prefixBuffer, focusable=False),width=22),
Window(content=BufferControl(buffer=userinputBuffer, focusable=True)),
Window(width=1, char='│', style="fg:#9D1D75"),
Window(content=BufferControl(buffer=resultBuffer, focusable=True)),
HSplit(children=[
Window(content=BufferControl(buffer=arrivalBuffer, focusable=True)),
Window(height=1, char='-', style="fg:#9D1D75"),
Window(content=BufferControl(buffer=departureBuffer, focusable=True)),
]),
]),
Window(height=1, char='-', style="bg:#9D1D75 fg:#FFFFFF"),
])
layout = Layout(root_container, focused_element=inputBuffer)
layout = Layout(root_container, focused_element=userinputBuffer)
@keyB.add("enter")
def handle_enter(event):
global app_state
user_input = inputBuffer.text.replace("Haltestelle eingeben: ", "").strip()
user_input = userinputBuffer.text.strip()
if user_input:
app_state = "RESULTS"
# TODO: handle case of no results gracefully
results = hafas.getStationNames(user_input)
inputBuffer.insert_line_below()
for station in results:
inputBuffer.insert_text(f"\n {station[0]}")
resultBuffer.text = f"Ergebnisse für: {results[0][1]}\n\n"
station = results[0][1] if results else "Keine Ergebnisse gefunden."
result = hafas.getArrDep(station, arrdep="ARR", count=3)
inputBuffer.insert_line_below()
for entry in result:
resultBuffer.insert_text(f"\n {result[0]}")
else:
if user_input:
app_state = "RESULTS"
results = hafas.getStationNames(user_input)
if not results:
arrivalBuffer.text = f"Keine Stationen gefunden für: {user_input}"
departureBuffer.text= ""
userinputBuffer.text=""
app_state = "MONITOR"
else:
station_name = results[0][1]
station_id = results[0][0]
arrivals = hafas.getArrDep(station_id, arrdep="ARR", count=3)
arrivalBuffer.text = f"Ankünfte für {station_name}:\n"
for arrival in arrivals:
arrivalBuffer.insert_text(f"\n {arrival}")
departures = hafas.getArrDep(station_id, arrdep="DEP", count=3)
departureBuffer.text = f"Abfahrten für {station_name}:\n"
for departure in departures:
departureBuffer.insert_text(f"\n {departure}")
userinputBuffer.text = ""
else:
app_state = "MONITOR"
@keyB.add("c-q")
def exit_(event):