310 lines
11 KiB
Python
310 lines
11 KiB
Python
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
import os
|
|
import json
|
|
import dbus
|
|
from urllib.parse import parse_qs
|
|
from ac_tracks import change_track, get_all_tracks, get_configs, get_preview_image, get_outline_image, get_current_track
|
|
from ac_cars import get_all_cars, get_car_image, update_cars, get_current_cars
|
|
from ac_gamemodes import get_gamemodes, update_gamemodes
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.path = self.path.replace("%20", " ") # handle spaces in URLs
|
|
if (self.handle_GET_path()):
|
|
return
|
|
|
|
self.send_error(404)
|
|
|
|
def do_POST(self):
|
|
if (self.handle_POST_path()):
|
|
return
|
|
|
|
self.send_error(404)
|
|
|
|
def handle_GET_path(self) -> bool:
|
|
"""
|
|
handles the path entered in the GET request.
|
|
|
|
Returns: true if the calling function should return, false otherwise
|
|
"""
|
|
# Serve preview images
|
|
if self.path.startswith("/img/"):
|
|
return self.handle_get_img_path()
|
|
|
|
if self.path.startswith("/track/"):
|
|
return self.handle_get_track_path()
|
|
|
|
if self.path == "/cars":
|
|
return self.handle_get_cars_path()
|
|
if self.path == "/currentcars":
|
|
return self.handle_get_current_cars_path()
|
|
if self.path == "/currenttrack":
|
|
return self.handle_get_current_track_path()
|
|
if self.path == "/gamemodes":
|
|
return self.handle_get_gamemodes_path()
|
|
|
|
if self.path == "/" or self.path == "/index.html":
|
|
return self.handle_get_root_path()
|
|
|
|
return False
|
|
|
|
def handle_get_img_path(self):
|
|
parts = self.path.split("/")
|
|
parts = [p for p in parts if p] # remove empty
|
|
|
|
img_type = parts[1] if len(parts) > 1 else ""
|
|
|
|
img_path = ""
|
|
if (img_type == "preview"):
|
|
track, config = self.extract_track_and_config(parts)
|
|
img_path = get_preview_image(track, config)
|
|
elif (img_type == "outline"):
|
|
track, config = self.extract_track_and_config(parts)
|
|
img_path = get_outline_image(track, config)
|
|
if img_path == "":
|
|
|
|
img_path = get_preview_image(track, config)
|
|
elif (img_type == "car"):
|
|
car = parts[2] if len(parts) > 2 else ""
|
|
skin = parts[3] if len(parts) > 3 else ""
|
|
img_path = get_car_image(car, skin)
|
|
else:
|
|
self.send_error(404)
|
|
return True
|
|
|
|
self.send_image(img_path)
|
|
|
|
return True
|
|
|
|
def handle_get_track_path(self):
|
|
track = self.path.replace("/track/", "").strip("/")
|
|
configs = get_configs(track)
|
|
|
|
data = {
|
|
"track": track,
|
|
"configs": configs,
|
|
"image": f"/img/preview/{track}",
|
|
"outline": f"/img/outline/{track}"
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
return True
|
|
|
|
def handle_get_current_track_path(self):
|
|
track, config = get_current_track()
|
|
data = {
|
|
"track": track,
|
|
"config": config
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
return True
|
|
|
|
def handle_get_cars_path(self):
|
|
cars = get_all_cars()
|
|
|
|
data = {
|
|
"cars": cars
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
return True
|
|
|
|
def handle_get_current_cars_path(self):
|
|
cars = get_current_cars()
|
|
data = {
|
|
"cars": cars
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
return True
|
|
|
|
def handle_get_gamemodes_path(self):
|
|
gamemodes = get_gamemodes()
|
|
data = {
|
|
"gamemodes": {
|
|
"practice_minutes": gamemodes[0],
|
|
"qualify_minutes": gamemodes[1],
|
|
"race_laps": gamemodes[2]
|
|
}
|
|
}
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data).encode())
|
|
return True
|
|
|
|
|
|
def handle_get_root_path(self):
|
|
with open("index.html", "r") as f:
|
|
html = f.read()
|
|
|
|
tracks = get_all_tracks()
|
|
track_options = "".join([f'<option value="{t}">{t}</option>' for t in tracks])
|
|
|
|
html = html.replace("{{tracks}}", track_options)
|
|
html = html.replace("{{configs}}", "") # empty at load
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "text/html")
|
|
self.end_headers()
|
|
self.wfile.write(html.encode())
|
|
return True
|
|
|
|
def handle_POST_path(self):
|
|
if (self.path.startswith("/changetrack/")):
|
|
return self.handle_post_change_track()
|
|
if (self.path.startswith("/changecars")):
|
|
return self.handle_post_change_car()
|
|
if (self.path.startswith("/changegamemodes")):
|
|
return self.handle_post_change_gamemodes()
|
|
return False
|
|
|
|
def handle_post_change_track(self):
|
|
parts = self.path.split("/")
|
|
parts = [p for p in parts if p] # remove empty
|
|
track, config = self.extract_track_and_config(parts, 1)
|
|
print(f"Changing track to '{track}' with config '{config}'")
|
|
|
|
success, message = change_track(track, config)
|
|
if success:
|
|
success, message = restart_ac_server()
|
|
print(f"restart succeeded: {success}, {message}")
|
|
|
|
if success:
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "success", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
else:
|
|
self.send_response(400)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "error", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
|
|
return True
|
|
|
|
def handle_post_change_car(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
|
|
# 2. Lees de body
|
|
body = self.rfile.read(length).decode("utf-8")
|
|
|
|
# 3. Parse JSON
|
|
import json
|
|
try:
|
|
data = json.loads(body)
|
|
print("Received cars:", data)
|
|
success, message = update_cars(data.get("cars", []))
|
|
if success:
|
|
success, message = restart_ac_server()
|
|
print(f"restart succeeded: {success}, {message}")
|
|
|
|
if success:
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "success", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
else:
|
|
self.send_response(400)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "error", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
return True
|
|
except json.JSONDecodeError:
|
|
self.send_response(400)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"error":"invalid json"}')
|
|
print("Invalid JSON received")
|
|
return False
|
|
|
|
def handle_post_change_gamemodes(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
|
|
body = self.rfile.read(length).decode("utf-8")
|
|
|
|
import json
|
|
try:
|
|
data = json.loads(body)
|
|
gamemodes = data.get("gamemodes", {})
|
|
print("Received gamemodes:", gamemodes)
|
|
practice = gamemodes.get("practice_minutes", 0)
|
|
qualify = gamemodes.get("qualify_minutes", 0)
|
|
race = gamemodes.get("race_laps", 0)
|
|
success, message = update_gamemodes(practice, qualify, race)
|
|
if success:
|
|
success, message = restart_ac_server()
|
|
print(f"restart succeeded: {success}, {message}")
|
|
|
|
if success:
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "success", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
else:
|
|
self.send_response(400)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
response = {"status": "error", "message": message}
|
|
self.wfile.write(json.dumps(response).encode())
|
|
return True
|
|
except json.JSONDecodeError:
|
|
self.send_response(400)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"error":"invalid json"}')
|
|
print("Invalid JSON received")
|
|
return False
|
|
|
|
def send_image(self, img_path: str):
|
|
if os.path.exists(img_path):
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "image/png")
|
|
self.end_headers()
|
|
with open(img_path, "rb") as f:
|
|
self.wfile.write(f.read())
|
|
else:
|
|
self.send_error(404)
|
|
|
|
def extract_track_and_config(self, parts: list, track_index: int = 2) -> tuple[str, str]:
|
|
print(parts)
|
|
track = parts[track_index] if len(parts) > track_index else ""
|
|
config = parts[track_index + 1] if len(parts) > track_index + 1 else ""
|
|
|
|
return track, config
|
|
|
|
def restart_ac_server() -> tuple[bool, str]:
|
|
sysbus = dbus.SystemBus()
|
|
systemd1 = sysbus.get_object('org.freedesktop.systemd1', '/org/freedesktop/systemd1')
|
|
manager = dbus.Interface(systemd1, 'org.freedesktop.systemd1.Manager')
|
|
try:
|
|
job = manager.RestartUnit('assetto-corsa-server.service', 'replace')
|
|
if job:
|
|
return True, "Successfully restarted assetto-corsa-server.service"
|
|
else:
|
|
return False, "Failed to restart assetto-corsa-server.service"
|
|
except dbus.DBusException as e:
|
|
return False, f"Could not restart server: {str(e)}"
|
|
|
|
server = HTTPServer(("0.0.0.0", 10303), Handler)
|
|
print("Server running on port 10303")
|
|
server.serve_forever()
|