1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
#!/usr/bin/env python3
# vim:tabstop=4 softtabstop=4 shiftwidth=4 textwidth=160 smarttab expandtab colorcolumn=160
#
# Copyright (C) 2021 Daniel Friesel
#
# SPDX-License-Identifier: BSD-2-Clause
import argparse
import aiohttp
from aiohttp import web
from datetime import datetime
import dateutil.parser
from jinja2 import Environment, FileSystemLoader, select_autoescape
import json
import os
headers = {
"Access-Control-Allow-Origin": "*",
"Content-Type": "text/html; charset=utf-8",
}
db_rest_api = os.getenv("NVM_DB_REST_API", "https://v5.db.transport.rest")
env = Environment(loader=FileSystemLoader("templates"), autoescape=select_autoescape())
class Departure:
def __init__(self, obj):
self.__dict__.update(obj)
if not "cancelled" in obj:
self.cancelled = False
self.classes = str()
if "," in self.direction:
self.direction, self.suffix = self.direction.split(",", maxsplit=1)
else:
self.suffix = None
if "line" in obj:
self.line = Line(self.line)
try:
self.when = dateutil.parser.parse(self.when)
except TypeError:
self.when = None
try:
self.plannedWhen = dateutil.parser.parse(self.plannedWhen)
except TypeError:
self.plannedWhen = None
if self.cancelled:
self.classes += " cancelled"
if self.when:
self.sort_by = self.when.timestamp()
elif self.plannedWhen:
self.sort_by = self.plannedWhen.timestamp()
else:
self.sort_by = 0
def set_relative(self, now):
minutes = (self.sort_by - now) // 60
if minutes < 1:
self.relativeWhen = "sofort"
elif minutes < 60:
self.relativeWhen = f"{minutes:.0f} min"
else:
self.relativeWhen = f"{minutes//60:.0f}h {minutes%60:.0f}min"
class Line:
def __init__(self, obj):
self.__dict__.update(obj)
self.css_class = str()
if self.product == "tram":
self.name = self.name.removeprefix("STR ")
self.css_class = "tram"
elif self.product == "suburban":
self.css_class = "suburban"
elif self.product == "subway":
self.css_class = "subway"
elif self.product == "bus":
self.name = self.name.removeprefix("Bus ")
self.css_class = "bus"
async def show_departure_board(request):
try:
eva = int(request.match_info.get("eva"))
except ValueError:
return web.HTTPBadRequest(text="EVA must be a number at the moment")
async with aiohttp.ClientSession() as session:
async with session.get(
f"{db_rest_api}/stops/{eva}/departures?results=60&duration=120&stopovers=true"
) as response:
departures = await response.text()
departures = json.loads(departures)
if type(departures) is dict and departures.get("error", False):
return web.HTTPNotFound(body=json.dumps(departures), headers=headers)
departures = list(map(Departure, departures))
now = datetime.now().timestamp()
for departure in departures:
departure.set_relative(now)
departure_board = env.get_template("departure_list.html")
return web.Response(
body=departure_board.render(title="Noot", departures=departures),
headers=headers,
)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="eva to efa gateway")
parser.add_argument("--port", type=int, metavar="PORT", default=8080)
parser.add_argument("--prefix", type=str, metavar="PATH", default="/")
args = parser.parse_args()
app = web.Application()
app.router.add_get(f"{args.prefix}board/{{eva}}", show_departure_board)
app.router.add_static(f"{args.prefix}static", "static")
web.run_app(app, host="localhost", port=args.port)
|
