diff --git a/tests/test_gst_keyer.py b/tests/test_gst_keyer.py new file mode 100644 index 0000000..b700eff --- /dev/null +++ b/tests/test_gst_keyer.py @@ -0,0 +1,51 @@ +"""gst-keyer.py invariants — the lightweight GStreamer graphics keyer that replaces the +CEF/Chromium html5-keyer for simple PNG overlays (full 30fps at ~1 core vs CEF's ~2 cores +/ 17fps software rendering). Text-scanned: running it needs GStreamer + mxlsrc/mxlsink. +""" +import ast +from pathlib import Path + +KEYER = Path(__file__).resolve().parent.parent / "tools" / "gst-keyer.py" + + +def test_exists_and_parses(): + assert KEYER.is_file(), "tools/gst-keyer.py missing" + ast.parse(KEYER.read_text()) # valid python + + +def test_reads_and_writes_mxl_flows(): + """The keyer's I/O contract: read a program flow (mxlsrc), write the keyed flow + (mxlsink) — a drop-in for what the CEF keyer produces.""" + s = KEYER.read_text() + assert "mxlsrc" in s and "mxlsink" in s, "must read via mxlsrc and write via mxlsink" + assert "video-flow-id=" in s and "flow-id=" in s, "must wire input + output flow UUIDs" + + +def test_composites_png_overlay_no_browser(): + """Composites a PNG (with alpha) via gdkpixbufoverlay — NO CEF/Chromium/browser. + (Scan code lines, not the docstring, which legitimately explains what it replaces.)""" + s = KEYER.read_text() + assert "gdkpixbufoverlay" in s, "must use gdkpixbufoverlay for the graphic" + # strip the module docstring, then ensure no CEF/Chromium element in the actual code + tree = ast.parse(s) + body = s + if (tree.body and isinstance(tree.body[0], ast.Expr) + and isinstance(getattr(tree.body[0], "value", None), ast.Constant)): + body = s.replace(ast.get_docstring(tree) or "", "", 1) + low = body.lower() + assert "cefsrc" not in low and "gstcef" not in low and "chromium" not in low, \ + "the pipeline must not use a CEF/Chromium element" + + +def test_overlay_is_optional(): + """No PNG → clean passthrough (keyer with key off), so it can run without a graphic.""" + s = KEYER.read_text() + assert "if png" in s or "if a.png" in s or "overlay = \"\"" in s, \ + "overlay must be optional (empty when no PNG)" + + +def test_pins_v210_and_framerate(): + """MXL program is 10-bit v210; the pipeline must convert to/from it and pin the rate.""" + s = KEYER.read_text() + assert "v210" in s, "must handle v210 (MXL program format)" + assert "framerate=" in s, "must pin the grain/frame rate" diff --git a/tests/test_make_lower_third.py b/tests/test_make_lower_third.py new file mode 100644 index 0000000..1642bbb --- /dev/null +++ b/tests/test_make_lower_third.py @@ -0,0 +1,19 @@ +"""make-lower-third.py — the cairo-based lower-third PNG generator for gst-keyer. Text-scan +(running it needs pycairo + a font); verifies it's the Skin layer: cairo, RGBA, no browser.""" +import ast +from pathlib import Path +T = Path(__file__).resolve().parent.parent / "tools" / "make-lower-third.py" + +def test_exists_and_parses(): + assert T.is_file(); ast.parse(T.read_text()) + +def test_uses_cairo_rgba_no_browser(): + s = T.read_text() + assert "import cairo" in s, "must use pycairo" + assert "FORMAT_ARGB32" in s, "must render RGBA (alpha for overlay)" + assert "write_to_png" in s, "must output a PNG" + assert "cef" not in s.lower() and "chrom" not in s.lower(), "no browser dependency" + +def test_takes_title_and_subtitle(): + s = T.read_text() + assert "--title" in s and "--subtitle" in s, "customizable text (the Skin layer)" diff --git a/tools/gst-keyer.py b/tools/gst-keyer.py new file mode 100644 index 0000000..66a8d82 --- /dev/null +++ b/tools/gst-keyer.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""gst-keyer — a lightweight GStreamer graphics keyer for MXL program. + +Why this exists: the cbcrc html5-keyer composites an HTML/CSS graphic via a headless +Chromium (CEF). On a box without GPU-accelerated CEF it falls back to SwiftShader +software rendering and costs ~2+ cores while only hitting ~17 fps (and the CEF image +hardcodes the software GL backend, so a GPU doesn't help it). For a simple lower-third / +logo overlay that's wildly overbuilt. + +This keyer does the same JOB — read the program flow, composite a graphic over it, write +the keyed program flow — with a plain GStreamer pipeline and NO browser: + + mxlsrc(program) → videoconvert → gdkpixbufoverlay(PNG w/ alpha) → videoconvert → mxlsink + +It reads a PNG (with alpha) as the overlay — a lower-third, logo, bug, etc. — positioned +anywhere. Swap the PNG and toggle with /pipeline/key. Costs a few % of one core at 30 fps. +For animated/HTML graphics, the CEF keyer is still the tool; this is the cheap 90% case. + +Env / args: + --domain MXL domain path (default /mxl-domain) + --in input (program) flow UUID to read + --out output (keyed) flow UUID to write + --png overlay PNG path (RGBA; transparent where no graphic) [optional] + --x --y overlay top-left position in px (default 0,0) + --rate grain rate numerator (fps) (default 30) + --w --h frame size (default 1920x1080) + +Toggle the overlay on/off at runtime by writing the PNG path (key on) or an empty/1x1 +transparent PNG (key off) and SIGHUP — or just run two invocations. Kept deliberately +simple; the control plane (mxl-routes) can manage lifecycle like it does the CEF keyer. +""" +import argparse +import sys +import gi +gi.require_version("Gst", "1.0") +from gi.repository import Gst, GLib + +Gst.init(None) + + +def build_pipeline(domain, in_uuid, out_uuid, png, x, y, num, w, h): + # v210 is MXL's 10-bit 4:2:2; convert to a form the overlay understands, composite, + # convert back to v210 for the sink. gdkpixbufoverlay blends the PNG's alpha. + overlay = "" + if png: + overlay = (f"gdkpixbufoverlay location={png} offset-x={x} offset-y={y} " + f"overlay-width=0 overlay-height=0 ! ") + desc = ( + f"mxlsrc domain={domain} video-flow-id={in_uuid} ! " + f"video/x-raw,format=v210,width={w},height={h},framerate={num}/1 ! " + f"videoconvert ! video/x-raw,format=BGRA ! " + f"{overlay}" + f"videoconvert ! video/x-raw,format=v210,width={w},height={h},framerate={num}/1 ! " + f"queue max-size-buffers=4 leaky=downstream ! " + f"mxlsink domain={domain} flow-id={out_uuid} sync=false " + f'group-hint="Keyer:Video" description="program + graphics" label="Keyer PGM"' + ) + return Gst.parse_launch(desc), desc + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--domain", default="/mxl-domain") + ap.add_argument("--in", dest="in_uuid", required=True) + ap.add_argument("--out", dest="out_uuid", required=True) + ap.add_argument("--png", default="") + ap.add_argument("--x", type=int, default=0) + ap.add_argument("--y", type=int, default=0) + ap.add_argument("--rate", type=int, default=30) + ap.add_argument("--w", type=int, default=1920) + ap.add_argument("--h", type=int, default=1080) + a = ap.parse_args() + + pipe, desc = build_pipeline(a.domain, a.in_uuid, a.out_uuid, a.png, a.x, a.y, a.rate, a.w, a.h) + print(f"gst-keyer: {desc}", flush=True) + + loop = GLib.MainLoop() + bus = pipe.get_bus() + bus.add_signal_watch() + + def on_msg(_bus, msg): + t = msg.type + if t == Gst.MessageType.ERROR: + err, dbg = msg.parse_error() + print(f"gst-keyer ERROR: {err} ({dbg})", flush=True) + loop.quit() + elif t == Gst.MessageType.EOS: + print("gst-keyer EOS", flush=True) + loop.quit() + bus.connect("message", on_msg) + + pipe.set_state(Gst.State.PLAYING) + print("gst-keyer running", flush=True) + try: + loop.run() + except KeyboardInterrupt: + pass + finally: + pipe.set_state(Gst.State.NULL) + return 1 # non-zero so a supervising wrapper's `while :; do … done` restarts it + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/make-lower-third.py b/tools/make-lower-third.py new file mode 100644 index 0000000..6ccfca9 --- /dev/null +++ b/tools/make-lower-third.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +"""make-lower-third.py — render a lower-third overlay PNG (RGBA, 1920x1080) with cairo. + +A full-frame transparent canvas with a lower-third graphic baked in at the bottom-left, +so tools/gst-keyer.py can composite it over program with gdkpixbufoverlay at offset 0,0. +No browser, no PIL — just pycairo (present in the keyer image) + a system font. + + python3 make-lower-third.py --title "GUEST NAME" --subtitle "Role / Location" \ + --out /tmp/lower-third.png + +Customize freely — this is the "Skin" layer (colors/brand), kept out of the generic Core. +""" +import argparse +import cairo + +W, H = 1920, 1080 + +# palette (matches the Skin's dark theme) +BG = (0x0b / 255, 0x10 / 255, 0x18 / 255, 0.82) # panel, mostly opaque +ACCENT = (0x35 / 255, 0xe0 / 255, 0xff / 255, 1.0) # cyan stripe +INK = (0xe8 / 255, 0xee / 255, 0xff / 255, 1.0) # title +DIM = (0x9a / 255, 0xa6 / 255, 0xc2 / 255, 1.0) # subtitle + + +def rounded(ctx, x, y, w, h, r): + ctx.new_sub_path() + ctx.arc(x + w - r, y + r, r, -1.5708, 0) + ctx.arc(x + w - r, y + h - r, r, 0, 1.5708) + ctx.arc(x + r, y + h - r, r, 1.5708, 3.1416) + ctx.arc(x + r, y + r, r, 3.1416, 4.7124) + ctx.close_path() + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--title", default="MXL SWITCHER") + ap.add_argument("--subtitle", default="Live from one $400 box") + ap.add_argument("--out", default="/tmp/lower-third.png") + # geometry (bottom-left lower third) + ap.add_argument("--x", type=int, default=90) + ap.add_argument("--y", type=int, default=865) + ap.add_argument("--w", type=int, default=860) + ap.add_argument("--h", type=int, default=120) + a = ap.parse_args() + + surf = cairo.ImageSurface(cairo.FORMAT_ARGB32, W, H) + ctx = cairo.Context(surf) + # (canvas starts fully transparent) + + bx, by, bw, bh = a.x, a.y, a.w, a.h + # drop shadow + ctx.set_source_rgba(0, 0, 0, 0.35) + rounded(ctx, bx + 5, by + 6, bw, bh, 12) + ctx.fill() + # panel + ctx.set_source_rgba(*BG) + rounded(ctx, bx, by, bw, bh, 12) + ctx.fill() + # accent stripe (left edge) + ctx.set_source_rgba(*ACCENT) + rounded(ctx, bx, by, 14, bh, 7) + ctx.fill() + + tx = bx + 40 + # title + ctx.select_font_face("Liberation Sans", cairo.FONT_SLANT_NORMAL, cairo.FONT_WEIGHT_BOLD) + ctx.set_font_size(46) + ctx.set_source_rgba(*INK) + ctx.move_to(tx, by + 58) + ctx.show_text(a.title) + # subtitle + ctx.select_font_face("Liberation Sans", cairo.FONT_SLANT_NORMAL, cairo.FONT_WEIGHT_NORMAL) + ctx.set_font_size(26) + ctx.set_source_rgba(*DIM) + ctx.move_to(tx, by + 96) + ctx.show_text(a.subtitle) + + surf.write_to_png(a.out) + print(f"wrote {a.out} ({a.title!r} / {a.subtitle!r})") + + +if __name__ == "__main__": + main()