Stefan Bethke
0adabf7758
All checks were successful
docker-image / docker (push) Successful in 10m11s
By specifying a layout, you can format the page directly, instead of having to overrride write_row.
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
"""
|
|
Pull departure info from https://trains.xatlabs.com and display.
|
|
|
|
See also https://github.com/derf/db-fakedisplay/blob/main/README.md
|
|
"""
|
|
import datetime
|
|
import logging
|
|
from threading import Thread
|
|
from time import sleep
|
|
|
|
from deutschebahn.db_infoscreen import DBInfoscreen
|
|
|
|
from buba.bubaanimator import BubaAnimation, LineLayoutColumn
|
|
from buba.bubacmd import BubaCmd
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
class DBFAnimation(BubaAnimation):
|
|
dbf_layout = [
|
|
LineLayoutColumn(12, BubaCmd.ALIGN_LEFT),
|
|
LineLayoutColumn(81, BubaCmd.ALIGN_LEFT),
|
|
LineLayoutColumn(27, BubaCmd.ALIGN_RIGHT),
|
|
]
|
|
|
|
def __init__(self, buba: BubaCmd, ds100="AHST", station="Holstenstraße", count=3):
|
|
super().__init__(buba)
|
|
self.dbi = DBInfoscreen("trains.xatlabs.com")
|
|
self.ds100 = ds100
|
|
self.station = station
|
|
self.trains = []
|
|
self.count = count
|
|
Thread(target=self.update, daemon=True).start()
|
|
|
|
def __repr__(self):
|
|
return f"<{type(self).__name__}, {self.ds100}>"
|
|
|
|
def fetch(self):
|
|
trains = self.dbi.calc_real_times(self.dbi.get_trains(self.ds100)) # Station
|
|
# trains = [t for t in trains] # if t['platform'] == "1"] # platform gleis
|
|
trains.sort(key=self.dbi.time_sort)
|
|
self.trains = trains
|
|
self.log.info(f"Fetched {len(trains)} trains")
|
|
|
|
def update(self):
|
|
while True:
|
|
try:
|
|
self.fetch()
|
|
except Exception as e:
|
|
self.log.warning(f"Unable to fetch {self.station}: {e}")
|
|
pass
|
|
sleep(60)
|
|
|
|
@staticmethod
|
|
def countdown(dt: datetime, cancelled:bool):
|
|
if cancelled:
|
|
return "--"
|
|
now = datetime.datetime.now().astimezone()
|
|
try:
|
|
day = now.strftime("%y-%m-%d")
|
|
departure = datetime.datetime.strptime(f"{day} {dt}", "%y-%m-%d %H:%M").astimezone()
|
|
except ValueError as e:
|
|
return "--"
|
|
|
|
dep_td = departure - now
|
|
if dep_td.total_seconds() <= -3600:
|
|
# dep_date += datetime.timedelta(days=1)
|
|
departure += datetime.timedelta(days=1)
|
|
if dep_td.total_seconds() >= 3600 * 23:
|
|
# dep_date -= datetime.timedelta(days=1)
|
|
departure -= datetime.timedelta(days=1)
|
|
|
|
return BubaAnimation.countdown(departure)
|
|
|
|
@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"
|
|
if station == "Wedel(Holst)":
|
|
station = "Wedel"
|
|
return station
|
|
|
|
@staticmethod
|
|
def short_train(train: str) -> str:
|
|
if train.startswith("ICE"):
|
|
train = "ICE"
|
|
if train.startswith("EC"):
|
|
train = "EC"
|
|
if train.startswith("EN"):
|
|
train = "EN"
|
|
if train.startswith("ME"):
|
|
train = "ME"
|
|
if train.startswith("NJ"):
|
|
train = "NJ"
|
|
if train.startswith("RB"):
|
|
train = "RB"
|
|
if train.startswith("RE"):
|
|
train = "RE"
|
|
train = train.replace(" ", "")
|
|
return train
|
|
|
|
def run(self):
|
|
self.pages(self.station, [[
|
|
self.short_train(train['train']),
|
|
self.short_station(train['destination']),
|
|
self.countdown(train['actualDeparture'], train['isCancelled']),
|
|
] for train in self.trains[:self.count]], layout=self.dbf_layout)
|