#!/usr/bin/env python3 """ treppe-gl: render Shadertoy style GLSL shaders onto the treppe2000. The shader is rendered into a 16x1 float framebuffer (x = stair step), packed as RGBW16 and sent to the priority UDP port of trepped. An optional debug window shows the shader at full resolution and the 16 steps as they are sent. """ import argparse import os import re import signal import socket import sys import time import numpy as np import moderngl STEPS = 16 STEPS_ACTIVE = 13 # trepped only drives 13 of the 16 channels HOST_DEFAULT = "treppe.club.berlin.ccc.de" PORT_DEFAULT = 3124 # priority port, overrides 3123 for the gracetime CMD_FRAME = 0x42 FLAG_RGBA = 0x02 FLAG_BITS_16 = 0x08 WHITE_MODES = ("extract", "alpha", "off") COMPACT_MODES = ("mean", "max", "row") # --- Protocol -------------------------------------------------------------- def to_rgbw(pixels, white="extract", brightness=100.0): """ Convert float RGBA pixels (STEPS x 4, 0..1) to RGBW (STEPS x 4). extract: w = min(r, g, b), subtracted from r, g, b alpha: the shader's alpha channel is the white value off: w = 0 brightness (0..100, percent) scales the result linearly. """ scale = np.float32(min(max(brightness, 0.0), 100.0) / 100.0) px = np.clip(np.asarray(pixels, dtype=np.float32), 0.0, 1.0) rgb = px[:, :3] if white == "extract": w = rgb.min(axis=1, keepdims=True) return np.hstack([rgb - w, w]) * scale if white == "alpha": return np.hstack([rgb, px[:, 3:4]]) * scale if white == "off": return np.hstack([rgb, np.zeros((len(px), 1), dtype=np.float32)]) * scale raise ValueError("unknown white mode: {}".format(white)) def encode_frame(rgbw): """Encode STEPS x 4 floats as a trepped RGBW16 (big endian) frame.""" data = np.clip(np.asarray(rgbw, dtype=np.float32), 0.0, 1.0) data = np.floor(data * 65535.0 + 0.5).astype(">u2") return bytes([CMD_FRAME, FLAG_RGBA | FLAG_BITS_16]) + data.tobytes() BLACK_FRAME = encode_frame(np.zeros((STEPS, 4), dtype=np.float32)) # --- 2D canvas ------------------------------------------------------------ CANVAS_RE = re.compile( r"^\s*//\s*canvas:\s*(\d+)x(\d+)(?:\s+compact:\s*(\w+))?", re.M) def parse_canvas(text): """Parse '256x32' into (256, 32); the width must fit the 16 steps.""" m = re.fullmatch(r"(\d+)x(\d+)", text.strip()) if not m: raise ValueError("canvas must look like 256x32, got {!r}".format(text)) w, h = int(m.group(1)), int(m.group(2)) if w < STEPS or h < 1 or w % STEPS != 0: raise ValueError( "canvas width must be a multiple of {}, got {}x{}".format( STEPS, w, h)) return (w, h) def directive(source): """ Read '// canvas: 256x32 compact: mean' from the top of a shader. Returns (canvas or None, compact or None). """ m = CANVAS_RE.search("\n".join(source.splitlines()[:15])) if not m: return None, None compact = m.group(3) if compact is not None and compact not in COMPACT_MODES: raise ValueError("unknown compact mode: {}".format(compact)) return parse_canvas("{}x{}".format(m.group(1), m.group(2))), compact def compact_canvas(pixels, mode="mean"): """ Compact a 2D canvas (H x W x 4 floats) to STEPS x 4, one step per W/STEPS wide group of columns. mean: average of the whole column group max: brightest value per channel in the group row: average of the middle row only """ h, w, _ = pixels.shape cols = w // STEPS if mode == "row": return pixels[h // 2].reshape(STEPS, cols, 4).mean(axis=1) groups = pixels.reshape(h, STEPS, cols, 4) if mode == "max": return groups.max(axis=(0, 2)) return groups.mean(axis=(0, 2)) # --- Shaders --------------------------------------------------------------- VERTEX = """ #version 330 in vec2 pos; out vec2 v_uv; void main() { v_uv = pos * 0.5 + 0.5; gl_Position = vec4(pos, 0.0, 1.0); } """ PRELUDE = """#version 330 uniform vec3 iResolution; uniform float iTime; uniform float iTimeDelta; uniform int iFrame; out vec4 _fragColor; """ EPILOGUE = """ void main() { vec4 c = vec4(0.0, 0.0, 0.0, 1.0); mainImage(c, gl_FragCoord.xy); _fragColor = c; } """ # Draw a texture as is. BLIT = """ #version 330 uniform sampler2D tex; in vec2 v_uv; out vec4 color; void main() { color = vec4(texture(tex, v_uv).rgb, 1.0); } """ # 16 blocks with step numbers, a bar for the white channel and # the inactive steps greyed out. STAIRS = """ #version 330 uniform sampler2D strip; uniform vec2 res; in vec2 v_uv; out vec4 color; // 3x5 digit font, row major, top row in the highest bits const int FONT[10] = int[10]( 31599, 11415, 29671, 29647, 23497, 31183, 31215, 29257, 31727, 31695); float digit(int d, vec2 p) { ivec2 c = ivec2(floor(p)); if (d < 0 || d > 9 || c.x < 0 || c.x > 2 || c.y < 0 || c.y > 4) { return 0.0; } int bit = c.y * 3 + (2 - c.x); return float((FONT[d] >> bit) & 1); } float number(int n, vec2 p) { // p in font pixels, two digits wide if (n >= 10) { return max(digit(n / 10, p), digit(n % 10, p - vec2(4.0, 0.0))); } return digit(n, p - vec2(2.0, 0.0)); } void main() { vec2 px = v_uv * res; float cell = res.x / 16.0; int i = int(floor(px.x / cell)); vec2 local = vec2(px.x - float(i) * cell, px.y); vec3 bg = vec3(0.06); vec3 col = bg; bool is_on = i < 13; float bar_h = res.y * 0.12; float num_h = res.y * 0.22; float gap = 3.0; vec4 s = texelFetch(strip, ivec2(i, 0), 0); float w = min(min(s.r, s.g), s.b); if (px.y > num_h + bar_h + gap) { // colour block if (local.x > gap && local.x < cell - gap) { col = clamp(s.rgb, 0.0, 1.0); if (!is_on) { col = mix(col * 0.25, vec3(0.2), 0.5); } } } else if (px.y > num_h) { // white bar (extracted white = min(rgb)) if (local.x > gap && local.x < cell - gap && local.y - num_h < bar_h) { float f = (local.x - gap) / (cell - 2.0 * gap); col = f < w ? vec3(0.95) : vec3(0.15); if (!is_on) { col *= 0.4; } } } else { // step number float scale = max(1.0, floor(num_h / 7.0)); vec2 p = (local - vec2(0.0, num_h * 0.5 - 2.5 * scale)) / scale; p.x -= (cell / scale - 7.0) * 0.5; float on = number(i, p); col = mix(bg, is_on ? vec3(0.9) : vec3(0.35), on); } color = vec4(col, 1.0); } """ def wrap_shader(source): return PRELUDE + source + EPILOGUE class Renderer: """Renders a Shadertoy style shader into the stairs strip.""" def __init__(self, ctx, size=(512, 288)): self.ctx = ctx self.quad = ctx.buffer( np.array([-1, -1, 3, -1, -1, 3], dtype="f4").tobytes()) self.size = size self.canvas = None self.compact = "mean" self.strip_tex = ctx.texture((STEPS, 1), 4, dtype="f4") self.strip_tex.filter = (moderngl.NEAREST, moderngl.NEAREST) self.strip_fbo = ctx.framebuffer([self.strip_tex]) self.full_tex = None self.full_fbo = None self._make_full(size, "f1") self.program = None self.vao_strip = None self.vao_full = None self.time = 0.0 self.frame = 0 def _make_full(self, size, dtype): if self.full_tex is not None: self.full_fbo.release() self.full_tex.release() self.size = size self.full_tex = self.ctx.texture(size, 4, dtype=dtype) self.full_tex.filter = (moderngl.NEAREST if self.canvas else moderngl.LINEAR,) * 2 self.full_fbo = self.ctx.framebuffer([self.full_tex]) def resize(self, size): """Resize the preview. In 2D canvas mode the canvas size is fixed.""" if self.canvas: return size = (max(1, int(size[0])), max(1, int(size[1]))) if size != self.size: self._make_full(size, "f1") def configure(self, canvas=None, compact="mean"): """Use a 2D canvas (W, H) compacted to the steps, or None for 1D.""" if canvas is not None: canvas = parse_canvas("{}x{}".format(*canvas)) if compact not in COMPACT_MODES: raise ValueError("unknown compact mode: {}".format(compact)) self.compact = compact if canvas == self.canvas: return self.canvas = canvas if canvas: self._make_full(canvas, "f4") else: self._make_full((512, 288), "f1") def load(self, source): """Compile a shader. Raises on error and keeps the old one.""" program = self.ctx.program( vertex_shader=VERTEX, fragment_shader=wrap_shader(source)) if self.program is not None: self.program.release() self.program = program self.vao = self.ctx.vertex_array( self.program, [(self.quad, "2f", "pos")]) def _set(self, name, value): if name in self.program: self.program[name].value = value def _draw(self, fbo, res, t, dt, frame): fbo.use() fbo.viewport = (0, 0, res[0], res[1]) self._set("iResolution", (float(res[0]), float(res[1]), 1.0)) self._set("iTime", t) self._set("iTimeDelta", dt) self._set("iFrame", frame) self.vao.render(moderngl.TRIANGLES, vertices=3) def render_strip(self, t, dt=0.0, frame=0): """Render the 16x1 strip, returns STEPS x 4 float pixels.""" if self.canvas: # Render the 2D canvas and compact it down to the steps. self._draw(self.full_fbo, self.canvas, t, dt, frame) raw = self.full_fbo.read(components=4, dtype="f4") w, h = self.canvas pixels = np.frombuffer(raw, dtype="f4").reshape(h, w, 4) strip = compact_canvas(pixels, self.compact).astype("f4") self.strip_tex.write(strip.tobytes()) return strip self._draw(self.strip_fbo, (STEPS, 1), t, dt, frame) raw = self.strip_fbo.read(components=4, dtype="f4") return np.frombuffer(raw, dtype="f4").reshape(STEPS, 4).copy() def render_full(self, t, dt=0.0, frame=0): if self.canvas: return # already rendered by render_strip self._draw(self.full_fbo, self.size, t, dt, frame) class DebugView: """Composites the full resolution shader and the stairs strip.""" def __init__(self, ctx, renderer): self.ctx = ctx self.renderer = renderer self.blit = ctx.program(vertex_shader=VERTEX, fragment_shader=BLIT) self.stairs = ctx.program(vertex_shader=VERTEX, fragment_shader=STAIRS) self.vao_blit = ctx.vertex_array( self.blit, [(renderer.quad, "2f", "pos")]) self.vao_stairs = ctx.vertex_array( self.stairs, [(renderer.quad, "2f", "pos")]) def draw(self, width, height): top_h = int(height * 0.65) bottom_h = height - top_h self.renderer.resize((width, top_h)) screen = self.ctx.screen screen.use() self.ctx.clear(0.0, 0.0, 0.0) screen.viewport = (0, bottom_h, width, top_h) self.renderer.full_tex.use(0) self.blit["tex"].value = 0 self.vao_blit.render(moderngl.TRIANGLES, vertices=3) screen.viewport = (0, 0, width, bottom_h) self.renderer.strip_tex.use(0) self.stairs["strip"].value = 0 self.stairs["res"].value = (float(width), float(bottom_h)) self.vao_stairs.render(moderngl.TRIANGLES, vertices=3) # --- Runtime --------------------------------------------------------------- class ShaderFile: """Reloads a shader file when it changes.""" def __init__(self, path): self.path = path self.mtime = None def poll(self): """Returns new source if the file changed, else None.""" try: mtime = os.stat(self.path).st_mtime except OSError as err: if self.mtime is None: raise print("cannot read {}: {}".format(self.path, err), file=sys.stderr) return None if mtime == self.mtime: return None self.mtime = mtime with open(self.path) as f: return f.read() def create_context(window): """Returns (ctx, glfw_window or None).""" if not window: return moderngl.create_standalone_context(backend="egl"), None import glfw if not glfw.init(): raise RuntimeError("could not initialise glfw") glfw.window_hint(glfw.CONTEXT_VERSION_MAJOR, 3) glfw.window_hint(glfw.CONTEXT_VERSION_MINOR, 3) glfw.window_hint(glfw.OPENGL_PROFILE, glfw.OPENGL_CORE_PROFILE) glfw.window_hint(glfw.OPENGL_FORWARD_COMPAT, True) win = glfw.create_window(1024, 512, "treppe-gl", None, None) if not win: glfw.terminate() raise RuntimeError("could not create window") glfw.make_context_current(win) glfw.swap_interval(0) return moderngl.create_context(), win def resolve(host, port): info = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_DGRAM) return info[0][4] def run(args): shader = ShaderFile(args.shader) target = None sock = None if not args.dry_run: target = resolve(args.host, args.port) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) print("sending to {}:{} ({}), overriding the stairs".format( args.host, target[1], target[0]), file=sys.stderr) else: print("dry run, not sending anything", file=sys.stderr) ctx, win = create_context(args.window) renderer = Renderer(ctx) view = DebugView(ctx, renderer) if win else None stop = [] signal.signal(signal.SIGINT, lambda *_: stop.append(True)) signal.signal(signal.SIGTERM, lambda *_: stop.append(True)) import glfw t0 = time.time() last = t0 frame = 0 next_reload = 0.0 period = 1.0 / args.fps try: while not stop: now = time.time() if now >= next_reload: next_reload = now + 0.25 source = shader.poll() if source is not None: try: canvas, compact = directive(source) if args.canvas: canvas = parse_canvas(args.canvas) renderer.load(source) renderer.configure( canvas, args.compact or compact or "mean") print("loaded {}".format(args.shader), file=sys.stderr) except Exception as err: print("shader error:\n{}".format(err), file=sys.stderr) if renderer.program is None: raise SystemExit(1) t = now - t0 dt = now - last last = now pixels = renderer.render_strip(t, dt, frame) rgbw = to_rgbw(pixels, args.white, args.brightness) if sock: sock.sendto(encode_frame(rgbw), target) if win: if glfw.window_should_close(win): break w, h = glfw.get_framebuffer_size(win) if w > 0 and h > 0: renderer.render_full(t, dt, frame) view.draw(w, h) glfw.swap_buffers(win) glfw.poll_events() frame += 1 sleep = period - (time.time() - now) if sleep > 0: time.sleep(sleep) finally: if sock: for _ in range(3): sock.sendto(BLACK_FRAME, target) print("sent black frame", file=sys.stderr) if win: glfw.terminate() def parse_args(argv=None): p = argparse.ArgumentParser(description=__doc__) p.add_argument("shader", help="Shadertoy style GLSL file (mainImage)") p.add_argument("--host", default=HOST_DEFAULT) p.add_argument("--port", type=int, default=PORT_DEFAULT) p.add_argument("--fps", type=float, default=60.0) p.add_argument("--white", choices=WHITE_MODES, default="extract") p.add_argument("--canvas", metavar="WxH", help="render a 2D canvas and compact it to the steps " "(overrides '// canvas: WxH' in the shader)") p.add_argument("--compact", choices=COMPACT_MODES, help="how the canvas is compacted (default mean)") p.add_argument("--brightness", type=float, default=100.0, help="0..100 percent, scales what is sent (default 100)") p.add_argument("--dry-run", action="store_true", help="do not send anything over the network") p.add_argument("--no-window", dest="window", action="store_false", help="headless, no debug window") return p.parse_args(argv) if __name__ == "__main__": run(parse_args())