"""
DoodleBook — HF ZeroGPU Version
Free T4 GPU on Hugging Face Spaces!
No Modal needed.
"""
import gradio as gr
import os
import sys
import torch
try:
import spaces
except ModuleNotFoundError:
# `spaces` only exists on HF ZeroGPU. Off-HF (local/dev) provide a no-op so
# the app still runs; generation then uses whatever local GPU/CPU exists.
class _SpacesShim:
@staticmethod
def GPU(*args, **kwargs):
if args and callable(args[0]): # bare @spaces.GPU
return args[0]
def deco(fn): # @spaces.GPU(duration=...)
return fn
return deco
spaces = _SpacesShim()
import json
import time
import tempfile
import logging
import struct
import re
sys.path.insert(0, os.path.dirname(__file__))
from config import (
FLUX_MODEL, STORY_MODEL, TTS_MODEL,
GENERATION_PARAMS, SAMPLE_BOOK_PATH, BASE_SEED, page_seed,
DEFAULT_VOICE, voice_design,
)
from book_builder import (
build_book_html, export_pdf, magic_loader_html,
build_coloring_html, export_coloring_pdf,
)
from ui.layout import create_layout
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ZeroGPU sets SPACES_ZERO_GPU. On the Space we load models on cuda at IMPORT
# (a CUDA-emulation layer makes that work without a real GPU); lazy-loading
# inside @spaces.GPU is explicitly discouraged and was why FLUX kept failing
# → sketch. Guarded so a local/dev import doesn't try to pull ~20GB of weights.
ON_ZEROGPU = bool(os.environ.get("SPACES_ZERO_GPU"))
_FLUX_PIPE = None
_STORY_MODEL = None
_STORY_TOKENIZER = None
_TTS_MODEL = None
_LOAD_ERRORS = {}
def load_flux():
"""FLUX image pipeline placed on cuda at module scope (the ZeroGPU pattern).
No enable_model_cpu_offload() — that fights ZeroGPU's device management."""
global _FLUX_PIPE
if _FLUX_PIPE is None:
from diffusers import Flux2KleinPipeline
logger.info(f"Loading image model: {FLUX_MODEL.hub_id}")
pipe = Flux2KleinPipeline.from_pretrained(
FLUX_MODEL.hub_id, torch_dtype=torch.bfloat16,
)
pipe.to("cuda")
_FLUX_PIPE = pipe
return _FLUX_PIPE
def load_story():
global _STORY_MODEL, _STORY_TOKENIZER
if _STORY_MODEL is None:
from transformers import AutoTokenizer, AutoModelForCausalLM
logger.info(f"Loading story model: {STORY_MODEL.hub_id}")
_STORY_TOKENIZER = AutoTokenizer.from_pretrained(
STORY_MODEL.hub_id, trust_remote_code=True,
)
_STORY_MODEL = AutoModelForCausalLM.from_pretrained(
STORY_MODEL.hub_id, torch_dtype=torch.float16, trust_remote_code=True,
).to("cuda").eval()
return _STORY_MODEL, _STORY_TOKENIZER
def load_tts():
global _TTS_MODEL
if _TTS_MODEL is None:
from voxcpm import VoxCPM
logger.info(f"Loading TTS model: {TTS_MODEL.hub_id}")
_TTS_MODEL = VoxCPM.from_pretrained(TTS_MODEL.hub_id, load_denoiser=False)
return _TTS_MODEL
if ON_ZEROGPU:
for _name, _loader in (("flux", load_flux), ("story", load_story), ("tts", load_tts)):
try:
_loader()
except Exception as _e: # keep the Space booting
_LOAD_ERRORS[_name] = repr(_e)
logger.exception(f"Module-level load failed for {_name}")
COLOR_ART_STYLE = (
"children's crayon storybook illustration, bold black outlines, "
"flat bright colors, simple shapes"
)
COLOR_PAGE_SUFFIX = "full colorful background scene, the character clearly visible."
LINE_ART_STYLE = (
"children's coloring book page, pure black ink outlines on pure white paper, "
"clean contour lines, no color, no gray, no shading, no texture, "
"no hatching, no pencil marks, open spaces to color"
)
LINE_ART_SUFFIX = (
"simple clean background shapes, same composition, thick readable outlines, "
"no filled black areas, no extra sketch marks."
)
THEME_TEMPLATES = {
"brave adventure": [
("{hero} loved exploring new places.", "{hero} standing at the start of a bright adventure trail"),
("One morning, {hero} discovered something glowing nearby.", "{hero} spotting a magical glow in the distance"),
("Taking a deep breath, {hero} bravely went closer.", "{hero} walking forward with courage"),
("There, a new friend needed help.", "{hero} finding a small friend in trouble"),
("{hero} helped with kindness and a clever idea.", "{hero} helping the friend together"),
("Everyone cheered, and {hero} felt proud and brave.", "{hero} celebrating at sunset with the new friend"),
],
"making a new friend": [
("{hero} was playing alone in a sunny place.", "{hero} playing under a bright sky"),
("Then {hero} noticed someone shy nearby.", "{hero} seeing a shy new friend nearby"),
("{hero} smiled and said hello.", "{hero} waving with a friendly smile"),
("Soon they were sharing stories and laughs.", "{hero} and the new friend laughing together"),
("They played games all afternoon.", "{hero} and the new friend playing together"),
("By sunset, {hero} had made a wonderful new friend.", "{hero} and the new friend smiling together at sunset"),
],
}
FEW_SHOT_EXEMPLAR = """
Write a 6-page children's storybook for age 5 about Luna the cat with theme: brave adventure.
Return ONLY valid JSON:
{
"title": "Luna's Brave Adventure",
"character_description": "A small orange tabby cat named Luna with big green eyes, whiskers, and a tiny red scarf",
"pages": [
{"page": 1, "text": "Luna was a small orange cat who loved to explore.", "scene": "Luna sitting by the window looking outside"},
{"page": 2, "text": "One sunny morning, Luna saw something sparkling in the forest.", "scene": "Luna spotting a glow in the trees"},
{"page": 3, "text": "Bravely, Luna crept into the forest to investigate.", "scene": "Luna walking cautiously through trees"},
{"page": 4, "text": "It was a tiny fairy stuck in a spider web!", "scene": "Luna discovering a fairy in trouble"},
{"page": 5, "text": "Luna gently freed the fairy with her paw.", "scene": "Luna carefully helping the fairy"},
{"page": 6, "text": "The fairy thanked Luna and they became friends forever.", "scene": "Luna and fairy playing together at sunset"}
]
}
"""
def build_story_prompt(hero_name: str, theme: str, age: int) -> str:
return f"""{FEW_SHOT_EXEMPLAR}
Write a 6-page children's storybook for age {age} about {hero_name} with theme: {theme}.
Return ONLY valid JSON:
"""
def _validate_story_structure(story: dict) -> bool:
required_keys = ["title", "character_description", "pages"]
if not all(k in story for k in required_keys):
return False
pages = story.get("pages", [])
if not isinstance(pages, list) or len(pages) < 1:
return False
first_page = pages[0]
return all(k in first_page for k in ["page", "text", "scene"])
def _repair_json(json_str: str) -> str:
json_str = re.sub(r',\s*([}\]])', r'\1', json_str)
json_str = re.sub(r'//.*?$', '', json_str, flags=re.MULTILINE)
json_str = re.sub(r'/\*[\s\S]*?\*/', '', json_str)
json_str = re.sub(r'(?<=")\n(?=")', '\\n', json_str)
json_str = re.sub(r'(\s)(\w+)(\s*:)', r'\1"\2"\3', json_str)
return json_str
def parse_story_json(raw_output: str) -> dict | None:
match = re.search(r'\{[\s\S]*\}', raw_output or "")
if not match:
return None
raw_json = match.group(0)
for candidate in (raw_json, _repair_json(raw_json)):
try:
story = json.loads(candidate)
if _validate_story_structure(story):
return story
except Exception:
continue
return None
def _normalize_story(story: dict) -> dict:
pages = list(story.get("pages", []))[:6]
while len(pages) < 6:
pages.append({
"page": len(pages) + 1,
"text": "And the adventure continued happily.",
"scene": "Continuing adventure",
})
story["pages"] = pages
story.setdefault("title", "A Wonderful Adventure")
story.setdefault(
"character_description",
"A friendly children's storybook hero with bright colors and cheerful features",
)
return story
def build_story_locally(hero_name: str, theme: str) -> dict:
"""Fast, deterministic fallback story that avoids any Modal dependency."""
hero = (hero_name or "Little Hero").strip() or "Little Hero"
beats = THEME_TEMPLATES.get(theme, THEME_TEMPLATES["brave adventure"])
pages = [
{"page": i + 1, "text": text.format(hero=hero), "scene": scene.format(hero=hero)}
for i, (text, scene) in enumerate(beats)
]
return {
"title": f"{hero}'s Storybook Adventure",
"character_description": (
f"{hero}, a friendly children's storybook hero with bright colors, "
"bold outlines, and a cheerful expressive face"
),
"pages": pages,
}
def silent_wav_bytes(duration_seconds: int = 2, sample_rate: int = 24000) -> bytes:
"""Return a short silent WAV so the UI remains stable if TTS is unavailable."""
num_samples = sample_rate * duration_seconds
data_size = num_samples * 2
header = struct.pack(
"<4sI4s4sIHHIIHH4sI",
b"RIFF", 36 + data_size, b"WAVE",
b"fmt ", 16, 1, 1, sample_rate, sample_rate * 2, 2, 16,
b"data", data_size,
)
return header + (b"\x00" * data_size)
def _with_heartbeat(blocking_fn, frame_fn, poll=4.0):
import threading
box = {}
def _run():
try:
box["val"] = blocking_fn()
except BaseException as e:
box["err"] = e
th = threading.Thread(target=_run, daemon=True)
th.start()
t0 = time.time()
while th.is_alive():
th.join(timeout=poll)
if th.is_alive():
yield ("hb", frame_fn(int(time.time() - t0)))
if "err" in box:
raise box["err"]
yield ("done", box["val"])
# ============================================================================
# SAMPLE BOOK (loads instantly, no GPU needed)
# ============================================================================
SAMPLE_BOOK_HTML = None
def load_sample_book() -> str:
"""Load pre-generated sample book (C3: always ship sample)."""
global SAMPLE_BOOK_HTML
if SAMPLE_BOOK_HTML:
return SAMPLE_BOOK_HTML
sample_path = os.path.join(SAMPLE_BOOK_PATH, "sample.html")
if os.path.exists(sample_path):
with open(sample_path, "r", encoding="utf-8") as f:
SAMPLE_BOOK_HTML = f.read()
return SAMPLE_BOOK_HTML
return "
Loading sample book...
"
# ============================================================================
# ZEROGPU INFERENCE FUNCTIONS
# ============================================================================
@spaces.GPU(duration=60)
def generate_story_gpu(hero_name: str, theme: str, age: int = 5) -> dict:
"""Generate a story on ZeroGPU, falling back to a deterministic local story."""
try:
model, tok = load_story()
prompt = build_story_prompt(hero_name, theme, age)
inputs = tok.apply_chat_template(
[{"role": "user", "content": prompt}],
add_generation_prompt=True,
enable_thinking=False,
return_dict=True,
return_tensors="pt",
).to("cuda")
with torch.no_grad():
out = model.generate(
**inputs,
max_new_tokens=GENERATION_PARAMS.max_story_tokens,
do_sample=False,
)
response = tok.decode(
out[0][inputs["input_ids"].shape[1]:],
skip_special_tokens=True,
)
parsed = parse_story_json(response)
if parsed:
return _normalize_story(parsed)
logger.warning("Story parser failed; using deterministic local fallback")
except Exception as e:
logger.warning(f"ZeroGPU story generation failed: {e}")
return _normalize_story(build_story_locally(hero_name, theme))
@spaces.GPU(duration=150)
def generate_images_gpu(
character_desc: str,
scenes: list,
doodle_bytes: bytes = None,
seed: int = 42,
) -> list:
"""Generate all story pages with FLUX on ZeroGPU (two-stage: canonical
character from the doodle, then the same character in each scene)."""
import io
from PIL import Image
pipe = load_flux()
num_steps, guidance = 6, 1.0
canonical = None
if doodle_bytes:
try:
ref = Image.open(io.BytesIO(doodle_bytes)).convert("RGB")
canonical = pipe(
prompt=(f"Turn this child's drawing into a clean, friendly, full-body cartoon "
f"character for a children's storybook. Keep the EXACT same creature, "
f"face, and features as the drawing. {COLOR_ART_STYLE}, "
f"plain white background, full character visible, centered."),
image=ref, height=768, width=768, guidance_scale=guidance,
num_inference_steps=num_steps,
generator=torch.Generator("cuda").manual_seed(seed),
).images[0]
logger.info("Canonical character built from doodle")
except Exception as e:
logger.warning(f"Canonical build failed ({e}); text2img fallback")
canonical = None
images = []
for i, scene in enumerate(scenes):
if canonical is not None:
prompt = f"The same character. {scene}. {COLOR_ART_STYLE}, {COLOR_PAGE_SUFFIX}"
kw = dict(image=canonical, prompt=prompt)
else:
prompt = (f"{character_desc}. Scene: {scene}. {COLOR_ART_STYLE}, "
f"white background, centered, full character visible")
kw = dict(prompt=prompt)
kw.update(height=768, width=768, guidance_scale=guidance,
num_inference_steps=num_steps,
generator=torch.Generator("cuda").manual_seed(seed + i + 1))
images.append(pipe(**kw).images[0])
logger.info(f"Generated page {i+1}/{len(scenes)}")
return images
@spaces.GPU(duration=150)
def generate_coloring_images_gpu(
character_desc: str,
scenes: list,
doodle_bytes: bytes = None,
seed: int = 42,
) -> list:
"""Generate coloring pages directly with FLUX as line art (no tracing)."""
import io
from PIL import Image
pipe = load_flux()
num_steps, guidance = 6, 1.0
canonical = None
if doodle_bytes:
try:
ref = Image.open(io.BytesIO(doodle_bytes)).convert("RGB")
canonical = pipe(
prompt=(f"Turn this child's drawing into a clean, friendly, full-body cartoon "
f"character for a children's coloring book. Keep the EXACT same creature, "
f"face, and features as the drawing. {LINE_ART_STYLE}, "
f"plain white background, full character visible, centered."),
image=ref, height=768, width=768, guidance_scale=guidance,
num_inference_steps=num_steps,
generator=torch.Generator("cuda").manual_seed(seed),
).images[0]
logger.info("Line-art canonical character built from doodle")
except Exception as e:
logger.warning(f"Line-art canonical build failed ({e}); text2img fallback")
canonical = None
images = []
for i, scene in enumerate(scenes):
if canonical is not None:
prompt = f"The same character. {scene}. {LINE_ART_STYLE}, {LINE_ART_SUFFIX}"
kw = dict(image=canonical, prompt=prompt)
else:
prompt = (f"{character_desc}. Scene: {scene}. {LINE_ART_STYLE}, "
f"white background, centered, full character visible")
kw = dict(prompt=prompt)
kw.update(height=768, width=768, guidance_scale=guidance,
num_inference_steps=num_steps,
generator=torch.Generator("cuda").manual_seed(seed + i + 101))
images.append(pipe(**kw).images[0])
logger.info(f"Generated coloring page {i+1}/{len(scenes)}")
return images
@spaces.GPU(duration=120)
def generate_tts_gpu(text: str, voice: str = DEFAULT_VOICE) -> bytes:
"""Narrate the book with VoxCPM2. Raises on failure so the caller can show
the real reason instead of silently shipping a silent clip."""
import io
import numpy as np
try:
model = load_tts()
design = voice_design(voice)
import re
chunks = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
if not chunks:
chunks = [text.strip() or "The end."]
sr = model.tts_model.sample_rate
pause = np.zeros(int(sr * 0.35), dtype=np.float32)
pieces = []
for i, sentence in enumerate(chunks):
wav = model.generate(
text=f"{design} {sentence}",
cfg_value=2.0,
inference_timesteps=10,
)
pieces.append(np.asarray(wav, dtype=np.float32))
if i < len(chunks) - 1:
pieces.append(pause)
audio = np.concatenate(pieces)
import soundfile as sf
buf = io.BytesIO()
sf.write(buf, audio, sr, format="WAV")
return buf.getvalue()
except Exception as e:
# Surface the real reason (e.g. missing model) instead of a silent clip
# that looks like it worked. create_book records this in the trace.
logger.exception("TTS failed")
raise
# ============================================================================
# MAIN BOOK CREATION (Generator for streaming)
# ============================================================================
def create_book(doodle_image, character_name, theme, hero_name, voice=DEFAULT_VOICE, make_coloring=False):
"""ZeroGPU book flow: story → images → narration → PDFs → coloring book,
each a sequential @spaces.GPU call (ZeroGPU has one GPU per request)."""
t_total = time.perf_counter()
character_name = (character_name or "").strip() or "Little Hero"
hero_name = (hero_name or "").strip() or character_name
trace_data = {
"backend": "zerogpu",
"hero_name": hero_name,
"theme": theme,
"voice": voice,
"make_coloring": make_coloring,
"seed": BASE_SEED,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
}
if _LOAD_ERRORS:
trace_data["model_load_errors"] = _LOAD_ERRORS
_no = gr.update(visible=False)
_keep = gr.update()
yield (
magic_loader_html("story", hero_name),
"Writing the story…",
None, _keep, {}, "", json.dumps(trace_data, indent=2),
_no, _keep,
)
t_story = time.perf_counter()
try:
story = generate_story_gpu(hero_name, theme)
except Exception as e:
logger.error(f"Story generation failed: {e}")
yield (
f"Error: {e}
",
f"Error: {e}",
None, _keep, {}, "", "",
_no, _keep,
)
return
trace_data["story_sec"] = round(time.perf_counter() - t_story, 2)
pages = story.get("pages", [])
char_desc = story.get("character_description", "")
title = story.get("title", "Untitled Story")
page_texts = [p.get("text", "") for p in pages]
scenes = [p.get("scene", "") for p in pages]
trace_data["title"] = title
trace_data["character_description"] = char_desc
yield (
magic_loader_html("images", hero_name),
f"{title} — illustrating on ZeroGPU…",
None, _keep, story, "", json.dumps(trace_data, indent=2),
_no, _keep,
)
doodle_bytes = None
if doodle_image is not None:
import io
from PIL import Image
img = Image.fromarray(doodle_image)
buf = io.BytesIO()
img.save(buf, format="PNG")
doodle_bytes = buf.getvalue()
full_text = f"{title}. {' '.join(page_texts)}"
# ---- IMAGES (FLUX on ZeroGPU) ----
img_bytes, engine = None, "sketch"
t_images = time.perf_counter()
try:
for kind, payload in _with_heartbeat(
lambda: generate_images_gpu(char_desc, scenes, doodle_bytes, BASE_SEED),
lambda s: (
magic_loader_html("images", hero_name),
f"{title} — illustrating on ZeroGPU… {s}s",
None, _keep, story, "", json.dumps(trace_data, indent=2), _no, _keep,
),
):
if kind == "hb":
yield payload
else:
images = payload
import io
img_bytes = []
for img in images:
buf = io.BytesIO()
img.save(buf, format="PNG")
img_bytes.append(buf.getvalue())
engine = "flux"
except Exception as e:
logger.exception("Image generation failed")
trace_data["image_error"] = repr(e)
from services.images import generate_placeholder_images
img_bytes = generate_placeholder_images(char_desc, scenes, doodle_bytes)
engine = "sketch"
trace_data["images_sec"] = round(time.perf_counter() - t_images, 2)
trace_data["engine"] = engine
book_html = build_book_html(img_bytes, page_texts, title, engine)
# ---- NARRATION (VoxCPM2 on ZeroGPU) — sequential: one GPU per request ----
audio_path = None
t_tts = time.perf_counter()
try:
for kind, payload in _with_heartbeat(
lambda: generate_tts_gpu(full_text, voice),
lambda s: (
book_html,
f"{title} — recording the narration… {s}s",
None, _keep, story, "", json.dumps(trace_data, indent=2), _no, _keep,
),
):
if kind == "hb":
yield payload
else:
voice_bytes = payload
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
tmp.write(voice_bytes)
audio_path = tmp.name
except Exception as e:
logger.exception("TTS failed")
trace_data["tts_error"] = repr(e)
trace_data["tts_sec"] = round(time.perf_counter() - t_tts, 2)
pdf_path = None
t_pdf = time.perf_counter()
try:
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
pdf_path = export_pdf(img_bytes, page_texts, title, tmp.name)
except Exception as e:
logger.warning(f"PDF failed: {e}")
trace_data["pdf_sec"] = round(time.perf_counter() - t_pdf, 2)
coloring_html = ""
coloring_pdf_path = None
if make_coloring:
t_coloring = time.perf_counter()
try:
from services.coloring import _crispen
for kind, payload in _with_heartbeat(
lambda: generate_coloring_images_gpu(char_desc, scenes, doodle_bytes, BASE_SEED),
lambda s: (
book_html,
f"{title} — building coloring book… {s}s",
audio_path,
_keep,
story,
"",
json.dumps(trace_data, indent=2),
_no,
_keep,
),
):
if kind == "hb":
yield payload
else:
coloring_images = payload
import io
outlines = []
for img in coloring_images:
buf = io.BytesIO()
img.save(buf, format="PNG")
outlines.append(_crispen(buf.getvalue()))
coloring_html = build_coloring_html(outlines, page_texts, title)
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
coloring_pdf_path = export_coloring_pdf(outlines, page_texts, title, tmp.name)
trace_data["coloring_book"] = True
trace_data["coloring_engine"] = "flux-direct-lineart"
except Exception as e:
logger.warning(f"Direct FLUX coloring book failed ({e}); using traced fallback")
try:
from services.coloring import derive_coloring_pages
outlines = derive_coloring_pages(img_bytes)
coloring_html = build_coloring_html(outlines, page_texts, title)
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
coloring_pdf_path = export_coloring_pdf(outlines, page_texts, title, tmp.name)
trace_data["coloring_book"] = True
trace_data["coloring_engine"] = "trace-fallback"
except Exception as e2:
logger.warning(f"Coloring book fallback failed: {e2}")
trace_data["coloring_sec"] = round(time.perf_counter() - t_coloring, 2)
trace_data["completed"] = True
trace_data["pages_generated"] = len(img_bytes)
trace_data["total_sec"] = round(time.perf_counter() - t_total, 2)
pdf_update = gr.update(value=pdf_path) if pdf_path else _keep
coloring_pdf_update = gr.update(value=coloring_pdf_path) if coloring_pdf_path else _keep
coloring_display_update = (gr.update(visible=True, value=coloring_html) if coloring_html
else _no)
yield (
book_html,
f"Complete: {title} — {len(img_bytes)} pages · {'FLUX (ZeroGPU)' if engine == 'flux' else 'local sketch fallback'} · voice: {voice} · total {trace_data['total_sec']}s",
audio_path,
pdf_update,
story,
f"Pages: {len(img_bytes)} | Seed: {BASE_SEED} | Engine: {engine} | Story {trace_data.get('story_sec', 0)}s | Images {trace_data.get('images_sec', 0)}s | PDF {trace_data.get('pdf_sec', 0)}s | Coloring {trace_data.get('coloring_sec', 0)}s",
json.dumps(trace_data, indent=2),
coloring_display_update,
coloring_pdf_update,
)
# ============================================================================
# MAIN
# ============================================================================
if __name__ == "__main__":
demo = create_layout(
load_sample_fn=load_sample_book,
create_book_fn=create_book,
)
demo.queue(default_concurrency_limit=2, max_size=8)
demo.launch(share=False, allowed_paths=[tempfile.gettempdir()])