109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
import threading
|
|
from wsgiref.simple_server import make_server, WSGIRequestHandler
|
|
from operator import itemgetter
|
|
from departures import Departure
|
|
from bottle import route, error, run, get, ServerAdapter, static_file, request
|
|
|
|
class Server(ServerAdapter):
|
|
server = None
|
|
def run(self, handler):
|
|
if self.quiet:
|
|
class QuietHandler(WSGIRequestHandler):
|
|
def log_request(self, *args, **kw):
|
|
pass
|
|
self.options['handler_class'] = QuietHandler
|
|
self.server = make_server(self.host, self.port, handler, **self.options)
|
|
self.server.serve_forever()
|
|
def stop(self):
|
|
self.server.shutdown()
|
|
|
|
class API:
|
|
|
|
def __init__(self, main):
|
|
|
|
@route('/departures/<stop_id>')
|
|
def departures(stop_id: str):
|
|
resp = []
|
|
for d in Departure.get(stop_id):
|
|
resp.append(d.json())
|
|
resp.sort(key=itemgetter("departure"))
|
|
return {'departures': resp}
|
|
|
|
@route('/stops')
|
|
def stop():
|
|
return {'stops': main.config["stops"]}
|
|
|
|
@get("/static/<file>")
|
|
def static(file: str):
|
|
return static_file(file, root="static")
|
|
|
|
@get("/")
|
|
def index():
|
|
return static_file("index.html", root="static")
|
|
|
|
@get("/admin")
|
|
def admin():
|
|
return static_file("admin.html", root="static")
|
|
|
|
@route('/admin/devices', method='POST')
|
|
def admin_devices():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
return {'devices': main.controller.json()}
|
|
|
|
@route('/admin/devices/resend', method='POST')
|
|
def admin_devices_resend():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
for i, d in enumerate(main.controller.devices):
|
|
if d.stop_id and str(i) == request.forms.get('index'):
|
|
for dep in Departure.get(d.stop_id):
|
|
dep.resend(d.id)
|
|
return {'success': True}
|
|
|
|
@route('/admin/devices/clear', method='POST')
|
|
def admin_devices_clear():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
for i, d in enumerate(main.controller.devices):
|
|
if str(i) == request.forms.get('index'):
|
|
d.clear()
|
|
return {'success': True}
|
|
|
|
@route('/admin/devices/change_stop', method='POST')
|
|
def admin_devices_change_stop():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
for i, d in enumerate(main.controller.devices):
|
|
if str(i) == request.forms.get('index'):
|
|
d.set_stop(request.forms.get('stop_id'))
|
|
return {'success': True}
|
|
|
|
@route('/admin/devices/change_filter', method='POST')
|
|
def admin_devices_change_filter():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
for i, d in enumerate(main.controller.devices):
|
|
if str(i) == request.forms.get('index'):
|
|
d.set_filter(request.forms.get('filter'))
|
|
return {'success': True}
|
|
|
|
@route('/admin/devices/toggle', method='POST')
|
|
def admin_devices_toggle():
|
|
if main.config["admin"]["secret"] != request.forms.get('secret'):
|
|
return {'error': "unauthorized"}
|
|
for i, d in enumerate(main.controller.devices):
|
|
if str(i) == request.forms.get('index'):
|
|
d.toggle(request.forms.get('state') == "true")
|
|
return {'success': True}
|
|
|
|
@error(404)
|
|
def error404(err):
|
|
return ''
|
|
|
|
self.server = Server(host=main.config["http"]["host"], port=main.config["http"]["port"])
|
|
self.thread = threading.Thread(target=self.start)
|
|
self.thread.start()
|
|
|
|
def start(self):
|
|
run(quiet=True, server=self.server) |