297 lines
8.6 KiB
Python
297 lines
8.6 KiB
Python
import json
|
|
import re
|
|
from enum import Enum
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from socketserver import ThreadingMixIn
|
|
|
|
# SETUP #
|
|
#
|
|
hostAddress = "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.")
|