← back to Paul Conrad Cartoons Rebrand
generator/gen_shadowman.py
186 lines
#!/usr/bin/env python3
"""Shadow Man editorial-cartoon renderer (TK-12179 step 2). Local SDXL, $0.
~/.venvs/sdxl/bin/python generator/gen_shadowman.py generator/out/batch.json [--limit 20]
Adapted from crazy-news-channel/tools/gen_images.py (same local SDXL checkpoint +
diffusers pipeline; that file is not modified). Differences:
* ink editorial-cartoon style, square-ish single panel, no photo style;
* post-process to high-contrast black ink, add a caption band, and stamp the
signature "Shadow Man" lower-right on EVERY final image (PIL);
* one image at a time with a memory guard: before loading the model and before
every image it checks kern.memorystatus_vm_pressure_level and the
memory_pressure free %, and aborts cleanly (exit 3) if pressure is critical,
so the exo vision instance and other LLM lanes on this box are not starved;
* appends each finished image to generator/out/manifest.json immediately
(a crash mid-batch still leaves a truthful manifest).
"""
import argparse, datetime, gc, json, os, re, subprocess, sys, time
HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.environ.get("SHADOWMAN_OUT") or os.path.join(HERE, "out")
MANIFEST = os.path.join(OUT, "manifest.json")
CKPT = os.environ.get("SDXL_CKPT") or \
"/Volumes/Henry/mac2-archive/2026-06-26-reclaim/ComfyUI-checkpoints/sd_xl_base_1.0.safetensors"
MODEL = "sd_xl_base_1.0 (local, diffusers StableDiffusionXLPipeline, MPS fp16)"
SIGNATURE = "Shadow Man"
NEGATIVE = ("text, words, letters, caption, speech bubble text, watermark, logo, signature, "
"artist name, typography, color, colorful, photograph, photorealistic, 3d render, "
"blurry, lowres, deformed hands, extra fingers, multiple panels, comic strip")
BANNED = re.compile(r"conrad|in the style of|pulitzer", re.I)
MIN_FREE_PCT = int(os.environ.get("SHADOWMAN_MIN_FREE_PCT", "12"))
FONT_CAPTION = "/System/Library/Fonts/Supplemental/Georgia Bold Italic.ttf"
FONT_SIG = "/System/Library/Fonts/Supplemental/Bradley Hand Bold.ttf"
def memory_ok():
"""Return (ok, detail). Critical pressure (level 4) or free% below floor -> not ok."""
level, free = None, None
try:
level = int(subprocess.run(["sysctl", "-n", "kern.memorystatus_vm_pressure_level"],
capture_output=True, text=True, timeout=10).stdout.strip())
except Exception:
pass
try:
out = subprocess.run(["memory_pressure", "-Q"], capture_output=True, text=True, timeout=20).stdout
m = re.search(r"free percentage:\s*(\d+)%", out)
free = int(m.group(1)) if m else None
except Exception:
pass
detail = f"pressure_level={level} free={free}%"
if level is None and free is None:
return False, detail + " (NOT MEASURED — refusing to run blind)"
if level is not None and level >= 4:
return False, detail + " (CRITICAL)"
if free is not None and free < MIN_FREE_PCT:
return False, detail + f" (< {MIN_FREE_PCT}% floor)"
return True, detail
def load_manifest():
if os.path.exists(MANIFEST):
return json.load(open(MANIFEST))
return {"signature": SIGNATURE, "note": "AI-generated original editorial cartoons signed Shadow Man. $0 local SDXL.", "items": []}
def save_manifest(m):
tmp = MANIFEST + ".tmp"
json.dump(m, open(tmp, "w"), indent=2)
os.replace(tmp, MANIFEST)
def wrap(draw, text, font, width):
words, lines, cur = text.split(), [], ""
for w in words:
t = (cur + " " + w).strip()
if draw.textlength(t, font=font) <= width:
cur = t
else:
lines.append(cur)
cur = w
if cur:
lines.append(cur)
return lines
def finish(img, caption):
"""Ink post-process + caption band + Shadow Man signature."""
from PIL import Image, ImageDraw, ImageFont, ImageOps, ImageEnhance
g = ImageOps.grayscale(img)
g = ImageEnhance.Contrast(g).enhance(1.6)
g = g.point(lambda p: 255 if p > 200 else (0 if p < 45 else p)) # push toward ink/paper
art = g.convert("RGB")
W, H = art.size
d = ImageDraw.Draw(art)
# signature: lower-right corner of the drawing, ink-style hand lettering on a paper patch
sf = ImageFont.truetype(FONT_SIG, max(30, W // 26))
sw = d.textlength(SIGNATURE, font=sf)
bbox = d.textbbox((0, 0), SIGNATURE, font=sf)
sh = bbox[3] - bbox[1]
x, y = W - sw - W // 30, H - sh - H // 22
d.rectangle([x - 12, y - 6, x + sw + 12, y + sh + 16], fill=(255, 255, 255))
d.text((x, y - bbox[1]), SIGNATURE, font=sf, fill=(10, 10, 10))
d.line([x, y + sh + 8, x + sw, y + sh + 4], fill=(10, 10, 10), width=3) # brush underline
# caption band under the panel
cf = ImageFont.truetype(FONT_CAPTION, max(26, W // 34))
lines = wrap(d, caption, cf, W - 80)
lh = int(cf.size * 1.35)
band = 40 + lh * len(lines)
canvas = Image.new("RGB", (W + 24, H + band + 24), (255, 255, 255))
canvas.paste(art, (12, 12))
cd = ImageDraw.Draw(canvas)
cd.rectangle([10, 10, W + 13, H + 13], outline=(0, 0, 0), width=3) # panel border
for i, ln in enumerate(lines):
tw = cd.textlength(ln, font=cf)
cd.text(((W + 24 - tw) / 2, H + 12 + 22 + i * lh), ln, font=cf, fill=(0, 0, 0))
return canvas
def main():
ap = argparse.ArgumentParser()
ap.add_argument("batch")
ap.add_argument("--limit", type=int, default=20)
ap.add_argument("--steps", type=int, default=30)
a = ap.parse_args()
items = json.load(open(a.batch))["items"][: a.limit]
for it in items:
if BANNED.search(it["prompt"]) or BANNED.search(it["caption"]):
sys.exit(f"refusing: banned term in {it['id']}")
os.makedirs(OUT, exist_ok=True)
man = load_manifest()
done = {m["id"] for m in man["items"]}
todo = [it for it in items if it["id"] not in done and not os.path.exists(os.path.join(OUT, it["id"] + ".png"))]
print(f"{len(items)} briefs, {len(todo)} to generate (limit {a.limit})", flush=True)
if not todo:
return 0
ok, det = memory_ok()
print(f"memory pre-load: {det}", flush=True)
if not ok:
print("ABORT: memory guard tripped before model load", flush=True)
return 3
import torch
from diffusers import StableDiffusionXLPipeline
t_load = time.time()
device = "mps" if torch.backends.mps.is_available() else "cpu"
pipe = StableDiffusionXLPipeline.from_single_file(
CKPT, torch_dtype=torch.float16 if device == "mps" else torch.float32)
pipe.to(device)
pipe.set_progress_bar_config(disable=True)
print(f"model loaded on {device} in {time.time() - t_load:.0f}s", flush=True)
t_batch = time.time()
for i, it in enumerate(todo, 1):
ok, det = memory_ok()
if not ok:
print(f"ABORT before {it['id']}: {det}", flush=True)
return 3
t0 = time.time()
img = pipe(prompt=it["prompt"], negative_prompt=NEGATIVE, width=1152, height=896,
num_inference_steps=a.steps, guidance_scale=7.0,
generator=torch.Generator("cpu").manual_seed(int(it["seed"]))).images[0]
final = finish(img, it["caption"])
final.save(os.path.join(OUT, it["id"] + ".png"), optimize=True)
secs = round(time.time() - t0, 1)
man["items"].append({
"id": it["id"], "created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
"theme": it["theme"], "era": it["era"], "caption": it["caption"], "prompt": it["prompt"],
"negative_prompt": NEGATIVE, "seed": it["seed"], "model": MODEL, "steps": a.steps,
"signature": SIGNATURE, "source": it.get("source"), "gen_seconds": secs, "cost_usd": 0,
"file": it["id"] + ".png",
})
save_manifest(man)
print(f"[{i}/{len(todo)}] {it['id']} {secs}s {det}", flush=True)
del img, final
gc.collect()
if device == "mps":
torch.mps.empty_cache()
print(f"batch done: {len(todo)} images in {time.time() - t_batch:.0f}s (+ load)", flush=True)
return 0
if __name__ == "__main__":
sys.exit(main())