Add glyph rendering and region preprocessing

vendor/pixtext/preprocessing/compositing.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import torch
+
+from pixtext.preprocessing.geometry import GroupGeometry, ImageSize
+
+
+def _validate_generated_image(image: torch.Tensor, *, expected_size: ImageSize, label: str) -> None:
+    if not isinstance(image, torch.Tensor):
+        raise TypeError(f"{label} must be a torch.Tensor")
+    if image.ndim != 3 or image.shape[0] != 3:
+        raise ValueError(
+            f"{label} must be one RGB tensor in (C, H, W) layout, got shape {tuple(image.shape)}"
+        )
+    actual_size = tuple(image.shape[-2:])
+    planned_size = (expected_size.height, expected_size.width)
+    if actual_size != planned_size:
+        raise ValueError(
+            f"{label} size does not match its geometry: got {actual_size}, expected {planned_size}"
+        )
+
+
+def compose_native_edit_items(
+    clean_reference: torch.Tensor, generated_items: Sequence[torch.Tensor], geometry: GroupGeometry
+) -> torch.Tensor:
+    """Paste generated item crops on an undegraded reference canvas."""
+    _validate_generated_image(
+        clean_reference, expected_size=geometry.reference.size, label="clean reference image"
+    )
+    if len(generated_items) != len(geometry.items):
+        raise ValueError(
+            f"generated item count does not match geometry: got {len(generated_items)}, expected {len(geometry.items)}"
+        )
+    composite = clean_reference.detach().float().clone().contiguous()
+    for item_index, (generated_item, item_geometry) in enumerate(
+        zip(generated_items, geometry.items, strict=True), start=1
+    ):
+        _validate_generated_image(
+            generated_item,
+            expected_size=item_geometry.crop.size,
+            label=f"generated item {item_index}",
+        )
+        if generated_item.device != composite.device:
+            raise ValueError(
+                f"generated item {item_index} is on {generated_item.device}, but the generated main image is on {composite.device}"
+            )
+        top = item_geometry.crop.top
+        left = item_geometry.crop.left
+        bottom = top + item_geometry.crop.size.height
+        right = left + item_geometry.crop.size.width
+        composite[:, top:bottom, left:right] = generated_item.detach().float()
+    return composite

vendor/pixtext/preprocessing/geometry.py

@@ -0,0 +1,211 @@
+"""Deterministic geometry planning for group-image JiT inputs.
+
+This module computes pixel-coordinate transforms only.  Image interpolation and
+tensor slicing are deliberately left to data adapters so the geometry contract
+does not depend on PIL, NumPy, or PyTorch.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable
+from dataclasses import dataclass
+from fractions import Fraction
+from numbers import Real
+
+
+@dataclass(frozen=True, slots=True)
+class ImageSize:
+    """An image size in ``(height, width)`` order."""
+
+    height: int
+    width: int
+
+    def __post_init__(self) -> None:
+        for name, value in (("height", self.height), ("width", self.width)):
+            if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+                raise ValueError(f"{name} must be a positive integer, got {value!r}")
+
+
+@dataclass(frozen=True, slots=True)
+class BoundingBox:
+    """A half-open pixel-boundary box in ``[x0, y0, x1, y1)`` form."""
+
+    x0: Real
+    y0: Real
+    x1: Real
+    y1: Real
+
+    def __post_init__(self) -> None:
+        for name, value in (("x0", self.x0), ("y0", self.y0), ("x1", self.x1), ("y1", self.y1)):
+            if isinstance(value, bool) or not isinstance(value, Real) or (not math.isfinite(value)):
+                raise ValueError(f"{name} must be a finite number, got {value!r}")
+        if self.x1 <= self.x0 or self.y1 <= self.y0:
+            raise ValueError(f"bbox must be non-degenerate, got {self!r}")
+
+
+@dataclass(frozen=True, slots=True)
+class ReferenceGeometry:
+    """The reference canvas and its scale from the logical canvas."""
+
+    size: ImageSize
+    scale_h: float
+    scale_w: float
+
+
+@dataclass(frozen=True, slots=True)
+class MainImageGeometry:
+    """The main image size and its actual scale from the reference canvas."""
+
+    size: ImageSize
+    scale_h: float
+    scale_w: float
+
+
+@dataclass(frozen=True, slots=True)
+class CropGeometry:
+    """An integer slice rectangle on the reference image."""
+
+    top: int
+    left: int
+    size: ImageSize
+
+
+@dataclass(frozen=True, slots=True)
+class ItemGeometry:
+    """One item's bbox in logical and reference coordinates."""
+
+    logical_bbox: BoundingBox
+    reference_bbox: BoundingBox
+    crop: CropGeometry
+
+
+@dataclass(frozen=True, slots=True)
+class GroupGeometry:
+    """All deterministic geometry needed to prepare one group."""
+
+    logical_size: ImageSize
+    reference: ReferenceGeometry
+    main: MainImageGeometry
+    items: tuple[ItemGeometry, ...]
+
+
+def _round_to_nearest_patch(value: float, patch_size: int) -> int:
+    return patch_size * max(1, math.floor(value / patch_size + 0.5))
+
+
+def _validate_bbox_in_canvas(bbox: BoundingBox, canvas: ImageSize) -> None:
+    if bbox.x0 < 0 or bbox.y0 < 0 or bbox.x1 > canvas.width or (bbox.y1 > canvas.height):
+        raise ValueError(f"bbox {bbox!r} lies outside logical canvas {canvas!r}")
+
+
+def _as_exact_coordinate(coordinate: Real) -> Fraction:
+    if isinstance(coordinate, Fraction):
+        return coordinate
+    if isinstance(coordinate, int):
+        return Fraction(coordinate)
+    return Fraction.from_float(float(coordinate))
+
+
+def _scale_coordinate_exact(
+    coordinate: float, logical_extent: int, reference_extent: int
+) -> Fraction:
+    return _as_exact_coordinate(coordinate) * reference_extent / logical_extent
+
+
+def _aligned_axis_crop(
+    bbox_start: Fraction, bbox_end: Fraction, canvas_extent: int, patch_size: int
+) -> tuple[int, int]:
+    pixel_start = math.floor(bbox_start)
+    pixel_end = math.ceil(bbox_end)
+    required_extent = pixel_end - pixel_start
+    crop_extent = (required_extent + patch_size - 1) // patch_size * patch_size
+    feasible_start = max(0, pixel_end - crop_extent)
+    feasible_end = min(pixel_start, canvas_extent - crop_extent)
+    if feasible_start > feasible_end:
+        raise RuntimeError("no in-canvas crop can contain a valid bbox")
+    centered_start = math.floor((bbox_start + bbox_end - crop_extent) / 2 + Fraction(1, 2))
+    crop_start = min(max(centered_start, feasible_start), feasible_end)
+    return (crop_start, crop_extent)
+
+
+def plan_group_geometry(
+    logical_size: ImageSize,
+    item_bboxes: Iterable[BoundingBox] = (),
+    *,
+    patch_size: int = 16,
+    main_target_area: float = 1048576,
+    clamp_main_to_reference: bool = True,
+) -> GroupGeometry:
+    """Plan the reference canvas and main image for one group.
+
+    ``clamp_main_to_reference`` caps the main scale at 1.0 so the main is never
+    upsampled past the reference canvas, which keeps the RoPE main-grid step at
+    or above the item-grid step of 1.
+    """
+    if isinstance(patch_size, bool) or not isinstance(patch_size, int) or patch_size <= 0:
+        raise ValueError(f"patch_size must be a positive integer, got {patch_size!r}")
+    if (
+        isinstance(main_target_area, bool)
+        or not isinstance(main_target_area, (int, float))
+        or (not math.isfinite(main_target_area))
+        or (main_target_area <= 0)
+    ):
+        raise ValueError(
+            f"main_target_area must be a positive finite number, got {main_target_area!r}"
+        )
+    if not isinstance(clamp_main_to_reference, bool):
+        raise TypeError("clamp_main_to_reference must be a bool")
+    reference_size = ImageSize(
+        height=_round_to_nearest_patch(logical_size.height, patch_size),
+        width=_round_to_nearest_patch(logical_size.width, patch_size),
+    )
+    main_scale = math.sqrt(main_target_area / (reference_size.height * reference_size.width))
+    if clamp_main_to_reference:
+        main_scale = min(main_scale, 1.0)
+    main_size = ImageSize(
+        height=_round_to_nearest_patch(reference_size.height * main_scale, patch_size),
+        width=_round_to_nearest_patch(reference_size.width * main_scale, patch_size),
+    )
+    items_list = []
+    for bbox in item_bboxes:
+        _validate_bbox_in_canvas(bbox, logical_size)
+        reference_x0 = _scale_coordinate_exact(bbox.x0, logical_size.width, reference_size.width)
+        reference_y0 = _scale_coordinate_exact(bbox.y0, logical_size.height, reference_size.height)
+        reference_x1 = _scale_coordinate_exact(bbox.x1, logical_size.width, reference_size.width)
+        reference_y1 = _scale_coordinate_exact(bbox.y1, logical_size.height, reference_size.height)
+        reference_bbox = BoundingBox(
+            x0=reference_x0, y0=reference_y0, x1=reference_x1, y1=reference_y1
+        )
+        (crop_left, crop_width) = _aligned_axis_crop(
+            reference_x0, reference_x1, reference_size.width, patch_size
+        )
+        (crop_top, crop_height) = _aligned_axis_crop(
+            reference_y0, reference_y1, reference_size.height, patch_size
+        )
+        items_list.append(
+            ItemGeometry(
+                logical_bbox=bbox,
+                reference_bbox=reference_bbox,
+                crop=CropGeometry(
+                    top=crop_top,
+                    left=crop_left,
+                    size=ImageSize(height=crop_height, width=crop_width),
+                ),
+            )
+        )
+    items = tuple(items_list)
+    return GroupGeometry(
+        logical_size=logical_size,
+        reference=ReferenceGeometry(
+            size=reference_size,
+            scale_h=reference_size.height / logical_size.height,
+            scale_w=reference_size.width / logical_size.width,
+        ),
+        main=MainImageGeometry(
+            size=main_size,
+            scale_h=main_size.height / reference_size.height,
+            scale_w=main_size.width / reference_size.width,
+        ),
+        items=items,
+    )

vendor/pixtext/preprocessing/glyphs.py

@@ -0,0 +1,549 @@
+"""Canonical per-item glyph rendering and font coverage validation.
+
+The renderer owns the inference glyph seam.  It deliberately owns shaping identity, font
+measurement, bbox-local placement, rasterization, and failure behaviour.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import ctypes.util
+import hashlib
+import json
+import math
+import unicodedata
+from dataclasses import asdict, dataclass
+from functools import lru_cache
+from pathlib import Path
+
+import numpy as np
+import torch
+from PIL import Image, ImageDraw, ImageFont, features
+
+from pixtext.preprocessing.geometry import BoundingBox, CropGeometry, ItemGeometry
+
+CANONICAL_GLYPH_POLICY = "whole_item_glyph"
+CANONICAL_FONT_RESOLUTION_POLICY = "first_font_covering_entire_item"
+CANONICAL_FONT_PATH = (
+    Path(__file__).resolve().parents[2] / "assets/fonts/canonical/AlibabaPuHuiTi-3-55-Regular.ttf"
+)
+CANONICAL_FONT_SHA256 = "be33bc8d45ced30fa5c542f6f056565baf85003d4e079522110d76016f8a04e2"
+CANONICAL_FALLBACK_FONT_PATH = (
+    Path(__file__).resolve().parents[2] / "assets/fonts/canonical/GoNotoCJKCore-v7.0.ttf"
+)
+CANONICAL_FALLBACK_FONT_SHA256 = "bfe18b47612e58ff28b568f32b55e35fde4505d6137581bf11d2363e257e2ad4"
+CANONICAL_LANGUAGE = "zh-CN"
+CANONICAL_FEATURES = ("locl",)
+CANONICAL_OCCUPANCY = 0.85
+
+
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class FontCoverageResult:
+    """Observable shaping/coverage verdict for one unchanged Unicode string."""
+
+    text: str
+    missing_codepoints: tuple[int, ...]
+    notdef_glyph_count: int
+    expected_visible_ink: bool
+    rendered_ink_nonempty: bool
+
+    @property
+    def covered(self) -> bool:
+        return (
+            not self.missing_codepoints
+            and self.notdef_glyph_count == 0
+            and (not self.expected_visible_ink or self.rendered_ink_nonempty)
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class GlyphFontIdentity:
+    index: int
+    role: str
+    path: str
+    sha256: str
+
+
+@dataclass(frozen=True, slots=True)
+class GlyphRendererIdentity:
+    policy: str
+    font_resolution_policy: str
+    font_chain: tuple[GlyphFontIdentity, ...]
+    occupancy: float
+    rasterization: str
+    layout_engine: str
+    language: str
+    opentype_features: tuple[str, ...]
+    pillow_version: str
+    freetype_version: str
+    raqm_version: str
+    polarity: str
+    orientation: str
+
+    @property
+    def digest(self) -> str:
+        payload = json.dumps(
+            asdict(self), ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        return hashlib.sha256(payload).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class GlyphCondition:
+    """The exact coverage tensor consumed by the model plus its provenance."""
+
+    image: torch.Tensor
+    text: str
+    pre_alignment_bbox: BoundingBox
+    aligned_crop: CropGeometry
+    renderer_identity: str
+    selected_font_index: int
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.image, torch.Tensor):
+            raise TypeError("glyph image must be a torch.Tensor")
+        expected = (1, self.aligned_crop.size.height, self.aligned_crop.size.width)
+        if self.image.shape != expected:
+            raise ValueError(
+                f"glyph image shape {tuple(self.image.shape)} does not match {expected}"
+            )
+        if not self.image.is_floating_point():
+            raise ValueError("glyph image must use a floating-point dtype")
+        torch._assert_async(
+            torch.isfinite(self.image).all() & self.image.ge(0).all() & self.image.le(1).all(),
+            "glyph coverage must be finite and lie in [0, 1]",
+        )
+        if not isinstance(self.text, str):
+            raise ValueError("glyph text must be a string")
+        if not self.text:
+            if not self.renderer_identity.endswith(":blank-v1"):
+                raise ValueError("empty glyph text requires explicit blank conditioning")
+            torch._assert_async(self.image.eq(0).all(), "blank glyph must have zero ink")
+        if not isinstance(self.renderer_identity, str) or not self.renderer_identity:
+            raise ValueError("renderer_identity must be non-empty")
+        if (
+            isinstance(self.selected_font_index, bool)
+            or not isinstance(self.selected_font_index, int)
+            or self.selected_font_index < 0
+        ):
+            raise ValueError("selected_font_index must be a non-negative integer")
+
+    def pin_memory(self) -> GlyphCondition:
+        if self.image.device.type != "cpu":
+            raise ValueError("only CPU glyph conditions can be pinned")
+        return GlyphCondition(
+            image=self.image.pin_memory(),
+            text=self.text,
+            pre_alignment_bbox=self.pre_alignment_bbox,
+            aligned_crop=self.aligned_crop,
+            renderer_identity=self.renderer_identity,
+            selected_font_index=self.selected_font_index,
+        )
+
+
+class _HarfBuzzGlyphInfo(ctypes.Structure):
+    _fields_ = (
+        ("codepoint", ctypes.c_uint32),
+        ("mask", ctypes.c_uint32),
+        ("cluster", ctypes.c_uint32),
+        ("var1", ctypes.c_uint32),
+        ("var2", ctypes.c_uint32),
+    )
+
+
+class FontCoverageChecker:
+    """Check cmap and shaped glyph IDs through the renderer's pinned font."""
+
+    def __init__(self, font_path: Path = CANONICAL_FONT_PATH) -> None:
+        self.font_path = Path(font_path)
+        library_path = ctypes.util.find_library("harfbuzz")
+        if library_path is None:
+            raise RuntimeError("HarfBuzz is required for canonical glyph coverage")
+        self._hb = ctypes.CDLL(library_path)
+        self._bind_harfbuzz()
+        self._font_bytes = self.font_path.read_bytes()
+        self._font_buffer = ctypes.create_string_buffer(self._font_bytes)
+        self._blob = self._hb.hb_blob_create(
+            self._font_buffer, len(self._font_bytes), 2, None, None
+        )
+        if not self._blob:
+            raise RuntimeError(f"HarfBuzz could not read font: {self.font_path}")
+        self._face = self._hb.hb_face_create(self._blob, 0)
+        self._font = self._hb.hb_font_create(self._face)
+        if not self._face or not self._font:
+            self.close()
+            raise RuntimeError("HarfBuzz could not create the canonical font")
+        self._hb.hb_ot_font_set_funcs(self._font)
+
+    def _bind_harfbuzz(self) -> None:
+        hb = self._hb
+        hb.hb_blob_create.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_uint,
+            ctypes.c_int,
+            ctypes.c_void_p,
+            ctypes.c_void_p,
+        ]
+        hb.hb_blob_create.restype = ctypes.c_void_p
+        hb.hb_blob_destroy.argtypes = [ctypes.c_void_p]
+        hb.hb_face_create.argtypes = [ctypes.c_void_p, ctypes.c_uint]
+        hb.hb_face_create.restype = ctypes.c_void_p
+        hb.hb_face_destroy.argtypes = [ctypes.c_void_p]
+        hb.hb_font_create.argtypes = [ctypes.c_void_p]
+        hb.hb_font_create.restype = ctypes.c_void_p
+        hb.hb_font_destroy.argtypes = [ctypes.c_void_p]
+        hb.hb_ot_font_set_funcs.argtypes = [ctypes.c_void_p]
+        hb.hb_font_get_nominal_glyph.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_uint32,
+            ctypes.POINTER(ctypes.c_uint32),
+        ]
+        hb.hb_font_get_nominal_glyph.restype = ctypes.c_int
+        hb.hb_buffer_create.restype = ctypes.c_void_p
+        hb.hb_buffer_destroy.argtypes = [ctypes.c_void_p]
+        hb.hb_buffer_add_utf8.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_char_p,
+            ctypes.c_int,
+            ctypes.c_uint,
+            ctypes.c_int,
+        ]
+        hb.hb_buffer_guess_segment_properties.argtypes = [ctypes.c_void_p]
+        hb.hb_language_from_string.argtypes = [ctypes.c_char_p, ctypes.c_int]
+        hb.hb_language_from_string.restype = ctypes.c_void_p
+        hb.hb_buffer_set_language.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
+        hb.hb_feature_from_string.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p]
+        hb.hb_feature_from_string.restype = ctypes.c_int
+        hb.hb_shape.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint]
+        hb.hb_buffer_get_glyph_infos.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint)]
+        hb.hb_buffer_get_glyph_infos.restype = ctypes.POINTER(_HarfBuzzGlyphInfo)
+
+    @staticmethod
+    def _requires_nominal_glyph(character: str) -> bool:
+        codepoint = ord(character)
+        category = unicodedata.category(character)
+        return not (
+            unicodedata.combining(character)
+            or category in {"Cf", "Cc", "Cs", "Zl", "Zp"}
+            or 65024 <= codepoint <= 65039
+            or (917760 <= codepoint <= 917999)
+            or (codepoint == 8205)
+        )
+
+    @staticmethod
+    def _expects_visible_ink(text: str) -> bool:
+        return any(
+            not character.isspace() and unicodedata.category(character)[0] not in {"C", "Z"}
+            for character in text
+        )
+
+    @lru_cache(maxsize=65536)
+    def check(self, text: str) -> FontCoverageResult:
+        if not isinstance(text, str) or not text:
+            raise ValueError("coverage text must be a non-empty string")
+        missing: list[int] = []
+        for character in text:
+            if not self._requires_nominal_glyph(character):
+                continue
+            glyph = ctypes.c_uint32()
+            if (
+                not self._hb.hb_font_get_nominal_glyph(
+                    self._font, ord(character), ctypes.byref(glyph)
+                )
+                or glyph.value == 0
+            ):
+                missing.append(ord(character))
+        encoded = text.encode("utf-8")
+        buffer = self._hb.hb_buffer_create()
+        if not buffer:
+            raise RuntimeError("HarfBuzz could not allocate a shaping buffer")
+        try:
+            self._hb.hb_buffer_add_utf8(buffer, encoded, len(encoded), 0, len(encoded))
+            self._hb.hb_buffer_guess_segment_properties(buffer)
+            language = self._hb.hb_language_from_string(CANONICAL_LANGUAGE.encode(), -1)
+            self._hb.hb_buffer_set_language(buffer, language)
+            feature = (ctypes.c_uint32 * 4)()
+            if not self._hb.hb_feature_from_string(b"locl=1", -1, ctypes.byref(feature)):
+                raise RuntimeError("HarfBuzz rejected the pinned locl feature")
+            self._hb.hb_shape(self._font, buffer, ctypes.byref(feature), 1)
+            length = ctypes.c_uint()
+            infos = self._hb.hb_buffer_get_glyph_infos(buffer, ctypes.byref(length))
+            notdef_count = sum(infos[index].codepoint == 0 for index in range(length.value))
+        finally:
+            self._hb.hb_buffer_destroy(buffer)
+        expected_ink = self._expects_visible_ink(text)
+        rendered_ink = True
+        if expected_ink:
+            font = ImageFont.truetype(
+                str(self.font_path), size=128, layout_engine=ImageFont.Layout.RAQM
+            )
+            mask = font.getmask(
+                text, mode="L", language=CANONICAL_LANGUAGE, features=list(CANONICAL_FEATURES)
+            )
+            rendered_ink = mask.getbbox() is not None
+        return FontCoverageResult(
+            text=text,
+            missing_codepoints=tuple(sorted(set(missing))),
+            notdef_glyph_count=notdef_count,
+            expected_visible_ink=expected_ink,
+            rendered_ink_nonempty=rendered_ink,
+        )
+
+    def close(self) -> None:
+        font = getattr(self, "_font", None)
+        face = getattr(self, "_face", None)
+        blob = getattr(self, "_blob", None)
+        if font:
+            self._hb.hb_font_destroy(font)
+            self._font = None
+        if face:
+            self._hb.hb_face_destroy(face)
+            self._face = None
+        if blob:
+            self._hb.hb_blob_destroy(blob)
+            self._blob = None
+
+    def __del__(self) -> None:
+        try:
+            self.close()
+        except Exception:
+            pass
+
+
+@dataclass(frozen=True, slots=True)
+class FontResolution:
+    """Whole-item result from an ordered, pinned font chain."""
+
+    text: str
+    selected_font_index: int | None
+    attempts: tuple[FontCoverageResult, ...]
+
+    @property
+    def covered(self) -> bool:
+        return self.selected_font_index is not None
+
+    @property
+    def selected_coverage(self) -> FontCoverageResult | None:
+        if self.selected_font_index is None:
+            return None
+        return self.attempts[self.selected_font_index]
+
+
+class PinnedFontResolver:
+    """Select the first font covering the entire unchanged item string."""
+
+    def __init__(self, font_paths: tuple[Path, ...]) -> None:
+        if not font_paths:
+            raise ValueError("canonical glyph font chain must not be empty")
+        self.checkers = tuple(FontCoverageChecker(path) for path in font_paths)
+
+    @lru_cache(maxsize=65536)
+    def resolve(self, text: str) -> FontResolution:
+        attempts: list[FontCoverageResult] = []
+        for index, checker in enumerate(self.checkers):
+            result = checker.check(text)
+            attempts.append(result)
+            if result.covered:
+                return FontResolution(
+                    text=text, selected_font_index=index, attempts=tuple(attempts)
+                )
+        return FontResolution(text=text, selected_font_index=None, attempts=tuple(attempts))
+
+    def check(self, text: str) -> FontResolution:
+        """Compatibility alias for callers interested only in chain coverage."""
+        return self.resolve(text)
+
+    def close(self) -> None:
+        for checker in self.checkers:
+            checker.close()
+
+
+class CanonicalItemGlyphRenderer:
+    """Render upright, bbox-centered 1x grayscale canonical glyphs."""
+
+    def __init__(
+        self,
+        *,
+        font_path: Path = CANONICAL_FONT_PATH,
+        expected_font_sha256: str = CANONICAL_FONT_SHA256,
+        fallback_font_path: Path = CANONICAL_FALLBACK_FONT_PATH,
+        expected_fallback_font_sha256: str = CANONICAL_FALLBACK_FONT_SHA256,
+        font_resolution_policy: str = CANONICAL_FONT_RESOLUTION_POLICY,
+        occupancy: float = CANONICAL_OCCUPANCY,
+    ) -> None:
+        self.font_paths = (Path(font_path), Path(fallback_font_path))
+        expected_sha256s = (expected_font_sha256, expected_fallback_font_sha256)
+        actual_sha256s: list[str] = []
+        for index, (path, expected_sha256) in enumerate(
+            zip(self.font_paths, expected_sha256s, strict=True)
+        ):
+            if not path.is_file():
+                raise ValueError(f"canonical font {index} is not readable: {path}")
+            actual_sha256 = _sha256_file(path)
+            if actual_sha256 != expected_sha256:
+                raise ValueError(
+                    f"canonical font {index} SHA-256 mismatch: got {actual_sha256}, expected {expected_sha256}"
+                )
+            actual_sha256s.append(actual_sha256)
+        if font_resolution_policy != CANONICAL_FONT_RESOLUTION_POLICY:
+            raise ValueError("unsupported canonical font resolution policy")
+        if not math.isfinite(occupancy) or not 0 < occupancy <= 1:
+            raise ValueError("glyph occupancy must lie in (0, 1]")
+        if not features.check_feature("raqm"):
+            raise RuntimeError("canonical glyph rendering requires Pillow RAQM")
+        self.occupancy = float(occupancy)
+        self.coverage = PinnedFontResolver(self.font_paths)
+        import PIL
+
+        self.identity = GlyphRendererIdentity(
+            policy=CANONICAL_GLYPH_POLICY,
+            font_resolution_policy=font_resolution_policy,
+            font_chain=tuple(
+                GlyphFontIdentity(
+                    index=index, role=role, path=str(path.resolve()), sha256=actual_sha256s[index]
+                )
+                for (index, (role, path)) in enumerate(
+                    zip(("primary", "fallback"), self.font_paths, strict=True)
+                )
+            ),
+            occupancy=self.occupancy,
+            rasterization="direct_1x_grayscale",
+            layout_engine="RAQM",
+            language=CANONICAL_LANGUAGE,
+            opentype_features=CANONICAL_FEATURES,
+            pillow_version=PIL.__version__,
+            freetype_version=features.version_module("freetype2") or "unknown",
+            raqm_version=features.version_feature("raqm") or "unknown",
+            polarity="black_0_white_1",
+            orientation="horizontal_upright_uniform_scale",
+        )
+
+    @lru_cache(maxsize=1024)
+    def _font(self, font_index: int, size: int) -> ImageFont.FreeTypeFont:
+        return ImageFont.truetype(
+            str(self.font_paths[font_index]), size=max(1, size), layout_engine=ImageFont.Layout.RAQM
+        )
+
+    @staticmethod
+    def _ink_bbox(font: ImageFont.FreeTypeFont, text: str) -> tuple[int, int, int, int]:
+        (mask, offset) = font.getmask2(
+            text, mode="L", language=CANONICAL_LANGUAGE, features=list(CANONICAL_FEATURES)
+        )
+        mask_bbox = mask.getbbox()
+        if mask_bbox is None:
+            raise ValueError(f"canonical glyph has an empty ink bbox: {text!r}")
+        return (
+            offset[0] + mask_bbox[0],
+            offset[1] + mask_bbox[1],
+            offset[0] + mask_bbox[2],
+            offset[1] + mask_bbox[3],
+        )
+
+    def render_blank(self, item_geometry: ItemGeometry) -> GlyphCondition:
+        """Present, zero-ink condition for erasure; distinct from absent CFG glyphs."""
+        crop = item_geometry.crop
+        return GlyphCondition(
+            image=torch.zeros((1, crop.size.height, crop.size.width)),
+            text="",
+            pre_alignment_bbox=item_geometry.reference_bbox,
+            aligned_crop=crop,
+            renderer_identity=self.identity.digest + ":blank-v1",
+            selected_font_index=0,
+        )
+
+    def render(self, text: str, item_geometry: ItemGeometry) -> GlyphCondition:
+        if not isinstance(text, str) or not text:
+            raise ValueError("canonical glyph text must be non-empty")
+        if not isinstance(item_geometry, ItemGeometry):
+            raise TypeError("item_geometry must be an ItemGeometry")
+        resolution = self.coverage.resolve(text)
+        if not resolution.covered:
+            attempts = tuple(
+                {
+                    "font_index": index,
+                    "missing": tuple(f"U+{value:04X}" for value in result.missing_codepoints),
+                    "notdef": result.notdef_glyph_count,
+                    "nonempty": result.rendered_ink_nonempty,
+                }
+                for (index, result) in enumerate(resolution.attempts)
+            )
+            raise ValueError(f"canonical font chain coverage failed for {text!r}: {attempts}")
+        font_index = resolution.selected_font_index
+        assert font_index is not None
+        coverage = resolution.selected_coverage
+        assert coverage is not None
+        crop = item_geometry.crop
+        bbox = item_geometry.reference_bbox
+        local_x0 = float(bbox.x0) - crop.left
+        local_y0 = float(bbox.y0) - crop.top
+        local_x1 = float(bbox.x1) - crop.left
+        local_y1 = float(bbox.y1) - crop.top
+        target_width = (local_x1 - local_x0) * self.occupancy
+        target_height = (local_y1 - local_y0) * self.occupancy
+        if target_width <= 0 or target_height <= 0:
+            raise ValueError("canonical glyph bbox is degenerate after localization")
+        probe_size = 128
+        probe_bbox = self._ink_bbox(self._font(font_index, probe_size), text)
+        probe_width = probe_bbox[2] - probe_bbox[0]
+        probe_height = probe_bbox[3] - probe_bbox[1]
+        fitted_size = max(
+            1,
+            math.floor(probe_size * min(target_width / probe_width, target_height / probe_height)),
+        )
+        while fitted_size > 1:
+            font = self._font(font_index, fitted_size)
+            (left, top, right, bottom) = self._ink_bbox(font, text)
+            if right - left <= target_width and bottom - top <= target_height:
+                break
+            fitted_size -= 1
+        else:
+            font = self._font(font_index, 1)
+            (left, top, right, bottom) = self._ink_bbox(font, text)
+        if right - left > target_width or bottom - top > target_height:
+            raise ValueError(
+                f"canonical glyph cannot fit the bbox occupancy at the minimum font size: text={text!r}, target={target_width}x{target_height}"
+            )
+        while True:
+            next_font = self._font(font_index, fitted_size + 1)
+            next_bbox = self._ink_bbox(next_font, text)
+            if (
+                next_bbox[2] - next_bbox[0] > target_width
+                or next_bbox[3] - next_bbox[1] > target_height
+            ):
+                break
+            fitted_size += 1
+            font = next_font
+            (left, top, right, bottom) = next_bbox
+        center_x = (local_x0 + local_x1) / 2
+        center_y = (local_y0 + local_y1) / 2
+        draw_x = center_x - (left + right) / 2
+        draw_y = center_y - (top + bottom) / 2
+        canvas = Image.new("L", (crop.size.width, crop.size.height), 0)
+        ImageDraw.Draw(canvas).text(
+            (draw_x, draw_y),
+            text,
+            font=font,
+            fill=255,
+            language=CANONICAL_LANGUAGE,
+            features=list(CANONICAL_FEATURES),
+        )
+        array = np.asarray(canvas, dtype=np.float32)
+        image = torch.from_numpy(array.copy()).unsqueeze(0).div_(255.0)
+        if coverage.expected_visible_ink and (not bool(image.gt(0).any())):
+            raise ValueError(f"canonical glyph rasterized to empty ink: {text!r}")
+        return GlyphCondition(
+            image=image.contiguous(),
+            text=text,
+            pre_alignment_bbox=bbox,
+            aligned_crop=crop,
+            renderer_identity=self.identity.digest,
+            selected_font_index=font_index,
+        )

vendor/pixtext/preprocessing/images.py

@@ -0,0 +1,55 @@
+"""Apply deterministic group geometry plans to RGB image tensors."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import torch
+from torchvision.transforms import InterpolationMode
+from torchvision.transforms.functional import resize
+
+from pixtext.preprocessing.geometry import GroupGeometry
+
+
+@dataclass(frozen=True, slots=True)
+class PreparedGroupImages:
+    """Reference, main, and ordered item tensors for one group."""
+
+    reference: torch.Tensor
+    main: torch.Tensor
+    items: tuple[torch.Tensor, ...]
+
+
+def _resize_bicubic(image: torch.Tensor, height: int, width: int) -> torch.Tensor:
+    return resize(
+        image, [height, width], interpolation=InterpolationMode.BICUBIC, antialias=True
+    ).contiguous()
+
+
+def prepare_group_images(image: torch.Tensor, geometry: GroupGeometry) -> PreparedGroupImages:
+    """Resize and crop one RGB image according to a geometry plan."""
+    if not isinstance(image, torch.Tensor):
+        raise TypeError(f"image must be a torch.Tensor, got {type(image).__name__}")
+    if image.ndim != 3 or image.shape[0] != 3:
+        raise ValueError(
+            f"image must be one RGB tensor in (C, H, W) layout, got shape {tuple(image.shape)}"
+        )
+    actual_size = tuple(image.shape[-2:])
+    expected_size = (geometry.logical_size.height, geometry.logical_size.width)
+    if actual_size != expected_size:
+        raise ValueError(
+            f"image size does not match the geometry logical canvas: got {actual_size}, expected {expected_size}"
+        )
+    reference = _resize_bicubic(
+        image, geometry.reference.size.height, geometry.reference.size.width
+    )
+    main = _resize_bicubic(reference, geometry.main.size.height, geometry.main.size.width)
+    items = tuple(
+        reference[
+            :,
+            item.crop.top : item.crop.top + item.crop.size.height,
+            item.crop.left : item.crop.left + item.crop.size.width,
+        ].contiguous()
+        for item in geometry.items
+    )
+    return PreparedGroupImages(reference=reference, main=main, items=items)

vendor/pixtext/preprocessing/main_condition.py

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+import cv2
+import numpy as np
+import torch
+
+from pixtext.preprocessing.geometry import CropGeometry, GroupGeometry, ImageSize, ItemGeometry
+
+
+@dataclass(frozen=True)
+class NativeEditConfig:
+    degradation_region: str = "strict_ocr_bbox"
+    operators: tuple[str, ...] = ("defocus", "gaussian_blur")
+    required_operators: tuple[str, ...] = ()
+    min_operators: int = 2
+    max_operators: int = 2
+    severity_min: float = 0.7
+    severity_max: float = 1.0
+
+
+class NativeEditMainConditioner:
+    def __init__(self, config: NativeEditConfig):
+        self.config = config
+
+    def __call__(self, clean_main, geometry, rng):
+        self._validate_inputs(clean_main, geometry)
+        cv2.setNumThreads(1)
+        seed = int(torch.randint(0, 2**63 - 1, (), dtype=torch.int64, generator=rng).item())
+        local_rng = torch.Generator(device="cpu").manual_seed(seed)
+        (operators, severities) = self._sample_operator_plan(local_rng)
+        operator_seeds = torch.randint(
+            0, 2**31 - 1, (len(operators),), generator=local_rng
+        ).tolist()
+        original = clean_main.permute(1, 2, 0).contiguous().numpy()
+        candidate = original.copy()
+        for operator, severity, operator_seed in zip(
+            operators, severities, operator_seeds, strict=True
+        ):
+            candidate = self._apply_operator(candidate, operator, severity, operator_seed)
+        footprint = self._replacement_footprint(geometry)
+        result = np.where(footprint[:, :, None], candidate, original).astype(np.uint8, copy=False)
+        return torch.from_numpy(np.ascontiguousarray(result)).permute(2, 0, 1).contiguous()
+
+    @staticmethod
+    def _validate_inputs(clean_main: torch.Tensor, geometry: GroupGeometry) -> None:
+        if not isinstance(clean_main, torch.Tensor):
+            raise TypeError("clean_main must be a torch.Tensor")
+        if not isinstance(geometry, GroupGeometry):
+            raise TypeError("geometry must be a GroupGeometry")
+        expected_shape = (3, geometry.main.size.height, geometry.main.size.width)
+        if clean_main.shape != expected_shape or clean_main.dtype is not torch.uint8:
+            raise ValueError(
+                f"clean_main must be uint8 RGB matching geometry.main.size, got {tuple(clean_main.shape)} and {clean_main.dtype}"
+            )
+        if clean_main.device.type != "cpu":
+            raise ValueError("native edit CPU conditioning requires a CPU tensor")
+        if len(geometry.items) == 0:
+            raise ValueError("native edit conditioning requires at least one item")
+
+    def _sample_operator_plan(
+        self, rng: torch.Generator
+    ) -> tuple[tuple[str, ...], tuple[float, ...]]:
+        required = self.config.required_operators
+        if required:
+            minimum_count = max(self.config.min_operators, len(required))
+            operator_count = int(
+                torch.randint(
+                    minimum_count, self.config.max_operators + 1, (), generator=rng
+                ).item()
+            )
+            optional = tuple(
+                operator for operator in self.config.operators if operator not in required
+            )
+            optional_count = operator_count - len(required)
+            optional_permutation = torch.randperm(len(optional), generator=rng)[
+                :optional_count
+            ].tolist()
+            selected = required + tuple(optional[index] for index in optional_permutation)
+            order = torch.randperm(len(selected), generator=rng).tolist()
+            operators = tuple(selected[index] for index in order)
+        else:
+            operator_count = int(
+                torch.randint(
+                    self.config.min_operators, self.config.max_operators + 1, (), generator=rng
+                ).item()
+            )
+            permutation = torch.randperm(len(self.config.operators), generator=rng)[
+                :operator_count
+            ].tolist()
+            operators = tuple(self.config.operators[index] for index in permutation)
+        severities = tuple(
+            float(value)
+            for value in (
+                torch.rand(operator_count, generator=rng)
+                * (self.config.severity_max - self.config.severity_min)
+                + self.config.severity_min
+            ).tolist()
+        )
+        return (operators, severities)
+
+    def _footprint_crop(self, item: ItemGeometry) -> CropGeometry:
+        """Resolve the reference-space rectangle this item degrades."""
+        if self.config.degradation_region == "aligned_item_crop":
+            return item.crop
+        bbox = item.reference_bbox
+        top = math.floor(bbox.y0)
+        left = math.floor(bbox.x0)
+        return CropGeometry(
+            top=top,
+            left=left,
+            size=ImageSize(height=math.ceil(bbox.y1) - top, width=math.ceil(bbox.x1) - left),
+        )
+
+    @staticmethod
+    def _projected_main_box(
+        crop: CropGeometry, geometry: GroupGeometry, main_h: int, main_w: int
+    ) -> tuple[int, int, int, int]:
+        reference_h = geometry.reference.size.height
+        reference_w = geometry.reference.size.width
+        left = math.floor(crop.left * main_w / reference_w)
+        top = math.floor(crop.top * main_h / reference_h)
+        right = math.ceil((crop.left + crop.size.width) * main_w / reference_w)
+        bottom = math.ceil((crop.top + crop.size.height) * main_h / reference_h)
+        return (
+            min(max(left, 0), main_w),
+            min(max(top, 0), main_h),
+            min(max(right, 0), main_w),
+            min(max(bottom, 0), main_h),
+        )
+
+    def _replacement_footprint(self, geometry: GroupGeometry) -> np.ndarray:
+        main_h = geometry.main.size.height
+        main_w = geometry.main.size.width
+        mask = np.zeros((main_h, main_w), dtype=np.bool_)
+        for item in geometry.items:
+            (left, top, right, bottom) = self._projected_main_box(
+                self._footprint_crop(item), geometry, main_h, main_w
+            )
+            mask[top:bottom, left:right] = True
+        return mask
+
+    @staticmethod
+    def _apply_operator(
+        image: np.ndarray, operator: str, severity: float, seed: int, scale: float | None = None
+    ) -> np.ndarray:
+        if operator == "solid_black":
+            return np.zeros_like(image)
+        (height, width) = image.shape[:2]
+        if scale is None:
+            scale = min(height, width)
+        if operator == "gaussian_blur":
+            sigma = scale * (0.005 + 0.025 * severity)
+            radius = max(1, math.ceil(3 * sigma))
+            kernel = 2 * radius + 1
+            return cv2.GaussianBlur(
+                image,
+                (kernel, kernel),
+                sigmaX=sigma,
+                sigmaY=sigma,
+                borderType=cv2.BORDER_REFLECT_101,
+            )
+        if operator == "defocus":
+            radius = max(1, math.ceil(scale * (0.005 + 0.025 * severity)))
+            coordinates = np.arange(-radius, radius + 1, dtype=np.float32)
+            (grid_y, grid_x) = np.meshgrid(coordinates, coordinates)
+            kernel = (grid_x**2 + grid_y**2 <= radius**2).astype(np.float32)
+            kernel /= kernel.sum()
+            alias_sigma = 0.1 + 0.4 * severity
+            kernel = cv2.GaussianBlur(
+                kernel, (3, 3), sigmaX=alias_sigma, borderType=cv2.BORDER_REFLECT_101
+            )
+            kernel /= kernel.sum()
+            return cv2.filter2D(image, -1, kernel, borderType=cv2.BORDER_REFLECT_101)
+        raise ValueError(f"Unsupported blur operator: {operator}")