"""ImageRegion — a detected/proposed sub-region of an image + its crop embedding. The storage backbone of the crop pipeline (#114). A region is a normalized bbox plus the embedding of its crop: - kind='face' / 'figure' → embedded by CCIP for cross-artist character identity. - kind='concept' → embedded by SigLIP, a localized instance for a concept head's bag-of-embeddings (a concept is "present if ANY instance matches"). One row carries the embedding appropriate to its kind (the other is null). The bbox doubles as grounded-tag provenance (hover a tag → highlight its region; a wrong box is a precise negative). The GPU agent writes these via the job API; the few-shot character matcher + bag scorer read them — both server-side, no GPU. """ from datetime import datetime from pgvector.sqlalchemy import Vector from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, func from sqlalchemy.orm import Mapped, mapped_column from .base import Base CCIP_DIM = 768 # deepghs/imgutils CCIP character embedding SIGLIP_DIM = 1152 # matches image_record.siglip_embedding class ImageRegion(Base): __tablename__ = "image_region" id: Mapped[int] = mapped_column(Integer, primary_key=True) image_record_id: Mapped[int] = mapped_column( ForeignKey("image_record.id", ondelete="CASCADE"), index=True ) # 'frame' (a whole video frame → SigLIP bag) | 'face' | 'figure' (→ CCIP # character id) | 'concept' (→ SigLIP head bag) | 'panel' (a comic panel crop, # also SigLIP → the bag). Free String, not an enum — proposers can add kinds # without a migration; the bag scorer keys on a non-null siglip_embedding, not # the kind, so any SigLIP-embedded region joins the bag. kind: Mapped[str] = mapped_column(String(16), nullable=False) # For video/animated media: the source frame's timestamp in SECONDS. NULL for # static images. Lets a video be a BAG of per-frame instances (fixes the # mean-embedding muddle) + grounds a tag to "appears at 0:42". frame_time: Mapped[float | None] = mapped_column(Float, nullable=True) # Normalized bbox in [0,1]: top-left (rx, ry) + size (rw, rh). Named rx/ry/… # rather than x/y/by to dodge SQL keyword ambiguity ('by'). rx: Mapped[float] = mapped_column(Float, nullable=False) ry: Mapped[float] = mapped_column(Float, nullable=False) rw: Mapped[float] = mapped_column(Float, nullable=False) rh: Mapped[float] = mapped_column(Float, nullable=False) # Proposer/detector confidence (null for deterministic proposers). score: Mapped[float | None] = mapped_column(Float, nullable=True) # Version stamps so a re-detect / re-crop / re-embed can be gated (compute # once; only redo when the producing model version changes). detector_version: Mapped[str | None] = mapped_column(String(64), nullable=True) crop_version: Mapped[str | None] = mapped_column(String(64), nullable=True) embedding_version: Mapped[str | None] = mapped_column(String(128), nullable=True) # Exactly one is set, per kind. ccip_embedding: Mapped[list[float] | None] = mapped_column( Vector(CCIP_DIM), nullable=True ) siglip_embedding: Mapped[list[float] | None] = mapped_column( Vector(SIGLIP_DIM), nullable=True ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() )