""" gifsmith.py — the render harness for the `gif` skill. You write ONLY the drawing. This module owns every part that has historically broken renders: the scratch directory and its disk budget, frame naming and the continuous counter, a LOCKED canvas size (frames that differ by one pixel break the encoder), per-scene crash isolation, the GIF-vs-MP4 decision, and the critique contact sheet. from gifsmith import Animation anim = Animation(width=640, height=480, fps=15) # --- cast: define each recurring character ONCE --- def draw_steve(ax, x, y, angry=0.0): ... @anim.scene("Steve watches the printer", seconds=2.5) def s0(ax, t): # t goes 0.0 -> 1.0 across the scene draw_steve(ax, 2, 1, angry=t) art = anim.render() # -> {"path": "/workspace/final.gif", "mime": ..., "bytes": ...} `ax` is a fresh matplotlib Axes at exactly (width, height) pixels, x in [0,100], y in [0, 100*height/width], origin bottom-left, no margins. Draw in those units and every frame lines up. PIL instead of matplotlib: @anim.scene("title card", seconds=1.5, mode="pil") def s1(img, draw, t): # img: RGB Image at the locked size; draw: ImageDraw draw.text((40, 40), "THE END", fill=(255, 255, 255), font=anim.font(48)) render() prints a report and returns the artifact dict. If one scene raises, the OTHER scenes still render and ship — the report names the broken one so you can fix just that scene. It also writes /workspace/still.png: a labelled contact sheet, one tile per scene, for the critique pass. """ import glob import math import os import shutil import subprocess import sys import traceback import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 from PIL import Image, ImageDraw, ImageFont # noqa: E402 # Re-exported so one import line covers the whole drawing vocabulary — a missing # `from matplotlib.patches import ...` is otherwise a NameError that costs a pass. from matplotlib.patches import ( # noqa: E402,F401 Arc, Circle, Ellipse, FancyBboxPatch, Polygon, Rectangle, Wedge) import numpy as np # noqa: E402,F401 # A GIF longer than this balloons past what's worth shipping as a GIF. GIF_MAX_SECONDS = 20 # Keep a GIF comfortably under Discord's ~10 MiB inline limit. GIF_MAX_BYTES = 9 * 1024 * 1024 # Measured on busy 640x480 matplotlib frames in this sandbox: ~86 KiB per PNG at # compress_level=1, i.e. 88064 / 307200 px. This decides whether a render streams # instead of staging frames on disk, so it must not read low — erring low # reintroduces the ENOSPC this harness exists to prevent, and erring high only # turns a borderline short piece into an MP4. PNG_BYTES_PER_PIXEL = 0.29 _FONT_CANDIDATES = [ "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", ] def _which(prog): return shutil.which(prog) is not None def _free_mb(path): try: st = os.statvfs(path) return st.f_bavail * st.f_frsize / 1024 / 1024 except OSError: return 0.0 def _pick_scratch(): """Largest writable scratch dir. /tmp is a small tmpfs in this sandbox; a SUBDIR of /workspace is bigger and is not returned in files_out (only TOP-LEVEL /workspace files are), so it is safe to fill with frames.""" best, best_free = None, -1.0 for base in ("/workspace/.frames", "/tmp/frames", "/var/tmp/frames"): parent = os.path.dirname(base) if not os.path.isdir(parent) or not os.access(parent, os.W_OK): continue free = _free_mb(parent) if free > best_free: best, best_free = base, free if best is None: raise RuntimeError("no writable scratch directory found") shutil.rmtree(best, ignore_errors=True) os.makedirs(best, exist_ok=True) return best, best_free class _Scene: def __init__(self, fn, title, seconds, mode, hold): self.fn, self.title, self.seconds = fn, title, float(seconds), self.mode, self.hold = mode, float(hold) self.frames = 0 self.error = None self.first_frame = None class Animation: """Collects scenes, renders them to a locked-size frame sequence, encodes.""" def __init__(self, width=640, height=480, fps=15, dpi=100, bg="white"): # H.264 needs even dimensions; lock them here so the encoder never fails. self.width = int(width) // 2 * 2 self.height = int(height) // 2 * 2 self.fps = int(fps) self.dpi = int(dpi) self.bg = bg self.xmax = 100.0 self.ymax = 100.0 * self.height / self.width self._scenes = [] self._count = 0 self._scratch = None self._peak_mb = 0.0 self._streaming = False self._pipe = None # ---------------------------------------------------------------- authoring def scene(self, title, seconds=2.0, mode="mpl", hold=0.0): """Decorator: register a scene. `hold` freezes the last frame for that many extra seconds — use it to let a caption stay readable.""" def deco(fn): self._scenes.append(_Scene(fn, title, seconds, mode, hold)) return fn return deco def font(self, size): for p in _FONT_CANDIDATES: if os.path.exists(p): return ImageFont.truetype(p, size) return ImageFont.load_default() def text(self, ax, s, size=18, y=None, color="white", box="black"): """Caption helper: centred, boxed, legible at GIF resolution.""" ax.text(self.xmax / 2, self.ymax * 0.06 if y is None else y, s, ha="center", va="center", fontsize=size, color=color, zorder=999, bbox=dict(boxstyle="round,pad=0.4", fc=box, ec="none", alpha=0.75)) def add_frame(self, img, times=1): """Escape hatch for hand-built frames (PIL compositing, pasted photos, anything the scene decorators can't express). Still size-locked.""" for _ in range(max(1, int(times))): self._emit(img) # ------------------------------------------------------------------ drawing def _new_axes(self): fig = plt.figure(figsize=(self.width / self.dpi, self.height / self.dpi), dpi=self.dpi) ax = fig.add_axes([0, 0, 1, 1]) ax.set_xlim(0, self.xmax) ax.set_ylim(0, self.ymax) ax.set_axis_off() ax.set_facecolor(self.bg) fig.patch.set_facecolor(self.bg) return fig, ax def _emit(self, img): """Emit one frame at exactly the locked size. Never bbox_inches='tight'. Short pieces go to PNG files (gifski wants a file list). Long pieces are piped straight into ffmpeg so a multi-minute render never touches disk — the sandbox scratch would otherwise cap you at well under a minute.""" if img.size != (self.width, self.height): img = img.resize((self.width, self.height), Image.LANCZOS) img = img.convert("RGB") if self._pipe is not None: img.save(self._pipe.stdin, format="PNG", compress_level=1) else: img.save(os.path.join(self._scratch, f"frame_{self._count:05d}.png"), compress_level=1) self._count += 1 def _render_mpl(self, sc, t): fig, ax = self._new_axes() try: sc.fn(ax, t) fig.canvas.draw() img = Image.frombuffer( "RGBA", fig.canvas.get_width_height(), fig.canvas.buffer_rgba(), "raw", "RGBA", 0, 1).convert("RGB") finally: plt.close(fig) return img def _render_pil(self, sc, t): img = Image.new("RGB", (self.width, self.height), self.bg if isinstance(self.bg, (str, tuple)) else "white") sc.fn(img, ImageDraw.Draw(img), t) return img # ------------------------------------------------------------------- render def render(self, long_edge=None, force=None, audio=None): if not self._scenes: raise RuntimeError("no scenes registered — decorate at least one " "function with @anim.scene(...)") self._scratch, free_mb = _pick_scratch() long_edge = long_edge or max(self.width, self.height) total_s = sum(s.seconds + s.hold for s in self._scenes) est_frames = int(total_s * self.fps) est_mb = est_frames * self.width * self.height * PNG_BYTES_PER_PIXEL / 1048576 # A piece too long for a GIF is an MP4 anyway, and an MP4 can be encoded # in one streaming pass — so pipe the frames into ffmpeg and keep NOTHING # on disk. That is what makes multi-minute pieces possible at all: the # sandbox scratch tops out around a minute of busy frames. self._streaming = (force == "mp4" or audio is not None or (force != "gif" and (total_s > GIF_MAX_SECONDS or est_mb > free_mb * 0.6))) self._pipe = None if self._streaming: self._pipe = self._open_pipe("/workspace/final.mp4", long_edge, audio) print(f"[gifsmith] streaming to MP4 (projected {est_mb:.0f} MiB of " f"frames vs {free_mb:.0f} MiB scratch) — nothing hits disk", flush=True) elif est_mb > free_mb * 0.6: print(f"[gifsmith] WARNING projected frames ~{est_mb:.0f} MiB vs " f"{free_mb:.0f} MiB free in {os.path.dirname(self._scratch)} — " f"dropping fps or canvas would be safer", file=sys.stderr) print(f"[gifsmith] {len(self._scenes)} scene(s), {total_s:.1f}s at " f"{self.fps}fps, canvas {self.width}x{self.height}", flush=True) for i, sc in enumerate(self._scenes): start = self._count n = max(1, int(round(sc.seconds * self.fps))) try: for k in range(n): t = k / max(n - 1, 1) img = (self._render_pil if sc.mode == "pil" else self._render_mpl)(sc, t) if sc.first_frame is None and k >= n // 2: sc.first_frame = img.copy() self._emit(img) for _ in range(int(round(sc.hold * self.fps))): self._emit(img) except Exception: # noqa: BLE001 sc.error = traceback.format_exc(limit=6) if not self._streaming: # Drop this scene's partial frames so it can't ship half-drawn. # (Streamed frames are already in the encoder — a broken scene # leaves its partial beat in the MP4; the report says so.) for p in sorted(glob.glob( os.path.join(self._scratch, "frame_*.png")))[start:]: os.remove(p) self._count = start print(f"[gifsmith] SCENE {i} ({sc.title!r}) FAILED — skipped, " f"other scenes continue:\n{sc.error}", file=sys.stderr, flush=True) continue sc.frames = self._count - start print(f"[gifsmith] scene {i}: {sc.frames} frames {sc.title!r}", flush=True) self._renumber_if_needed() used = sum(os.path.getsize(p) for p in glob.glob(os.path.join(self._scratch, "frame_*.png"))) self._peak_mb = used / 1024 / 1024 if self._count == 0: raise RuntimeError( "every scene failed — see the per-scene tracebacks above; " "nothing was encoded. Fix the error they all share (usually one " "cast helper called with a keyword it does not take, or a name " "that differs by a character) and re-run the whole program in a " "fresh code_exec call, remembering gifsmith.py in files_in.") try: art = self._encode(total_s, long_edge, force, audio) except Exception: # noqa: BLE001 # A half-written final.* at the top of /workspace still comes back as # a files_out file_id, and a broken artifact that LOOKS deliverable is # worse than none — the agent would ship it. Take it with us. for leftover in ("/workspace/final.mp4", "/workspace/final.gif"): if os.path.exists(leftover): os.remove(leftover) shutil.rmtree(self._scratch, ignore_errors=True) raise self._contact_sheet() broken = [f"{i}:{s.title!r}" for i, s in enumerate(self._scenes) if s.error] print(f"[gifsmith] {self._count} frames, scratch " + ("streamed (0 MiB on disk)" if self._streaming else f"peak {self._peak_mb:.1f} MiB")) print(f"[gifsmith] scene errors: {', '.join(broken) if broken else 'none'}") print(f"[gifsmith] ART {art}") print("[gifsmith] still: /workspace/still.png (contact sheet — call " "image_describe on ITS file_id from files_out)") shutil.rmtree(self._scratch, ignore_errors=True) return art def _renumber_if_needed(self): """A skipped scene leaves no gaps because _count rewinds; kept as a no-op hook so the counter contract is explicit.""" return # ------------------------------------------------------------------ encoding def _open_pipe(self, out, long_edge, audio): """ffmpeg reading a PNG stream on stdin — the streaming MP4 path. stderr goes to a FILE, not a pipe: nobody drains a stderr pipe until the encode is done, so a chatty ffmpeg would fill the 64 KiB pipe buffer, block, stop reading stdin, and deadlock the render halfway through.""" cmd = ["ffmpeg", "-y", "-loglevel", "error", "-f", "image2pipe", "-framerate", str(self.fps), "-i", "-"] if audio: cmd += ["-i", audio, "-c:a", "aac", "-b:a", "128k", "-shortest"] cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "23", "-vf", f"scale='min({long_edge},iw)':-2:flags=lanczos", "-movflags", "+faststart", out] self._pipe_err = open(os.path.join(self._scratch, "ffmpeg.log"), "w+b") return subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=self._pipe_err) def _close_pipe(self): try: self._pipe.stdin.close() except OSError: # ffmpeg already died; its exit code + log below say why. pass rc = self._pipe.wait() self._pipe_err.seek(0) err = self._pipe_err.read().decode(errors="replace") self._pipe_err.close() self._pipe = None if rc != 0: raise RuntimeError(f"ffmpeg (streaming) exited {rc}:\n{err[-2000:]}") def _encode(self, seconds, long_edge, force, audio=None): gif, mp4 = "/workspace/final.gif", "/workspace/final.mp4" if self._streaming: self._close_pipe() return _art(mp4, "video/mp4") fs = sorted(glob.glob(os.path.join(self._scratch, "frame_*.png"))) want_mp4 = (force == "mp4" or audio is not None or (force != "gif" and seconds > GIF_MAX_SECONDS)) if want_mp4: self._mp4(mp4, long_edge, audio) return _art(mp4, "video/mp4") self._gif(fs, gif, long_edge) if os.path.getsize(gif) <= GIF_MAX_BYTES: return _art(gif, "image/gif") print(f"[gifsmith] GIF {os.path.getsize(gif)/1048576:.1f} MiB is too big " f"for Discord — re-encoding as MP4", flush=True) self._mp4(mp4, long_edge, audio) os.remove(gif) return _art(mp4, "video/mp4") def _gif(self, fs, out, long_edge): if _which("gifski"): _run(["gifski", "-o", out, "--fps", str(self.fps), "--quality", "90", "--width", str(min(long_edge, self.width))] + fs) else: # -framerate applies to the INPUT. Without it ffmpeg reads a PNG # sequence at its default 25fps, and a fps= filter then resamples # 25 -> self.fps: on the #1510 render that silently dropped 91 of # 207 frames (44%) and played the rest too fast. _mp4 below always # passed -framerate; this path never did. With the input rate set # correctly the fps= filter is redundant — and removing it is what # takes the output from 205 frames back to all 207. vf = (f"scale='min({long_edge},iw)':-1:flags=lanczos," f"split[s0][s1];[s0]palettegen=stats_mode=diff[p];" f"[s1][p]paletteuse=dither=bayer:bayer_scale=3") _run(["ffmpeg", "-y", "-loglevel", "error", "-framerate", str(self.fps), "-pattern_type", "glob", "-i", os.path.join(self._scratch, "frame_*.png"), "-vf", vf, out]) if _which("gifsicle"): # --colors 256 is load-bearing for DISCORD, not for size. # # gifski gives every frame its OWN local colour table (that is where # its quality comes from). Stack gifsicle's cross-frame transparency # on top and the result is a GIF that gifsicle itself refuses to # reopen: "too complex to unoptimize — local colour tables or complex # transparency". PIL, ffmpeg and macOS Preview all render it fine; # Discord's decoder does not, and paints later scenes with stale rows, # ghosted captions and holes punched in solid shapes (#1510). # # Collapsing to ONE global palette costs ~6/255 mean colour error # (visible as slight banding in gradients, since one palette now # spans every scene) and is the price of playing in the client that # actually matters. It also came out SMALLER on the reported file: # 1210 KB -> 915 KB. _run(["gifsicle", "--colors", "256", "-O3", "--lossy=60", out, "-o", out]) def _mp4(self, out, long_edge, audio=None): cmd = ["ffmpeg", "-y", "-loglevel", "error", "-framerate", str(self.fps), "-pattern_type", "glob", "-i", os.path.join(self._scratch, "frame_*.png")] if audio: cmd += ["-i", audio, "-c:a", "aac", "-b:a", "128k", "-shortest"] cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "23", "-vf", f"scale='min({long_edge},iw)':-2:flags=lanczos", "-movflags", "+faststart", out] _run(cmd) # ------------------------------------------------------------- critique aid def _contact_sheet(self, path="/workspace/still.png"): tiles = [(i, s) for i, s in enumerate(self._scenes) if s.first_frame is not None] if not tiles: return if len(tiles) == 1: tiles[0][1].first_frame.save(path) return cols = min(3, len(tiles)) rows = math.ceil(len(tiles) / cols) tw = 300 th = max(1, round(self.height * tw / self.width)) sheet = Image.new("RGB", (cols * tw, rows * (th + 20)), (20, 20, 24)) d = ImageDraw.Draw(sheet) f = self.font(14) for k, (i, sc) in enumerate(tiles): x, y = (k % cols) * tw, (k // cols) * (th + 20) sheet.paste(sc.first_frame.resize((tw, th), Image.LANCZOS), (x, y + 20)) d.text((x + 4, y + 3), f"scene {i}: {sc.title[:38]}", fill=(235, 235, 240), font=f) sheet.save(path) def _run(cmd): p = subprocess.run(cmd, capture_output=True, text=True) if p.returncode != 0: raise RuntimeError(f"{cmd[0]} failed:\n{p.stderr[-2000:]}") def _art(path, mime): b = os.path.getsize(path) print(f"[gifsmith] encoded {path} {mime} {b / 1024 / 1024:.2f} MiB") return {"path": path, "mime": mime, "bytes": b}