#!/usr/bin/env python3 import argparse import colorsys import datetime import json import math import os import re import socket import sys import threading import time import urllib.parse 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" } ICON_DIR = "icons" def scan_icons(): icons = [] if os.path.exists(ICON_DIR): for file in os.listdir(ICON_DIR): if file.lower().endswith(".png"): icons.append(file[:-4]) return sorted(icons) 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, "icon_layer", None): overlay.paste(base_banner.icon_layer, (0, 0), base_banner.icon_layer) 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() if "{icon:" not in text: 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: mask = Image.new("RGBA", (1, target_height), (0, 0, 0, 0)) mask.icon_layer = None return mask cropped = temp_img.crop(bbox) scale_factor = target_height / float(cropped.size[1]) new_w = max(1, int(cropped.size[0] * scale_factor)) mask = cropped.resize((new_w, target_height), Image.Resampling.LANCZOS) mask.icon_layer = None return mask parts = re.split(r'(\{icon:[^}]+\})', text) elements = [] for part in parts: if not part: continue if part.startswith("{icon:") and part.endswith("}"): icon_name = part[6:-1] icon_path = os.path.join(ICON_DIR, f"{icon_name}.png") if os.path.exists(icon_path): try: icon_img = Image.open(icon_path).convert("RGBA") scale = target_height / float(icon_img.size[1]) new_w = max(1, int(icon_img.size[0] * scale)) resized_icon = icon_img.resize((new_w, target_height), Image.Resampling.NEAREST) elements.append(("icon", resized_icon)) continue except Exception: pass temp_img = Image.new("RGBA", (3000, 250), (0, 0, 0, 0)) temp_draw = ImageDraw.Draw(temp_img) temp_draw.text((0, 10), part, font=font, fill=(255, 255, 255, 255)) bbox = temp_img.getbbox() if bbox: cropped = temp_img.crop(bbox) scale = target_height / float(cropped.size[1]) new_w = max(1, int(cropped.size[0] * scale)) resized_text = cropped.resize((new_w, target_height), Image.Resampling.LANCZOS) elements.append(("text", resized_text)) total_w = sum(img.size[0] for _, img in elements) + 20 if total_w <= 20: total_w = 21 base_mask = Image.new("RGBA", (total_w, target_height), (0, 0, 0, 0)) icon_layer = Image.new("RGBA", (total_w, target_height), (0, 0, 0, 0)) current_x = 10 for etype, img in elements: if etype == "text": base_mask.paste(img, (current_x, 0), img) current_x += img.size[0] + 5 elif etype == "icon": icon_layer.paste(img, (current_x, 0), img) current_x += img.size[0] + 5 base_mask.icon_layer = icon_layer return base_mask 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)) mask = cropped.resize((new_w, new_h), Image.Resampling.LANCZOS) mask.icon_layer = None return mask 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 img.icon_layer = None 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) img.icon_layer = None 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() self.ambient_params = None if not os.path.exists(ICON_DIR): os.makedirs(ICON_DIR) def start_animation(self, params): with self.lock: if float(params.get("duration", 0.0)) == 0.0: self.ambient_params = dict(params) 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: self.ambient_params = None 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 duration = float(params.get("duration", 0.0)) end_time = (time.time() + duration) if duration > 0 else None 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 end_time and time.time() >= end_time: break 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 else: 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(): if end_time and time.time() >= end_time: break for x_pos in range(width, -banner_width - 1, -1): if self.stop_event.is_set() or (end_time and time.time() >= end_time): 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() and not (end_time and time.time() >= end_time): pause_end = time.time() + sweep_pause while time.time() < pause_end and not self.stop_event.is_set(): if end_time and time.time() >= end_time: break 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 if not self.stop_event.is_set(): if duration > 0 and self.ambient_params: return self._render_loop(dict(self.ambient_params)) else: clear_matrix(self.sock, ip, port, width, height) # --- HTML Documentation Template --- HTML_DOCS_TEMPLATE = """