Add station departure display

buba/animations/dbf.py

@@ -0,0 +1,123 @@
+"""
+Pull departure info from https://trains.xatlabs.com and display.
+
+See also https://github.com/derf/db-fakedisplay/blob/main/README.md
+"""
+import datetime
+from itertools import islice
+from threading import Thread
+from time import sleep
+
+from deutschebahn.db_infoscreen import DBInfoscreen
+
+from buba.bubaanimator import BubaAnimation
+from buba.bubacmd import BubaCmd
+
+
+class DBFAnimation(BubaAnimation):
+    def __init__(self, buba: BubaCmd, ds100="AHST", station="Holstenstraße", count=3):
+        super().__init__(buba)
+        self.ds100 = ds100
+        self.station = station
+        self.trains = []
+        self.count = count
+        Thread(target=self.update, daemon=True).start()
+
+    def fetch(self):
+        dbi = DBInfoscreen("trains.xatlabs.com")
+        trains = dbi.calc_real_times(dbi.get_trains(self.ds100))  # Station
+        trains = [t for t in trains]  # if t['platform'] == "1"]  # platform gleis
+        trains.sort(key=dbi.time_sort)
+        self.trains = trains
+
+    def update(self):
+        self.fetch()
+        sleep(60)
+
+    @staticmethod
+    def countdown(train):
+        now = datetime.datetime.now()
+        try:
+            dep_time = datetime.datetime.strptime(train['actualDeparture'], "%H:%M").time()
+        except TypeError as e:
+            return -1
+
+        # First, assume all departure times are on the current day
+        dep_date = now.date()
+
+        # Calculate timedelta under the above assumption
+        dep_td = datetime.datetime.combine(dep_date, dep_time) - now
+
+        # If the calculated timedelta is more than one hour in the past,
+        # assume that the day should actually be the next day
+        # (This will be the case e.g. when a train departs at 00:15 and it's currently 23:50)
+        if dep_td.total_seconds() <= -3600:
+            dep_date += datetime.timedelta(days=1)
+
+        # If the calculated timedelta is more than 23 hours in the future,
+        # assume that the day should actually be the previous day.
+        # (This will be the case e.g. when a train should have departed at 23:50 but it's already 00:15)
+        if dep_td.total_seconds() >= 3600 * 23:
+            dep_date -= datetime.timedelta(days=1)
+
+        # Recalculate the timedelta
+        dep_td = datetime.datetime.combine(dep_date, dep_time) - now
+        return round(dep_td.total_seconds() / 60)
+
+    @staticmethod
+    def short_station(station: str) -> str:
+        station = station.strip()
+        if station.startswith("Hamburg-") or station.startswith("Hamburg "):
+            station = station[8:]
+        if station.endswith("(S)"):
+            station = station[:-3]
+        if station.endswith("(S-Bahn)"):
+            station = station[:-8]
+        if station == "Hbf":
+            station = "Hauptbahnhof"
+        return station
+
+    @staticmethod
+    def short_train(train: str) -> str:
+        if train.startswith("ICE"):
+            train = "ICE"
+        if train.startswith("ME"):
+            train = "ME"
+        if train.startswith("RE"):
+            train = "RE"
+        train = train.replace(" ", "")
+        return train
+
+    @staticmethod
+    def chunk(it, size):
+        it = iter(it)
+        return iter(lambda: tuple(islice(it, size)), ())
+
+    def run(self):
+        all = self.trains[:self.count]
+        all_len = int((len(all)+1)/3)
+
+        if len(self.trains) == 0:
+            sleep(5)
+
+        for page, trains in enumerate(self.chunk(all, 3)):
+            if all_len == 1:
+                title = self.station
+            else:
+                title = f"{self.station} ({page+1}/{all_len})"
+            self.buba.text(page=0, row=0, col_start=0, col_end=120, text=title, align=BubaCmd.ALIGN_LEFT)
+            for i, train in enumerate(trains):
+                if train['isCancelled']:
+                    when = "--"
+                else:
+                    when = self.countdown(train)
+                    if when < 0:
+                        when = 0
+                self.buba.text(page=0, row=i + 1, col_start=0, col_end=120, text=self.short_train(train['train']))
+                self.buba.text(page=0, row=i + 1, col_start=12, col_end=76, text=self.short_station(train['destination']))
+                self.buba.text(page=0, row=i + 1, col_start=105, col_end=120,
+                               text=str(when), align=BubaCmd.ALIGN_RIGHT)
+            self.buba.set_page(0)
+            for i in range(10):
+                self.buba.text(page=0, row=0, col_start=93, col_end=120, text=datetime.datetime.now().strftime("%H:%M:%S"), align=BubaCmd.ALIGN_RIGHT)
+                sleep(1)