#!/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 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") # --- 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)) # --- 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.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.resize(size) self.program = None self.vao_strip = None self.vao_full = None self.time = 0.0 self.frame = 0 def resize(self, size): size = (max(1, int(size[0])), max(1, int(size[1]))) if self.full_fbo is not None and size == self.size: return self.size = size if self.full_tex is not None: self.full_fbo.release() self.full_tex.release() self.full_tex = self.ctx.texture(size, 4, dtype="f1") self.full_tex.filter = (moderngl.LINEAR, moderngl.LINEAR) self.full_fbo = self.ctx.framebuffer([self.full_tex]) 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.""" 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): 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: renderer.load(source) 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("--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())