from flask import Flask, request, jsonify, Response import threading import socket import json from config import DISPLAY_MAX_X, DISPLAY_MAX_Y UDP_PORT = 4210 COLS = DISPLAY_MAX_X + 1 # 48 ROWS = DISPLAY_MAX_Y + 1 # 16 FRAME_SIZE = COLS * ROWS * 3 # 2304 bytes app = Flask(__name__) # Shared state — written by Flask thread, read by main loop state = { 'brightness': 100, 'program_top_text_enabled': True, 'program_bottom_text_enabled': True, 'program_vehicles_enabled': True, 'force_vehicles': False, } _lock = threading.Lock() _frame = b'\x00' * FRAME_SIZE _frame_lock = threading.Lock() def _udp_listener(): global _frame sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(('127.0.0.1', UDP_PORT)) while True: data, _ = sock.recvfrom(FRAME_SIZE + 1) if len(data) == FRAME_SIZE: with _frame_lock: _frame = data INDEX_HTML = """ LED Panel Control

LED Panel Control

Top TextWitamy w PTI
Bottom TextSchool programs
VehiclesVehicle interruptions
100
""" @app.route('/') def index(): return Response(INDEX_HTML, mimetype='text/html') @app.route('/display', methods=['GET']) def handle_display(): with _frame_lock: data = _frame pixels = [] off = 0 for y in range(ROWS): row = [] for x in range(COLS): row.append([data[off], data[off + 1], data[off + 2]]) off += 3 pixels.append(row) return Response(json.dumps(pixels), mimetype='application/json') @app.route('/brightness', methods=['POST']) def handle_brightness(): value = request.form.get('value') if value is not None: with _lock: state['brightness'] = max(0, min(100, int(value))) return 'OK' @app.route('/toggle-program', methods=['POST']) def handle_toggle(): program = request.form.get('program') enabled_str = request.form.get('enabled') if not program or enabled_str is None: return 'Missing program or enabled parameter', 400 enabled = enabled_str.lower() == 'true' key_map = { 'top_text': 'program_top_text_enabled', 'bottom_text': 'program_bottom_text_enabled', 'vehicles': 'program_vehicles_enabled', } if program not in key_map: return 'Unknown program', 400 with _lock: state[key_map[program]] = enabled return 'OK' @app.route('/force-vehicles', methods=['POST']) def handle_force_vehicles(): with _lock: state['force_vehicles'] = True return 'OK' @app.route('/program-status', methods=['GET']) def handle_status(): with _lock: s = dict(state) return jsonify({ 'top_text': s['program_top_text_enabled'], 'bottom_text': s['program_bottom_text_enabled'], 'vehicles': s['program_vehicles_enabled'], 'brightness': s['brightness'], }) def start_server(host='0.0.0.0', port=80): threading.Thread(target=_udp_listener, daemon=True).start() threading.Thread(target=lambda: app.run(host=host, port=port, use_reloader=False), daemon=True).start()