Merge branch 'main' of https://git.miaig.dev/mia/hafas-terminal-app
This commit is contained in:
commit
d0a67d5a18
4 changed files with 303 additions and 70 deletions
100
backend.py
100
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'})
|
||||
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 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 departuresRequest(station, departures=1):
|
||||
def arrDepRequest(self, station, arrdep, count):
|
||||
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'})
|
||||
# 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(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'})
|
||||
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)
|
||||
|
||||
|
||||
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()
|
||||
|
|
|
|||
87
main.py
87
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()
|
||||
|
|
|
|||
|
|
@ -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,6 +27,20 @@ from prompt_toolkit.shortcuts import yes_no_dialog
|
|||
from prompt_toolkit.shortcuts import input_dialog
|
||||
from prompt_toolkit import prompt
|
||||
|
||||
import backend
|
||||
|
||||
|
||||
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: ")
|
||||
|
|
@ -53,13 +73,11 @@ drivetimeBuffer = Buffer()
|
|||
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
|
||||
|
|
@ -72,7 +90,6 @@ root_container = HSplit(children=[
|
|||
#Window(Content)
|
||||
]),
|
||||
Window(height=1, char=' ', style="bg:#2A71D5 fg:black"),
|
||||
#Window(content=BufferControl(buffer=userBuffer),height=4),
|
||||
])
|
||||
|
||||
|
||||
|
|
@ -87,7 +104,11 @@ def _(event):
|
|||
|
||||
|
||||
layout = Layout(root_container)
|
||||
|
||||
app = Application(layout=layout, key_bindings=kb, full_screen=True)
|
||||
|
||||
try:
|
||||
app.run()
|
||||
except KeyboardInterrupt:
|
||||
# graceful exit on Ctrl-C
|
||||
return
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue