#!/usr/bin/env python3 import os import sys import json import queue import threading import socket import time try: import tomllib except ModuleNotFoundError: try: import tomli as tomllib except ModuleNotFoundError: print("tomli/tomllib not found. Install with: pip install tomli") sys.exit(1) try: import tomli_w except ModuleNotFoundError: tomli_w = None from flask import Flask, render_template, jsonify, Response, request BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_PATH = os.path.join(BASE_DIR, "config.toml") with open(CONFIG_PATH, "rb") as f: config = tomllib.load(f) PORT = 50743 app = Flask(__name__) scores = {"left": 0, "right": 0} subscribers = [] sub_lock = threading.Lock() def get_local_ip(): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("10.255.255.255", 1)) ip = s.getsockname()[0] except: ip = "127.0.0.1" finally: s.close() return ip def udp_beacon(): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) while True: try: payload = json.dumps({ "display_name": config.get("display_name", ""), "ip": get_local_ip(), "port": PORT, }) sock.sendto(payload.encode(), ("255.255.255.255", 50744)) except: pass time.sleep(10) threading.Thread(target=udp_beacon, daemon=True).start() def broadcast(data): dead = [] with sub_lock: for q in subscribers: try: q.put_nowait(json.dumps(data)) except: dead.append(q) for q in dead: subscribers.remove(q) def cors(resp): resp.headers["Access-Control-Allow-Origin"] = "*" resp.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS" resp.headers["Access-Control-Allow-Headers"] = "Content-Type" return resp def save_config(): if tomli_w is None: return with open(CONFIG_PATH, "wb") as f: tomli_w.dump(config, f) @app.route("/") def index(): return render_template( "index.html", display_id=config["display_id"], header=config.get("header", ""), header_font_size=config.get("header_font_size", 0), content_padding_top=config.get("content_padding_top", 0), content_gap=config.get("content_gap", 1), left_name=config["left"]["name"], left_color=config["left"]["color"], left_subtitle=config["left"].get("subtitle", ""), left_name_font_size=config["left"].get("name_font_size", 0), left_subtitle_font_size=config["left"].get("subtitle_font_size", 0), left_score_font_size=config["left"].get("score_font_size", 0), right_name=config["right"]["name"], right_color=config["right"]["color"], right_subtitle=config["right"].get("subtitle", ""), right_name_font_size=config["right"].get("name_font_size", 0), right_subtitle_font_size=config["right"].get("subtitle_font_size", 0), right_score_font_size=config["right"].get("score_font_size", 0), left_score=scores["left"], right_score=scores["right"], ) @app.route("/events") def sse(): def stream(): q = queue.Queue() with sub_lock: subscribers.append(q) try: while True: data = q.get() yield f"data: {data}\n\n" except GeneratorExit: with sub_lock: if q in subscribers: subscribers.remove(q) return Response(stream(), mimetype="text/event-stream") @app.route("/api/status") def api_status(): return cors(jsonify({ "display_id": config["display_id"], "display_name": config.get("display_name", ""), "header": config.get("header", ""), "header_font_size": config.get("header_font_size", 0), "content_padding_top": config.get("content_padding_top", 0), "content_gap": config.get("content_gap", 1), "ip": get_local_ip(), "port": PORT, "scores": dict(scores), "config": { "left": { "name": config["left"]["name"], "subtitle": config["left"].get("subtitle", ""), "color": config["left"]["color"], "name_font_size": config["left"].get("name_font_size", 0), "subtitle_font_size": config["left"].get("subtitle_font_size", 0), "score_font_size": config["left"].get("score_font_size", 0), }, "right": { "name": config["right"]["name"], "subtitle": config["right"].get("subtitle", ""), "color": config["right"]["color"], "name_font_size": config["right"].get("name_font_size", 0), "subtitle_font_size": config["right"].get("subtitle_font_size", 0), "score_font_size": config["right"].get("score_font_size", 0), }, }, })) @app.route("/api/score/", methods=["PUT", "OPTIONS"]) def api_set_score(side): if request.method == "OPTIONS": return cors(jsonify({})) if side not in scores: return cors((jsonify({"error": "invalid side"}), 400)) data = request.get_json(silent=True) or {} value = data.get("value") if value is None or not isinstance(value, int) or value < 0: return cors((jsonify({"error": "value must be a non-negative integer"}), 400)) scores[side] = value broadcast({"type": "score", "scores": dict(scores), "side": side}) return cors(jsonify(scores)) @app.route("/api/score//reset", methods=["POST", "OPTIONS"]) def api_reset_side(side): if request.method == "OPTIONS": return cors(jsonify({})) if side in scores: scores[side] = 0 broadcast({"type": "score", "scores": dict(scores), "side": side}) return cors(jsonify(scores)) @app.route("/api/score/reset", methods=["POST", "OPTIONS"]) def api_reset_all(): if request.method == "OPTIONS": return cors(jsonify({})) scores["left"] = 0 scores["right"] = 0 broadcast({"type": "score", "scores": dict(scores), "side": None}) return cors(jsonify(scores)) @app.route("/api/score//increment", methods=["POST", "OPTIONS"]) def api_increment(side): if request.method == "OPTIONS": return cors(jsonify({})) if side in scores: scores[side] += 1 broadcast({"type": "score", "scores": dict(scores), "side": side}) return cors(jsonify(scores)) @app.route("/api/score//decrement", methods=["POST", "OPTIONS"]) def api_decrement(side): if request.method == "OPTIONS": return cors(jsonify({})) if side in scores and scores[side] > 0: scores[side] -= 1 broadcast({"type": "score", "scores": dict(scores), "side": side}) return cors(jsonify(scores)) @app.route("/api/config", methods=["PUT", "OPTIONS"]) def api_update_config(): if request.method == "OPTIONS": return cors(jsonify({})) data = request.get_json(silent=True) or {} if "display_name" in data: config["display_name"] = data["display_name"] if "header" in data: config["header"] = data["header"] if "header_font_size" in data: config["header_font_size"] = data["header_font_size"] if "content_padding_top" in data: config["content_padding_top"] = data["content_padding_top"] if "content_gap" in data: config["content_gap"] = data["content_gap"] for side in ("left", "right"): if side in data: for key in ("name", "subtitle", "color", "name_font_size", "subtitle_font_size", "score_font_size"): if key in data[side]: config[side][key] = data[side][key] save_config() broadcast({"type": "config", "config": data}) return cors(jsonify({"status": "ok"})) @app.route("/increment/", methods=["GET", "POST"]) def increment(side): if side in scores: scores[side] += 1 broadcast({"type": "score", "scores": dict(scores), "side": side}) return jsonify(scores) @app.route("/reset", methods=["GET", "POST"]) def reset(): scores["left"] = 0 scores["right"] = 0 broadcast({"type": "score", "scores": dict(scores), "side": None}) return jsonify(scores) @app.route("/reset/", methods=["GET", "POST"]) def reset_side(side): if side in scores: scores[side] = 0 broadcast({"type": "score", "scores": dict(scores), "side": side}) return jsonify(scores) if __name__ == "__main__": print(f" CuteBoard Display #{config['display_id']}") print(f" {config['left']['name']} ({config['left']['color']}) vs {config['right']['name']} ({config['right']['color']})") print(f" URL: http://0.0.0.0:{PORT}") app.run(host="0.0.0.0", port=PORT, debug=False, threaded=True)