import builtins #builtins.print = lambda *args, **kwargs: None import v, _html from v import * #### TESTING 2026 gtfs_lock = threading.Lock() #### TESTING 2026 def _get_deviations(country, operator, station): if country == "se": if operator == "sl": return sl_get_deviations(country, operator, station) def add_error(error_string): print(error_string) if not error_string in v.errors.values(): v.errors[time.mktime(time.localtime())] = error_string time.sleep(1) def validate_request(country, operator, station): v.operators.setdefault(country, {}).setdefault(operator, {}).setdefault(station, {"timestamp": 0, "departures": ""}) # "timestamp":time.mktime(time.localtime())} v.online.setdefault(country, {}).setdefault(operator, False) def get_departures(country, operator, station): try: _delay = v.departures_update_delay[str(operator)] except: _delay = v.departures_update_delay["default"] print("DELAY: ", _delay) validate_request(country, operator, station) try: #if not time.mktime(time.localtime()) > v.operators[country][operator][station]["timestamp"] + v.departures_update_delay: if not time.mktime(time.localtime()) > v.operators[country][operator][station]["timestamp"] + _delay: return v.operators[country][operator][station] else: v.operators[country][operator][station]["timestamp"] = time.mktime(time.localtime()) except: v.operators[country][operator][station]["timestamp"] = time.mktime(time.localtime()) v.online[country][operator] = 2 try: fetch_departures(country, operator, station) v.online[country][operator] = True except: v.online[country][operator] = False add_error(traceback.format_exc().splitlines()) def fetch_departures(country, operator, station): pprint(">> " + color["red"] + "*" + color["end"] + "Loading departures: " + station) if country == "se": if operator == "vt": return vt_load_departures(station, operator) elif operator == "fe": return ferries_load_departures(station, operator) elif operator == "sj": return sj_load_departures(station, operator) elif operator == "sl": return sl_load_departures(station, operator=operator) elif operator == "norrbotten": return resrobot_load_departures(station, operator=operator) elif country == "fr": if operator == "pa": return pa_load_departures(station, operator=operator) elif country == "uk": if operator == "lo": return lo_load_departures(station, operator=operator) elif country == "at": if operator == "wl": return wl_load_departures(station, operator=operator) elif country == "de": if operator == "dr": return dr_load_departures(station, operator=operator) if operator == "rmv": return rmv_load_departures(station, operator=operator) if operator == "kvv": return kvv_load_departures(station, operator=operator) if operator == "db": return db_load_departures(station, operator=operator) if operator == "db_trains": return db_trains_load_departures(station, operator=operator) if operator == "db2": return db2_load_departures(station, operator=operator) if operator == "be": return be_load_departures(station, operator=operator) if operator == "kvv": return kvv_load_departures(station, operator=operator) if operator == "vrr": return vrr_load_departures(station, operator=operator) elif country == "nl": if operator == "ns": return ns_load_departures(station, operator=operator) elif country == "ch": return zu_load_departures(station, operator=operator) elif country == "dk": if operator == "kb": return kb_load_departures(station, operator=operator) if operator == "dk": return kb_load_departures(station, operator=operator) elif country == "no": return no_load_departures(country, operator, station) elif country == "fi": if operator == "hsl": return hsl_load_departures(operator, station, country) with gtfs_lock: gtfs_load_departures(country, station, operator) return v.operators[country][operator][station] def pprint(result = "", **kwargs): return for arg in kwargs.values(): result += arg print(result + color["end"]) def load_deviations(): pprint(">> " + color["white"] + "Loading deviations...") v.deviations_timestamp = time.mktime(time.localtime()) if v.callout_message: data = [{"priority": {"importance_level": 10,"influence_level": 10,"urgency_level": 10}, "message_variants": [{"header": str(v.callout_message)}], "scope": {"lines": [{"name": "%"}]}}] v.deviations_list = [str(v.callout_message)] else: data = json.loads(fetch_data(host=servers["deviations"][0], args=servers["deviations"][1], port=80)) v.deviations_dict = [] v.deviations_list = [] for i in data: importance = i["priority"]["importance_level"] * i["priority"]["influence_level"] * i["priority"]["urgency_level"] if importance > 20: try: _deviation_message = i["scope"]["lines"][0]["name"] + ": " + i["message_variants"][0]["header"] if "Närtrafiken" in _deviation_message: continue if not _deviation_message in v.deviations_list: v.deviations_list.append(_deviation_message) v.deviations_dict.append(i) except: pass def convert_to_v4(data, t_types): v4_format = {"StatusCode": 0, "ResponseData": {"LatestUpdate": time.strftime('%Y-%m-%dT%H:%M:%S'), "DataAge": 0, "Metros": [], "Buses": [], "Trains": [], "Trams": [], "Ships": []}} for i in data["departures"]: for t_type in v.t_types: if t_type in i["line"]["transport_mode"] and t_type.lower() in t_types: v4_format["ResponseData"][v.t_types[t_type]].append({"Destination":i["destination"], "LineNumber":str(i["line"]["id"]), "JourneyDirection":i["direction_code"], "ExpectedDateTime":i["expected"], "Deviations":None}) return v4_format def check_deviations(): pprint(">> Requested deviation... ") if not len(v.deviations_list): load_deviations() if time.mktime(time.localtime()) > v.deviations_timestamp + v.deviations_update_delay: v.deviations_timestamp = time.mktime(time.localtime()) v.deviations_list.pop(0) if not len(v.deviations_list): load_deviations() def settings_page(request = None, live=False): #try: del sys.modules["_html"]#, _html #except: import _html try: reload(_html.__name__) except Exception as e: print(e) for char in v.spec_dict: _html.content = _html.content.replace(char, v.spec_dict[char]) if not live: _html.content = _html.content.replace("setInterval(updateGraph, 1000);", "setInterval(updateGraph, 100000);") if request and "shop=" in request.path: return _html.params(request) if request and "?" in request.path: return _html.params(request) return _html.content def stats_page(): settings_page() return json.dumps(v.stats_dict) def parse_headers(headers): headers_dict = {} _headers = str(headers).splitlines() for i in _headers: try: headers_dict[i.split(":")[0].strip()] = i.split(": ")[1] except: pass return headers_dict def timestamp_client(headers, siteid, ip=None, port=None, mystation=None): if not str(siteid) == "0": #if int(siteid) > 0: v.clients[headers["User-Agent"]] = {"timestamp" : time.mktime(time.localtime()), "station": siteid, "port":port} if mystation: v.clients[headers["User-Agent"]]["mystation"] = mystation for i in headers: v.clients[headers["User-Agent"]][i] = headers[i] if ip: v.clients[headers["User-Agent"]]["ip"] = ip pprint(">> Requested station: " + color["yellow"] + str(siteid)) def handle_request(request, siteid = 0, out = "Server-X: Access denied"): if "favicon" in request.path: return headers = parse_headers(request.headers) for char in v.html_decode: request.path = request.path.replace(char, v.html_decode[char]) pprint(color["green"] + "Connection from: " + str(request.client_address[0]) + color["gray"] + " " + request.path) if request.server.server_port == 89: if "/settings" in request.path or "/shop" in request.path: out = settings_page(request) elif "/livesettings" in request.path: out = settings_page(request, live=True) elif "/lookup" in request.path: out = v.lookup(request.path) else: if "msg" in request.path: check_deviations() message = v.deviations_list[0] if headers["User-Agent"].replace("0x","") in v.direct_messages: message = "%: " + v.direct_messages[headers["User-Agent"].replace("0x","")] out = json.dumps({"StatusCode": -2, "Message": message}) elif "/?siteid" in request.path: if "30010" in request.path or "30310" in request.path: siteid = siteid = str(request.path).split("siteid=")[1][:9] siteid = (siteid[len(siteid)-4:]) else: siteid = str(request.path).split("siteid=")[1][:4] t_types = [] for i in v.t_types: t_type = v.t_types[i][0:-1].lower().replace("buse","bus") if t_type in request.path: t_types.append(t_type) if len(t_types) == 0: t_types = ["metro"] #check_departures(siteid) get_departures(country="se", operator="sl", station=siteid) try: out = json.dumps(convert_to_v4(v.operators["se"]["sl"][siteid], t_types)) except: out = json.dumps({"StatusCode": -2, "Message": "Ingen data för hÃ¥llplats"}) elif "/v1" in request.path: if "/sites/" in request.path: siteid = request.path.split("sites/")[1].split("/")[0] get_departures(country="se", operator="sl", station=siteid) try: out = json.dumps(v.operators["se"]["sl"][siteid]) except: out = json.dumps({"departures": [],"stop_deviations": []}) elif "/messages" in request.path: check_deviations() message = v.deviations_dict out = json.dumps(message) if headers["User-Agent"].replace("0x","") in v.direct_messages: message = v.direct_messages[headers["User-Agent"].replace("0x","")] out = json.dumps([{"priority": {"importance_level": 10,"influence_level": 10,"urgency_level": 10}, "message_variants": [{"header": str(message)}], "scope": {"lines": [{"name": "%"}]}}]) else: pprint(">>" + color["red"] + out) timestamp_client(headers, siteid, request.client_address[0], request.server.server_port) http_timestamp = formatdate(timeval=None, localtime=True, usegmt=True) request.wfile.write(bytearray("""HTTP/1.0 200 OK\n""" + """Date: """ + str(http_timestamp) + """\r\n""" + """Access-Control-Allow-Origin: *\r\n""" + """Connection: close\r\n\r\n""" + out + "\n", "utf-8")) return if "/restart_server" in request.path: print("Restarting...") with open("db.dat","w") as f: f.write(json.dumps(v.operators)) import subprocess script_path = "/var/www/html/restart.sh" try: subprocess.run([script_path], shell=True) except Exception as e: return str(e) out = "OK" try: country = request.path.split("country=")[1].split("&")[0] operator = request.path.split("operator=")[1].split("&")[0] station = request.path.split("station=")[1].split("&")[0] try: if headers["Operator"] == "sj": _station = bytes.fromhex(station) station = _station.decode("utf-8") except Exception as e: pass validate_request(country, operator, station) except: request.path = "" out = "Missing parameters" try: mystation = request.path.split("from=")[1].split("&")[0] except: mystation = None if "/search_stop" in request.path: pprint(">> Lookup: [" + color["yellow"] + country + "] " + operator + " / " + station) out = json.dumps(search(country, operator, station, gtfs_lock)) if "/get_deviations" in request.path: pprint(">> Requested deviations: " + color["yellow"] + country.upper() + "/" + color["cyan"] + operator + color["end"] + " | station: " + color["cyan"] + station) deviations = sl_get_deviations(country, operator, station) deviations = deviations if deviations else v.deviations_list out = json.dumps(deviations) if "/get_departures" in request.path: pprint(">> Requested departures: " + color["yellow"] + country.upper() + "/" + color["cyan"] + operator + color["end"] + " | station: " + color["cyan"] + station) get_departures(country, operator, station) out = json.dumps(v.operators[country][operator][station]) def char_to_number(input_string): output = [str(ord(char) - ord('a') + 1) for char in input_string.lower()] return ''.join(output) siteid = station timestamp_client(headers, siteid, request.client_address[0], request.server.server_port, mystation) timezone = None if country == "fi": timezone = time.time() + 3600 if country == "uk": timezone = time.time() - 3600 http_timestamp = formatdate(timeval=timezone, localtime=True, usegmt=True) request.wfile.write(bytearray("""HTTP/1.0 200 OK\n""" + """Date: """ + str(http_timestamp) + """\r\n""" + """Connection: close\r\n\r\n""" + out + "\n", "utf-8")) def error_logger(request): v.print_memory_usage() with threading.Lock(): try: handle_request(request) except: add_error(traceback.format_exc().splitlines()) request.close_connection = True SimpleHTTPRequestHandler.do_GET = error_logger SimpleHTTPRequestHandler.timeout = 11 for port in [90, 89]: try: server = ThreadingHTTPServer(("", port), SimpleHTTPRequestHandler) server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = True server_thread.start() except Exception as e: print(f"Not starting on port {port} due to: {e}") #try: search("no","os","oslo") #except: print("Failed to initialize Oslo") while True: try: time.sleep(.01) except KeyboardInterrupt: sys.exit()