diff --git a/backend.py b/backend.py index 8bb755d..d5eb4e5 100644 --- a/backend.py +++ b/backend.py @@ -1,6 +1,6 @@ """ haTerm - v0.1 -(c) Kieler Mia, Chiara Wohlwend, Sophia Schmidhofer +(c) Kieler Mia haTerm is a Terminal Based hafas client, using the ivb endpoint. Prompt_toolkit (https://github.com/prompt-toolkit/python-prompt-toolkit) will be used to render a terminal user interface (TUI). @@ -13,38 +13,82 @@ from datetime import datetime from prompt_toolkit import Application from prompt_toolkit.completion import WordCompleter -def stationRequest(station): - res = session.post("https://fahrplan.ivb.at/bin/mgate.exe", - data=json.dumps({"svcReqL": [{"req": {"input": {"field": "S", "loc": {"name": station, "type": "S"}}}, "meth": "LocMatch"}], "client": {"id": "VAO", "name": "webapp", "type": "WEB"}, "ver": "1.32", "lang": "de", "auth": {"type": "AID", "aid": "wf7mcf9bv3nv8g5f"}}), - headers={'User-Agent': "IOS",'Content-Type': 'application/json'}) - return json.loads(res.text) +class HafasClient: + def __init__(self): + self.session = requests.session() + self.clientInfo = {"id": "VAO", "name": "webapp", "type": "WEB"} + self.version = "1.32" + self.language = "de" + self.auth = {"type": "AID", "aid": "wf7mcf9bv3nv8g5f"} + self.headers = {'User-Agent': "IOS",'Content-Type': 'application/json'} + self.baseUrl = "https://fahrplan.ivb.at/bin/mgate.exe" -def departuresRequest(station, departures=1): - time = datetime.now() - print(time.strftime("%Y%m%d")) - res = session.post("https://fahrplan.ivb.at/bin/mgate.exe", - data=json.dumps({"svcReqL": [{"req": {"type": "DEP", "stbLoc": {"lid": f"A=1@L={station}@"}, "dirLoc": None, "maxJny": departures, "date": time.strftime("%Y%m%d"), "time": time.strftime("%H%M%S"), "dur": -1, "jnyFltrL": [{"type": "PROD", "mode": "INC", "value": "4087"}]}, "meth": "StationBoard"}], "client": {"id": "VAO", "name": "webapp", "type": "WEB"}, "ver": "1.32", "lang": "de", "auth": {"type": "AID", "aid": "wf7mcf9bv3nv8g5f"}}), - headers={'User-Agent': "IOS",'Content-Type': 'application/json'}) - return json.loads(res.text) + def stationRequest(self, station): + res = self.session.post(self.baseUrl, + data=json.dumps({"svcReqL": [{"req": {"input": {"field": "S", "loc": {"name": station, "type": "S"}}}, "meth": "LocMatch"}], "client": self.clientInfo, "ver": self.version, "lang": self.language, "auth": self.auth}), + headers=self.headers) + return json.loads(res.text) + + def arrDepRequest(self, station, arrdep, count): + time = datetime.now() + # print(time.strftime("%Y%m%d")) + res = self.session.post(self.baseUrl, + data=json.dumps({"svcReqL": [{"req": {"type": arrdep, "stbLoc": {"lid": f"A=1@L={station}@"}, "dirLoc": None, "maxJny": count, "date": time.strftime("%Y%m%d"), "time": time.strftime("%H%M%S"), "dur": -1, "jnyFltrL": [{"type": "PROD", "mode": "INC", "value": "4087"}]}, "meth": "StationBoard"}], "client": self.clientInfo, "ver": self.version, "lang": self.language, "auth": self.auth}), + headers=self.headers) + return json.loads(res.text) + + def tripRequest(self, departure): + res = self.session.post(self.baseUrl, + data = json.dumps({"svcReqL": [{"req": {"jid": departure}, "meth": "JourneyDetails"}], "client": self.clientInfo, "ver": self.version, "lang": self.language, "auth": self.auth}), + headers=self.headers) + return json.loads(res.text) + + + + def getStationNames(self, stationString): + res = self.stationRequest(stationString) + return [(station["name"], station["extId"]) for station in res["svcResL"][0]["res"]["match"]["locL"]] + + def getArrDep(self, stationId, arrdep="DEP", count=1): + res = self.arrDepRequest(stationId, arrdep, count) + return res["svcResL"][0]["res"]["jnyL"] + + def getTrip(self, departure): + res = self.tripRequest(departure) + return res["svcResL"][0]["res"] + + + + def runTests(self): + stationInput = input("Enter Station Name: ") + station = self.getStationNames(stationInput) + print(station) + departures = self.getArrDep(station[0][1], arrdep="DEP", count=5) + print(departures) + + print("\n)------------------\n") + arrivals = self.getArrDep(station[0][1], arrdep="ARR", count=5) + print(arrivals) + + print("\n)------------------\n") + trip = self.getTrip(departures[0]["jid"]) + __import__('pprint').pprint(trip) + + + # streq = self.stationRequest(station)["svcResL"][0]["res"]["match"]["locL"] + # + # selectedStation = streq[0] # should be selected using ptk + # __import__('pprint').pprint(selectedStation) + # print(selectedStation["lid"]) + # dpreq = self.departuresRequest(selectedStation["lid"], departuresCount=5) + # + # nextDepartures = dpreq["svcResL"][0]["res"]["jnyL"] + # __import__('pprint').pprint(nextDepartures) + # for departure in nextDepartures: + # print(f"{datetime.strptime(departure["stbStop"]["dTimeS"], "%H%M%S").strftime("%H:%M")} / {datetime.strptime(departure["stbStop"]["dTimeR"], "%H%M%S").strftime("%H:%M")} {departure["dirFlg"]} --> {departure["dirTxt"]}") + # print(nextDepartures) -def tripRequest(departure): - res = session.post("https://fahrplan.ivb.at/bin/mgate.exe", - data = json.dumps({"svcReqL": [{"req": {"jid": departure}, "meth": "JourneyDetails"}], "client": {"id": "VAO", "name": "webapp", "type": "WEB"}, "ver": "1.32", "lang": "de", "auth": {"type": "AID", "aid": "wf7mcf9bv3nv8g5f"}}), - headers={'User-Agent': "IOS",'Content-Type': 'application/json'}) - return json.loads(res.text) if __name__ == '__main__': - session = requests.session() - station = input("Enter Station Name: ") - streq = stationRequest(station)["svcResL"][0]["res"]["match"]["locL"] - - selectedStation = streq[0] # should be selected using ptk - __import__('pprint').pprint(selectedStation) - print(selectedStation["lid"]) - dpreq = departuresRequest(selectedStation["lid"], departures=5) - - nextDepartures = dpreq["svcResL"][0]["res"]["jnyL"] - __import__('pprint').pprint(nextDepartures) - for departure in nextDepartures: - print(f"{datetime.strptime(departure["stbStop"]["dTimeS"], "%H%M%S").strftime("%H:%M")} / {datetime.strptime(departure["stbStop"]["dTimeR"], "%H%M%S").strftime("%H:%M")} {departure["dirFlg"]} --> {departure["dirTxt"]}") - print(nextDepartures) + client = HafasClient() + client.runTests() diff --git a/main.py b/main.py index b43be72..b33ec64 100644 --- a/main.py +++ b/main.py @@ -6,3 +6,90 @@ haTerm is a Terminal Based hafas client, using the ivb endpoint. Prompt_toolkit (https://github.com/prompt-toolkit/python-prompt-toolkit) will be used to render a terminal user interface (TUI). The application will provide routing information and station departures/arrivals. """ + +from backend import HafasClient +import route_planning +import station_monitor + +from prompt_toolkit import Application +from prompt_toolkit.layout import Layout, HSplit, VSplit, Window +from prompt_toolkit.widgets import Button, Box +from prompt_toolkit.key_binding import KeyBindings + + +def pk_menu(client: HafasClient): + """Display a centered menu with two buttons selectable by arrow keys. + + The function runs a full-screen prompt_toolkit application. Left/right + arrows move focus between the buttons; Enter activates the focused + button. The function returns the chosen mode as a string or None when + the user quits. + """ + choice = {'value': None} + + def on_rp(): + choice['value'] = 'rp' + app.exit() + + def on_sm(): + choice['value'] = 'sm' + app.exit() + + btn_rp = Button("Route Planning", handler=on_rp, width=20) + btn_sm = Button("Station Monitor", handler=on_sm, width=20) + + buttons = [btn_rp, btn_sm] + index = {'i': 0} + + kb = KeyBindings() + + def focus_current(): + app.layout.focus(buttons[index['i']]) + + @kb.add('left') + def _left(event): + index['i'] = max(0, index['i'] - 1) + focus_current() + + @kb.add('right') + def _right(event): + index['i'] = min(len(buttons) - 1, index['i'] + 1) + focus_current() + + @kb.add('q') + @kb.add('c-q') + @kb.add('c-c') + def _quit(event): + app.exit() + + root = HSplit(children=[ + Window(), + Box(VSplit(children=[ + Window(), + Box(VSplit(children=buttons, padding=6), padding=2, style="bg:#2A71D5 fg:#FFFFFF"), + Window() + ])), + Window() + ], align='CENTER') + + app = Application(layout=Layout(root, focused_element=btn_rp), key_bindings=kb, full_screen=True) + app.run() + return choice['value'] + + +def main(): + client = HafasClient() + + while True: + result = pk_menu(client) + if result == 'rp': + route_planning.run_route_planning(client) + elif result == 'sm': + station_monitor.run_station_monitor(client) + else: + print("Exiting.") + break + + +if __name__ == '__main__': + main() diff --git a/route_planning.py b/route_planning.py index 8217d02..126cfd8 100644 --- a/route_planning.py +++ b/route_planning.py @@ -5,10 +5,16 @@ haTerm - v0.1 haTerm is a Terminal Based hafas client, using the ivb endpoint. Prompt_toolkit (https://github.com/prompt-toolkit/python-prompt-toolkit) will be used to render a terminal user interface (TUI). The application will provide routing information and station departures/arrivals. -""" -#routenplanung mit promt tool kit -import prompt_toolkit as pk +Route planning UI. + +Wrapped in a function so it can be launched from `main.py` with a shared +`HafasClient` instance from `backend.py`. + +Note: This module is a minimal TUI stub. Many features are not implemented +yet (station lookup, route rendering). Add TODOs where appropriate. +""" + import time from prompt_toolkit import Application from prompt_toolkit.buffer import Buffer @@ -21,45 +27,57 @@ from prompt_toolkit.shortcuts import yes_no_dialog from prompt_toolkit.shortcuts import input_dialog from prompt_toolkit import prompt -kb = KeyBindings() +import backend -start = prompt("Start: ") -end = prompt("Ende: ") + +def run_route_planning(hafas: backend.HafasClient | None = None): + """Run the route planning TUI. + + Parameters: + - hafas: optional shared HafasClient instance. If omitted, a new one is + created. Prefer passing the shared client from `main.py`. + """ + if hafas is None: + hafas = backend.HafasClient() + + # TODO: integrate actual station search + routing using `hafas`. + kb = KeyBindings() + + start = prompt("Start: ") + end = prompt("Ende: ") -startBuffer = Buffer() -startBuffer.text = start + startBuffer = Buffer() + startBuffer.text = start -etdBuffer = Buffer() -etdBuffer.text = "13:00" #testweise, später mit tatsächlicher Abfahrtszeit von Hafas ersetzen -#etdBuffer.text = (ungefähre abfahrtszeit) + etdBuffer = Buffer() + etdBuffer.text = "13:00" #testweise, später mit tatsächlicher Abfahrtszeit von Hafas ersetzen + #etdBuffer.text = (ungefähre abfahrtszeit) -endBuffer = Buffer() -endBuffer.text = end + endBuffer = Buffer() + endBuffer.text = end -etaBuffer = Buffer() -etaBuffer.text = "13:45" #testweise, später mit tatsächlicher Ankunftszeit von Hafas ersetzen -#etaBuffer.text = (ungefähre ankunftszeit) + etaBuffer = Buffer() + etaBuffer.text = "13:45" #testweise, später mit tatsächlicher Ankunftszeit von Hafas ersetzen + #etaBuffer.text = (ungefähre ankunftszeit) -drivetimeBuffer = Buffer() -#drivetimeBuffer.text = Fahrzeit von Hafas. + drivetimeBuffer = Buffer() + #drivetimeBuffer.text = Fahrzeit von Hafas. -infoBuffer = Buffer() -infoBuffer.text = f"Routenplanung von {start} nach {end}" - - -root_container = HSplit(children=[ + infoBuffer = Buffer() + infoBuffer.text = f"Routenplanung von {start} nach {end}" + root_container = HSplit(children=[ + HSplit(children=[ Window(height=2, content=BufferControl(buffer=infoBuffer, focusable=False), style="fg:#2A71D5"), #informationen über Route Window(width=1, height = 1, char='-', style="fg:#2A71D5"), #Trennlinie - Window(content=BufferControl(buffer=startBuffer, focusable=True)), #start Station Window(content=BufferControl(buffer=etdBuffer, focusable=False)), #abfahrtszeit Window(width = 1, height = 2, char= ".", style= "fg:#A86FD6"), #Trennlinie @@ -70,24 +88,27 @@ root_container = HSplit(children=[ #Window(content=BufferControl(buffer=drivetimeBuffer, focusable=False)), #Window(Content) - ]), - Window(height=1, char=' ', style="bg:#2A71D5 fg:black"), - #Window(content=BufferControl(buffer=userBuffer),height=4), -]) + ]), + Window(height=1, char=' ', style="bg:#2A71D5 fg:black"), + ]) -@kb.add('c-e') -def _(event): - result = yes_no_dialog( - title='Programm beenden', - text='Fenster schließen?').run() - if result == True: - event.app.exit() + @kb.add('c-e') + def _(event): + result = yes_no_dialog( + title='Programm beenden', + text='Fenster schließen?').run() + if result == True: + event.app.exit() -layout = Layout(root_container) + layout = Layout(root_container) + app = Application(layout=layout, key_bindings=kb, full_screen=True) -app = Application(layout=layout, key_bindings=kb, full_screen=True) + try: + app.run() + except KeyboardInterrupt: + # graceful exit on Ctrl-C + return -app.run() \ No newline at end of file diff --git a/station_monitor.py b/station_monitor.py index ffa236c..4450f4c 100644 --- a/station_monitor.py +++ b/station_monitor.py @@ -6,3 +6,84 @@ haTerm is a Terminal Based hafas client, using the ivb endpoint. Prompt_toolkit (https://github.com/prompt-toolkit/python-prompt-toolkit) will be used to render a terminal user interface (TUI). The application will provide routing information and station departures/arrivals. """ + +import backend +from prompt_toolkit import Application +from prompt_toolkit.buffer import Buffer +from prompt_toolkit.layout.layout import Layout +from prompt_toolkit.layout.containers import Window, VSplit, HSplit +from prompt_toolkit.layout.controls import BufferControl +from prompt_toolkit.key_binding import KeyBindings +import threading +import time + + +def run_station_monitor(hafas: backend.HafasClient | None = None): + """Run the station monitor TUI. + + Parameters: + - hafas: optional shared HafasClient instance. If omitted, a new one is + created. Prefer passing the shared client from `main.py`. + + Note: Some parts of this monitor are incomplete; see TODO markers. + """ + if hafas is None: + hafas = backend.HafasClient() + + keyB = KeyBindings() + inputBuffer = Buffer() + resultBuffer = Buffer() + stop_event = threading.Event() + app_state = "MONITOR" + + inputBuffer.text = "Haltestelle eingeben: " + inputBuffer.cursor_position = len(inputBuffer.text) + resultBuffer.text = "Ausgabe: " + + root_container = HSplit(children=[ + VSplit(children=[ + Window(content=BufferControl(buffer=inputBuffer, focusable=True)), + Window(width=1, char='│', style="fg:#9D1D75"), + Window(content=BufferControl(buffer=resultBuffer, focusable=True)), + ]), + Window(height=1, char='-', style="bg:#9D1D75 fg:#FFFFFF"), + ]) + + layout = Layout(root_container, focused_element=inputBuffer) + + @keyB.add("enter") + def handle_enter(event): + global app_state + + user_input = inputBuffer.text.replace("Haltestelle eingeben: ", "").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: + app_state = "MONITOR" + + @keyB.add("c-q") + def exit_(event): + stop_event.set() + event.app.exit() + + app = Application(layout=layout, full_screen=True, key_bindings=keyB) + + try: + app.run() + except KeyboardInterrupt: + pass + finally: + stop_event.set()