steve
7906a1ab4c
Two changes, both driven by measurement against 128 production gifsmith runs and a local replica of the agent loop (docs/audits/2026-07-24-gifsmith-quality.md in mort). Staging rules. The renders that "worked" still read badly, always the same four ways: the subject drawn 5-10% of the frame height under an empty sky, long runs of pixel-identical frames, captions narrating action that was never drawn, and flat untextured shapes. The recipe never asked for anything else. SKILL.md now states four checkable rules — fill the frame, every frame moves, show it don't caption it, give it depth — plus easing/anticipation/squash-and-stretch, and the self-critique makes a "no" on depiction, subject size or caption legibility a mandatory re-render instead of a suggestion. Render harness. 26% of production code_exec calls end in a traceback, and every crash class lives in plumbing the agent re-derives each run: scratch dir, frame loop, numbering, canvas size, contact sheet, encode. scripts/gifsmith.py owns all of it and the agent writes only @anim.scene drawing functions. A scene that raises is now skipped while the others still render and ship, the canvas is size-locked so ffmpeg cannot fail on jittering frames, and the drawing vocabulary is re-exported so a missing import cannot NameError. The harness also unblocks the multi-minute pieces this skill advertises but could never produce: /tmp is a 16 MiB tmpfs (~13s of frames) and /workspace 64 MiB (~51s), which is why "No space left on device" is 15% of all production crashes. Anything past the GIF length threshold is an MP4, and an MP4 encodes in one streaming pass, so those frames now go straight into ffmpeg's stdin and never touch disk — verified at 1350 frames / 90 s with 0 MiB on disk. encode.py is removed; the harness supersedes it.
395 lines
17 KiB
Python
395 lines
17 KiB
Python
"""
|
|
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
|
|
|
|
_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)
|
|
# ~90 KiB per 640x480 PNG at compress_level=1; budget 60% of free space.
|
|
est_mb = est_frames * (self.width * self.height * 9e-5)
|
|
|
|
# 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.")
|
|
|
|
art = self._encode(total_s, long_edge, force, audio)
|
|
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."""
|
|
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]
|
|
return subprocess.Popen(cmd, stdin=subprocess.PIPE,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
|
|
|
|
def _close_pipe(self):
|
|
self._pipe.stdin.close()
|
|
err = self._pipe.stderr.read().decode(errors="replace")
|
|
if self._pipe.wait() != 0:
|
|
raise RuntimeError(f"ffmpeg (streaming) failed:\n{err[-2000:]}")
|
|
self._pipe = None
|
|
|
|
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:
|
|
vf = (f"fps={self.fps},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", "-pattern_type", "glob",
|
|
"-i", os.path.join(self._scratch, "frame_*.png"), "-vf", vf, out])
|
|
if _which("gifsicle"):
|
|
_run(["gifsicle", "-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}
|