import json import os import re from enum import Enum from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn # SETUP # # hostAddress = os.environ.get("SIMULATION_BIND_ADDRESS", "0.0.0.0") hostPort = 8080 network = re.compile(r"(127)\.(0)\.(0)\..+") # ############# passedCarCount = 0 safetyBit = 0 validBit = 1 class Color(Enum): red = 0 yellow = 1 green = 2 off = 3 class TrafficLight: def __init__(self): self.carCount = 0 self.color = Color.off def toJSON(self, isColor): if isColor: return '"{}"'.format( self.color.name if str(validBit) == "1" else Color.off.name ) else: return str(self.carCount) def fromJSON(self, isColor, JSON): if isColor: if JSON != "red" and JSON != "yellow" and JSON != "green" and JSON != "off": raise ValueError("non-existent state") else: self.color = Color[JSON] else: try: float(JSON) except ValueError: raise ValueError("expected a number") self.carCount = JSON class Crossroads: def __init__(self): self.N = TrafficLight() self.E = TrafficLight() self.S = TrafficLight() self.W = TrafficLight() def toJSON(self, isColor): return '{{ "N": {}, "E": {}, "S": {}, "W": {} }}'.format( self.N.toJSON(isColor), self.E.toJSON(isColor), self.S.toJSON(isColor), self.W.toJSON(isColor), ) def fromJSON(self, isColor, JSON): self.N.fromJSON(isColor, JSON["N"]) self.E.fromJSON(isColor, JSON["E"]) self.S.fromJSON(isColor, JSON["S"]) self.W.fromJSON(isColor, JSON["W"]) class RoadSystem: def __init__(self): self.NW = Crossroads() self.NE = Crossroads() self.SE = Crossroads() self.SW = Crossroads() def toJSON(self, isColor): return '{{ "NW": {}, "NE": {}, "SE": {}, "SW": {} }}'.format( self.NW.toJSON(isColor), self.NE.toJSON(isColor), self.SE.toJSON(isColor), self.SW.toJSON(isColor), ) def fromJSON(self, isColor, JSON): self.NW.fromJSON(isColor, JSON["NW"]) self.NE.fromJSON(isColor, JSON["NE"]) self.SE.fromJSON(isColor, JSON["SE"]) self.SW.fromJSON(isColor, JSON["SW"]) reTrafficLights = re.compile(r"/trafficLights.*") reRoads = re.compile(r"/roads.*") reSecondary = re.compile(r"(NW|NE|SW|SE)") rePrimary = re.compile(r"[NESW]") def processURI(path): isTrafficLight = False if reTrafficLights.match(path): path = path[len("/trafficLights/") :] isTrafficLight = True elif reRoads.match(path): path = path[len("/roads/") :] else: return False if not path: return {"levels": 1, "isTrafficLight": isTrafficLight} else: crossroadDirection = path[0:2] if reSecondary.match(crossroadDirection): path = path[3:] if not path: return { "levels": 2, "isTrafficLight": isTrafficLight, "crossroadDirection": crossroadDirection, } else: trafficLightDirection = path[0] if rePrimary.match(trafficLightDirection) and len(path) == 1: return { "levels": 3, "isTrafficLight": isTrafficLight, "crossroadDirection": crossroadDirection, "trafficLightDirection": trafficLightDirection, } else: return False else: return False class server(BaseHTTPRequestHandler): def respondWith(self, code): self.send_response(code) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, PUT, OPTIONS") self.send_header("Access-Control-Allow-Headers", "*") self.send_header("Content-type", "application/json") self.end_headers() def do_GET(self): processedURI = processURI(self.path) if self.path == "/passedCarCount": self.respondWith(200) self.wfile.write(bytes(str(passedCarCount), "utf-8")) elif self.path == "/safetyBit": self.respondWith(200) self.wfile.write(bytes(str(safetyBit), "utf-8")) elif self.path == "/validBit": self.respondWith(200) self.wfile.write(bytes(str(validBit), "utf-8")) elif self.path == "/all": self.respondWith(200) self.wfile.write( bytes( '{{ "passedCarCount": {}, "roads": {}, "trafficLights": {} }}'.format( str(passedCarCount), roadSystem.toJSON(False), roadSystem.toJSON(True), ), "utf-8", ) ) elif processedURI: isColor = processedURI["isTrafficLight"] if processedURI["levels"] == 1: response = roadSystem.toJSON(isColor) else: crossroads = getattr(roadSystem, processedURI["crossroadDirection"]) if processedURI["levels"] == 2: response = crossroads.toJSON(isColor) elif processedURI["levels"] == 3: trafficLight = getattr( crossroads, processedURI["trafficLightDirection"] ) response = trafficLight.toJSON(isColor) self.respondWith(200) self.wfile.write(bytes(response, "utf-8")) else: self.respondWith(404) def do_PUT(self): global passedCarCount, safetyBit, validBit if not network.match(self.client_address[0]): print( "Unauthorized access: " + str(self.client_address[0]) + ":" + str(self.client_address[1]) ) self.respondWith(401) return length = int(self.headers["Content-Length"]) data = self.rfile.read(length).decode("utf-8") processedURI = processURI(self.path) if self.path == "/passedCarCount": try: int(data) passedCarCount = data except ValueError: print("expected a number") self.respondWith(400) return self.respondWith(204) elif self.path == "/safetyBit": safetyBit = data self.respondWith(204) elif self.path == "/validBit": validBit = data self.respondWith(204) elif processedURI: isColor = processedURI["isTrafficLight"] try: if processedURI["levels"] < 3: data = json.loads(data) if processedURI["levels"] == 1: roadSystem.fromJSON(isColor, data) else: crossroads = getattr(roadSystem, processedURI["crossroadDirection"]) if processedURI["levels"] == 2: crossroads.fromJSON(isColor, data) elif processedURI["levels"] == 3: trafficLight = getattr( crossroads, processedURI["trafficLightDirection"] ) trafficLight.fromJSON(isColor, data) except Exception as e: print(e) self.respondWith(400) return self.respondWith(204) else: self.respondWith(404) def do_OPTIONS(self): self.respondWith(200) def log_request(code="-", size="-"): pass class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): """Handle requests in a separate thread.""" roadSystem = RoadSystem() try: backendServer = ThreadedHTTPServer((hostAddress, hostPort), server) print( "The TRAFFICLIGHT server has been started. {}:{}".format(hostAddress, hostPort) ) backendServer.serve_forever() except KeyboardInterrupt: backendServer.server_close() print("The TRAFFICLIGHT server has been stopped.")