import os from flask import Flask, request, jsonify, Response from flask_sock import Sock 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 TEXTS_FILE = os.path.join(os.path.dirname(__file__), 'texts.json') DEFAULT_TEXTS = { 'top': [ { 'text': 'Witamy w PTI', 'colors': [ {'r': 40, 'g': 40, 'b': 40, 'from': 0}, {'r': 0, 'g': 0, 'b': 50, 'from': 9}, ], } ], 'bottom': [ {'text': 'Technik informatyk + ai, robotyka', 'colors': [{'r': 0, 'g': 0, 'b': 50, 'from': 0}]}, {'text': 'Technik programista', 'colors': [{'r': 50, 'g': 40, 'b': 0, 'from': 0}]}, {'text': 'Technik poligrafii i grafiki komputerowej', 'colors': [{'r': 0, 'g': 45, 'b': 0, 'from': 0}]}, {'text': 'Technik reklamy', 'colors': [{'r': 60, 'g': 20, 'b': 0, 'from': 0}]}, ], } def _load_texts(): try: with open(TEXTS_FILE) as f: return json.load(f) except Exception: return DEFAULT_TEXTS app = Flask(__name__) _sock = Sock(app) # 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, 'force_dvd': False, 'texts': _load_texts(), } _lock = threading.Lock() _frame = b'\x00' * FRAME_SIZE _frame_cond = threading.Condition() 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_cond: _frame = data _frame_cond.notify_all() INDEX_HTML = """ LED Panel Control

LED Panel Control

Disconnected

Text Editor ▼

Top TextWitamy w PTI

Bottom TextSchool programs

VehiclesVehicle interruptions

Brightness

100

""" @_sock.route('/ws/display') def ws_display(ws): streaming = [True] def _reader(): try: while True: msg = ws.receive() if msg == 'stop': streaming[0] = False elif msg == 'start': streaming[0] = True except Exception: streaming[0] = False t = threading.Thread(target=_reader, daemon=True) t.start() try: while t.is_alive(): with _frame_cond: _frame_cond.wait(timeout=0.5) data = _frame if streaming[0]: ws.send(data) except Exception: pass @app.route('/') def index(): return Response(INDEX_HTML, mimetype='text/html') @app.route('/display', methods=['GET']) def handle_display(): with _frame_cond: 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('/texts', methods=['GET']) def handle_get_texts(): with _lock: return jsonify(state['texts']) @app.route('/texts', methods=['POST']) def handle_post_texts(): data = request.get_json(silent=True) if not isinstance(data, dict) or 'top' not in data or 'bottom' not in data: return 'Invalid format', 400 try: with open(TEXTS_FILE, 'w') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: return str(e), 500 with _lock: state['texts'] = data return 'OK' @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('/force-dvd', methods=['POST']) def handle_force_dvd(): with _lock: state['force_dvd'] = 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'], 'force_vehicles': s['force_vehicles'], 'force_dvd': s['force_dvd'], }) 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()