Spaces:

Pixelkram
/

biegen

Paused

biegen / engine.py
Deploy app
1624300 verified 6 days ago
26.2 kB
"""Model loading, Settings and the custom denoising loop.
Every rule the workshop talks about is a line in this file. The pipeline object
is only a container for the components and a few static helpers. We never call
pipe(...) during the workshop.
"""

from __future__ import annotations

import math
import os
import threading
import time
from collections import OrderedDict
from contextlib import contextmanager
from dataclasses import dataclass, field

import numpy as np
import torch
from PIL import Image

from family import ACTIVE as FAMILY

# Defaults come from the model family (family.py), environment variables override them.
MODEL_ID = os.environ.get("MODEL_ID", FAMILY.model_id)
DEFAULT_SIZE = int(os.environ.get("DEFAULT_SIZE", FAMILY.default_size))
DEFAULT_STEPS = int(os.environ.get("DEFAULT_STEPS", FAMILY.default_steps))
GRID_N = int(os.environ.get("GRID_N", FAMILY.grid_n))     # FLUX: 9 images take 30 s on the L40S, 4 take 13 s
MAX_BATCH = int(os.environ.get("MAX_BATCH", FAMILY.max_batch))
MAX_SEQ = int(os.environ.get("MAX_SEQ", FAMILY.max_seq))
GRID_SIZES = tuple(k * k for k in range(2, 9))             # 2 x 2 to 8 x 8, one choice for the whole app

STANDARD_LAYERS = FAMILY.standard_layers
EXPERIMENTS = ("seed_grid", "guidance_sweep", "interpolation")
# Pixels per VAE decode call. Decoding in float32 is memory hungry at 1024 px.
DECODE_PIXELS = int(os.environ.get("DECODE_PIXELS", str(4 * 512 * 512)))
EMB_CACHE_SIZE = 32

BEND_OPS = ("none", "amplify", "invert", "ablate", "rotate", "flip", "shift", "noise", "threshold")
SPATIAL_OPS = ("rotate", "flip", "shift")   # image stream only
# Offsets from the seed, so each source of randomness has its own stream.
SEED_SHUFFLE, SEED_EMB_NOISE, SEED_ABLATE, SEED_BEND_NOISE, SEED_WEIGHTS = 3, 5, 7, 8, 9


@dataclass(frozen=True)
class Settings:
    prompt_a: str
    prompt_b: str
    experiment: str                 # "seed_grid" | "guidance_sweep" | "interpolation"
    seed: int
    steps: int
    t: float = 0.0                  # blend A -> B, used by seed_grid and guidance_sweep
    guidance: float = 4.0           # used by seed_grid and interpolation
    g_from: float = 1.5             # guidance_sweep
    g_to: float = 7.0
    t_from: float = 0.0             # interpolation
    t_to: float = 1.0
    width: int = DEFAULT_SIZE
    height: int = DEFAULT_SIZE
    emb_scale: float = 1.0          # R4
    emb_noise: float = 0.0          # R4
    emb_shuffle: bool = False       # R4
    te_layers: tuple = STANDARD_LAYERS  # R5
    sigma_mult: float = 1.0         # R6
    sigma_offset: float = 0.0       # R6
    stop_at: float = 1.0            # R7, fraction of steps actually run
    show_x0: bool = False           # R7, decode predicted clean image instead of x_t
    bend_block: str | None = None   # R8, "double.0".."double.4" | "single.0".."single.19"
    bend_stream: str = "image"      # "image" | "text"
    bend_op: str = "none"
    bend_strength: float = 0.0
    bend_from: float = 0.0          # active window as fraction of the schedule
    bend_to: float = 1.0
    weight_block: str | None = None # R9
    weight_noise: float = 0.0       # R9, relative to each weight tensor's std
    grid: int | None = None         # images in the grid, None = GRID_N
    tools: tuple = ()               # Tab 4: tools_sd15.ToolUse entries, SD 1.5 only


@dataclass
class Result:
    items: list                     # [(PIL.Image, caption)] for gr.Gallery
    grid: Image.Image               # composite for the board
    warnings: list = field(default_factory=list)
    seconds: float = 0.0
    timings: dict = field(default_factory=dict)


class Clock:
    """Progress through the schedule, read by the bending hooks."""

    progress = 0.0


def fmt_num(x: float, decimals: int) -> str:
    """German number formatting with a decimal comma."""
    return f"{x:.{decimals}f}".replace(".", ",")


def clamp_size(x: int) -> int:
    return max(16, int(round(x / 16)) * 16)


def sample_specs(s: Settings, n: int | None = None) -> list[tuple[int, float, float, str]]:
    """One (seed, t, guidance, caption) tuple per image of the grid."""
    n = n or s.grid or GRID_N
    if s.experiment == "seed_grid":
        return [(s.seed + i, s.t, s.guidance, f"Seed {s.seed + i}") for i in range(n)]
    if s.experiment == "guidance_sweep":
        gs = np.linspace(s.g_from, s.g_to, n)
        return [(s.seed, s.t, float(g), f"Guidance {fmt_num(g, 1)}") for g in gs]
    if s.experiment == "interpolation":
        ts = np.linspace(s.t_from, s.t_to, n)
        return [(s.seed, float(t), s.guidance, f"t = {fmt_num(t, 3)}") for t in ts]
    raise ValueError(f"unknown experiment {s.experiment!r}")


def make_grid(images: list[Image.Image], cols: int | None = None, cell: int = 256,
              gap: int = 4, bg=(255, 255, 255)) -> Image.Image:
    """Composite of all images, the long side of each cell is `cell` px (smaller for wide grids)."""
    n = len(images)
    cols = cols or math.ceil(math.sqrt(n))
    rows = math.ceil(n / cols)
    cell = min(cell, 1040 // cols)                 # a 6 x 6 grid stays about as wide as a 4 x 4 one
    w0, h0 = images[0].size
    scale = cell / max(w0, h0)
    cw, ch = max(1, round(w0 * scale)), max(1, round(h0 * scale))
    grid = Image.new("RGB", (cols * cw + (cols - 1) * gap, rows * ch + (rows - 1) * gap), bg)
    for i, im in enumerate(images):
        r, c = divmod(i, cols)
        grid.paste(im.convert("RGB").resize((cw, ch), Image.LANCZOS), (c * (cw + gap), r * (ch + gap)))
    return grid


def block_ids(transformer) -> list[str]:
    """"double.0".."double.4" (text and image still separate), then "single.0".."single.19"."""
    return ([f"double.{k}" for k in range(len(transformer.transformer_blocks))]
            + [f"single.{k}" for k in range(len(transformer.single_transformer_blocks))])


def get_block(transformer, block_id: str):
    kind, k = block_id.split(".")
    blocks = {"double": transformer.transformer_blocks, "single": transformer.single_transformer_blocks}[kind]
    return blocks[int(k)]


def bend_tensor(h: torch.Tensor, op: str, s: float, lat_hw: tuple[int, int] | None = None,
                mask: torch.Tensor | None = None, gen: torch.Generator | None = None) -> torch.Tensor:
    """R8: one bending operation on tokens h (B, L, D). Spatial ops need lat_hw = (lat_h, lat_w)."""
    if op == "amplify":
        return h * (1 + 4 * s)
    if op == "invert":
        return h * (1 - 2 * s)
    if op == "ablate":
        return h * mask
    if op == "noise":
        z = torch.randn(h.shape, generator=gen, device=h.device, dtype=torch.float32)
        return h + (2 * s * h.float().std() * z).to(h.dtype)
    if op == "threshold":
        return torch.where(h.abs() > 2 * s * h.float().std(), h, torch.zeros_like(h))
    B, L, D = h.shape
    lat_h, lat_w = lat_hw
    x = h.reshape(B, lat_h, lat_w, D).permute(0, 3, 1, 2)          # spatial view (B, D, h, w)
    if op == "rotate":
        # 90 degrees keeps the shape only for square latents, otherwise turn by 180.
        x = torch.lerp(x, torch.rot90(x, 1 if lat_h == lat_w else 2, dims=(-2, -1)), s)
    elif op == "flip":
        x = torch.lerp(x, torch.flip(x, dims=(-1,)), s)
    elif op == "shift":
        x = torch.roll(x, shifts=round(s * lat_w / 2), dims=-1)
    else:
        raise ValueError(f"unknown bending op {op!r}")
    return x.permute(0, 2, 3, 1).reshape(B, L, D)


class Engine:
    """FLUX.2 [klein] engine. engine_sd15.SD15Engine reuses render, conditioning and the interventions."""

    latent_factor = 16              # pixels per packed latent token: VAE scale 8 x patch 2
    standard_layers = (9, 18, 27)   # Qwen3 hidden states the pipeline reads

    def __init__(self, pipe, device: str | torch.device = "cuda"):
        from diffusers import FlowMatchEulerDiscreteScheduler
        from diffusers.pipelines.flux2.pipeline_flux2_klein import (
            Flux2KleinPipeline, compute_empirical_mu, retrieve_timesteps)

        self.pipe = pipe
        self.device = torch.device(device)
        self.P = Flux2KleinPipeline
        self._Scheduler = FlowMatchEulerDiscreteScheduler
        self._mu = compute_empirical_mu
        self._retrieve_timesteps = retrieve_timesteps
        self.clock = Clock()
        self._emb_cache: OrderedDict = OrderedDict()
        self._lock = threading.Lock()
        self.post_step: list = []       # functions latent -> latent after every sampler step (Tab 4)
        self.encode_tag = ""            # part of the encoding cache key while a tool changes the text encoder
        self.run_notes: list = []       # notes for the status line from the current render call
        for m in (pipe.transformer, pipe.text_encoder, pipe.vae):
            m.requires_grad_(False)
            m.eval()

    @classmethod
    def from_pretrained(cls, model_id: str = MODEL_ID, device: str = "cuda") -> "Engine":
        from diffusers import Flux2KleinPipeline

        pipe = Flux2KleinPipeline.from_pretrained(model_id, dtype=torch.bfloat16).to(device)
        # Extreme latents overflow less in float32. The VAE config sets force_upcast anyway.
        pipe.vae.to(torch.float32)
        # Tiles only above the VAE's sample_size (1024 px), so smaller images are unchanged.
        # Without it a float32 decode at 1536 px runs out of memory on a 24 GB card.
        pipe.vae.enable_tiling()
        return cls(pipe, device)

    def block(self, block_id: str):
        return get_block(self.pipe.transformer, block_id)

    def text_bend_note(self, s: Settings) -> str | None:
        """A status line when a text-stream bend cannot act, else None."""
        if s.bend_stream == "text" and s.bend_block == block_ids(self.pipe.transformer)[-1]:
            # The transformer drops the text tokens after its last block.
            return ("Nach dem letzten Block liest das Netz den Text-Strom nicht mehr. "
                    "Dieser Eingriff ändert deshalb nichts. Wählt einen früheren Block.")
        return None

    def warmup(self, size: int = DEFAULT_SIZE, n: int = 2, steps: int = 2) -> float:
        s = Settings("warm-up", "warm-up", "seed_grid", seed=0, steps=steps, width=size, height=size,
                     te_layers=self.standard_layers)
        return self.render(s, sample_specs(s, n)).seconds

    # ---------------------------------------------------------------- text

    def encode(self, prompt: str, layers: tuple = STANDARD_LAYERS, max_len: int = MAX_SEQ) -> torch.Tensor:
        """Stacked Qwen3 hidden states (1, max_len, 3 * 2560), cached."""
        key = (prompt, tuple(layers), max_len)
        with self._lock:
            if key in self._emb_cache:
                self._emb_cache.move_to_end(key)
                return self._emb_cache[key]
        with torch.inference_mode():
            emb = self.P._get_qwen3_prompt_embeds(
                self.pipe.text_encoder, self.pipe.tokenizer, prompt, dtype=torch.bfloat16,
                device=self.device, max_sequence_length=max_len, hidden_states_layers=tuple(layers))
        with self._lock:
            self._emb_cache[key] = emb
            while len(self._emb_cache) > EMB_CACHE_SIZE:
                self._emb_cache.popitem(last=False)
        return emb

    def clear_cache(self) -> None:
        with self._lock:
            self._emb_cache.clear()

    def conditioning(self, s: Settings, ts: list[float], max_len: int) -> tuple[torch.Tensor, torch.Tensor]:
        """cond (n, L, D) and uncond (1, L, D), both bfloat16."""
        n_layers = self.pipe.text_encoder.config.num_hidden_layers   # hidden_states has n_layers + 1 entries
        if any(not 0 <= k <= n_layers for k in s.te_layers):
            raise ValueError(f"text encoder layers must lie in 0..{n_layers}, got {s.te_layers}")
        emb_a, emb_b = self.prompt_embeddings(s, max_len)
        uncond = self.encode("", s.te_layers, max_len)
        # R2: a linear blend in the stacked hidden-state space, padding positions included.
        t = torch.tensor(ts, device=self.device, dtype=torch.float32).view(-1, 1, 1)
        cond = emb_a + t * (emb_b - emb_a)
        # R4 acts on cond only, in this order: shuffle, scale, noise. uncond stays untouched.
        if s.emb_shuffle:
            cond = self.shuffle_cond(s, cond, max_len)
        if s.emb_scale != 1.0:
            cond = cond * s.emb_scale
        if s.emb_noise != 0:
            # The same noise for every sample, so a seed grid stays comparable.
            z = torch.randn(cond.shape[1:], generator=torch.Generator("cpu").manual_seed(s.seed + SEED_EMB_NOISE))
            cond = cond + s.emb_noise * z.to(self.device) * cond.std(dim=-1, keepdim=True)
        return cond.to(torch.bfloat16), uncond

    def prompt_embeddings(self, s: Settings, max_len: int) -> tuple[torch.Tensor, torch.Tensor]:
        return (self.encode(s.prompt_a, s.te_layers, max_len).float(),
                self.encode(s.prompt_b, s.te_layers, max_len).float())

    def shuffle_cond(self, s: Settings, cond: torch.Tensor, max_len: int) -> torch.Tensor:
        """R4: permute the hidden states at the prompt's token positions (template and padding stay)."""
        pos = sorted(set(self.content_positions(s.prompt_a, max_len))
                     | set(self.content_positions(s.prompt_b, max_len)))
        if len(pos) > 1:
            perm = torch.randperm(len(pos), generator=torch.Generator("cpu").manual_seed(s.seed + SEED_SHUFFLE))
            idx = torch.tensor(pos, device=self.device)
            cond[:, idx] = cond[:, idx[perm.to(self.device)]]
        return cond

    def content_positions(self, prompt: str, max_len: int = MAX_SEQ) -> list[int]:
        """Token positions of the user's prompt inside the chat template (no template tokens, no padding)."""
        tok = self.pipe.tokenizer

        def template(content: str) -> str:
            return tok.apply_chat_template([{"role": "user", "content": content}], tokenize=False,
                                           add_generation_prompt=True, enable_thinking=False)

        text = template(prompt)
        start = len(template("\x00").split("\x00")[0])
        end = start + len(prompt)
        if text[start:end] != prompt:
            raise ValueError("chat template does not contain the prompt verbatim")
        enc = tok(text, return_offsets_mapping=True, padding="max_length", truncation=True, max_length=max_len)
        return [i for i, (a, b) in enumerate(enc["offset_mapping"]) if b > a and a >= start and b <= end]

    # ---------------------------------------------------------------- loop

    @contextmanager
    def interventions(self, s: Settings, lat_h: int, lat_w: int, txt_len: int):
        """Hooks (R8) and weight noise (R9) for exactly one render call, removed on any exit.
        Safe only because the Space runs one GPU job at a time.
        """
        with self.weight_noise(s), self.bending(s, lat_h, lat_w, txt_len):
            yield

    @contextmanager
    def bending(self, s: Settings, lat_h: int, lat_w: int, txt_len: int):
        """R8: a forward hook on one block changes its image or text tokens inside a phase window."""
        if not s.bend_block or s.bend_op == "none" or s.bend_strength == 0:
            yield
            return
        op, st, image = s.bend_op, float(s.bend_strength), s.bend_stream == "image"
        if op not in BEND_OPS:
            raise ValueError(f"unknown bending op {op!r}")
        if not image and op in SPATIAL_OPS:
            raise ValueError(f"{op} works on the image stream only")
        block = self.block(s.bend_block)
        double = s.bend_block.startswith("double")
        mask = gen = None
        if op == "ablate":
            dim = self.pipe.transformer.config.num_attention_heads * self.pipe.transformer.config.attention_head_dim
            off = torch.randperm(dim, generator=torch.Generator("cpu").manual_seed(s.seed + SEED_ABLATE))
            mask = torch.ones(dim)
            mask[off[:round(min(max(st, 0.0), 1.0) * dim)]] = 0.0   # typed values may leave 0..1
            mask = mask.to(self.device, torch.bfloat16)
        elif op == "noise":
            gen = torch.Generator(self.device).manual_seed(s.seed + SEED_BEND_NOISE)
        clock = self.clock

        def bend(h, spatial):
            return bend_tensor(h, op, st, (lat_h, lat_w) if spatial else None, mask, gen)

        def hook(module, args, output):
            if not (s.bend_from <= clock.progress <= s.bend_to):
                return None                                    # outside the window: unchanged
            if double:                                         # (text, image)
                txt, img = output
                return (txt, bend(img, True)) if image else (bend(txt, False), img)
            txt, img = output[:, :txt_len], output[:, txt_len:]   # single stream: [text, image]
            return torch.cat([txt, bend(img, True)] if image else [bend(txt, False), img], dim=1)

        handle = block.register_forward_hook(hook)
        try:
            yield
        finally:
            handle.remove()

    @contextmanager
    def weight_noise(self, s: Settings):
        """R9: add noise to one block's weight matrices, relative to each tensor's std. Restored on exit."""
        if not s.weight_block or s.weight_noise == 0:
            yield
            return
        block = self.block(s.weight_block)
        gen = torch.Generator(self.device).manual_seed(s.seed + SEED_WEIGHTS)
        saved = []
        try:
            with torch.no_grad():
                for p in block.parameters():
                    if p.dim() < 2:
                        continue
                    saved.append((p, p.detach().clone()))
                    z = torch.randn(p.shape, generator=gen, device=p.device, dtype=torch.float32)
                    w = p.float()
                    p.copy_((w + s.weight_noise * w.std() * z).to(p.dtype))
            yield
        finally:
            with torch.no_grad():
                for p, orig in saved:
                    p.copy_(orig)

    def denoise(self, s: Settings, cond: torch.Tensor, uncond: torch.Tensor, seeds: list[int],
                guidances: list[float], on_step=None) -> torch.Tensor:
        """Run the sampler for one chunk. Returns packed latents (n, h*w, 128) in float32."""
        P, dev, N = self.P, self.device, s.steps
        n = len(seeds)
        lat_h, lat_w = s.height // 16, s.width // 16          # vae scale 8 x patch 2
        noise = torch.stack([torch.randn((128, lat_h, lat_w), generator=torch.Generator("cpu").manual_seed(sd))
                             for sd in seeds])                 # float32, device independent
        img_ids = P._prepare_latent_ids(noise).to(dev)
        lat = P._pack_latents(noise).to(dev)                   # (n, h*w, 128), stays float32

        sched = self._Scheduler.from_config(self.pipe.scheduler.config)   # fresh: the scheduler is stateful
        sigmas = np.linspace(1.0, 1.0 / N, N)
        mu = self._mu(image_seq_len=lat.shape[1], num_steps=N)
        timesteps, _ = self._retrieve_timesteps(sched, N, dev, sigmas=sigmas, mu=mu)
        sched.set_begin_index(0)
        sig = sched.sigmas.float().cpu().tolist()              # shifted sigmas, N + 1 values ending in 0
        n_run = max(1, round(N * s.stop_at))

        enc = torch.cat([cond, uncond.expand(n, -1, -1)]).to(torch.bfloat16)
        txt_ids = P._prepare_text_ids(enc).to(dev)
        ids2 = torch.cat([img_ids, img_ids])
        g = torch.tensor(guidances, device=dev, dtype=torch.float32).view(-1, 1, 1)
        x0_pred = None
        for i, t in enumerate(timesteps[:n_run]):
            self.clock.progress = i / N
            sigma = sig[i]
            t_model = sigma * s.sigma_mult + s.sigma_offset    # R6: the model is told a different time
            x = torch.cat([lat, lat]).to(torch.bfloat16)
            v = self.pipe.transformer(
                hidden_states=x, timestep=torch.full((2 * n,), t_model, device=dev), guidance=None,
                encoder_hidden_states=enc, txt_ids=txt_ids, img_ids=ids2, return_dict=False)[0].float()
            v_c, v_u = v.chunk(2)
            v = v_u + g * (v_c - v_u)                          # R3: per-sample guidance
            x0_pred = lat - sigma * v
            lat = sched.step(v, t, lat, return_dict=False)[0]  # float32 stays float32 because v is float32
            if on_step is not None:
                on_step()
        return x0_pred if (s.show_x0 and n_run < N) else lat

    def decode(self, final: torch.Tensor, lat_h: int, lat_w: int) -> tuple[list[Image.Image], list[bool]]:
        vae = self.pipe.vae
        n = final.shape[0]
        finite = torch.isfinite(final).flatten(1).all(1).tolist()
        final = torch.nan_to_num(final.float(), 0.0, 0.0, 0.0)
        lat = final.view(n, lat_h, lat_w, 128).permute(0, 3, 1, 2)   # row-major, as _prepare_latent_ids
        mean = vae.bn.running_mean.view(1, -1, 1, 1).to(lat.device, torch.float32)
        std = torch.sqrt(vae.bn.running_var.view(1, -1, 1, 1) + vae.config.batch_norm_eps).to(lat.device, torch.float32)
        lat = self.P._unpatchify_latents(lat * std + mean)
        per_call = max(1, DECODE_PIXELS // (lat_h * lat_w * 256))
        images = []
        for k in range(0, n, per_call):
            img = vae.decode(lat[k:k + per_call].to(vae.dtype), return_dict=False)[0]
            img = torch.nan_to_num(img.float(), 0.0, 0.0, 0.0)
            images += self.pipe.image_processor.postprocess(img, output_type="pil")
        return images, finite

    # ---------------------------------------------------------------- entry points

    def render(self, s: Settings, specs: list[tuple[int, float, float, str]], max_seq: int = MAX_SEQ,
               per_chunk: int | None = None, progress=None) -> Result:
        """Render explicit per-sample specs. `progress(done, total)` is called once per step.
        Images per forward pass default to MAX_BATCH at DEFAULT_SIZE, scaled by pixel count.
        """
        s = _checked(s)
        t0 = time.perf_counter()
        lat_h, lat_w = s.height // self.latent_factor, s.width // self.latent_factor
        n = len(specs)
        per_chunk = per_chunk or max(1, math.floor(MAX_BATCH * DEFAULT_SIZE ** 2 / (s.width * s.height)))
        chunks = [specs[k:k + per_chunk] for k in range(0, n, per_chunk)]
        n_run = max(1, round(s.steps * s.stop_at))
        total, done = len(chunks) * n_run, 0

        def on_step():
            nonlocal done
            done += 1
            if progress is not None:
                progress(done, total)

        timings = {}
        self.run_notes = []
        try:
            with self.interventions(s, lat_h, lat_w, max_seq), torch.inference_mode():
                _sync()
                t1 = time.perf_counter()
                finals = []
                cond_all, uncond = self.conditioning(s, [sp[1] for sp in specs], max_seq)
                _sync()
                timings["encode"] = time.perf_counter() - t1
                t1 = time.perf_counter()
                for k, chunk in enumerate(chunks):
                    lo = k * per_chunk
                    finals.append(self.denoise(s, cond_all[lo:lo + len(chunk)], uncond,
                                               [sp[0] for sp in chunk], [sp[2] for sp in chunk], on_step))
                _sync()
                timings["denoise"] = time.perf_counter() - t1
                t1 = time.perf_counter()
                images, finite = self.decode(torch.cat(finals), lat_h, lat_w)
                _sync()
                timings["decode"] = time.perf_counter() - t1
        except torch.cuda.OutOfMemoryError:
            torch.cuda.empty_cache()
            import gradio as gr
            raise gr.Error("Zu wenig GPU-Speicher. Wählt eine kleinere Bildgröße.")

        items, warnings = [], []
        for im, ok, sp in zip(images, finite, specs):
            items.append((im, sp[3] if ok else f"{sp[3]} (numerischer Überlauf)"))
        note = self.text_bend_note(s) if (s.bend_block and s.bend_op != "none" and s.bend_strength != 0) else None
        if note:
            warnings.append(note)
        warnings += self.run_notes
        n_bad = finite.count(False)
        if n_bad:
            warnings.append(f"Bei {n_bad} von {n} Bildern ist die Rechnung numerisch übergelaufen. "
                            "Diese Bilder zeigen nur einen Ersatz für die ungültigen Werte.")
        cols = math.ceil(math.sqrt(n))
        return Result(items=items, grid=make_grid(images, cols), warnings=warnings,
                      seconds=time.perf_counter() - t0, timings=timings)

    def generate(self, s: Settings, progress=None) -> Result:
        return self.render(s, sample_specs(s), progress=progress)


def load_engine(device: str = "cuda") -> Engine:
    """The engine of the active model family."""
    if FAMILY.key == "sd15":
        from engine_sd15 import SD15Engine

        return SD15Engine.from_pretrained(MODEL_ID, device)
    return Engine.from_pretrained(MODEL_ID, device)


def _checked(s: Settings) -> Settings:
    from dataclasses import replace

    if s.experiment not in EXPERIMENTS:
        raise ValueError(f"unknown experiment {s.experiment!r}")
    if s.bend_op not in BEND_OPS:
        raise ValueError(f"unknown bending op {s.bend_op!r}")
    if s.bend_stream not in ("image", "text"):
        raise ValueError(f"unknown stream {s.bend_stream!r}")
    if s.tools:
        from tools_sd15 import TOOLS

        unknown = [u.key for u in s.tools if u.key not in TOOLS]
        if unknown:
            raise ValueError(f"unknown tools {unknown}")
    return replace(s, prompt_a=s.prompt_a or "", prompt_b=s.prompt_b or "",
                   width=clamp_size(s.width), height=clamp_size(s.height),
                   steps=max(1, int(s.steps)), seed=int(s.seed),
                   te_layers=tuple(int(k) for k in s.te_layers),
                   stop_at=min(1.0, max(0.0, float(s.stop_at))),
                   grid=None if s.grid is None else max(1, min(max(GRID_SIZES), int(s.grid))))


def _sync() -> None:
    if torch.cuda.is_available():
        torch.cuda.synchronize()