130 lines
4.7 KiB
Python
130 lines
4.7 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
|
|
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 __repr__(self):
|
|
return f"<{type(self).__name__}, {self.ds100}>"
|
|
|
|
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("NJ"):
|
|
train = "NJ"
|
|
if train.startswith("RB"):
|
|
train = "RB"
|
|
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=92, 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=11, text=self.short_train(train['train']))
|
|
self.buba.text(page=0, row=i + 1, col_start=12, col_end=104, text=self.short_station(train['destination']))
|
|
self.buba.text(page=0, row=i + 1, col_start=105, col_end=119,
|
|
text=str(when), align=BubaCmd.ALIGN_RIGHT)
|
|
self.buba.set_page(0)
|
|
for i in range(5):
|
|
self.buba.text(page=0, row=0, col_start=93, col_end=119, text=datetime.datetime.now().strftime("%H:%M"), align=BubaCmd.ALIGN_RIGHT)
|
|
sleep(2)
|