143 lines
6.2 KiB
Python
143 lines
6.2 KiB
Python
"""
|
|
haTerm - v0.1
|
|
(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).
|
|
The application will provide routing information and station departures/arrivals.
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from prompt_toolkit import Application
|
|
from prompt_toolkit.completion import WordCompleter
|
|
|
|
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 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 _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)
|
|
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": re.search(r'#ZB#([^#]+)', departure.get("jid")).group(1).strip(),
|
|
"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)
|
|
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)
|
|
|
|
def monitorTest(self):
|
|
stationInput = input("Enter Station Name: ")
|
|
station = self.getStationNames(stationInput)
|
|
print(station)
|
|
# departures = self.getArrDep(station[0][1], arrdep="DEP", count=1)
|
|
print("\n)------------------\n")
|
|
departures = self.getArrDep(station[0][1], arrdep="DEP", count=2)
|
|
__import__('pprint').pprint(departures)
|
|
print("\n)------------------\n")
|
|
arrivals = self.getArrDep(station[0][1], arrdep="ARR", count=2)
|
|
__import__('pprint').pprint(arrivals)
|
|
#arrivalsRaw = self.arrDepRequest(station[0][1], arrdep="ARR", count=1)
|
|
#__import__('pprint').pprint(arrivalsRaw)
|
|
|
|
|
|
# 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__':
|
|
client = HafasClient()
|
|
# client.runTests()
|
|
client.monitorTest()
|