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
|
#!/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 "line" in obj:
self.line = Line(self.line)
self.when = dateutil.parser.parse(self.when)
self.plannedWhen = dateutil.parser.parse(self.plannedWhen)
class Line:
def __init__(self, obj):
self.__dict__.update(obj)
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))
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)
|
