first draft

main.py

@@ -0,0 +1,148 @@
+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()