#!/usr/bin/env python3 import argparse import colorsys import datetime import json import math import os import socket import sys import threading import time from http.server import BaseHTTPRequestHandler, HTTPServer from PIL import Image, ImageDraw, ImageEnhance, ImageFont # --- DDP Network Engine --- CHUNK_SIZE = 1440 COLOR_MAP = { "red": "#FF0000", "green": "#00FF00", "blue": "#0000FF", "cyan": "#00FFFF", "magenta": "#FF00FF", "yellow": "#FFFF00", "white": "#FFFFFF", "black": "#000000", "orange": "#FF7F00", "purple": "#800080", "pink": "#FFC0CB", "hotpink": "#FF1493", "lime": "#00FF00", "teal": "#008080", "navy": "#000080", "amber": "#FFBF00", "violet": "#8A2BE2", "turquoise": "#40E0D0" } def is_monospace_font(path): try: font = ImageFont.truetype(path, 20) w_i = font.getlength('i') w_w = font.getlength('W') w_m = font.getlength('m') return abs(w_i - w_w) < 0.1 and abs(w_i - w_m) < 0.1 except Exception: name = os.path.basename(path).lower() return any(k in name for k in ["mono", "code", "term", "courier", "consolas", "fixed"]) def scan_system_fonts(): font_dirs = [ "/usr/share/fonts/truetype", "/usr/local/share/fonts", os.path.expanduser("~/.local/share/fonts"), os.path.expanduser("~/.fonts") ] all_fonts = {} mono_fonts = {} for directory in font_dirs: if os.path.exists(directory): for root, _, files in os.walk(directory): for file in files: if file.lower().endswith(".ttf"): path = os.path.join(root, file) key = file.lower().replace(".ttf", "") all_fonts[key] = path if is_monospace_font(path): mono_fonts[key] = path if not all_fonts: default_path = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" all_fonts["default"] = default_path mono_fonts["default"] = default_path return dict(sorted(all_fonts.items())), dict(sorted(mono_fonts.items())) ALL_FONT_MAP, MONO_FONT_MAP = scan_system_fonts() def parse_rgb(rgb_val): if isinstance(rgb_val, (list, tuple)): if len(rgb_val) == 3 and all(isinstance(x, int) and 0 <= x <= 255 for x in rgb_val): return tuple(rgb_val) raise ValueError("RGB tuple/list must contain 3 integers between 0 and 255.") val = str(rgb_val).strip().lower() if val in COLOR_MAP: val = COLOR_MAP[val] try: if val.startswith("#"): val = val.lstrip("#") if len(val) == 6: return tuple(int(val[i:i+2], 16) for i in (0, 2, 4)) parts = [int(x.strip()) for x in str(rgb_val).split(",")] if len(parts) == 3: return tuple(parts) except Exception: pass raise ValueError(f"Invalid color '{rgb_val}'. Use a valid name, hex '#RRGGBB', or 'R,G,B'.") def send_ddp_frame(sock, raw_rgb, ip, port, seq): total_bytes = len(raw_rgb) offset = 0 while offset < total_bytes: chunk = raw_rgb[offset : offset + CHUNK_SIZE] is_last = (offset + len(chunk)) >= total_bytes flags = 0x41 if is_last else 0x40 header = bytearray([ flags, seq & 0x0F, 0x01, 0x01, (offset >> 24) & 0xFF, (offset >> 16) & 0xFF, (offset >> 8) & 0xFF, offset & 0xFF, (len(chunk) >> 8) & 0xFF, len(chunk) & 0xFF ]) sock.sendto(header + chunk, (ip, port)) offset += len(chunk) def clear_matrix(sock, ip, port, width, height): send_ddp_frame(sock, bytes([0, 0, 0] * (width * height)), ip, port, 0) # --- Graphics & Render Engine --- def generate_effect_layer(w, h, effect, color1, color2, frame_idx): layer = Image.new("RGBA", (w, h), (0, 0, 0, 0)) draw = ImageDraw.Draw(layer) if effect == "solid": draw.rectangle([(0, 0), (w, h)], fill=color1 + (255,)) elif effect == "rainbow": for x in range(w): hue = ((x / float(max(1, w))) + (frame_idx * 0.02)) % 1.0 r, g, b = [int(c * 255) for c in colorsys.hsv_to_rgb(hue, 1.0, 1.0)] draw.line([(x, 0), (x, h)], fill=(r, g, b, 255)) elif effect == "gradient-h": for x in range(w): t = x / float(max(1, w - 1)) r = int(color1[0] + (color2[0] - color1[0]) * t) g = int(color1[1] + (color2[1] - color1[1]) * t) b = int(color1[2] + (color2[2] - color1[2]) * t) draw.line([(x, 0), (x, h)], fill=(r, g, b, 255)) elif effect == "gradient-v": for y in range(h): t = y / float(max(1, h - 1)) r = int(color1[0] + (color2[0] - color1[0]) * t) g = int(color1[1] + (color2[1] - color1[1]) * t) b = int(color1[2] + (color2[2] - color1[2]) * t) draw.line([(0, y), (w, y)], fill=(r, g, b, 255)) elif effect == "pulse": factor = (math.sin(frame_idx * 0.15) + 1.0) / 2.0 r = int(color1[0] * (0.15 + 0.85 * factor)) g = int(color1[1] * (0.15 + 0.85 * factor)) b = int(color1[2] * (0.15 + 0.85 * factor)) draw.rectangle([(0, 0), (w, h)], fill=(r, g, b, 255)) return layer def apply_text_effect(base_banner, effect, color1, color2, frame_idx): w, h = base_banner.size alpha_mask = base_banner.split()[3] overlay = generate_effect_layer(w, h, effect, color1, color2, frame_idx) overlay.putalpha(alpha_mask) if getattr(base_banner, "is_analog_clock", False): final = Image.alpha_composite(overlay, base_banner) final.putalpha(alpha_mask) return final return overlay def render_base_text(text, font_path, target_height): try: font = ImageFont.truetype(font_path, 80) except IOError: font = ImageFont.load_default() temp_img = Image.new("RGBA", (3000, 250), (0, 0, 0, 0)) temp_draw = ImageDraw.Draw(temp_img) temp_draw.text((10, 10), text, font=font, fill=(255, 255, 255, 255)) bbox = temp_img.getbbox() if not bbox: return Image.new("RGBA", (1, target_height), (0, 0, 0, 0)) cropped = temp_img.crop(bbox) scale_factor = target_height / float(cropped.size[1]) new_w = max(1, int(cropped.size[0] * scale_factor)) return cropped.resize((new_w, target_height), Image.Resampling.LANCZOS) def render_fitted_text(text, font_path, max_w, max_h): try: font = ImageFont.truetype(font_path, 80) except IOError: font = ImageFont.load_default() temp_img = Image.new("RGBA", (2000, 300), (0, 0, 0, 0)) temp_draw = ImageDraw.Draw(temp_img) temp_draw.text((10, 10), text, font=font, fill=(255, 255, 255, 255)) bbox = temp_img.getbbox() if not bbox: return Image.new("RGBA", (1, 1), (0, 0, 0, 0)) cropped = temp_img.crop(bbox) scale = min(max_w / float(cropped.size[0]), max_h / float(cropped.size[1])) new_w = max(1, int(cropped.size[0] * scale)) new_h = max(1, int(cropped.size[1] * scale)) return cropped.resize((new_w, new_h), Image.Resampling.LANCZOS) def render_analog_clock(width, height): now = datetime.datetime.now() img = Image.new("RGBA", (width, height), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) center_x, center_y = width // 2, height // 2 radius = min(center_x, center_y) - 1 draw.ellipse([center_x - radius, center_y - radius, center_x + radius, center_y + radius], outline=(255, 255, 255, 255), width=1) sec_angle = math.radians(now.second * 6 - 90) min_angle = math.radians(now.minute * 6 - 90) hr_angle = math.radians((now.hour % 12) * 30 + (now.minute / 2) - 90) draw.line([center_x, center_y, center_x + int(radius * 0.5) * math.cos(hr_angle), center_y + int(radius * 0.5) * math.sin(hr_angle)], fill=(255, 255, 255, 255), width=2) draw.line([center_x, center_y, center_x + int(radius * 0.8) * math.cos(min_angle), center_y + int(radius * 0.8) * math.sin(min_angle)], fill=(255, 255, 255, 255), width=1) draw.line([center_x, center_y, center_x + int(radius * 0.9) * math.cos(sec_angle), center_y + int(radius * 0.9) * math.sin(sec_angle)], fill=(255, 0, 0, 255), width=1) img.is_analog_clock = True return img def render_digital_clock(width, height, font_path, show_date=True, show_seconds=True, blink=True): now = datetime.datetime.now() img = Image.new("RGBA", (width, height), (0, 0, 0, 0)) fmt = "%H:%M:%S" if show_seconds else "%H:%M" time_str = now.strftime(fmt) if blink and now.second % 2 == 0: time_str = time_str.replace(":", " ") max_w = width - 2 max_h = (height // 2) - 2 if show_date else height - 2 time_img = render_fitted_text(time_str, font_path, max_w, max_h) t_x = (width - time_img.size[0]) // 2 t_y = ((height // 2) - time_img.size[1]) // 2 if show_date else (height - time_img.size[1]) // 2 img.paste(time_img, (t_x, t_y), time_img) if show_date: date_str = now.strftime("%a %d %b") date_img = render_fitted_text(date_str, font_path, max_w, max_h) d_x = (width - date_img.size[0]) // 2 d_y = (height // 2) + (((height // 2) - date_img.size[1]) // 2) img.paste(date_img, (d_x, d_y), date_img) return img # --- Thread-Safe Billboard Animation Manager --- class BillboardDaemon: def __init__(self, default_ip, default_port, default_width, default_height, default_font): self.default_ip = default_ip self.default_port = default_port self.default_width = default_width self.default_height = default_height self.default_font = default_font self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.stop_event = threading.Event() self.worker_thread = None self.lock = threading.Lock() def start_animation(self, params): with self.lock: if self.worker_thread and self.worker_thread.is_alive(): self.stop_event.set() self.worker_thread.join() self.stop_event.clear() self.worker_thread = threading.Thread( target=self._render_loop, args=(params,), daemon=True ) self.worker_thread.start() def stop_animation(self): with self.lock: if self.worker_thread and self.worker_thread.is_alive(): self.stop_event.set() self.worker_thread.join() clear_matrix(self.sock, self.default_ip, self.default_port, self.default_width, self.default_height) def _render_loop(self, params): mode = params.get("mode", "scroll") ip = params.get("ip", self.default_ip) port = params.get("port", self.default_port) width = params.get("width", self.default_width) height = params.get("height", self.default_height) font_path = params.get("font", self.default_font) effect = params.get("effect", "rainbow") color1 = params.get("color", (255, 0, 255)) color2 = params.get("color2", (0, 255, 255)) bg_effect = params.get("bg_effect", "solid") bg_color = params.get("bg_color", (0, 0, 0)) bg_color2 = params.get("bg_color2", (0, 0, 128)) brightness = max(0, min(255, params.get("brightness", 255))) / 255.0 seq = 0 frame_idx = 0 if mode in ["clock_digital", "clock_analog"]: show_date = params.get("clock_date", True) show_seconds = params.get("clock_seconds", True) blink = params.get("clock_blink", True) while not self.stop_event.is_set(): if mode == "clock_analog": base_mask = render_analog_clock(width, height) else: base_mask = render_digital_clock(width, height, font_path, show_date, show_seconds, blink) styled_mask = apply_text_effect(base_mask, effect, color1, color2, frame_idx) bg_layer = generate_effect_layer(width, height, bg_effect, bg_color, bg_color2, frame_idx).convert("RGB") bg_layer.paste(styled_mask, (0, 0), styled_mask) if brightness < 1.0: bg_layer = ImageEnhance.Brightness(bg_layer).enhance(brightness) send_ddp_frame(self.sock, bg_layer.tobytes(), ip, port, seq) seq = (seq + 1) % 16 frame_idx += 1 if self.stop_event.wait(0.1): break clear_matrix(self.sock, ip, port, width, height) return text = params.get("text", "WOMAN CAVE") speed = params.get("speed", 0.05) loops = params.get("loops", 0) sweep_pause = params.get("sweep_pause", 2.0) base_banner = render_base_text(text, font_path, height) banner_width = base_banner.size[0] loop_count = 0 while not self.stop_event.is_set(): for x_pos in range(width, -banner_width - 1, -1): if self.stop_event.is_set(): break styled_banner = apply_text_effect(base_banner, effect, color1, color2, frame_idx) bg_layer = generate_effect_layer(width, height, bg_effect, bg_color, bg_color2, frame_idx).convert("RGB") bg_layer.paste(styled_banner, (x_pos, 0), styled_banner) if brightness < 1.0: bg_layer = ImageEnhance.Brightness(bg_layer).enhance(brightness) send_ddp_frame(self.sock, bg_layer.tobytes(), ip, port, seq) seq = (seq + 1) % 16 frame_idx += 1 if self.stop_event.wait(speed): break if sweep_pause > 0 and not self.stop_event.is_set(): pause_end = time.time() + sweep_pause while time.time() < pause_end and not self.stop_event.is_set(): bg_layer = generate_effect_layer(width, height, bg_effect, bg_color, bg_color2, frame_idx).convert("RGB") if brightness < 1.0: bg_layer = ImageEnhance.Brightness(bg_layer).enhance(brightness) send_ddp_frame(self.sock, bg_layer.tobytes(), ip, port, seq) seq = (seq + 1) % 16 frame_idx += 1 if self.stop_event.wait(0.1): break loop_count += 1 if loops > 0 and loop_count >= loops: break clear_matrix(self.sock, ip, port, width, height) # --- HTML Documentation Template --- HTML_DOCS_TEMPLATE = """ WLED Billboard Daemon API

WLED Billboard Daemon Controller

Interactive Web Control Panel

WOMAN CAVE

Foreground / Text Style

Background Canvas Style

Dynamic REST API Payload (cURL)


        
""" # --- HTTP Daemon Request Handler --- class DaemonHTTPHandler(BaseHTTPRequestHandler): daemon_instance = None def log_message(self, format, *args): return def do_GET(self): if self.path.startswith("/font/"): font_key = self.path.split("/")[-1] if font_key in ALL_FONT_MAP: try: with open(ALL_FONT_MAP[font_key], "rb") as f: font_data = f.read() self.send_response(200) self.send_header("Content-Type", "font/ttf") self.end_headers() self.wfile.write(font_data) return except Exception: pass self.send_error(404, "Font not found") return html_page = HTML_DOCS_TEMPLATE.replace("__ALL_FONTS_JSON__", json.dumps(ALL_FONT_MAP)) html_page = html_page.replace("__MONO_FONTS_JSON__", json.dumps(MONO_FONT_MAP)) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(html_page.encode("utf-8")) def do_POST(self): content_length = int(self.headers.get("Content-Length", 0)) post_data = self.rfile.read(content_length) try: payload = json.loads(post_data.decode("utf-8")) except Exception as e: return self._send_error_response(f"Invalid JSON body: {str(e)}") parsed_params = {} if payload.get("action") in ["clear", "stop"]: self.daemon_instance.stop_animation() return self._send_success_response("Matrix cleared and animation stopped.") mode = str(payload.get("mode", "scroll")).lower() if mode not in ["scroll", "clock_digital", "clock_analog"]: mode = "scroll" parsed_params["mode"] = mode if mode == "scroll": if "text" not in payload or not str(payload.get("text", "")).strip(): return self._send_error_response("Missing or empty required field: 'text'") parsed_params["text"] = str(payload["text"]) try: parsed_params["sweep_pause"] = float(payload.get("sweep_pause", 2.0)) except ValueError: return self._send_error_response("Field 'sweep_pause' must be a float in seconds.") if mode == "clock_digital": parsed_params["clock_date"] = bool(payload.get("clock_date", True)) parsed_params["clock_seconds"] = bool(payload.get("clock_seconds", True)) parsed_params["clock_blink"] = bool(payload.get("clock_blink", True)) effect = payload.get("effect", "solid").lower() if effect not in ["solid", "rainbow", "gradient-h", "gradient-v", "pulse"]: return self._send_error_response(f"Invalid effect '{effect}'.") parsed_params["effect"] = effect bg_effect = payload.get("bg_effect", "solid").lower() if bg_effect not in ["solid", "rainbow", "gradient-h", "gradient-v", "pulse"]: return self._send_error_response(f"Invalid bg_effect '{bg_effect}'.") parsed_params["bg_effect"] = bg_effect font_val = str(payload.get("font", "")).strip() if mode == "clock_digital": if font_val in MONO_FONT_MAP: parsed_params["font"] = MONO_FONT_MAP[font_val] else: parsed_params["font"] = next(iter(MONO_FONT_MAP.values())) if MONO_FONT_MAP else next(iter(ALL_FONT_MAP.values())) else: if font_val in ALL_FONT_MAP: parsed_params["font"] = ALL_FONT_MAP[font_val] elif os.path.exists(font_val): parsed_params["font"] = font_val else: parsed_params["font"] = self.daemon_instance.default_font try: parsed_params["color"] = parse_rgb(payload.get("color", "magenta")) parsed_params["color2"] = parse_rgb(payload.get("color2", "cyan")) parsed_params["bg_color"] = parse_rgb(payload.get("bg_color", "black")) parsed_params["bg_color2"] = parse_rgb(payload.get("bg_color2", "navy")) except ValueError as ve: return self._send_error_response(str(ve)) try: parsed_params["brightness"] = int(payload.get("brightness", 255)) if not (0 <= parsed_params["brightness"] <= 255): raise ValueError() except ValueError: return self._send_error_response("Field 'brightness' must be an integer between 0 and 255.") try: parsed_params["speed"] = float(payload.get("speed", 0.05)) if parsed_params["speed"] < 0.001: raise ValueError() except ValueError: return self._send_error_response("Field 'speed' must be a positive float (delay in seconds).") try: parsed_params["loops"] = int(payload.get("loops", 0)) if parsed_params["loops"] < 0: raise ValueError() except ValueError: return self._send_error_response("Field 'loops' must be an integer >= 0.") if "ip" in payload: parsed_params["ip"] = str(payload["ip"]) if "port" in payload: parsed_params["port"] = int(payload["port"]) self.daemon_instance.start_animation(parsed_params) return self._send_success_response("Pushed payload to matrix.") def _send_success_response(self, message): self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() res = {"status": "success", "message": message} self.wfile.write(json.dumps(res).encode("utf-8")) def _send_error_response(self, error_msg): self.send_response(400) self.send_header("Content-Type", "application/json") self.end_headers() usage_instructions = { "status": "error", "error_detail": error_msg } self.wfile.write(json.dumps(usage_instructions, indent=2).encode("utf-8")) # --- CLI & Daemon Entrypoint --- def main(): parser = argparse.ArgumentParser(description="WLED DDP Matrix Billboard & Daemon") parser.add_argument("-d", "--daemon", action="store_true", help="Run as background HTTP daemon server") parser.add_argument("--listen-ip", type=str, default="0.0.0.0", help="HTTP daemon bind IP") parser.add_argument("--listen-port", type=int, default=8080, help="HTTP daemon listening port") parser.add_argument("-m", "--mode", type=str, choices=["scroll", "clock_digital", "clock_analog"], default="scroll", help="Display mode") parser.add_argument("-t", "--text", type=str, default="WOMAN CAVE", help="Text to scroll") parser.add_argument("--ip", type=str, default="192.168.195.216", help="WLED IP address") parser.add_argument("--port", type=int, default=4048, help="DDP UDP Port") parser.add_argument("-b", "--brightness", type=int, default=255, help="Global brightness scale (0-255)") parser.add_argument("--no-date", action="store_true", help="Hide date in digital clock mode") parser.add_argument("--no-seconds", action="store_true", help="Hide seconds in digital clock mode") parser.add_argument("--no-blink", action="store_true", help="Disable blinking colon in digital clock mode") parser.add_argument("-e", "--effect", type=str, choices=["solid", "rainbow", "gradient-h", "gradient-v", "pulse"], default="solid", help="Text effect") parser.add_argument("-c", "--color", type=str, default="magenta", help="Primary color") parser.add_argument("--color2", type=str, default="cyan", help="Secondary color for gradient effect") parser.add_argument("--bg-effect", type=str, choices=["solid", "rainbow", "gradient-h", "gradient-v", "pulse"], default="solid", help="Background canvas effect") parser.add_argument("--bg-color", type=str, default="black", help="Background primary color") parser.add_argument("--bg-color2", type=str, default="navy", help="Background secondary color for gradient effect") parser.add_argument("-s", "--speed", type=float, default=0.05, help="Scroll frame delay in seconds") parser.add_argument("--sweep-pause", type=float, default=2.0, help="Seconds to pause on blank screen after sweep") parser.add_argument("--font", type=str, default="default", help="Path or preset key for TTF font") parser.add_argument("--width", type=int, default=48, help="Matrix width in pixels") parser.add_argument("--height", type=int, default=32, help="Matrix height in pixels") parser.add_argument("-l", "--loops", type=int, default=0, help="Number of scroll loops (0 = infinite)") args = parser.parse_args() font_path = ALL_FONT_MAP.get(args.font, args.font) if not os.path.exists(font_path): font_path = next(iter(ALL_FONT_MAP.values())) if ALL_FONT_MAP else "" daemon_mgr = BillboardDaemon( default_ip=args.ip, default_port=args.port, default_width=args.width, default_height=args.height, default_font=font_path ) if args.daemon: DaemonHTTPHandler.daemon_instance = daemon_mgr server = HTTPServer((args.listen_ip, args.listen_port), DaemonHTTPHandler) print(f"Billboard Daemon active on http://{args.listen_ip}:{args.listen_port}") print(f"Found {len(ALL_FONT_MAP)} total fonts ({len(MONO_FONT_MAP)} monospace).") print(f"Targeting WLED at {args.ip}:{args.port} | Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down daemon...") daemon_mgr.stop_animation() server.server_close() print("Daemon stopped cleanly.") else: try: params = { "mode": args.mode, "text": args.text, "ip": args.ip, "port": args.port, "width": args.width, "height": args.height, "font": font_path, "effect": args.effect, "color": parse_rgb(args.color), "color2": parse_rgb(args.color2), "bg_effect": args.bg_effect, "bg_color": parse_rgb(args.bg_color), "bg_color2": parse_rgb(args.bg_color2), "brightness": args.brightness, "speed": args.speed, "sweep_pause": args.sweep_pause, "loops": args.loops, "clock_date": not args.no_date, "clock_seconds": not args.no_seconds, "clock_blink": not args.no_blink } except ValueError as ve: print(f"Error: {ve}") sys.exit(1) print(f"Streaming CLI payload to {args.ip}:{args.port} | Press Ctrl+C to stop.") daemon_mgr.start_animation(params) try: while daemon_mgr.worker_thread and daemon_mgr.worker_thread.is_alive(): time.sleep(0.5) except KeyboardInterrupt: print("\nStopping CLI stream...") daemon_mgr.stop_animation() if __name__ == "__main__": main()