"""Deterministic Petri-dish identity, vision, and process-state services.

This module deliberately keeps robot motion out of the image-processing path.
Vision produces reviewable observations and calibrated targets; a separately
commissioned motion controller decides whether movement is permitted.
"""

from __future__ import annotations

import binascii
import csv
import hashlib
import io
import json
import math
import os
import re
import sqlite3
import stat
import tempfile
import threading
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence
from typing import Tuple
from types import SimpleNamespace

_ID_PATTERN = re.compile(r"^[A-Z0-9][A-Z0-9._-]{2,63}$")
_ARTIFACT_COMPONENT_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
_QR_PREFIX = "labcobot:v1"
_CLUSTER_NAMESPACE = uuid.UUID("0aa47ca9-4254-492f-a5db-6654cc7bd96c")
_OBSERVATION_NAMESPACE = uuid.UUID("510a9ea4-b8de-48b7-856d-2295ef8ac0b7")
_MAX_PENDING_REVIEW_IMAGE_BYTES = 8 * 1024 * 1024
_MAX_PENDING_REVIEW_IMAGE_DIMENSION = 8192
_MAX_PENDING_REVIEW_IMAGE_PIXELS = 40_000_000
_MAX_PNG_CHUNKS = 1024
_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"


class PetriWorkflowError(RuntimeError):
    """Base class for rejected Petri workflow operations."""


class IdentityError(PetriWorkflowError):
    """Raised when a plate identity or QR payload is invalid."""


class VisionReviewRequired(PetriWorkflowError):
    """Raised when an operation requires an explicit human selection."""


@dataclass(frozen=True)
class PetriIdentity:
    """A non-personal, typo-detecting identity encoded in a QR payload."""

    plate_id: str
    qr_payload: str
    label: str
    created_at: str


@dataclass(frozen=True)
class CalibratedPetriTarget:
    """One reviewed image segment mapped into the robot base frame."""

    observation_id: str
    cluster_id: str
    workspace_xyz_mm: Tuple[float, float, float]
    image_xy: Tuple[float, float]
    confidence: float
    captured_at_s: float


@dataclass(frozen=True)
class MotionPose:
    """One finite xArm TCP pose in millimetres and degrees."""

    x_mm: float
    y_mm: float
    z_mm: float
    roll_deg: float
    pitch_deg: float
    yaw_deg: float

    @classmethod
    def from_values(cls, values: Sequence[Any], name: str) -> "MotionPose":
        vector = _finite_vector(values, 6, name)
        return cls(*vector)

    def to_tuple(self) -> Tuple[float, ...]:
        return (
            self.x_mm,
            self.y_mm,
            self.z_mm,
            self.roll_deg,
            self.pitch_deg,
            self.yaw_deg,
        )


@dataclass(frozen=True)
class WorkcellMotionProfile:
    """Validated commissioning data for the physical Petri workflow.

    Loading this object never moves hardware. The live controller must still
    apply its own joint, workspace, collision, velocity, and operator gates.
    """

    profile_id: str
    home_joints_deg: Tuple[float, ...]
    suction_start: MotionPose
    suction_end: MotionPose
    photo_pose: MotionPose
    transit_z_mm: float
    workspace_min_mm: Tuple[float, float, float]
    workspace_max_mm: Tuple[float, float, float]
    max_speed_mm_s: float
    max_acceleration_mm_s2: float

    @classmethod
    def from_dict(cls, raw: Mapping[str, Any]) -> "WorkcellMotionProfile":
        profile_id = _validated_id(raw.get("profile_id"), "profile_id")
        home = _finite_vector(raw.get("home_joints_deg"), 6, "home_joints_deg")
        workspace_min = _finite_vector(
            raw.get("workspace_min_mm"),
            3,
            "workspace_min_mm",
        )
        workspace_max = _finite_vector(
            raw.get("workspace_max_mm"),
            3,
            "workspace_max_mm",
        )
        if any(low >= high for low, high in zip(workspace_min, workspace_max)):
            raise PetriWorkflowError("workspace_min_mm must be below workspace_max_mm")
        poses = {
            name: MotionPose.from_values(raw.get(name), name)
            for name in ("suction_start", "suction_end", "photo_pose")
        }
        for name, pose in poses.items():
            for value, low, high, axis in zip(
                pose.to_tuple()[:3],
                workspace_min,
                workspace_max,
                ("x", "y", "z"),
            ):
                if not low <= value <= high:
                    raise PetriWorkflowError(
                        "{} {} coordinate is outside the commissioned "
                        "workspace".format(name, axis)
                    )
        transit_z = _finite_float(raw.get("transit_z_mm"), "transit_z_mm")
        highest_task_z = max(pose.z_mm for pose in poses.values())
        if transit_z < highest_task_z + 20.0:
            raise PetriWorkflowError(
                "transit_z_mm must provide at least 20 mm clearance"
            )
        if not workspace_min[2] <= transit_z <= workspace_max[2]:
            raise PetriWorkflowError(
                "transit_z_mm is outside the commissioned workspace"
            )
        speed = _positive_float(
            raw.get("max_speed_mm_s"),
            "max_speed_mm_s",
        )
        acceleration = _positive_float(
            raw.get("max_acceleration_mm_s2"),
            "max_acceleration_mm_s2",
        )
        if any(abs(angle) > 360.0 for angle in home):
            raise PetriWorkflowError("home_joints_deg contains an invalid angle")
        return cls(
            profile_id=profile_id,
            home_joints_deg=home,
            suction_start=poses["suction_start"],
            suction_end=poses["suction_end"],
            photo_pose=poses["photo_pose"],
            transit_z_mm=transit_z,
            workspace_min_mm=workspace_min,
            workspace_max_mm=workspace_max,
            max_speed_mm_s=speed,
            max_acceleration_mm_s2=acceleration,
        )

    @classmethod
    def load(cls, path: os.PathLike[str]) -> "WorkcellMotionProfile":
        source = Path(path)
        if source.is_symlink() or not source.is_file():
            raise PetriWorkflowError("motion profile must be a regular file")
        try:
            raw = json.loads(source.read_text(encoding="utf-8"))
        except (OSError, ValueError) as exc:
            raise PetriWorkflowError("motion profile is not valid JSON") from exc
        if not isinstance(raw, Mapping):
            raise PetriWorkflowError("motion profile must be a JSON object")
        return cls.from_dict(raw)


class PetriRegistry:
    """SQLite-backed plate identities, stable clusters, and observations."""

    def __init__(self, path: os.PathLike[str]) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.RLock()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(str(self.path), timeout=10.0)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 10000")
        return connection

    def _initialize(self) -> None:
        with self._lock, self._connect() as connection:
            connection.executescript("""
                PRAGMA journal_mode = WAL;
                CREATE TABLE IF NOT EXISTS petri_plates (
                    plate_id TEXT PRIMARY KEY,
                    qr_payload TEXT NOT NULL UNIQUE,
                    label TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS cell_clusters (
                    cluster_id TEXT PRIMARY KEY,
                    plate_id TEXT NOT NULL,
                    center_x REAL NOT NULL,
                    center_y REAL NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY (plate_id)
                        REFERENCES petri_plates (plate_id)
                );
                CREATE INDEX IF NOT EXISTS cluster_plate_idx
                    ON cell_clusters (plate_id, updated_at);
                CREATE TABLE IF NOT EXISTS cell_observations (
                    observation_id TEXT PRIMARY KEY,
                    cluster_id TEXT NOT NULL,
                    plate_id TEXT NOT NULL,
                    captured_at TEXT NOT NULL,
                    center_x REAL NOT NULL,
                    center_y REAL NOT NULL,
                    radius_px REAL NOT NULL,
                    area_px2 REAL NOT NULL,
                    confidence REAL NOT NULL,
                    status TEXT NOT NULL,
                    selected INTEGER NOT NULL DEFAULT 0,
                    crop_path TEXT NOT NULL,
                    annotated_path TEXT NOT NULL,
                    metadata_json TEXT NOT NULL,
                    FOREIGN KEY (cluster_id)
                        REFERENCES cell_clusters (cluster_id),
                    FOREIGN KEY (plate_id)
                        REFERENCES petri_plates (plate_id)
                );
                CREATE INDEX IF NOT EXISTS observation_history_idx
                    ON cell_observations (cluster_id, captured_at);
                """)

    def create_plate(
        self,
        plate_id: Optional[str] = None,
        *,
        label: str = "",
        created_at: Optional[str] = None,
    ) -> PetriIdentity:
        normalized = (
            _validated_id(plate_id, "plate_id")
            if plate_id is not None
            else "PLATE-{}".format(uuid.uuid4().hex[:12].upper())
        )
        timestamp = _timestamp(created_at)
        qr_payload = encode_plate_qr(normalized)
        safe_label = str(label).strip()[:200]
        with self._lock, self._connect() as connection:
            connection.execute(
                """
                INSERT INTO petri_plates (
                    plate_id, qr_payload, label, created_at
                ) VALUES (?, ?, ?, ?)
                ON CONFLICT(plate_id) DO UPDATE SET label = excluded.label
                """,
                (normalized, qr_payload, safe_label, timestamp),
            )
            row = connection.execute(
                "SELECT * FROM petri_plates WHERE plate_id = ?",
                (normalized,),
            ).fetchone()
        assert row is not None
        return PetriIdentity(**dict(row))

    def resolve_qr(self, qr_payload: str) -> PetriIdentity:
        plate_id = decode_plate_qr(qr_payload)
        with self._lock, self._connect() as connection:
            row = connection.execute(
                "SELECT * FROM petri_plates WHERE plate_id = ?",
                (plate_id,),
            ).fetchone()
        if row is None:
            raise IdentityError("QR plate is not registered")
        return PetriIdentity(**dict(row))

    def export_plate_table(self) -> str:
        output = io.StringIO()
        writer = csv.DictWriter(
            output,
            fieldnames=("plate_id", "qr_payload", "label", "created_at"),
        )
        writer.writeheader()
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                "SELECT * FROM petri_plates ORDER BY created_at, plate_id"
            ).fetchall()
        for row in rows:
            writer.writerow(dict(row))
        return output.getvalue()

    def import_plate_table(self, csv_text: str) -> int:
        reader = csv.DictReader(io.StringIO(str(csv_text)))
        required = {"plate_id", "qr_payload", "label", "created_at"}
        if not required.issubset(set(reader.fieldnames or ())):
            raise IdentityError("QR table is missing required columns")
        accepted = 0
        for row in reader:
            plate_id = decode_plate_qr(row["qr_payload"])
            if plate_id != _validated_id(row["plate_id"], "plate_id"):
                raise IdentityError("QR payload does not match plate_id")
            self.create_plate(
                plate_id,
                label=row.get("label") or "",
                created_at=row.get("created_at") or None,
            )
            accepted += 1
        return accepted

    def assign_cluster(
        self,
        plate_id: str,
        center_x: float,
        center_y: float,
        *,
        captured_at: str,
        match_radius_px: float,
    ) -> str:
        x_value = _finite_float(center_x, "center_x")
        y_value = _finite_float(center_y, "center_y")
        radius = _positive_float(match_radius_px, "match_radius_px")
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT cluster_id, center_x, center_y
                FROM cell_clusters WHERE plate_id = ?
                """,
                (plate_id,),
            ).fetchall()
            nearest = None
            for row in rows:
                distance = math.hypot(
                    x_value - float(row["center_x"]),
                    y_value - float(row["center_y"]),
                )
                if distance <= radius and (nearest is None or distance < nearest[0]):
                    nearest = (distance, str(row["cluster_id"]))
            if nearest is None:
                cluster_id = "CELL-{}".format(
                    uuid.uuid5(
                        _CLUSTER_NAMESPACE,
                        "{}:{:.3f}:{:.3f}".format(
                            plate_id,
                            x_value,
                            y_value,
                        ),
                    )
                    .hex[:16]
                    .upper()
                )
                connection.execute(
                    """
                    INSERT OR IGNORE INTO cell_clusters (
                        cluster_id, plate_id, center_x, center_y,
                        created_at, updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?)
                    """,
                    (
                        cluster_id,
                        plate_id,
                        x_value,
                        y_value,
                        captured_at,
                        captured_at,
                    ),
                )
            else:
                cluster_id = nearest[1]
                connection.execute(
                    """
                    UPDATE cell_clusters
                    SET center_x = ?, center_y = ?, updated_at = ?
                    WHERE cluster_id = ?
                    """,
                    (x_value, y_value, captured_at, cluster_id),
                )
        return cluster_id

    def add_observation(self, observation: Mapping[str, Any]) -> None:
        metadata = json.dumps(
            dict(observation.get("metadata") or {}),
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        )
        with self._lock, self._connect() as connection:
            connection.execute(
                """
                INSERT OR IGNORE INTO cell_observations (
                    observation_id, cluster_id, plate_id, captured_at,
                    center_x, center_y, radius_px, area_px2, confidence,
                    status, selected, crop_path, annotated_path, metadata_json
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    observation["observation_id"],
                    observation["cluster_id"],
                    observation["plate_id"],
                    observation["captured_at"],
                    observation["center_x"],
                    observation["center_y"],
                    observation["radius_px"],
                    observation["area_px2"],
                    observation["confidence"],
                    observation["status"],
                    1 if observation.get("selected") else 0,
                    observation["crop_path"],
                    observation["annotated_path"],
                    metadata,
                ),
            )

    def select_observation(self, observation_id: str) -> Mapping[str, Any]:
        normalized = _validated_id(observation_id, "observation_id")
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            updated = connection.execute(
                """
                UPDATE cell_observations
                SET selected = 1, status = 'accepted_manual'
                WHERE observation_id = ?
                  AND status = 'requires_review'
                  AND selected = 0
                """,
                (normalized,),
            )
            if updated.rowcount != 1:
                connection.rollback()
                raise VisionReviewRequired("observation is not pending manual review")
            row = connection.execute(
                """
                SELECT * FROM cell_observations
                WHERE observation_id = ?
                """,
                (normalized,),
            ).fetchone()
            if row is None:
                connection.rollback()
                raise VisionReviewRequired(
                    "observation selection could not be verified"
                )
            connection.commit()
        result = _observation_row(row)
        return result

    def operations_summary(self) -> Mapping[str, int]:
        """Return fixed aggregate counters for the operator status surface."""

        with self._lock, self._connect() as connection:
            row = connection.execute("""
                SELECT
                    (SELECT COUNT(*) FROM petri_plates) AS plate_count,
                    (SELECT COUNT(*) FROM cell_clusters) AS cluster_count,
                    (SELECT COUNT(*) FROM cell_observations)
                        AS observation_count,
                    (SELECT COUNT(*) FROM cell_observations
                        WHERE status = 'requires_review' AND selected = 0)
                        AS pending_review_count,
                    (SELECT COUNT(*) FROM cell_observations
                        WHERE status = 'accepted_manual' AND selected = 1)
                        AS accepted_manual_count
                """).fetchone()
        if row is None:
            raise PetriWorkflowError("registry summary is unavailable")
        return {
            "plate_count": int(row["plate_count"]),
            "cluster_count": int(row["cluster_count"]),
            "observation_count": int(row["observation_count"]),
            "pending_review_count": int(row["pending_review_count"]),
            "accepted_manual_count": int(row["accepted_manual_count"]),
        }

    def operations_snapshot(self, limit: int = 20) -> Mapping[str, Any]:
        """Return aggregate counts and bounded recent rows atomically.

        The public snapshot intentionally selects a closed column list. Local
        artifact paths and unbounded observation metadata never cross this
        backend boundary.
        """

        bounded_limit = _bounded_registry_limit(
            limit,
            "recent history limit",
            maximum=20,
        )
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN")
            summary = connection.execute("""
                SELECT
                    (SELECT COUNT(*) FROM petri_plates) AS plate_count,
                    (SELECT COUNT(*) FROM cell_clusters) AS cluster_count,
                    (SELECT COUNT(*) FROM cell_observations)
                        AS observation_count,
                    (SELECT COUNT(*) FROM cell_observations
                        WHERE status = 'requires_review' AND selected = 0)
                        AS pending_review_count,
                    (SELECT COUNT(*) FROM cell_observations
                        WHERE status = 'accepted_manual' AND selected = 1)
                        AS accepted_manual_count
                """).fetchone()
            rows = connection.execute(
                """
                SELECT observation_id, cluster_id, plate_id, captured_at,
                       area_px2, confidence, status, selected
                FROM cell_observations
                ORDER BY captured_at DESC, observation_id DESC
                LIMIT ?
                """,
                (bounded_limit,),
            ).fetchall()
            connection.commit()
        if summary is None:
            raise PetriWorkflowError("registry operations snapshot is unavailable")
        return {
            "schema_version": 1,
            "plate_count": int(summary["plate_count"]),
            "cluster_count": int(summary["cluster_count"]),
            "observation_count": int(summary["observation_count"]),
            "pending_review_count": int(summary["pending_review_count"]),
            "accepted_manual_count": int(summary["accepted_manual_count"]),
            "recent": [
                {
                    "observation_id": str(row["observation_id"]),
                    "cluster_id": str(row["cluster_id"]),
                    "plate_id": str(row["plate_id"]),
                    "captured_at": str(row["captured_at"]),
                    "area_px2": float(row["area_px2"]),
                    "confidence": float(row["confidence"]),
                    "status": str(row["status"]),
                    "selected": bool(row["selected"]),
                }
                for row in rows
            ],
        }

    def pending_reviews(self, limit: int = 8) -> List[Mapping[str, Any]]:
        """Return at most eight oldest unresolved review rows without paths."""

        bounded_limit = _bounded_registry_limit(
            limit,
            "pending review limit",
            maximum=8,
        )
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT observation_id, cluster_id, plate_id, captured_at,
                       area_px2, confidence, status
                FROM cell_observations
                WHERE status = 'requires_review' AND selected = 0
                ORDER BY captured_at ASC, observation_id ASC
                LIMIT ?
                """,
                (bounded_limit,),
            ).fetchall()
        return [
            {
                "observation_id": str(row["observation_id"]),
                "cluster_id": str(row["cluster_id"]),
                "plate_id": str(row["plate_id"]),
                "captured_at": str(row["captured_at"]),
                "area_px2": float(row["area_px2"]),
                "confidence": float(row["confidence"]),
                "status": str(row["status"]),
            }
            for row in rows
        ]

    def pending_review_image(
        self,
        observation_id: str,
        artifact_root: os.PathLike[str],
        max_bytes: int = _MAX_PENDING_REVIEW_IMAGE_BYTES,
    ) -> bytes:
        """Read one pending-review PNG through a descriptor-relative path.

        The immediate transaction serializes this read with manual selection.
        It also keeps the database-only path private: callers receive either
        verified PNG bytes or a bounded domain error, never a local path.
        """

        normalized = _validated_id(observation_id, "observation_id")
        bounded_max = _bounded_registry_limit(
            max_bytes,
            "pending review image max_bytes",
            maximum=_MAX_PENDING_REVIEW_IMAGE_BYTES,
        )
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            try:
                row = connection.execute(
                    """
                    SELECT crop_path FROM cell_observations
                    WHERE observation_id = ?
                      AND status = 'requires_review'
                      AND selected = 0
                    """,
                    (normalized,),
                ).fetchone()
                if row is None:
                    raise VisionReviewRequired(
                        "observation is not pending manual review"
                    )
                payload = _read_pending_review_png(
                    artifact_root,
                    row["crop_path"],
                    max_bytes=bounded_max,
                )
                still_pending = connection.execute(
                    """
                    SELECT 1 FROM cell_observations
                    WHERE observation_id = ?
                      AND status = 'requires_review'
                      AND selected = 0
                    """,
                    (normalized,),
                ).fetchone()
                if still_pending is None:
                    raise VisionReviewRequired(
                        "observation review state changed during image read"
                    )
                connection.commit()
            except Exception:
                connection.rollback()
                raise
        return payload

    def recent_history(self, limit: int = 20) -> List[Mapping[str, Any]]:
        """Return at most twenty newest observation rows without local paths."""

        bounded_limit = _bounded_registry_limit(
            limit,
            "recent history limit",
            maximum=20,
        )
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT observation_id, cluster_id, plate_id, captured_at,
                       area_px2, confidence, status, selected
                FROM cell_observations
                ORDER BY captured_at DESC, observation_id DESC
                LIMIT ?
                """,
                (bounded_limit,),
            ).fetchall()
        return [
            {
                "observation_id": str(row["observation_id"]),
                "cluster_id": str(row["cluster_id"]),
                "plate_id": str(row["plate_id"]),
                "captured_at": str(row["captured_at"]),
                "area_px2": float(row["area_px2"]),
                "confidence": float(row["confidence"]),
                "status": str(row["status"]),
                "selected": bool(row["selected"]),
            }
            for row in rows
        ]

    def history(self, cluster_id: str) -> List[Mapping[str, Any]]:
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT * FROM cell_observations
                WHERE cluster_id = ?
                ORDER BY captured_at ASC, observation_id ASC
                """,
                (str(cluster_id),),
            ).fetchall()
        return [_observation_row(row) for row in rows]

    def cluster_dashboard(self, plate_id: str) -> List[Mapping[str, Any]]:
        with self._lock, self._connect() as connection:
            rows = connection.execute(
                """
                SELECT
                    c.cluster_id,
                    c.center_x,
                    c.center_y,
                    c.created_at,
                    c.updated_at,
                    COUNT(o.observation_id) AS observation_count,
                    MAX(o.area_px2) AS latest_area_px2,
                    AVG(o.confidence) AS average_confidence,
                    SUM(CASE WHEN o.status = 'requires_review'
                        THEN 1 ELSE 0 END) AS review_count
                FROM cell_clusters c
                LEFT JOIN cell_observations o
                    ON o.cluster_id = c.cluster_id
                WHERE c.plate_id = ?
                GROUP BY c.cluster_id
                ORDER BY c.updated_at DESC, c.cluster_id ASC
                """,
                (_validated_id(plate_id, "plate_id"),),
            ).fetchall()
        return [dict(row) for row in rows]

    def export_history_csv(
        self,
        *,
        cluster_id: Optional[str] = None,
        plate_id: Optional[str] = None,
    ) -> str:
        if not cluster_id and not plate_id:
            raise PetriWorkflowError("cluster_id or plate_id is required")
        clauses = []
        parameters: List[str] = []
        if cluster_id:
            clauses.append("cluster_id = ?")
            parameters.append(str(cluster_id))
        if plate_id:
            clauses.append("plate_id = ?")
            parameters.append(_validated_id(plate_id, "plate_id"))
        query = (
            "SELECT * FROM cell_observations WHERE "
            + " AND ".join(clauses)
            + " ORDER BY captured_at ASC, observation_id ASC"
        )
        with self._lock, self._connect() as connection:
            rows = connection.execute(query, parameters).fetchall()
        fieldnames = (
            "observation_id",
            "cluster_id",
            "plate_id",
            "captured_at",
            "center_x",
            "center_y",
            "radius_px",
            "area_px2",
            "confidence",
            "status",
            "selected",
            "crop_path",
            "annotated_path",
        )
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            record = dict(row)
            writer.writerow({key: record[key] for key in fieldnames})
        return output.getvalue()


class PetriVisionWorkflow:
    """Create bounded image artifacts and stable cluster observations."""

    def __init__(
        self,
        registry: PetriRegistry,
        artifact_root: os.PathLike[str],
        *,
        confidence_threshold: float = 0.72,
    ) -> None:
        self.registry = registry
        self.artifact_root = Path(artifact_root)
        self.artifact_root.mkdir(parents=True, exist_ok=True)
        threshold = _finite_float(
            confidence_threshold,
            "confidence_threshold",
        )
        if not 0.0 < threshold <= 1.0:
            raise ValueError("confidence_threshold must be in (0, 1]")
        self.confidence_threshold = threshold

    def process_image(
        self,
        image_bytes: bytes,
        *,
        plate_qr: str,
        detections: Iterable[Mapping[str, Any]],
        captured_at: Optional[str] = None,
        source_role: str = "tool",
    ) -> Mapping[str, Any]:
        if not image_bytes or len(image_bytes) > 40 * 1024 * 1024:
            raise PetriWorkflowError("image must contain 1 byte to 40 MiB")
        identity = self.registry.resolve_qr(plate_qr)
        timestamp = _timestamp(captured_at)
        normalized = [
            _normalize_detection(detection) for detection in list(detections)[:256]
        ]
        image, draw_module = _load_image(image_bytes)
        width, height = image.size
        for detection in normalized:
            _validate_detection_bounds(detection, width, height)
        ambiguous = _ambiguous_detection_indexes(normalized)
        annotated = image.copy()
        draw = draw_module.Draw(annotated)

        run_key = hashlib.sha256(
            "{}:{}:{}".format(
                identity.plate_id,
                timestamp,
                hashlib.sha256(image_bytes).hexdigest(),
            ).encode("utf-8")
        ).hexdigest()[:20]
        run_dir = self.artifact_root / identity.plate_id / run_key
        crop_dir = run_dir / "crops"
        crop_dir.mkdir(parents=True, exist_ok=True)
        observations: List[Mapping[str, Any]] = []

        for index, detection in enumerate(normalized):
            x_value = detection["x"]
            y_value = detection["y"]
            radius = detection["radius"]
            confidence = detection["confidence"]
            requires_review = (
                confidence < self.confidence_threshold or index in ambiguous
            )
            status = "requires_review" if requires_review else "accepted_auto"
            cluster_id = self.registry.assign_cluster(
                identity.plate_id,
                x_value,
                y_value,
                captured_at=timestamp,
                match_radius_px=max(25.0, radius * 0.75),
            )
            observation_id = "OBS-{}".format(
                uuid.uuid5(
                    _OBSERVATION_NAMESPACE,
                    "{}:{}:{:.3f}:{:.3f}".format(
                        cluster_id,
                        timestamp,
                        x_value,
                        y_value,
                    ),
                )
                .hex[:20]
                .upper()
            )
            bounds = (
                max(0, int(math.floor(x_value - radius))),
                max(0, int(math.floor(y_value - radius))),
                min(width, int(math.ceil(x_value + radius))),
                min(height, int(math.ceil(y_value + radius))),
            )
            crop_relative = Path(identity.plate_id) / run_key / "crops"
            crop_relative = crop_relative / "{}.png".format(observation_id)
            crop_path = self.artifact_root / crop_relative
            crop = image.crop(bounds)
            _save_png_atomic(crop, crop_path)
            color = "#ff9f1c" if requires_review else "#e02020"
            draw.rectangle(bounds, outline=color, width=4)
            draw.ellipse(bounds, outline=color, width=3)
            record: Dict[str, Any] = {
                "observation_id": observation_id,
                "cluster_id": cluster_id,
                "plate_id": identity.plate_id,
                "captured_at": timestamp,
                "center_x": x_value,
                "center_y": y_value,
                "radius_px": radius,
                "area_px2": math.pi * radius * radius,
                "confidence": confidence,
                "status": status,
                "selected": False,
                "crop_path": crop_relative.as_posix(),
                "annotated_path": "",
                "bounding_box": {
                    "x_min": bounds[0],
                    "y_min": bounds[1],
                    "x_max": bounds[2],
                    "y_max": bounds[3],
                },
                "metadata": {
                    "source_role": str(source_role),
                    "requires_review": requires_review,
                    "raw": detection.get("raw", {}),
                },
            }
            observations.append(record)

        annotated_relative = Path(identity.plate_id) / run_key / "annotated.png"
        annotated_path = self.artifact_root / annotated_relative
        _save_png_atomic(annotated, annotated_path)
        for observation in observations:
            observation["annotated_path"] = annotated_relative.as_posix()
            self.registry.add_observation(observation)

        manifest: Dict[str, Any] = {
            "schema": "labcobot.petri-segmentation/v1",
            "plate_id": identity.plate_id,
            "captured_at": timestamp,
            "source_role": str(source_role),
            "image_width": width,
            "image_height": height,
            "annotated_image_url": annotated_relative.as_posix(),
            "image_urls": [
                annotated_relative.as_posix(),
                *[str(item["crop_path"]) for item in observations],
            ],
            "segments": observations,
            "requires_review": [
                item["observation_id"]
                for item in observations
                if item["status"] == "requires_review"
            ],
        }
        manifest_relative = Path(identity.plate_id) / run_key / "manifest.json"
        _write_json_atomic(self.artifact_root / manifest_relative, manifest)
        manifest["manifest_url"] = manifest_relative.as_posix()
        return manifest


@dataclass(frozen=True)
class ProcessSnapshot:
    """Reader-facing progress state used by the Raspberry Pi UI."""

    state: str
    completed_steps: int
    total_steps: int
    progress_percent: float
    elapsed_seconds: float
    eta_seconds: Optional[float]
    message: str
    error_code: str
    recovery_required: bool


class ProcessRunTracker:
    """Thread-safe progress, ETA, cancellation, and recovery state."""

    def __init__(self, clock: Callable[[], float] = time.monotonic) -> None:
        self.clock = clock
        self._lock = threading.RLock()
        self._state = "idle"
        self._started_at: Optional[float] = None
        self._completed_steps = 0
        self._total_steps = 0
        self._estimated_seconds: Optional[float] = None
        self._message = "Ready"
        self._error_code = ""
        self._recovery_required = False

    def start(
        self,
        total_steps: int,
        *,
        estimated_seconds: Optional[float] = None,
    ) -> ProcessSnapshot:
        if isinstance(total_steps, bool) or int(total_steps) <= 0:
            raise ValueError("total_steps must be positive")
        estimate = (
            _positive_float(estimated_seconds, "estimated_seconds")
            if estimated_seconds is not None
            else None
        )
        with self._lock:
            if self._state == "running":
                raise PetriWorkflowError("process is already running")
            self._state = "running"
            self._started_at = float(self.clock())
            self._completed_steps = 0
            self._total_steps = int(total_steps)
            self._estimated_seconds = estimate
            self._message = "Process started"
            self._error_code = ""
            self._recovery_required = False
            return self.snapshot()

    def advance(self, message: str = "") -> ProcessSnapshot:
        with self._lock:
            if self._state != "running":
                raise PetriWorkflowError("process is not running")
            self._completed_steps = min(
                self._total_steps,
                self._completed_steps + 1,
            )
            if message:
                self._message = str(message)
            if self._completed_steps >= self._total_steps:
                self._state = "completed"
                self._message = message or "Process completed"
            return self.snapshot()

    def complete(self, message: str = "Process completed") -> ProcessSnapshot:
        """Complete a running process when its controller returns successfully."""

        with self._lock:
            if self._state != "running":
                return self.snapshot()
            self._completed_steps = self._total_steps
            self._state = "completed"
            self._message = str(message)[:400]
            return self.snapshot()

    def cancel(self, message: str = "Cancelled by operator") -> ProcessSnapshot:
        with self._lock:
            if self._state == "running":
                self._state = "cancelled"
                self._message = str(message)
            return self.snapshot()

    def fail(
        self,
        error_code: str,
        message: str,
        *,
        recovery_required: bool = True,
    ) -> ProcessSnapshot:
        with self._lock:
            self._state = "fault"
            self._error_code = str(error_code)[:80]
            self._message = str(message)[:400]
            self._recovery_required = bool(recovery_required)
            return self.snapshot()

    def mark_recovered(self) -> ProcessSnapshot:
        with self._lock:
            if self._state != "fault":
                raise PetriWorkflowError("no process fault requires recovery")
            self._state = "idle"
            self._started_at = None
            self._completed_steps = 0
            self._total_steps = 0
            self._estimated_seconds = None
            self._message = "Recovery confirmed; motion remains locked"
            self._error_code = ""
            self._recovery_required = False
            return self.snapshot()

    def snapshot(self) -> ProcessSnapshot:
        with self._lock:
            elapsed = (
                max(0.0, float(self.clock()) - self._started_at)
                if self._started_at is not None
                else 0.0
            )
            progress = (
                100.0 * self._completed_steps / self._total_steps
                if self._total_steps
                else 0.0
            )
            eta = None
            if self._state == "running":
                if self._completed_steps > 0:
                    per_step = elapsed / self._completed_steps
                    eta = per_step * (self._total_steps - self._completed_steps)
                elif self._estimated_seconds is not None:
                    eta = max(0.0, self._estimated_seconds - elapsed)
            return ProcessSnapshot(
                state=self._state,
                completed_steps=self._completed_steps,
                total_steps=self._total_steps,
                progress_percent=progress,
                elapsed_seconds=elapsed,
                eta_seconds=eta,
                message=self._message,
                error_code=self._error_code,
                recovery_required=self._recovery_required,
            )


def error_guidance(error_code: str) -> str:
    """Return bounded, deterministic operator help without remote code execution."""

    guidance = {
        "E_VISION_UNCERTAIN": (
            "Keep the tray still, verify even lighting, then select the correct "
            "orange cell-field outline manually."
        ),
        "E_CAMERA": (
            "Check camera power and lens visibility. The process remains stopped "
            "until both camera health indicators are running."
        ),
        "E_MOTION": (
            "Use the physical emergency stop if anything is moving. Inspect and "
            "clear the controller fault before requesting reviewed recovery."
        ),
        "E_NETWORK": (
            "Work can continue offline. Images and metrics stay in the durable "
            "outbox and synchronize when the network returns."
        ),
        "E_UI_ACTION": (
            "Read the device log and service status. Do not arm motion until the "
            "underlying fault has been inspected."
        ),
    }
    return guidance.get(
        str(error_code),
        "Keep motion locked, note the error code, and inspect the service status.",
    )


def map_segments_to_workspace(
    manifest: Mapping[str, Any],
    mapper: Any,
    *,
    captured_at_s: float,
    camera_role: str = "tool",
) -> List[CalibratedPetriTarget]:
    """Map accepted segments through a synchronized camera calibration.

    The caller must supply the exact frame timestamp used by the depth and
    robot-pose buffers. A stale timestamp, missing depth, unapproved hand-eye
    calibration, camera mismatch, or out-of-workspace point is rejected by the
    mapper. Review-required segments are never converted into motion targets.
    """

    timestamp = _finite_float(captured_at_s, "captured_at_s")
    readiness = getattr(mapper, "is_ready", None)
    if not callable(readiness) or readiness() is not True:
        raise PetriWorkflowError("camera workspace calibration is not ready")
    segments = manifest.get("segments")
    if not isinstance(segments, list):
        raise PetriWorkflowError("segmentation manifest has no segment list")
    targets = []
    for segment in segments:
        if not isinstance(segment, Mapping):
            raise PetriWorkflowError("segmentation manifest contains invalid data")
        status = str(segment.get("status") or "")
        if status == "requires_review":
            continue
        if status not in {"accepted_auto", "accepted_manual"}:
            raise PetriWorkflowError("segment has an unsupported review status")
        if status == "accepted_manual" and not bool(segment.get("selected")):
            continue
        center = (
            _finite_float(segment.get("center_x"), "segment.center_x"),
            _finite_float(segment.get("center_y"), "segment.center_y"),
        )
        track = SimpleNamespace(
            camera_role=str(camera_role),
            center_xy=center,
            captured_at_s=timestamp,
        )
        try:
            workspace = tuple(float(value) for value in mapper.to_workspace(track))
        except Exception as exc:
            raise PetriWorkflowError(
                "calibrated image-to-workspace mapping was rejected"
            ) from exc
        if len(workspace) != 3 or not all(math.isfinite(value) for value in workspace):
            raise PetriWorkflowError(
                "calibrated image-to-workspace mapping returned invalid data"
            )
        targets.append(
            CalibratedPetriTarget(
                observation_id=str(segment["observation_id"]),
                cluster_id=str(segment["cluster_id"]),
                workspace_xyz_mm=workspace,
                image_xy=center,
                confidence=_finite_float(
                    segment.get("confidence"),
                    "segment.confidence",
                ),
                captured_at_s=timestamp,
            )
        )
    return targets


def encode_plate_qr(plate_id: str) -> str:
    normalized = _validated_id(plate_id, "plate_id")
    checksum = hashlib.sha256(
        "{}:{}".format(_QR_PREFIX, normalized).encode("utf-8")
    ).hexdigest()[:12]
    return "{}:{}:{}".format(_QR_PREFIX, normalized, checksum)


def render_qr_label(
    identity: PetriIdentity,
    output_path: os.PathLike[str],
    *,
    box_size: int = 12,
    border: int = 4,
) -> Path:
    """Render one high-contrast printable PNG label with a human-readable ID."""

    if isinstance(box_size, bool) or not 4 <= int(box_size) <= 40:
        raise ValueError("box_size must be between 4 and 40")
    if isinstance(border, bool) or not 4 <= int(border) <= 12:
        raise ValueError("border must be between 4 and 12")
    try:
        import qrcode
        from PIL import Image, ImageDraw
    except ImportError as exc:
        raise PetriWorkflowError(
            "qrcode and Pillow are required to render printable labels"
        ) from exc
    qr = qrcode.QRCode(
        version=None,
        error_correction=qrcode.constants.ERROR_CORRECT_Q,
        box_size=int(box_size),
        border=int(border),
    )
    qr.add_data(identity.qr_payload)
    qr.make(fit=True)
    qr_image = qr.make_image(fill_color="black", back_color="white")
    qr_image = qr_image.convert("RGB")
    label_height = 72
    canvas = Image.new(
        "RGB",
        (qr_image.width, qr_image.height + label_height),
        "white",
    )
    canvas.paste(qr_image, (0, 0))
    draw = ImageDraw.Draw(canvas)
    display = identity.plate_id
    if identity.label:
        display = "{}  {}".format(display, identity.label)
    draw.text((12, qr_image.height + 12), display[:80], fill="black")
    target = Path(output_path)
    if target.suffix.lower() != ".png":
        raise ValueError("QR label output must use a .png suffix")
    _save_png_atomic(canvas, target)
    return target


def decode_plate_qr(payload: str) -> str:
    parts = str(payload).strip().split(":")
    if len(parts) != 4 or ":".join(parts[:2]) != _QR_PREFIX:
        raise IdentityError("QR payload has an unsupported format")
    plate_id = _validated_id(parts[2], "plate_id")
    if parts[3] != encode_plate_qr(plate_id).rsplit(":", 1)[1]:
        raise IdentityError("QR payload checksum does not match")
    return plate_id


def _timestamp(value: Optional[str]) -> str:
    if value is None:
        return datetime.now(timezone.utc).isoformat()
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError as exc:
        raise PetriWorkflowError("timestamp must be ISO-8601") from exc
    if parsed.tzinfo is None:
        raise PetriWorkflowError("timestamp must include a timezone")
    return parsed.astimezone(timezone.utc).isoformat()


def _validated_id(value: Any, name: str) -> str:
    normalized = str(value or "").strip().upper()
    if not _ID_PATTERN.fullmatch(normalized):
        raise IdentityError(
            "{} must contain 3-64 safe uppercase characters".format(name)
        )
    return normalized


def _bounded_registry_limit(value: Any, name: str, *, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise PetriWorkflowError("{} must be an integer".format(name))
    if not 1 <= value <= maximum:
        raise PetriWorkflowError("{} must be between 1 and {}".format(name, maximum))
    return value


def _artifact_path_components(value: Any) -> Tuple[str, ...]:
    if not isinstance(value, str) or not value or len(value) > 1024:
        raise PetriWorkflowError("pending review artifact path is invalid")
    if value.startswith("/") or "\\" in value or "\x00" in value:
        raise PetriWorkflowError("pending review artifact path is invalid")
    parts = tuple(value.split("/"))
    if not parts or any(
        part in ("", ".", "..") or _ARTIFACT_COMPONENT_PATTERN.fullmatch(part) is None
        for part in parts
    ):
        raise PetriWorkflowError("pending review artifact path is invalid")
    return parts


def _stat_fingerprint(value: os.stat_result) -> Tuple[int, ...]:
    return (
        int(value.st_dev),
        int(value.st_ino),
        int(value.st_mode),
        int(value.st_nlink),
        int(value.st_size),
        int(value.st_mtime_ns),
        int(value.st_ctime_ns),
    )


def _read_pending_review_png(
    artifact_root: os.PathLike[str],
    relative_path: Any,
    *,
    max_bytes: int,
) -> bytes:
    parts = _artifact_path_components(relative_path)
    try:
        root_path = os.fspath(artifact_root)
    except TypeError as exc:
        raise PetriWorkflowError("pending review artifact root is invalid") from exc
    if not isinstance(root_path, (str, bytes)) or not root_path:
        raise PetriWorkflowError("pending review artifact root is invalid")
    required_flags = ("O_DIRECTORY", "O_NOFOLLOW")
    if any(not hasattr(os, name) for name in required_flags):
        raise PetriWorkflowError("secure artifact access is unavailable")
    directory_flags = (
        os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
    )
    file_flags = (
        os.O_RDONLY
        | os.O_NOFOLLOW
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(os, "O_NONBLOCK", 0)
    )
    descriptors: List[int] = []
    directory_fingerprints: List[Tuple[int, Tuple[int, ...]]] = []
    try:
        root_fd = os.open(root_path, directory_flags)
        descriptors.append(root_fd)
        root_stat = os.fstat(root_fd)
        if not stat.S_ISDIR(root_stat.st_mode):
            raise PetriWorkflowError("pending review artifact root is not a directory")
        root_fingerprint = _stat_fingerprint(root_stat)
        directory_fingerprints.append((root_fd, root_fingerprint))
        parent_fd = root_fd
        for component in parts[:-1]:
            directory_fd = os.open(
                component,
                directory_flags,
                dir_fd=parent_fd,
            )
            descriptors.append(directory_fd)
            directory_stat = os.fstat(directory_fd)
            if not stat.S_ISDIR(directory_stat.st_mode):
                raise PetriWorkflowError("pending review artifact directory is invalid")
            directory_fingerprints.append(
                (directory_fd, _stat_fingerprint(directory_stat))
            )
            parent_fd = directory_fd

        file_fd = os.open(parts[-1], file_flags, dir_fd=parent_fd)
        descriptors.append(file_fd)
        initial_stat = os.fstat(file_fd)
        if not stat.S_ISREG(initial_stat.st_mode) or initial_stat.st_nlink != 1:
            raise PetriWorkflowError(
                "pending review artifact is not a private regular file"
            )
        size = int(initial_stat.st_size)
        if not 1 <= size <= max_bytes:
            raise PetriWorkflowError("pending review image size is invalid")
        initial_fingerprint = _stat_fingerprint(initial_stat)
        remaining = size
        chunks: List[bytes] = []
        while remaining:
            chunk = os.read(file_fd, min(remaining, 64 * 1024))
            if not chunk:
                raise PetriWorkflowError(
                    "pending review image changed while being read"
                )
            chunks.append(chunk)
            remaining -= len(chunk)
        if os.read(file_fd, 1):
            raise PetriWorkflowError("pending review image changed while being read")
        final_stat = os.fstat(file_fd)
        if (
            _stat_fingerprint(final_stat) != initial_fingerprint
            or not stat.S_ISREG(final_stat.st_mode)
            or final_stat.st_nlink != 1
        ):
            raise PetriWorkflowError("pending review image changed while being read")
        current_name_stat = os.stat(
            parts[-1],
            dir_fd=parent_fd,
            follow_symlinks=False,
        )
        if _stat_fingerprint(current_name_stat) != initial_fingerprint:
            raise PetriWorkflowError("pending review image changed while being read")
        for directory_fd, fingerprint in directory_fingerprints:
            if _stat_fingerprint(os.fstat(directory_fd)) != fingerprint:
                raise PetriWorkflowError(
                    "pending review artifact tree changed while being read"
                )
        current_root_stat = os.stat(root_path, follow_symlinks=False)
        if (
            not stat.S_ISDIR(current_root_stat.st_mode)
            or _stat_fingerprint(current_root_stat) != root_fingerprint
        ):
            raise PetriWorkflowError(
                "pending review artifact root changed while being read"
            )
        payload = b"".join(chunks)
    except PetriWorkflowError:
        raise
    except (OSError, TypeError, ValueError):
        raise PetriWorkflowError("pending review image is unavailable") from None
    finally:
        for descriptor in reversed(descriptors):
            try:
                os.close(descriptor)
            except OSError:
                pass
    _validate_pending_review_png(payload)
    return payload


def _validate_pending_review_png(payload: bytes) -> None:
    if not isinstance(payload, bytes) or not payload.startswith(_PNG_SIGNATURE):
        raise PetriWorkflowError("pending review image is not a valid PNG")
    if len(payload) < 45:
        raise PetriWorkflowError("pending review image is not a valid PNG")
    view = memoryview(payload)
    offset = len(_PNG_SIGNATURE)
    chunk_count = 0
    color_type: Optional[int] = None
    saw_idat = False
    idat_closed = False
    idat_bytes = 0
    saw_plte = False
    saw_iend = False
    while offset < len(payload):
        chunk_count += 1
        if chunk_count > _MAX_PNG_CHUNKS or offset + 12 > len(payload):
            raise PetriWorkflowError("pending review PNG structure is invalid")
        length = int.from_bytes(view[offset : offset + 4], "big")
        chunk_type = bytes(view[offset + 4 : offset + 8])
        data_start = offset + 8
        data_end = data_start + length
        chunk_end = data_end + 4
        if chunk_end > len(payload) or not all(
            65 <= value <= 90 or 97 <= value <= 122 for value in chunk_type
        ):
            raise PetriWorkflowError("pending review PNG structure is invalid")
        if not 65 <= chunk_type[2] <= 90:
            raise PetriWorkflowError("pending review PNG structure is invalid")
        expected_crc = int.from_bytes(view[data_end:chunk_end], "big")
        actual_crc = binascii.crc32(chunk_type)
        actual_crc = binascii.crc32(view[data_start:data_end], actual_crc)
        if actual_crc & 0xFFFFFFFF != expected_crc:
            raise PetriWorkflowError("pending review PNG checksum is invalid")

        if chunk_count == 1:
            if chunk_type != b"IHDR" or length != 13:
                raise PetriWorkflowError("pending review PNG header is invalid")
            width = int.from_bytes(view[data_start : data_start + 4], "big")
            height = int.from_bytes(
                view[data_start + 4 : data_start + 8],
                "big",
            )
            bit_depth = int(view[data_start + 8])
            color_type = int(view[data_start + 9])
            compression = int(view[data_start + 10])
            filter_method = int(view[data_start + 11])
            interlace = int(view[data_start + 12])
            valid_depths = {
                0: (1, 2, 4, 8, 16),
                2: (8, 16),
                3: (1, 2, 4, 8),
                4: (8, 16),
                6: (8, 16),
            }
            if (
                not 1 <= width <= _MAX_PENDING_REVIEW_IMAGE_DIMENSION
                or not 1 <= height <= _MAX_PENDING_REVIEW_IMAGE_DIMENSION
                or width * height > _MAX_PENDING_REVIEW_IMAGE_PIXELS
                or color_type not in valid_depths
                or bit_depth not in valid_depths[color_type]
                or compression != 0
                or filter_method != 0
                or interlace not in (0, 1)
            ):
                raise PetriWorkflowError(
                    "pending review PNG dimensions or format are invalid"
                )
        elif chunk_type == b"IHDR":
            raise PetriWorkflowError("pending review PNG structure is invalid")

        if 65 <= chunk_type[0] <= 90 and chunk_type not in {
            b"IHDR",
            b"PLTE",
            b"IDAT",
            b"IEND",
        }:
            raise PetriWorkflowError("pending review PNG chunk is unsupported")
        if chunk_type == b"PLTE":
            if (
                saw_plte
                or saw_idat
                or color_type in (0, 4)
                or length < 3
                or length > 768
                or length % 3
            ):
                raise PetriWorkflowError("pending review PNG palette is invalid")
            saw_plte = True
        elif chunk_type == b"IDAT":
            if idat_closed:
                raise PetriWorkflowError("pending review PNG structure is invalid")
            saw_idat = True
            idat_bytes += length
        elif chunk_type == b"IEND":
            if length != 0 or not saw_idat or idat_bytes <= 0:
                raise PetriWorkflowError("pending review PNG ending is invalid")
            saw_iend = True
            offset = chunk_end
            break
        elif saw_idat:
            idat_closed = True
        offset = chunk_end
    if not saw_iend or offset != len(payload) or (color_type == 3 and not saw_plte):
        raise PetriWorkflowError("pending review PNG structure is invalid")


def _finite_float(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise PetriWorkflowError("{} must be finite".format(name))
    try:
        result = float(value)
    except (TypeError, ValueError) as exc:
        raise PetriWorkflowError("{} must be finite".format(name)) from exc
    if not math.isfinite(result):
        raise PetriWorkflowError("{} must be finite".format(name))
    return result


def _positive_float(value: Any, name: str) -> float:
    result = _finite_float(value, name)
    if result <= 0.0:
        raise PetriWorkflowError("{} must be positive".format(name))
    return result


def _finite_vector(
    value: Any,
    length: int,
    name: str,
) -> Tuple[float, ...]:
    if not isinstance(value, (tuple, list)) or len(value) != length:
        raise PetriWorkflowError("{} must contain {} values".format(name, length))
    return tuple(_finite_float(item, name) for item in value)


def _normalize_detection(detection: Mapping[str, Any]) -> Dict[str, Any]:
    if not isinstance(detection, Mapping):
        raise PetriWorkflowError("each detection must be an object")
    x_value = _finite_float(detection.get("x"), "detection.x")
    y_value = _finite_float(detection.get("y"), "detection.y")
    radius = _positive_float(detection.get("radius"), "detection.radius")
    confidence = _finite_float(
        detection.get("confidence", 0.0),
        "detection.confidence",
    )
    if not 0.0 <= confidence <= 1.0:
        raise PetriWorkflowError("detection.confidence must be in [0, 1]")
    safe_raw = {
        str(key)[:80]: value
        for key, value in detection.items()
        if key not in {"image", "file_data"}
        and isinstance(value, (str, int, float, bool, type(None)))
    }
    return {
        "x": x_value,
        "y": y_value,
        "radius": radius,
        "confidence": confidence,
        "raw": safe_raw,
    }


def _validate_detection_bounds(
    detection: Mapping[str, Any],
    width: int,
    height: int,
) -> None:
    if not (
        0.0 <= float(detection["x"]) < width and 0.0 <= float(detection["y"]) < height
    ):
        raise PetriWorkflowError("detection center is outside the image")
    if float(detection["radius"]) > max(width, height):
        raise PetriWorkflowError("detection radius is larger than the image")


def _ambiguous_detection_indexes(
    detections: Sequence[Mapping[str, Any]],
) -> set:
    result = set()
    for first in range(len(detections)):
        for second in range(first + 1, len(detections)):
            left = detections[first]
            right = detections[second]
            distance = math.hypot(
                float(left["x"]) - float(right["x"]),
                float(left["y"]) - float(right["y"]),
            )
            smaller = min(float(left["radius"]), float(right["radius"]))
            if distance < smaller * 0.65:
                result.update((first, second))
    return result


def _load_image(image_bytes: bytes) -> Tuple[Any, Any]:
    try:
        from PIL import Image, ImageDraw
    except ImportError as exc:
        raise PetriWorkflowError(
            "Pillow is required for Petri image artifacts"
        ) from exc
    try:
        image = Image.open(io.BytesIO(image_bytes))
        image.load()
        return image.convert("RGB"), ImageDraw
    except Exception as exc:
        raise PetriWorkflowError("image could not be decoded") from exc


def _save_png_atomic(image: Any, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=".{}.".format(path.name),
        suffix=".tmp",
        dir=str(path.parent),
    )
    try:
        with os.fdopen(descriptor, "wb") as output:
            image.save(output, format="PNG")
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary_name, path)
    except Exception:
        try:
            os.unlink(temporary_name)
        except OSError:
            pass
        raise


def _write_json_atomic(path: Path, value: Mapping[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=".{}.".format(path.name),
        suffix=".tmp",
        dir=str(path.parent),
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as output:
            json.dump(
                value,
                output,
                allow_nan=False,
                indent=2,
                sort_keys=True,
            )
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary_name, path)
    except Exception:
        try:
            os.unlink(temporary_name)
        except OSError:
            pass
        raise


def _observation_row(row: sqlite3.Row) -> Dict[str, Any]:
    result = dict(row)
    result["selected"] = bool(result["selected"])
    result["metadata"] = json.loads(result.pop("metadata_json"))
    return result
