"""Ownership assignment and seam validation for multi-view marker textures (stdlib only).

The problem this solves
-----------------------
`uv_1` and `uv_2` are both *full-surface* layouts, and both carry the same pattern
(grounded on the sensing camera). Printing both therefore double-inks essentially the whole
part. Ownership is the fix: assign every loop triangle to exactly one production view, then
mask each output file to the fragments of the triangles it owns.

Why a triangle, and why that makes zero overlap a theorem
--------------------------------------------------------
A triangle is a disjoint surface element. If every triangle has exactly one owner and file
*k* emits only owner-*k* fragments, then no surface element is inked twice — regardless of
how well the two print passes are registered. Registration then only matters *at the seam*,
where an error produces a small local gap or overlap instead of a doubled pattern.

Two modes
---------
* ``computed`` — pick the first view in priority order that qualifies on each triangle
  (visible, within the incidence limit, above the footprint floor, and covering at least
  ``min_triangle_coverage`` of the triangle's area). The seam falls where the metric puts it.
* ``authored`` — the artist's ``use_seam`` edges cut the triangle adjacency graph into
  connected components, and each component is assigned whole to one view. Disjointness is
  again a theorem (components partition the triangles), and the seam is exactly where it was
  drawn.

Both are deterministic: triangles are visited in sorted index order, views in a fixed
priority order, and there is no scoring and no float tie-break.

Measured on `sensors/finger/r01` (2026-09-27, blend sha256 61093B1A…)
---------------------------------------------------------------------
The authored seam is the y = 0 plane, which is exactly the boundary where
``incidence_uv1 == incidence_uv2`` because the two cameras are mirrored about XZ. So
``authored`` and ``computed`` agree to within 0.0014 of coverage at a 60 degree incidence
limit and are identical from 62 degrees up. Neither mode is a compromise here.

A trap this module guards against
---------------------------------
Assigning a component by *count of visible triangles* does not work on this mesh: both views
see ~100% of each component's triangles, so the count ties and the naive rule silently hands
both components to the first view. Component assignment therefore uses **qualifying footprint**
(which folds in both the incidence limit and the footprint floor), reported per view so the
decision is auditable.
"""

from __future__ import annotations

from collections import Counter, defaultdict
from dataclasses import dataclass, field
from typing import Iterable, Mapping, Sequence

# Defaults, calibrated on `sensors/finger/r01`. See the module docstring and the plan:
# measured hole fraction is 0.0218 at 60 deg and 0.0000 at 62 deg, so 60 fails the 0.02 gate.
DEFAULT_INCIDENCE_LIMIT_DEG = 62.0
DEFAULT_MIN_FOOTPRINT_MM2 = 2.0e-3
DEFAULT_MIN_TRIANGLE_COVERAGE = 0.60
DEFAULT_MAX_HOLE_FRACTION = 0.02

MODE_COMPUTED = "computed"
MODE_AUTHORED = "authored"
_MODES = (MODE_COMPUTED, MODE_AUTHORED)

POLICY_COMPUTED = "computed"
POLICY_BEST_COVERAGE = "best_coverage"
POLICY_PINNED = "pinned"


class MergeConfigError(ValueError):
    """Raised for an unusable merge policy or seam. Always names the offender."""


@dataclass(frozen=True)
class ViewFragments:
    """One view's fragments, parallel arrays as produced by `build_patch_geometry`.

    ``triangle`` is the loop-triangle index each fragment belongs to; ``incidence_deg`` and
    ``footprint_mm2`` are per fragment. Every emitted fragment is visible by construction
    (`cell_fragments` drops or rejects invisible ones), so there is no visibility flag here.
    """

    triangle: Sequence[int]
    incidence_deg: Sequence[float]
    footprint_mm2: Sequence[float]

    def __post_init__(self) -> None:
        if not (len(self.triangle) == len(self.incidence_deg) == len(self.footprint_mm2)):
            raise MergeConfigError(
                "ViewFragments arrays must be parallel: got "
                f"triangle={len(self.triangle)} incidence={len(self.incidence_deg)} "
                f"footprint={len(self.footprint_mm2)}"
            )

    @classmethod
    def from_meta(cls, meta: Mapping) -> "ViewFragments":
        """Build from a `build_patch_geometry` ``_meta`` block."""
        return cls(
            triangle=list(meta["fragment_triangle"]),
            incidence_deg=list(meta["fragment_incidence_deg"]),
            footprint_mm2=list(meta["fragment_footprint_mm2"]),
        )

    def by_triangle(self) -> dict[int, list[tuple[float, float]]]:
        out: dict[int, list[tuple[float, float]]] = defaultdict(list)
        for t, inc, foot in zip(self.triangle, self.incidence_deg, self.footprint_mm2):
            out[t].append((inc, foot))
        return dict(out)


@dataclass(frozen=True)
class MergePolicy:
    """Thresholds for `assign_ownership`. Bounds are inclusive (>= / <=)."""

    mode: str = MODE_COMPUTED
    incidence_limit_deg: float = DEFAULT_INCIDENCE_LIMIT_DEG
    min_footprint_mm2: float = DEFAULT_MIN_FOOTPRINT_MM2
    min_triangle_coverage: float = DEFAULT_MIN_TRIANGLE_COVERAGE
    max_hole_fraction: float = DEFAULT_MAX_HOLE_FRACTION
    component_policy: str = POLICY_BEST_COVERAGE

    def __post_init__(self) -> None:
        if self.mode not in _MODES:
            raise MergeConfigError(f"mode: must be one of {_MODES} (got {self.mode!r})")
        if self.component_policy not in (POLICY_BEST_COVERAGE, POLICY_PINNED):
            raise MergeConfigError(
                "component_policy: must be 'best_coverage' or 'pinned' "
                f"(got {self.component_policy!r})"
            )
        if not 0.0 <= self.min_triangle_coverage <= 1.0:
            raise MergeConfigError(
                f"min_triangle_coverage: must be in [0, 1] (got {self.min_triangle_coverage!r})"
            )
        if not 0.0 <= self.incidence_limit_deg <= 90.0:
            raise MergeConfigError(
                f"incidence_limit_deg: must be in [0, 90] (got {self.incidence_limit_deg!r})"
            )
        if self.min_footprint_mm2 < 0.0:
            raise MergeConfigError(
                f"min_footprint_mm2: must be >= 0 (got {self.min_footprint_mm2!r})"
            )


def edge_key(a: int, b: int) -> tuple[int, int]:
    """Canonical undirected vertex-pair key."""
    return (a, b) if a < b else (b, a)


def triangle_edge_keys(vertices: Sequence[int]) -> tuple[tuple[int, int], ...]:
    """The three undirected edge keys of a triangle given its vertex indices."""
    if len(vertices) != 3:
        raise MergeConfigError(f"triangle_vertices: expected 3 indices, got {len(vertices)}")
    a, b, c = vertices
    return (edge_key(a, b), edge_key(b, c), edge_key(c, a))


def edge_use_counts(
    triangle_vertices: Sequence[Sequence[int]],
) -> tuple[Counter, set[tuple[int, int]], set[tuple[int, int]]]:
    """Return (use counts, boundary edges, non-manifold edges)."""
    counts: Counter = Counter()
    for vertices in triangle_vertices:
        counts.update(triangle_edge_keys(vertices))
    boundary = {k for k, n in counts.items() if n == 1}
    non_manifold = {k for k, n in counts.items() if n > 2}
    return counts, boundary, non_manifold


def components_from_seam(
    triangle_vertices: Sequence[Sequence[int]],
    seam_edges: Iterable[tuple[int, int]],
) -> list[int]:
    """Connected components of the triangle adjacency graph with seam edges cut.

    Two triangles are adjacent when they share an edge that is *not* a seam edge. Returns
    one component id per triangle, numbered in order of first appearance (deterministic).
    """
    seam = {edge_key(*e) for e in seam_edges}
    incident: dict[tuple[int, int], list[int]] = defaultdict(list)
    for index, vertices in enumerate(triangle_vertices):
        for key in triangle_edge_keys(vertices):
            incident[key].append(index)

    parent = list(range(len(triangle_vertices)))

    def find(x: int) -> int:
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    for key, tris in incident.items():
        if key in seam:
            continue
        for other in tris[1:]:
            ra, rb = find(tris[0]), find(other)
            if ra != rb:
                parent[ra] = rb

    ids: dict[int, int] = {}
    out: list[int] = []
    for index in range(len(triangle_vertices)):
        root = find(index)
        if root not in ids:
            ids[root] = len(ids)
        out.append(ids[root])
    return out


def validate_seam(
    triangle_vertices: Sequence[Sequence[int]],
    seam_edges: Iterable[tuple[int, int]],
    components: Sequence[int],
) -> list[str]:
    """Return human-readable problems, empty when the seam is a usable partition.

    The load-bearing rule is #1: for a cut to *separate* anything, every seam edge must have
    its two adjacent faces in **different** components. That is the partition proof. Rules
    2-3 are diagnostics that catch the shapes which make rule 1 fail in non-obvious ways.
    """
    errors: list[str] = []
    seam = {edge_key(*e) for e in seam_edges}
    if not seam:
        errors.append("seam: no seam edges; a cut needs at least one")
        return errors
    if len(components) != len(triangle_vertices):
        errors.append(
            f"components: expected one per triangle ({len(triangle_vertices)}), "
            f"got {len(components)}"
        )
        return errors

    _, boundary, non_manifold = edge_use_counts(triangle_vertices)
    incident: dict[tuple[int, int], list[int]] = defaultdict(list)
    for index, vertices in enumerate(triangle_vertices):
        for key in triangle_edge_keys(vertices):
            incident[key].append(index)

    unknown = sorted(k for k in seam if k not in incident)
    if unknown:
        errors.append(
            f"seam: {len(unknown)} seam edge(s) are not edges of any triangle, e.g. {unknown[:3]}"
        )

    # Rule 1 — the partition proof.
    same_component = [
        k for k in sorted(seam)
        if k in incident and len(incident[k]) == 2
        and components[incident[k][0]] == components[incident[k][1]]
    ]
    if same_component:
        errors.append(
            f"seam: {len(same_component)} seam edge(s) join two triangles in the SAME "
            f"component, so the cut does not separate there, e.g. {same_component[:3]} — "
            "a dangling or branching seam leaves the surface connected around its end"
        )

    # Rule 2 — every seam edge must be interior, so both sides are surface.
    on_boundary = [k for k in sorted(seam) if len(incident.get(k, ())) < 2]
    if on_boundary:
        errors.append(
            f"seam: {len(on_boundary)} seam edge(s) are mesh-boundary or missing edges and "
            f"cannot separate two faces, e.g. {on_boundary[:3]}"
        )

    # Rule 3 — degree form. A closed loop (all degrees 2) cuts a closed surface; on a
    # surface with boundary a simple open path also cuts, but only if BOTH of its free ends
    # land on the mesh boundary. Anything else (degree >= 3, or a free end in the interior)
    # is what usually makes rule 1 fail.
    degree: Counter = Counter()
    for a, b in seam:
        degree[a] += 1
        degree[b] += 1
    interior_free_ends = sorted(
        v for v, d in degree.items() if d == 1 and not any(v in k for k in boundary)
    )
    if interior_free_ends:
        errors.append(
            f"seam: {len(interior_free_ends)} free seam end(s) are NOT on the mesh boundary "
            f"(e.g. vertex {interior_free_ends[0]}); the cut is open and cannot separate"
        )
    branching = sorted(v for v, d in degree.items() if d >= 3)
    if branching:
        errors.append(
            f"seam: {len(branching)} seam vertex/vertices have degree >= 3 "
            f"(e.g. vertex {branching[0]}); the cut branches and is not a simple curve"
        )
    return errors


def component_areas(components: Sequence[int], triangle_area_mm2: Sequence[float]) -> dict[int, float]:
    """Total surface area per component."""
    out: dict[int, float] = defaultdict(float)
    for comp, area in zip(components, triangle_area_mm2):
        out[comp] += area
    return dict(out)


def qualifying_footprint(
    fragments: Sequence[tuple[float, float]],
    policy: MergePolicy,
) -> float:
    """Sum of footprint over fragments that pass the per-fragment gates."""
    return sum(
        foot
        for incidence, foot in fragments
        if incidence <= policy.incidence_limit_deg and foot >= policy.min_footprint_mm2
    )


def assign_components(
    view_fragments: Mapping[str, ViewFragments],
    priority: Sequence[str],
    components: Sequence[int],
    component_count: int,
    policy: MergePolicy,
    pins: Mapping[int, str] | None = None,
) -> tuple[dict[int, str], dict[int, dict[str, float]]]:
    """Assign each component to one view. Returns (component -> view, audit table).

    Uses **qualifying footprint**, not visible-triangle count: on `sensors/finger/r01` both
    views see ~100% of each component's triangles, so a count ties and would silently hand
    every component to the first view in priority order.
    """
    pins = dict(pins or {})
    by_triangle: dict[str, dict[int, list[tuple[float, float]]]] = {
        view: view_fragments[view].by_triangle() for view in priority
    }
    comp_tris: dict[int, list[int]] = defaultdict(list)
    for tri, comp in enumerate(components):
        comp_tris[comp].append(tri)

    scores: dict[int, dict[str, float]] = {}
    for comp in range(component_count):
        scores[comp] = {
            view: sum(
                qualifying_footprint(by_triangle[view].get(tri, ()), policy)
                for tri in comp_tris.get(comp, ())
            )
            for view in priority
        }

    owners: dict[int, str] = {}
    for comp in range(component_count):
        if comp in pins:
            view = pins[comp]
            if view not in priority:
                raise MergeConfigError(
                    f"pins: component {comp} pinned to unknown view {view!r}; "
                    f"known views are {list(priority)}"
                )
            owners[comp] = view
            continue
        ranked = sorted(scores[comp].items(), key=lambda kv: (-kv[1], priority.index(kv[0])))
        if ranked[0][1] <= 0.0:
            # No view can qualify anywhere in this component: leave it unowned so every
            # triangle lands in the reported hole census rather than being painted blind.
            continue
        owners[comp] = ranked[0][0]
    return owners, scores


def assign_ownership(
    triangle_area_mm2: Sequence[float],
    view_fragments: Mapping[str, ViewFragments],
    priority: Sequence[str],
    policy: MergePolicy | None = None,
    components: Sequence[int] | None = None,
    pins: Mapping[int, str] | None = None,
) -> dict:
    """Assign every triangle to at most one view; return the owner map plus a census.

    Returns a dict with ``owner`` (triangle -> view; absent means a hole), ``owned_mm2``,
    ``hole_mm2``, ``hole_fraction``, ``per_view_mm2``, and — for authored mode —
    ``component_owner`` and ``component_scores`` so the decision can be audited.
    """
    policy = policy or MergePolicy()
    priority = tuple(priority)
    if not priority:
        raise MergeConfigError("priority: at least one view is required")
    if len(set(priority)) != len(priority):
        raise MergeConfigError(f"priority: duplicate view id in {list(priority)}")
    missing = [view for view in priority if view not in view_fragments]
    if missing:
        raise MergeConfigError(f"view_fragments: missing data for view(s) {missing}")

    triangle_count = len(triangle_area_mm2)
    by_triangle = {view: view_fragments[view].by_triangle() for view in priority}
    owner: dict[int, str] = {}
    component_owner: dict[int, str] = {}
    scores: dict[int, dict[str, float]] = {}

    if policy.mode == MODE_AUTHORED:
        if components is None:
            raise MergeConfigError("mode 'authored' requires the component map from the seam")
        if len(components) != triangle_count:
            raise MergeConfigError(
                f"components: expected {triangle_count} entries, got {len(components)}"
            )
        component_count = (max(components) + 1) if triangle_count else 0
        component_owner, scores = assign_components(
            view_fragments, priority, components, component_count, policy, pins
        )
        for tri in range(triangle_count):
            view = component_owner.get(components[tri])
            if view is None:
                continue
            if qualifying_footprint(by_triangle[view].get(tri, ()), policy) <= 0.0:
                # The component's owner cannot qualify on this triangle: a reported hole,
                # never silently painted or handed to another view.
                continue
            owner[tri] = view
    else:
        for tri in range(triangle_count):
            area = triangle_area_mm2[tri]
            for view in priority:
                fragments = by_triangle[view].get(tri, ())
                covered = qualifying_footprint(fragments, policy)
                if covered <= 0.0:
                    continue
                if area > 0.0 and covered < policy.min_triangle_coverage * area:
                    continue
                owner[tri] = view
                break

    owned_mm2 = sum(triangle_area_mm2[t] for t in owner)
    total_mm2 = sum(triangle_area_mm2)
    per_view: dict[str, float] = defaultdict(float)
    for tri, view in owner.items():
        per_view[view] += triangle_area_mm2[tri]

    return {
        "mode": policy.mode,
        "owner": owner,
        "owned_mm2": owned_mm2,
        "hole_mm2": total_mm2 - owned_mm2,
        "total_mm2": total_mm2,
        "hole_fraction": (total_mm2 - owned_mm2) / total_mm2 if total_mm2 else 0.0,
        "per_view_mm2": dict(per_view),
        "per_view_triangles": {
            view: sum(1 for v in owner.values() if v == view) for view in priority
        },
        "component_owner": component_owner,
        "component_scores": scores,
        "policy": policy,
    }


def ownership_disjoint(owner: Mapping[int, str]) -> bool:
    """Trivially true for a mapping, but kept explicit: one owner per triangle key."""
    return len(set(owner)) == len(owner)


__all__ = [
    "DEFAULT_INCIDENCE_LIMIT_DEG",
    "DEFAULT_MAX_HOLE_FRACTION",
    "DEFAULT_MIN_FOOTPRINT_MM2",
    "DEFAULT_MIN_TRIANGLE_COVERAGE",
    "MODE_AUTHORED",
    "MODE_COMPUTED",
    "POLICY_BEST_COVERAGE",
    "POLICY_COMPUTED",
    "POLICY_PINNED",
    "MergeConfigError",
    "MergePolicy",
    "ViewFragments",
    "assign_components",
    "assign_ownership",
    "component_areas",
    "components_from_seam",
    "edge_key",
    "edge_use_counts",
    "ownership_disjoint",
    "qualifying_footprint",
    "triangle_edge_keys",
    "validate_seam",
]