149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
import requests
|
|
from datetime import datetime, timedelta
|
|
import xmltodict
|
|
from cachetools import cached, TTLCache
|
|
|
|
eva = 8002377
|
|
|
|
base_url = "https://iris.noncd.db.de/iris-tts/timetable"
|
|
|
|
@cached(cache=TTLCache(maxsize=128, ttl=60))
|
|
def _get_changes(eva):
|
|
url = f"{base_url}/fchg/{eva}"
|
|
x = requests.get(url)
|
|
changes = xmltodict.parse(x.text)['timetable']['s']
|
|
return changes
|
|
|
|
class Departure:
|
|
def __init__(self, event, station=None, eva=None):
|
|
dp = event['dp']
|
|
self.id = event['@id']
|
|
try:
|
|
self.line = f"{event['tl']['@c']}{dp['@l']}"
|
|
except KeyError:
|
|
self.line = '??'
|
|
self.time = datetime.strptime(dp['@pt'], "%y%m%d%H%M")
|
|
self.old_time = self.time
|
|
self.delayed = None
|
|
self.platform = dp['@pp']
|
|
path = dp['@ppth']
|
|
self.target = path.split('|')[-1]
|
|
self.path = path.split('|')
|
|
self.station = station
|
|
self.eva = eva
|
|
|
|
|
|
def fetch_actual(self, eva):
|
|
changes = _get_changes(eva)
|
|
for event in changes:
|
|
if event['@id'] == self.id:
|
|
self.time = datetime.strptime(event['dp']['@ct'], "%y%m%d%H%M")
|
|
self.delayed = self.time - self.old_time
|
|
self.delayed = int(self.delayed.total_seconds()/60)
|
|
if self.delayed == 0:
|
|
self.delayed = None
|
|
pass
|
|
|
|
def __repr__(self):
|
|
|
|
return f"{self.line} to {self.target}: {self.time.strftime("%Y-%m-%d %H:%M")}{f" ({self.delayed:+d})" if self.delayed is not None else ""} @ platform {self.platform}"
|
|
|
|
def main():
|
|
|
|
# current_timestamp = datetime(year=2025, month=8, day=22, hour=22, minute=51, second=0)
|
|
retrieve_and_print_schedule(eva, number_future_deps=100)
|
|
|
|
pass
|
|
|
|
def retrieve_and_print_schedule(eva, target_timestamp = None, future_only = True, number_future_deps=6):
|
|
current_timestamp = datetime.now() if target_timestamp is None else target_timestamp
|
|
|
|
timetable = fetch_timetable(current_timestamp, eva)
|
|
departures = extract_departures(timetable, eva)
|
|
|
|
if current_timestamp.minute < 10 and not future_only:
|
|
timetable = fetch_timetable(current_timestamp - timedelta(hours=1), eva)
|
|
departures = extract_departures(timetable, eva, departures)
|
|
# elif current_timestamp.minute > 50:
|
|
# # current_timestamp = timedelta(hours=1)
|
|
# timetable = fetch_timetable(current_timestamp + timedelta(hours=1), eva)
|
|
# departures = extract_departures(timetable, eva, departures)
|
|
|
|
if future_only:
|
|
split_index = -1
|
|
for di, dep in enumerate(departures):
|
|
if dep.time < current_timestamp:
|
|
split_index = di
|
|
departures = departures[split_index+1:]
|
|
|
|
cnt = 0
|
|
while len(departures) < number_future_deps and cnt <= 24:
|
|
current_timestamp += timedelta(hours=1)
|
|
timetable = fetch_timetable(current_timestamp, eva)
|
|
departures = extract_departures(timetable, eva, departures)
|
|
cnt += 1
|
|
|
|
# departures = departures[:number_future_deps]
|
|
|
|
print(f"{departures[0].station} ({departures[0].station})")
|
|
for abf in departures:
|
|
# if abf.time >= current_timestamp:
|
|
print(abf)
|
|
|
|
def fetch_timetable(target_datetime: datetime, eva):
|
|
|
|
"""
|
|
timetable:
|
|
@station: // station name
|
|
s: // stop, contains list of the following
|
|
@id // id
|
|
tl: // trip label
|
|
@f // distance class, F: Fern, N: Nah, S: Stadt
|
|
@t // trip type: e, p, z, s, h, n -> normally p
|
|
@o // owner: EVU-Number 800725 is S-Bahn München
|
|
@c // category: CE, IC, EC, IRE, RE, RB, S, MEX, TGJ, NJ, Bus
|
|
@n // train number
|
|
ar: // arrival
|
|
see dp
|
|
dp: // departure
|
|
@pt // yyMMddHHmm
|
|
@pp // platform number
|
|
@l // line. -> tl.@c + dp.@l make up the train line (eg S3)
|
|
@ppth // path, stations separated by |. for the last station in dp.@ppth is the destination, the first station in ar.@ppth is the origin
|
|
|
|
"""
|
|
|
|
daystamp = target_datetime.strftime(r"%y%m%d")
|
|
hourstamp = target_datetime.strftime(r"%H")
|
|
|
|
url = f"{base_url}/plan/{eva}/{daystamp}/{hourstamp}"
|
|
x = requests.get(url)
|
|
if x.status_code != 200:
|
|
return None
|
|
timetable = xmltodict.parse(x.text)['timetable']
|
|
return timetable
|
|
|
|
def extract_departures(timetable, eva, departures=None):
|
|
if departures is None:
|
|
departures = []
|
|
if timetable is None:
|
|
return departures
|
|
|
|
events = timetable['s']
|
|
if not isinstance(events, list):
|
|
events = [events]
|
|
for event in events:
|
|
if event['tl']['@f'] != 'S':
|
|
# only s-bahnen are supported for now
|
|
continue
|
|
departure = Departure(event, station=timetable['@station'], eva=eva)
|
|
departure.fetch_actual(eva)
|
|
departures.append(departure)
|
|
|
|
departures.sort(key=lambda abf: abf.time)
|
|
return departures
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|