Dxf

interface:
  display_name: "DXF"
  short_description: "Generate and validate 2D DXF drawings."
  default_prompt: "Use $dxf to create, regenerate, and validate explicit DXF drawing files from gen_dxf() Python sources, handing outputs to $cad-viewer when available."

MIT License

Copyright (c) 2026 earthtojake

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

--editable ./scripts/packages/cadpy
ezdxf

"""DXF generation CLI."""

from __future__ import annotations

import sys
from pathlib import Path

if __package__ in {None, ""}:
    tool_dir = Path(__file__).resolve().parent
    if str(tool_dir) not in sys.path:
        sys.path.insert(0, str(tool_dir))
    from cli import main
else:
    from .cli import main


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import argparse
from collections.abc import Sequence


def generate_dxf_targets(*args, **kwargs):
    from cadpy.generation import generate_dxf_targets as generate

    return generate(*args, **kwargs)


def _targets_include_output_pairs(targets: Sequence[str]) -> bool:
    return any("=" in str(target or "") for target in targets)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="dxf",
        description="Generate explicit DXF targets from Python sources.",
    )
    parser.add_argument(
        "targets",
        nargs="+",
        help="Explicit Python source file or SOURCE.py=OUTPUT.dxf pair defining gen_dxf() to generate.",
    )
    parser.add_argument(
        "-o",
        "--output",
        metavar="PATH",
        help="Write the generated DXF file to this path. Valid only with one plain generated Python target.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show detailed progress and timing information.",
    )
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(list(argv) if argv is not None else None)
    if args.output is not None:
        if _targets_include_output_pairs(args.targets):
            parser.error("--output cannot be combined with SOURCE=OUTPUT targets")
        if len(args.targets) != 1:
            parser.error("--output can only be used with exactly one target")
    return generate_dxf_targets(args.targets, output=args.output, verbose=bool(args.verbose))


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import math
from dataclasses import dataclass
from pathlib import Path

import ezdxf


DXF_RENDER_SCHEMA_VERSION = 1
SUPPORTED_ENTITY_TYPES = {"LINE", "ARC", "CIRCLE", "LWPOLYLINE"}
ANGLE_EPSILON = 1e-9


@dataclass(frozen=True)
class LineEntity:
    layer: str
    start: tuple[float, float]
    end: tuple[float, float]


@dataclass(frozen=True)
class ArcEntity:
    layer: str
    center: tuple[float, float]
    radius: float
    start_angle_deg: float
    sweep_angle_deg: float

    @property
    def end_angle_deg(self) -> float:
        return self.start_angle_deg + self.sweep_angle_deg


@dataclass(frozen=True)
class CircleEntity:
    layer: str
    center: tuple[float, float]
    radius: float


def _normalize_layer_name(value: object) -> str:
    text = str(value or "").strip()
    return text or "0"


def _semantic_kind_for_layer(layer_name: str) -> str:
    return "bend" if "bend" in layer_name.strip().lower() else "cut"


def _normalize_angle(angle_deg: float) -> float:
    value = math.fmod(angle_deg, 360.0)
    return value + 360.0 if value < 0.0 else value


def _angle_in_sweep(angle_deg: float, start_angle_deg: float, sweep_angle_deg: float) -> bool:
    abs_sweep_angle_deg = abs(sweep_angle_deg)
    if abs_sweep_angle_deg >= 360.0 - ANGLE_EPSILON:
        return True
    if sweep_angle_deg >= 0.0:
        normalized_delta = (_normalize_angle(angle_deg) - _normalize_angle(start_angle_deg)) % 360.0
        return normalized_delta <= abs_sweep_angle_deg + ANGLE_EPSILON
    normalized_delta = (_normalize_angle(start_angle_deg) - _normalize_angle(angle_deg)) % 360.0
    return normalized_delta <= abs_sweep_angle_deg + ANGLE_EPSILON


def _point_on_circle(center: tuple[float, float], radius: float, angle_deg: float) -> tuple[float, float]:
    radians = math.radians(angle_deg)
    return (
        center[0] + radius * math.cos(radians),
        center[1] + radius * math.sin(radians),
    )


def _arc_extrema_points(arc: ArcEntity) -> list[tuple[float, float]]:
    points = [
        _point_on_circle(arc.center, arc.radius, arc.start_angle_deg),
        _point_on_circle(arc.center, arc.radius, arc.end_angle_deg),
    ]
    for candidate_angle in (0.0, 90.0, 180.0, 270.0):
        if _angle_in_sweep(candidate_angle, arc.start_angle_deg, arc.sweep_angle_deg):
            points.append(_point_on_circle(arc.center, arc.radius, candidate_angle))
    return points


def _line_bounds(line: LineEntity) -> tuple[float, float, float, float]:
    xs = (line.start[0], line.end[0])
    ys = (line.start[1], line.end[1])
    return (min(xs), min(ys), max(xs), max(ys))


def _circle_bounds(circle: CircleEntity) -> tuple[float, float, float, float]:
    cx, cy = circle.center
    r = circle.radius
    return (cx - r, cy - r, cx + r, cy + r)


def _arc_bounds(arc: ArcEntity) -> tuple[float, float, float, float]:
    points = _arc_extrema_points(arc)
    xs = [point[0] for point in points]
    ys = [point[1] for point in points]
    return (min(xs), min(ys), max(xs), max(ys))


def _expand_bounds(
    current: tuple[float, float, float, float] | None,
    next_bounds: tuple[float, float, float, float],
) -> tuple[float, float, float, float]:
    if current is None:
        return next_bounds
    return (
        min(current[0], next_bounds[0]),
        min(current[1], next_bounds[1]),
        max(current[2], next_bounds[2]),
        max(current[3], next_bounds[3]),
    )


def _screen_point(point: tuple[float, float], *, min_x: float, max_y: float) -> tuple[float, float]:
    return (point[0] - min_x, max_y - point[1])


def _format_number(value: float) -> float:
    rounded = round(float(value), 6)
    return 0.0 if abs(rounded) < ANGLE_EPSILON else rounded


def _build_path_record(layer_name: str, semantic_kind: str, path_data: str) -> dict[str, object]:
    return {"layer": layer_name, "kind": semantic_kind, "d": path_data}


def _arc_from_bulge_segment(
    *,
    layer_name: str,
    start: tuple[float, float],
    end: tuple[float, float],
    bulge: float,
) -> ArcEntity | None:
    dx = end[0] - start[0]
    dy = end[1] - start[1]
    chord_length = math.hypot(dx, dy)
    if chord_length <= ANGLE_EPSILON or abs(bulge) <= ANGLE_EPSILON:
        return None

    included_angle_rad = 4.0 * math.atan(bulge)
    radius = (chord_length * (1.0 + bulge * bulge)) / (4.0 * abs(bulge))
    midpoint = ((start[0] + end[0]) / 2.0, (start[1] + end[1]) / 2.0)
    left_normal = (-dy / chord_length, dx / chord_length)
    center_offset = (chord_length * (1.0 - bulge * bulge)) / (4.0 * bulge)
    center = (
        midpoint[0] + left_normal[0] * center_offset,
        midpoint[1] + left_normal[1] * center_offset,
    )
    start_angle_deg = _normalize_angle(math.degrees(math.atan2(start[1] - center[1], start[0] - center[0])))
    return ArcEntity(
        layer=layer_name,
        center=center,
        radius=radius,
        start_angle_deg=start_angle_deg,
        sweep_angle_deg=math.degrees(included_angle_rad),
    )


def _lwpolyline_entities(
    entity,
    *,
    layer_name: str,
    dxf_path: Path,
) -> tuple[list[LineEntity], list[ArcEntity]]:
    vertices: list[tuple[tuple[float, float], float]] = []
    for point in entity:
        bulge = float(point[4]) if len(point) > 4 else 0.0
        vertices.append(((float(point[0]), float(point[1])), bulge))

    if len(vertices) < 2:
        raise ValueError(f"Invalid DXF LWPOLYLINE in {dxf_path.as_posix()}: expected at least 2 vertices")

    lines: list[LineEntity] = []
    arcs: list[ArcEntity] = []

    def add_segment(start_vertex: tuple[tuple[float, float], float], end_vertex: tuple[tuple[float, float], float]) -> None:
        start, bulge = start_vertex
        end, _ = end_vertex
        if start == end:
            return
        if abs(bulge) > ANGLE_EPSILON:
            arc = _arc_from_bulge_segment(layer_name=layer_name, start=start, end=end, bulge=bulge)
            if arc is not None:
                arcs.append(arc)
            return
        lines.append(LineEntity(layer=layer_name, start=start, end=end))

    for start_vertex, end_vertex in zip(vertices, vertices[1:]):
        add_segment(start_vertex, end_vertex)
    if entity.closed:
        add_segment(vertices[-1], vertices[0])
    return lines, arcs


def _load_dxf_entities(document, dxf_path: Path) -> tuple[list[LineEntity], list[ArcEntity], list[CircleEntity]]:
    modelspace = document.modelspace()
    lines: list[LineEntity] = []
    arcs: list[ArcEntity] = []
    circles: list[CircleEntity] = []

    for entity in modelspace:
        entity_type = entity.dxftype()
        if entity_type not in SUPPORTED_ENTITY_TYPES:
            raise ValueError(
                f"Unsupported DXF entity {entity_type} in {dxf_path.as_posix()}; "
                f"supported types: {', '.join(sorted(SUPPORTED_ENTITY_TYPES))}"
            )

        layer_name = _normalize_layer_name(entity.dxf.layer)
        if entity_type == "LINE":
            lines.append(
                LineEntity(
                    layer=layer_name,
                    start=(float(entity.dxf.start.x), float(entity.dxf.start.y)),
                    end=(float(entity.dxf.end.x), float(entity.dxf.end.y)),
                )
            )
            continue

        if entity_type == "LWPOLYLINE":
            polyline_lines, polyline_arcs = _lwpolyline_entities(entity, layer_name=layer_name, dxf_path=dxf_path)
            lines.extend(polyline_lines)
            arcs.extend(polyline_arcs)
            continue

        if entity_type == "ARC":
            radius = float(entity.dxf.radius)
            if radius <= 0.0:
                raise ValueError(f"Invalid DXF arc radius in {dxf_path.as_posix()}: {radius}")
            start_angle_deg = _normalize_angle(float(entity.dxf.start_angle))
            end_angle_deg = _normalize_angle(float(entity.dxf.end_angle))
            sweep_angle_deg = (end_angle_deg - start_angle_deg) % 360.0
            if sweep_angle_deg <= ANGLE_EPSILON:
                sweep_angle_deg = 360.0
            arcs.append(
                ArcEntity(
                    layer=layer_name,
                    center=(float(entity.dxf.center.x), float(entity.dxf.center.y)),
                    radius=radius,
                    start_angle_deg=start_angle_deg,
                    sweep_angle_deg=sweep_angle_deg,
                )
            )
            continue

        radius = float(entity.dxf.radius)
        if radius <= 0.0:
            raise ValueError(f"Invalid DXF circle radius in {dxf_path.as_posix()}: {radius}")
        circles.append(
            CircleEntity(
                layer=layer_name,
                center=(float(entity.dxf.center.x), float(entity.dxf.center.y)),
                radius=radius,
            )
        )

    if not lines and not arcs and not circles:
        raise ValueError(f"No supported DXF entities found in {dxf_path.as_posix()}")

    return lines, arcs, circles


def build_dxf_render_payload(dxf_path: Path, *, file_ref: str) -> dict[str, object]:
    source_path = dxf_path.resolve()
    document = ezdxf.readfile(source_path)
    lines, arcs, circles = _load_dxf_entities(document, source_path)

    raw_bounds: tuple[float, float, float, float] | None = None
    for line in lines:
        raw_bounds = _expand_bounds(raw_bounds, _line_bounds(line))
    for arc in arcs:
        raw_bounds = _expand_bounds(raw_bounds, _arc_bounds(arc))
    for circle in circles:
        raw_bounds = _expand_bounds(raw_bounds, _circle_bounds(circle))

    if raw_bounds is None:
        raise ValueError(f"Failed to compute DXF bounds for {source_path.as_posix()}")

    min_x, min_y, max_x, max_y = raw_bounds
    width = max(max_x - min_x, 0.0)
    height = max(max_y - min_y, 0.0)

    path_records: list[dict[str, object]] = []
    circle_records: list[dict[str, object]] = []
    layer_summary: dict[str, dict[str, object]] = {}

    def touch_layer(layer_name: str) -> dict[str, object]:
        summary = layer_summary.get(layer_name)
        if summary is not None:
            return summary
        summary = {
            "name": layer_name,
            "kind": _semantic_kind_for_layer(layer_name),
            "pathCount": 0,
            "circleCount": 0,
        }
        layer_summary[layer_name] = summary
        return summary

    for line in lines:
        start = _screen_point(line.start, min_x=min_x, max_y=max_y)
        end = _screen_point(line.end, min_x=min_x, max_y=max_y)
        path_records.append(
            _build_path_record(
                line.layer,
                _semantic_kind_for_layer(line.layer),
                (
                    f"M {_format_number(start[0])} {_format_number(start[1])} "
                    f"L {_format_number(end[0])} {_format_number(end[1])}"
                ),
            )
        )
        touch_layer(line.layer)["pathCount"] += 1

    for arc in arcs:
        start_point = _screen_point(
            _point_on_circle(arc.center, arc.radius, arc.start_angle_deg),
            min_x=min_x,
            max_y=max_y,
        )
        end_point = _screen_point(
            _point_on_circle(arc.center, arc.radius, arc.end_angle_deg),
            min_x=min_x,
            max_y=max_y,
        )
        large_arc_flag = 1 if abs(arc.sweep_angle_deg) > 180.0 + ANGLE_EPSILON else 0
        sweep_flag = 0 if arc.sweep_angle_deg >= 0.0 else 1
        path_records.append(
            _build_path_record(
                arc.layer,
                _semantic_kind_for_layer(arc.layer),
                (
                    f"M {_format_number(start_point[0])} {_format_number(start_point[1])} "
                    f"A {_format_number(arc.radius)} {_format_number(arc.radius)} 0 "
                    f"{large_arc_flag} {sweep_flag} {_format_number(end_point[0])} {_format_number(end_point[1])}"
                ),
            )
        )
        touch_layer(arc.layer)["pathCount"] += 1

    for circle in circles:
        center = _screen_point(circle.center, min_x=min_x, max_y=max_y)
        circle_records.append(
            {
                "layer": circle.layer,
                "kind": _semantic_kind_for_layer(circle.layer),
                "cx": _format_number(center[0]),
                "cy": _format_number(center[1]),
                "r": _format_number(circle.radius),
            }
        )
        touch_layer(circle.layer)["circleCount"] += 1

    return {
        "schemaVersion": DXF_RENDER_SCHEMA_VERSION,
        "fileRef": file_ref,
        "sourceUnits": int(getattr(document, "units", 0) or 0),
        "defaultThicknessMm": 0.0,
        "bounds": {
            "minX": 0.0,
            "minY": 0.0,
            "maxX": _format_number(width),
            "maxY": _format_number(height),
            "width": _format_number(width),
            "height": _format_number(height),
        },
        "counts": {
            "paths": len(path_records),
            "circles": len(circle_records),
            "entities": len(path_records) + len(circle_records),
        },
        "layers": [layer_summary[name] for name in sorted(layer_summary)],
        "geometry": {
            "lines": [
                {
                    "layer": line.layer,
                    "kind": _semantic_kind_for_layer(line.layer),
                    "start": [_format_number(line.start[0]), _format_number(line.start[1])],
                    "end": [_format_number(line.end[0]), _format_number(line.end[1])],
                }
                for line in lines
            ],
            "arcs": [
                {
                    "layer": arc.layer,
                    "kind": _semantic_kind_for_layer(arc.layer),
                    "center": [_format_number(arc.center[0]), _format_number(arc.center[1])],
                    "radius": _format_number(arc.radius),
                    "startAngleDeg": _format_number(arc.start_angle_deg),
                    "sweepAngleDeg": _format_number(arc.sweep_angle_deg),
                }
                for arc in arcs
            ],
            "circles": [
                {
                    "layer": circle.layer,
                    "kind": _semantic_kind_for_layer(circle.layer),
                    "center": [_format_number(circle.center[0]), _format_number(circle.center[1])],
                    "radius": _format_number(circle.radius),
                }
                for circle in circles
            ],
        },
        "paths": path_records,
        "circles": circle_records,
    }

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "cadpy"
version = "0.3.6"
description = "Shared STEP/GLB topology artifact generation runtime for CAD skills."
requires-python = ">=3.11"
dependencies = [
  "build123d",
  "cadquery-ocp",
]

[project.scripts]
cadpy-step-artifact = "cadpy.step_artifact:main"

[tool.setuptools.packages.find]
where = ["src"]
include = ["cadpy*"]

[tool.setuptools.package-data]
cadpy = ["py.typed"]

"""Shared CAD artifact generation runtime."""

__all__ = [
    "AssemblyHelper",
    "MateRelation",
    "MateTarget",
    "ensure_step_glb_artifact",
    "label_text",
    "label_shape",
    "target",
    "validate_step_glb_artifact",
]


def __getattr__(name: str):
    if name in {"ensure_step_glb_artifact", "validate_step_glb_artifact"}:
        from cadpy.api import ensure_step_glb_artifact, validate_step_glb_artifact

        return {
            "ensure_step_glb_artifact": ensure_step_glb_artifact,
            "validate_step_glb_artifact": validate_step_glb_artifact,
        }[name]
    if name in {"AssemblyHelper", "MateRelation", "MateTarget", "label_shape", "label_text", "target"}:
        from cadpy.assembly import AssemblyHelper, MateRelation, MateTarget, label_shape, label_text, target

        return {
            "AssemblyHelper": AssemblyHelper,
            "MateRelation": MateRelation,
            "MateTarget": MateTarget,
            "label_text": label_text,
            "label_shape": label_shape,
            "target": target,
        }[name]
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

from __future__ import annotations

import hashlib
import json
import math
from typing import Any

from . import lookup


AXIS_NAMES = ("x", "y", "z")
AXIS_INDEX = {name: index for index, name in enumerate(AXIS_NAMES)}
AXIS_ALIGNMENT_THRESHOLD = 0.985


def _float_triplet(value: object) -> tuple[float, float, float] | None:
    if not isinstance(value, (list, tuple)) or len(value) != 3:
        return None
    try:
        return (float(value[0]), float(value[1]), float(value[2]))
    except (TypeError, ValueError):
        return None


def _normalize(vector: tuple[float, float, float] | None) -> tuple[float, float, float] | None:
    if vector is None:
        return None
    length = math.sqrt(sum(component * component for component in vector))
    if length <= 1e-12:
        return None
    return tuple(component / length for component in vector)


def dominant_axis(
    vector: object,
    *,
    aligned_threshold: float = AXIS_ALIGNMENT_THRESHOLD,
) -> dict[str, object] | None:
    normalized = _normalize(_float_triplet(vector))
    if normalized is None:
        return None
    magnitudes = [abs(component) for component in normalized]
    axis_index = max(range(3), key=lambda index: magnitudes[index])
    component = normalized[axis_index]
    return {
        "axis": AXIS_NAMES[axis_index],
        "index": axis_index,
        "sign": 1 if component >= 0.0 else -1,
        "component": component,
        "magnitude": magnitudes[axis_index],
        "aligned": magnitudes[axis_index] >= aligned_threshold,
    }


def bbox_size(bbox: object) -> list[float] | None:
    if not isinstance(bbox, dict):
        return None
    min_point = _float_triplet(bbox.get("min"))
    max_point = _float_triplet(bbox.get("max"))
    if min_point is None or max_point is None:
        return None
    return [
        float(max_point[0] - min_point[0]),
        float(max_point[1] - min_point[1]),
        float(max_point[2] - min_point[2]),
    ]


def bbox_center(bbox: object) -> list[float] | None:
    if not isinstance(bbox, dict):
        return None
    min_point = _float_triplet(bbox.get("min"))
    max_point = _float_triplet(bbox.get("max"))
    if min_point is None or max_point is None:
        return None
    return [
        float((min_point[0] + max_point[0]) * 0.5),
        float((min_point[1] + max_point[1]) * 0.5),
        float((min_point[2] + max_point[2]) * 0.5),
    ]


def bbox_diag(bbox: object) -> float | None:
    size = bbox_size(bbox)
    if size is None:
        return None
    return math.sqrt(sum(component * component for component in size))


def bbox_facts(bbox: object) -> dict[str, object]:
    facts: dict[str, object] = {}
    if not isinstance(bbox, dict):
        return facts
    size = bbox_size(bbox)
    center = bbox_center(bbox)
    diag = bbox_diag(bbox)
    if size is not None:
        facts["size"] = size
        extent_axis = dominant_axis(size, aligned_threshold=0.0)
        if extent_axis is not None:
            facts["extentAxis"] = extent_axis["axis"]
    if center is not None:
        facts["center"] = center
    if diag is not None:
        facts["diag"] = diag
    return facts


def _dot(left: tuple[float, float, float], right: tuple[float, float, float]) -> float:
    return (left[0] * right[0]) + (left[1] * right[1]) + (left[2] * right[2])


def _transform_tuple(value: object) -> tuple[float, ...] | None:
    if not isinstance(value, (list, tuple)) or len(value) != 16:
        return None
    try:
        transform = tuple(float(item) for item in value)
    except (TypeError, ValueError):
        return None
    if not all(math.isfinite(item) for item in transform):
        return None
    return transform


def _axis_alignment_payload(vector: object) -> dict[str, object] | None:
    alignment = dominant_axis(vector)
    if alignment is None:
        return None
    return {
        "axis": alignment["axis"],
        "sign": alignment["sign"],
        "aligned": alignment["aligned"],
        "magnitude": alignment["magnitude"],
    }


def transform_frame_facts(transform: object) -> dict[str, object] | None:
    matrix = _transform_tuple(transform)
    if matrix is None:
        return None
    local_x = _normalize((matrix[0], matrix[4], matrix[8]))
    local_y = _normalize((matrix[1], matrix[5], matrix[9]))
    local_z = _normalize((matrix[2], matrix[6], matrix[10]))
    return {
        "translation": [matrix[3], matrix[7], matrix[11]],
        "localAxes": {
            "x": list(local_x) if local_x is not None else [matrix[0], matrix[4], matrix[8]],
            "y": list(local_y) if local_y is not None else [matrix[1], matrix[5], matrix[9]],
            "z": list(local_z) if local_z is not None else [matrix[2], matrix[6], matrix[10]],
        },
        "transform": list(matrix),
    }


def positioning_facts_for_row(
    selector_type: str,
    row: dict[str, object],
    index: lookup.SelectorIndex | None = None,
) -> dict[str, object]:
    bbox = row.get("bbox")
    center = _float_triplet(row.get("center")) or _float_triplet(bbox_center(bbox))
    facts: dict[str, object] = {
        "selectorType": selector_type,
    }
    if index is not None and row.get("id"):
        facts["selector"] = lookup.display_selector(str(row["id"]), index)
    if isinstance(bbox, dict):
        facts["bbox"] = bbox
        bbox_fact_payload = bbox_facts(bbox)
        if bbox_fact_payload:
            facts["bboxFacts"] = bbox_fact_payload
    if center is not None:
        facts["center"] = list(center)

    if selector_type == "occurrence":
        facts["kind"] = "frame"
        name = str(row.get("name") or row.get("sourceName") or "").strip()
        if name:
            facts["name"] = name
        frame = transform_frame_facts(row.get("transform"))
        if frame is not None:
            facts.update(frame)
        return facts

    if selector_type == "shape":
        facts["kind"] = str(row.get("kind") or "shape")
        if index is not None and row.get("occurrenceId"):
            facts["occurrenceId"] = lookup.display_selector(str(row["occurrenceId"]), index)
        return facts

    if selector_type == "vertex":
        facts["kind"] = "point"
        if center is not None:
            facts["point"] = list(center)
        if index is not None and row.get("occurrenceId"):
            facts["occurrenceId"] = lookup.display_selector(str(row["occurrenceId"]), index)
        return facts

    params = row.get("params") if isinstance(row.get("params"), dict) else {}
    if index is not None and row.get("occurrenceId"):
        facts["occurrenceId"] = lookup.display_selector(str(row["occurrenceId"]), index)
    if index is not None and row.get("shapeId"):
        facts["shapeId"] = lookup.display_selector(str(row["shapeId"]), index)

    if selector_type == "face":
        surface_type = str(row.get("surfaceType") or "").lower()
        facts["kind"] = surface_type or "face"
        if row.get("area") not in {None, ""}:
            facts["area"] = float(row["area"])

        normal = _normalize(
            _float_triplet(row.get("normal"))
            or _float_triplet(params.get("normal"))
            or _float_triplet(params.get("axis"))
        )
        point = _float_triplet(params.get("origin")) or center
        if point is not None:
            facts["origin"] = list(point)
        if normal is not None:
            facts["normal"] = list(normal)
            alignment = _axis_alignment_payload(normal)
            if alignment is not None:
                facts["axisAlignment"] = alignment
                facts["normalAxis"] = {
                    "axis": alignment["axis"],
                    "sign": alignment["sign"],
                    "aligned": alignment["aligned"],
                }
                if bool(alignment["aligned"]) and point is not None:
                    axis = str(alignment["axis"])
                    facts["axis"] = axis
                    facts["coordinate"] = float(point[AXIS_INDEX[axis]])
            if point is not None:
                facts["planeOffset"] = _dot(normal, point)

        radius = params.get("radius")
        if radius not in {None, ""}:
            facts["radius"] = float(radius)
        if surface_type in {"cylinder", "cone", "torus"}:
            axis_vector = _normalize(_float_triplet(params.get("axis")))
            if axis_vector is not None:
                facts["axisVector"] = list(axis_vector)
                facts["axisAlignment"] = _axis_alignment_payload(axis_vector)
        if surface_type == "sphere":
            sphere_center = _float_triplet(params.get("center")) or center
            if sphere_center is not None:
                facts["center"] = list(sphere_center)
        return facts

    curve_type = str(row.get("curveType") or "").lower()
    facts["kind"] = curve_type or "edge"
    if row.get("length") not in {None, ""}:
        facts["length"] = float(row["length"])
    if curve_type == "line":
        origin = _float_triplet(params.get("origin")) or center
        direction = _normalize(_float_triplet(params.get("direction")))
        if origin is not None:
            facts["origin"] = list(origin)
        if direction is not None:
            facts["direction"] = list(direction)
            alignment = _axis_alignment_payload(direction)
            if alignment is not None:
                facts["axisAlignment"] = alignment
                if bool(alignment["aligned"]) and center is not None:
                    axis = str(alignment["axis"])
                    facts["axis"] = axis
                    facts["coordinate"] = float(center[AXIS_INDEX[axis]])
    elif curve_type in {"circle", "ellipse"}:
        circle_center = _float_triplet(params.get("center")) or center
        axis_vector = _normalize(_float_triplet(params.get("axis")))
        if circle_center is not None:
            facts["center"] = list(circle_center)
        if axis_vector is not None:
            facts["axisVector"] = list(axis_vector)
            facts["axisAlignment"] = _axis_alignment_payload(axis_vector)
        for radius_key in ("radius", "majorRadius", "minorRadius"):
            if params.get(radius_key) not in {None, ""}:
                facts[radius_key] = float(params[radius_key])
    return facts


def positioning_point(facts: dict[str, object]) -> list[float] | None:
    for key in ("point", "origin", "center", "translation"):
        point = _float_triplet(facts.get(key))
        if point is not None:
            return list(point)
    bbox = facts.get("bbox")
    center = bbox_center(bbox)
    return center if center is not None else None


def positioning_coordinate(
    facts: dict[str, object],
    axis: str,
) -> tuple[float, str] | None:
    normalized_axis = str(axis or "").strip().lower()
    if normalized_axis not in AXIS_INDEX:
        return None
    if str(facts.get("axis") or "") == normalized_axis and facts.get("coordinate") not in {None, ""}:
        return float(facts["coordinate"]), "coordinate"
    point = positioning_point(facts)
    if point is not None:
        return float(point[AXIS_INDEX[normalized_axis]]), "point"
    return None


def infer_positioning_axis(*fact_payloads: dict[str, object]) -> str | None:
    axes: list[str] = []
    for facts in fact_payloads:
        alignment = facts.get("axisAlignment")
        if not isinstance(alignment, dict) or not bool(alignment.get("aligned")):
            continue
        axis = str(alignment.get("axis") or "")
        if axis in AXIS_INDEX:
            axes.append(axis)
    if axes and all(axis == axes[0] for axis in axes):
        return axes[0]
    for facts in fact_payloads:
        axis = str(facts.get("axis") or "")
        if axis in AXIS_INDEX:
            return axis
    return None


def vector_relationship(
    left: object,
    right: object,
    *,
    aligned_threshold: float = AXIS_ALIGNMENT_THRESHOLD,
) -> dict[str, object] | None:
    left_vector = _normalize(_float_triplet(left))
    right_vector = _normalize(_float_triplet(right))
    if left_vector is None or right_vector is None:
        return None
    dot = _dot(left_vector, right_vector)
    abs_dot = abs(dot)
    if dot <= -aligned_threshold:
        relation = "opposed"
    elif dot >= aligned_threshold:
        relation = "parallel"
    elif abs_dot <= 1.0 - aligned_threshold:
        relation = "perpendicular"
    else:
        relation = "angled"
    return {
        "relation": relation,
        "dot": dot,
        "aligned": abs_dot >= aligned_threshold,
    }


def geometry_facts_for_row(
    selector_type: str,
    row: dict[str, object],
    index: lookup.SelectorIndex | None = None,
) -> dict[str, object]:
    facts = bbox_facts(row.get("bbox"))
    if selector_type in {"occurrence", "shape"}:
        return facts

    params = row.get("params") if isinstance(row.get("params"), dict) else {}
    center = _float_triplet(row.get("center"))
    if center is not None:
        facts.setdefault("center", list(center))

    if selector_type == "vertex":
        if row.get("edgeCount") not in {None, ""}:
            facts["edgeCount"] = int(row["edgeCount"])
        if index is not None and row.get("id"):
            facts["selector"] = lookup.display_selector(str(row["id"]), index)
        return facts

    axis_vector = _float_triplet(params.get("axis"))
    direction_vector = _float_triplet(params.get("direction"))
    normal_vector = _float_triplet(row.get("normal")) or _float_triplet(params.get("normal")) or axis_vector

    if selector_type == "face":
        surface_type = str(row.get("surfaceType") or "")
        if surface_type:
            facts["surfaceType"] = surface_type
        if row.get("area") not in {None, ""}:
            facts["area"] = float(row["area"])
        normal_axis = dominant_axis(normal_vector)
        if normal_axis is not None:
            facts["normalAxis"] = {
                "axis": normal_axis["axis"],
                "sign": normal_axis["sign"],
                "aligned": normal_axis["aligned"],
            }
            if center is not None and bool(normal_axis["aligned"]):
                facts["planeCoordinate"] = center[int(normal_axis["index"])]
        radius = params.get("radius")
        if radius not in {None, ""}:
            facts["radius"] = float(radius)
        if surface_type == "plane" and axis_vector is not None:
            facts["axis"] = list(axis_vector)
        if index is not None and row.get("id"):
            facts["selector"] = lookup.display_selector(str(row["id"]), index)
        return facts

    curve_type = str(row.get("curveType") or "")
    if curve_type:
        facts["curveType"] = curve_type
    if row.get("length") not in {None, ""}:
        facts["length"] = float(row["length"])
    direction_axis = dominant_axis(direction_vector or axis_vector or bbox_size(row.get("bbox")))
    if direction_axis is not None:
        facts["directionAxis"] = {
            "axis": direction_axis["axis"],
            "sign": direction_axis["sign"],
            "aligned": direction_axis["aligned"],
        }
    radius = params.get("radius")
    if radius not in {None, ""}:
        facts["radius"] = float(radius)
    if index is not None and row.get("id"):
        facts["selector"] = lookup.display_selector(str(row["id"]), index)
    return facts


def _merge_bboxes(boxes: list[dict[str, object]]) -> dict[str, object]:
    min_x = min(float(box["min"][0]) for box in boxes)
    min_y = min(float(box["min"][1]) for box in boxes)
    min_z = min(float(box["min"][2]) for box in boxes)
    max_x = max(float(box["max"][0]) for box in boxes)
    max_y = max(float(box["max"][1]) for box in boxes)
    max_z = max(float(box["max"][2]) for box in boxes)
    return {
        "min": [min_x, min_y, min_z],
        "max": [max_x, max_y, max_z],
    }


def major_planar_face_groups(
    index: lookup.SelectorIndex,
    *,
    coordinate_tolerance: float = 1e-3,
    min_area_ratio: float = 0.05,
    limit: int = 12,
) -> list[dict[str, object]]:
    planar_rows = [
        row for row in index.faces if str(row.get("surfaceType") or "").lower() == "plane"
    ]
    total_planar_area = sum(float(row.get("area") or 0.0) for row in planar_rows)
    grouped: dict[tuple[str, int], dict[str, object]] = {}

    for row in planar_rows:
        facts = geometry_facts_for_row("face", row, index)
        normal_axis = facts.get("normalAxis")
        plane_coordinate = facts.get("planeCoordinate")
        if not isinstance(normal_axis, dict) or plane_coordinate in {None, ""}:
            continue
        axis = str(normal_axis.get("axis") or "")
        if axis not in AXIS_INDEX:
            continue
        coordinate = float(plane_coordinate)
        bucket = int(round(coordinate / coordinate_tolerance)) if coordinate_tolerance > 0 else 0
        key = (axis, bucket)
        bbox = row.get("bbox")
        if not isinstance(bbox, dict):
            continue
        group = grouped.get(key)
        if group is None:
            group = {
                "axis": axis,
                "coordinate": 0.0,
                "normalSign": int(normal_axis.get("sign") or 1),
                "faceCount": 0,
                "totalArea": 0.0,
                "bboxParts": [],
                "selectors": [],
            }
            grouped[key] = group
        area = float(row.get("area") or 0.0)
        group["coordinate"] = float(group["coordinate"]) + (coordinate * max(area, 1e-9))
        group["faceCount"] = int(group["faceCount"]) + 1
        group["totalArea"] = float(group["totalArea"]) + area
        group["bboxParts"].append(bbox)
        selector = lookup.display_selector(str(row.get("id") or ""), index)
        if selector:
            group["selectors"].append(selector)

    result: list[dict[str, object]] = []
    for group in grouped.values():
        total_area = float(group["totalArea"])
        if total_planar_area > 0.0 and total_area / total_planar_area < min_area_ratio:
            continue
        weighted_coordinate = float(group["coordinate"]) / max(total_area, 1e-9)
        merged_bbox = _merge_bboxes(list(group["bboxParts"]))
        result.append(
            {
                "axis": group["axis"],
                "coordinate": weighted_coordinate,
                "normalSign": group["normalSign"],
                "faceCount": group["faceCount"],
                "totalArea": total_area,
                "bbox": merged_bbox,
                "selectors": sorted(set(str(selector) for selector in group["selectors"] if selector)),
            }
        )

    result.sort(key=lambda item: (-float(item["totalArea"]), str(item["axis"]), float(item["coordinate"])))
    return result[: max(int(limit), 0)]


def _table_rows(manifest: dict[str, Any], table_name: str, columns_name: str) -> list[dict[str, Any]]:
    columns = manifest.get("tables", {}).get(columns_name)
    rows = manifest.get(table_name)
    if not isinstance(columns, list) or not isinstance(rows, list):
        return []
    materialized: list[dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, list):
            continue
        materialized.append({str(columns[index]): row[index] for index in range(min(len(columns), len(row)))})
    return materialized


def _stable_hash(payload: object) -> str:
    encoded = json.dumps(payload, separators=(",", ":"), sort_keys=True, ensure_ascii=True)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def selector_manifest_diff(
    old_manifest: dict[str, Any] | None,
    new_manifest: dict[str, Any],
) -> dict[str, object]:
    if old_manifest is None:
        return {
            "hasPrevious": False,
            "topologyChanged": False,
            "geometryChanged": False,
            "bboxChanged": False,
            "countDelta": {},
        }

    old_faces = _table_rows(old_manifest, "faces", "faceColumns")
    new_faces = _table_rows(new_manifest, "faces", "faceColumns")
    old_edges = _table_rows(old_manifest, "edges", "edgeColumns")
    new_edges = _table_rows(new_manifest, "edges", "edgeColumns")
    old_vertices = _table_rows(old_manifest, "vertices", "vertexColumns")
    new_vertices = _table_rows(new_manifest, "vertices", "vertexColumns")

    old_topology = {
        "occurrences": [row.get("id") for row in _table_rows(old_manifest, "occurrences", "occurrenceColumns")],
        "shapes": [row.get("id") for row in _table_rows(old_manifest, "shapes", "shapeColumns")],
        "faces": [row.get("id") for row in old_faces],
        "edges": [row.get("id") for row in old_edges],
        "vertices": [row.get("id") for row in old_vertices],
        "faceCount": old_manifest.get("stats", {}).get("faceCount"),
        "edgeCount": old_manifest.get("stats", {}).get("edgeCount"),
        "vertexCount": old_manifest.get("stats", {}).get("vertexCount"),
    }
    new_topology = {
        "occurrences": [row.get("id") for row in _table_rows(new_manifest, "occurrences", "occurrenceColumns")],
        "shapes": [row.get("id") for row in _table_rows(new_manifest, "shapes", "shapeColumns")],
        "faces": [row.get("id") for row in new_faces],
        "edges": [row.get("id") for row in new_edges],
        "vertices": [row.get("id") for row in new_vertices],
        "faceCount": new_manifest.get("stats", {}).get("faceCount"),
        "edgeCount": new_manifest.get("stats", {}).get("edgeCount"),
        "vertexCount": new_manifest.get("stats", {}).get("vertexCount"),
    }
    old_geometry = {
        "bbox": old_manifest.get("bbox"),
        "faces": [
            {
                "id": row.get("id"),
                "surfaceType": row.get("surfaceType"),
                "center": row.get("center"),
                "normal": row.get("normal"),
                "bbox": row.get("bbox"),
                "area": row.get("area"),
            }
            for row in old_faces
        ],
        "edges": [
            {
                "id": row.get("id"),
                "curveType": row.get("curveType"),
                "center": row.get("center"),
                "bbox": row.get("bbox"),
                "length": row.get("length"),
            }
            for row in old_edges
        ],
        "vertices": [
            {
                "id": row.get("id"),
                "center": row.get("center"),
                "bbox": row.get("bbox"),
            }
            for row in old_vertices
        ],
    }
    new_geometry = {
        "bbox": new_manifest.get("bbox"),
        "faces": [
            {
                "id": row.get("id"),
                "surfaceType": row.get("surfaceType"),
                "center": row.get("center"),
                "normal": row.get("normal"),
                "bbox": row.get("bbox"),
                "area": row.get("area"),
            }
            for row in new_faces
        ],
        "edges": [
            {
                "id": row.get("id"),
                "curveType": row.get("curveType"),
                "center": row.get("center"),
                "bbox": row.get("bbox"),
                "length": row.get("length"),
            }
            for row in new_edges
        ],
        "vertices": [
            {
                "id": row.get("id"),
                "center": row.get("center"),
                "bbox": row.get("bbox"),
            }
            for row in new_vertices
        ],
    }

    count_delta = {
        "faceCount": int(new_manifest.get("stats", {}).get("faceCount") or 0)
        - int(old_manifest.get("stats", {}).get("faceCount") or 0),
        "edgeCount": int(new_manifest.get("stats", {}).get("edgeCount") or 0)
        - int(old_manifest.get("stats", {}).get("edgeCount") or 0),
        "vertexCount": int(new_manifest.get("stats", {}).get("vertexCount") or 0)
        - int(old_manifest.get("stats", {}).get("vertexCount") or 0),
        "shapeCount": int(new_manifest.get("stats", {}).get("shapeCount") or 0)
        - int(old_manifest.get("stats", {}).get("shapeCount") or 0),
    }

    return {
        "hasPrevious": True,
        "topologyChanged": _stable_hash(old_topology) != _stable_hash(new_topology),
        "geometryChanged": _stable_hash(old_geometry) != _stable_hash(new_geometry),
        "bboxChanged": old_manifest.get("bbox") != new_manifest.get("bbox"),
        "countDelta": count_delta,
    }


def view_name_for_axis(axis: str, sign: int) -> str:
    normalized_sign = 1 if sign >= 0 else -1
    if axis == "x":
        return "right" if normalized_sign > 0 else "left"
    if axis == "y":
        return "top" if normalized_sign > 0 else "bottom"
    return "front" if normalized_sign > 0 else "back"


def aligned_view_name_for_facts(
    selector_type: str,
    facts: dict[str, object],
) -> str | None:
    if selector_type == "face":
        normal_axis = facts.get("normalAxis")
        if isinstance(normal_axis, dict):
            axis = str(normal_axis.get("axis") or "")
            sign = int(normal_axis.get("sign") or 1)
            if axis in AXIS_INDEX:
                return view_name_for_axis(axis, sign)
        return None

    if selector_type != "edge":
        return None
    direction_axis = facts.get("directionAxis")
    if not isinstance(direction_axis, dict):
        return None
    axis = str(direction_axis.get("axis") or "")
    if axis == "x":
        return "front"
    if axis == "y":
        return "front"
    if axis == "z":
        return "top"
    return None

from __future__ import annotations

from cadpy.step_artifacts import ensure_step_topology_artifact as ensure_step_glb_artifact
from cadpy.step_targets import validate_step_topology_artifact as validate_step_glb_artifact

__all__ = ["ensure_step_glb_artifact", "validate_step_glb_artifact"]

from __future__ import annotations

import os
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

from cadpy.assembly_spec import (
    IDENTITY_TRANSFORM,
    AssemblySpec,
    assembly_spec_children,
    multiply_transforms,
)
from cadpy.glb_topology import read_step_topology_manifest_from_glb
from cadpy.render import existing_part_glb_path, part_glb_path, relative_to_repo, sha256_file


ASSEMBLY_COMPOSITION_SCHEMA_VERSION = 1
TOPOLOGY_COUNT_KEYS = ("shapeCount", "faceCount", "edgeCount")
_SOURCE_TOPOLOGY_COUNTS_CACHE: dict[tuple[str, int, int], dict[str, int]] = {}
_SOURCE_ASSEMBLY_PAYLOAD_CACHE: dict[tuple[str, int, int], dict[str, Any] | None] = {}


class AssemblyCompositionError(ValueError):
    pass


def component_name(instance_path: Sequence[str]) -> str:
    return "__".join(str(part) for part in instance_path if str(part)) or "root"


def _relative_to_topology(topology_path: Path, target_path: Path) -> str:
    return os.path.relpath(target_path.resolve(), start=topology_path.resolve().parent).replace(os.sep, "/")


def _versioned_relative_url(topology_path: Path, target_path: Path, content_hash: str) -> str:
    suffix = f"?v={content_hash}" if content_hash else ""
    return f"{_relative_to_topology(topology_path, target_path)}{suffix}"


def _assembly_mesh_payload(topology_path: Path, mesh_path: Path) -> dict[str, Any]:
    mesh_hash = sha256_file(mesh_path) if mesh_path.exists() else ""
    return {
        "url": _versioned_relative_url(topology_path, mesh_path, mesh_hash)
        if mesh_hash
        else _relative_to_topology(topology_path, mesh_path),
        "hash": mesh_hash,
        "addressing": "gltf-node-extras",
        "occurrenceIdKey": "cadOccurrenceId",
    }


def build_linked_assembly_composition(
    *,
    cad_ref: str,
    topology_path: Path,
    topology_manifest: dict[str, Any],
    assembly_spec: AssemblySpec,
    entries_by_step_path: Mapping[Path, object],
    read_assembly_spec: Callable[[Path], AssemblySpec],
    mesh_path: Path,
) -> dict[str, Any]:
    occurrences = _rows(topology_manifest, "occurrences", "occurrenceColumns")
    if not occurrences:
        raise AssemblyCompositionError(f"Assembly topology has no occurrences: {cad_ref}")
    component_occurrences = _component_occurrences(topology_manifest)
    root_occurrence = occurrences[0]
    children = [
        _linked_instance_node(
            cad_ref=cad_ref,
            topology_path=topology_path,
            instance=instance,
            instance_path=(instance.instance_id,),
            target_occurrence=None,
            parent_world_transform=IDENTITY_TRANSFORM,
            parent_use_source_colors=True,
            all_occurrences=occurrences,
            component_occurrences=component_occurrences,
            entries_by_step_path=entries_by_step_path,
            read_assembly_spec=read_assembly_spec,
            stack=(assembly_spec.assembly_path.resolve().as_posix(),),
        )
        for instance in assembly_spec_children(assembly_spec)
    ]
    if not children:
        raise AssemblyCompositionError(f"Assembly {cad_ref} has no component instances")
    return {
        "schemaVersion": ASSEMBLY_COMPOSITION_SCHEMA_VERSION,
        "mode": "linked",
        "mesh": _assembly_mesh_payload(topology_path, mesh_path),
        "root": _assembly_root_node(cad_ref, root_occurrence, children),
    }


def build_native_assembly_composition(
    *,
    cad_ref: str,
    topology_path: Path,
    topology_manifest: dict[str, Any],
    mesh_path: Path,
) -> dict[str, Any]:
    occurrences = _rows(topology_manifest, "occurrences", "occurrenceColumns")
    if not occurrences:
        raise AssemblyCompositionError(f"Assembly topology has no occurrences: {cad_ref}")
    by_id = {
        str(row.get("id") or "").strip(): row
        for row in occurrences
        if str(row.get("id") or "").strip()
    }
    children_by_parent: dict[str, list[dict[str, Any]]] = {}
    top_level: list[dict[str, Any]] = []
    for row in occurrences:
        parent_id = str(row.get("parentId") or "").strip()
        if parent_id:
            children_by_parent.setdefault(parent_id, []).append(row)
        else:
            top_level.append(row)

    root_occurrence = top_level[0] if len(top_level) == 1 else occurrences[0]
    root_children = top_level
    if len(top_level) == 1 and not children_by_parent.get(str(top_level[0].get("id") or "").strip()):
        root_children = top_level
    elif len(top_level) == 1:
        root_children = children_by_parent.get(str(top_level[0].get("id") or "").strip(), [])

    children = [
        _native_occurrence_node(
            row,
            children_by_parent=children_by_parent,
            topology_path=topology_path,
            parent_world_transform=IDENTITY_TRANSFORM,
        )
        for row in root_children
    ]
    if not children:
        row = root_occurrence
        children = [
            _native_part_node(
                row,
                topology_path=topology_path,
                parent_world_transform=IDENTITY_TRANSFORM,
            )
        ]
    return {
        "schemaVersion": ASSEMBLY_COMPOSITION_SCHEMA_VERSION,
        "mode": "native",
        "mesh": _assembly_mesh_payload(topology_path, mesh_path),
        "root": _assembly_root_node(cad_ref, root_occurrence, children),
    }


def _linked_instance_node(
    *,
    cad_ref: str,
    topology_path: Path,
    instance: object,
    instance_path: tuple[str, ...],
    target_occurrence: Mapping[str, Any] | None,
    parent_world_transform: tuple[float, ...],
    parent_use_source_colors: bool,
    all_occurrences: Sequence[dict[str, Any]],
    component_occurrences: Sequence[dict[str, Any]],
    entries_by_step_path: Mapping[Path, object],
    read_assembly_spec: Callable[[Path], AssemblySpec],
    stack: tuple[str, ...],
) -> dict[str, Any]:
    instance_source_path = Path(instance.source_path).resolve() if instance.source_path is not None else None
    source_spec = entries_by_step_path.get(instance_source_path) if instance_source_path is not None else None
    child_kind = str(getattr(source_spec, "kind", "") or "")
    instance_transform = tuple(float(value) for value in instance.transform)
    world_transform = multiply_transforms(parent_world_transform, instance_transform)
    source_step_path = getattr(source_spec, "step_path", None) if source_spec is not None else None
    source_path = _relative_to_topology(topology_path, Path(source_step_path)) if source_step_path is not None else (instance.path or "")
    # Instance names are the authored assembly labels; fall back to source/path
    # stems only for legacy or incomplete specs.
    display_name = str(
        instance.name
        or (
            Path(source_step_path).stem
            if source_step_path is not None
            else Path(instance.path or "").stem
        )
        or instance_path[-1]
    ).strip()
    use_source_colors = parent_use_source_colors and bool(instance.use_source_colors)

    if instance.children:
        occurrence = target_occurrence or _find_occurrence_by_component_name(
            component_name(instance_path),
            all_occurrences,
            cad_ref,
        )
        occurrence_id = str(occurrence.get("id") or component_name(instance_path)).strip() if occurrence else component_name(instance_path)
        occurrence_world_transform = (
            tuple(float(value) for value in occurrence.get("transform"))
            if occurrence and isinstance(occurrence.get("transform"), list) and len(occurrence.get("transform")) == 16
            else world_transform
        )
        target_children_by_parent = _children_by_parent(all_occurrences)
        target_children = target_children_by_parent.get(occurrence_id, [])
        children = [
            _linked_instance_node(
                cad_ref=cad_ref,
                topology_path=topology_path,
                instance=child_instance,
                instance_path=(*instance_path, child_instance.instance_id),
                target_occurrence=_match_occurrence_child_for_instance(
                    target_children,
                    child_instance,
                    index,
                    (*instance_path, child_instance.instance_id),
                ),
                parent_world_transform=occurrence_world_transform,
                parent_use_source_colors=use_source_colors,
                all_occurrences=all_occurrences,
                component_occurrences=component_occurrences,
                entries_by_step_path=entries_by_step_path,
                read_assembly_spec=read_assembly_spec,
                stack=stack,
            )
            for index, child_instance in enumerate(instance.children)
        ]
        return _assembly_node(
            id=occurrence_id,
            occurrence_id=occurrence_id,
            display_name=display_name,
            source_kind="catalog",
            source_path=source_path,
            instance_path=".".join(instance_path),
            use_source_colors=use_source_colors,
            local_transform=instance_transform,
            world_transform=occurrence_world_transform,
            bbox=(occurrence.get("bbox") if occurrence else None) or _merge_bbox([child.get("bbox") for child in children]),
            topology_counts=_sum_public_counts(children),
            children=children,
        )

    if source_spec is None or instance_source_path is None:
        raise AssemblyCompositionError(
            f"{cad_ref} assembly component {component_name(instance_path)} references missing CAD source {instance.path}"
        )

    if child_kind == "assembly":
        stack_key = instance_source_path.as_posix()
        if stack_key in stack:
            cycle = " -> ".join((*stack, stack_key))
            raise AssemblyCompositionError(f"Assembly cycle detected: {cycle}")
        source_source_path = getattr(source_spec, "source_path", None)
        script_path = getattr(source_spec, "script_path", None)
        if source_source_path is None or script_path is None:
            raise AssemblyCompositionError(
                f"{cad_ref} nested assembly {instance.path} must be a generated assembly source"
            )
        child_spec = read_assembly_spec(Path(source_source_path))
        occurrence = target_occurrence or _find_occurrence_by_component_name(
            component_name(instance_path),
            all_occurrences,
            cad_ref,
        )
        occurrence_id = str(occurrence.get("id") or component_name(instance_path)).strip() if occurrence else component_name(instance_path)
        occurrence_world_transform = (
            tuple(float(value) for value in occurrence.get("transform"))
            if occurrence and isinstance(occurrence.get("transform"), list) and len(occurrence.get("transform")) == 16
            else world_transform
        )
        target_children_by_parent = _children_by_parent(all_occurrences)
        target_children = target_children_by_parent.get(occurrence_id, [])
        children = [
            _linked_instance_node(
                cad_ref=cad_ref,
                topology_path=topology_path,
                instance=child_instance,
                instance_path=(*instance_path, child_instance.instance_id),
                target_occurrence=_match_occurrence_child_for_instance(
                    target_children,
                    child_instance,
                    index,
                    (*instance_path, child_instance.instance_id),
                ),
                parent_world_transform=occurrence_world_transform,
                parent_use_source_colors=use_source_colors,
                all_occurrences=all_occurrences,
                component_occurrences=component_occurrences,
                entries_by_step_path=entries_by_step_path,
                read_assembly_spec=read_assembly_spec,
                stack=(*stack, stack_key),
            )
            for index, child_instance in enumerate(assembly_spec_children(child_spec))
        ]
        return _assembly_node(
            id=occurrence_id,
            occurrence_id=occurrence_id,
            display_name=display_name,
            source_kind="catalog",
            source_path=_relative_to_topology(topology_path, Path(source_step_path)) if source_step_path is not None else (instance.path or ""),
            instance_path=".".join(instance_path),
            use_source_colors=use_source_colors,
            local_transform=instance_transform,
            world_transform=occurrence_world_transform,
            bbox=(occurrence.get("bbox") if occurrence else None) or _merge_bbox([child.get("bbox") for child in children]),
            topology_counts=_sum_public_counts(children),
            children=children,
        )

    native_source_assembly = _source_assembly_payload(Path(source_step_path)) if source_step_path is not None else None
    if native_source_assembly is not None:
        occurrence = target_occurrence or _find_occurrence_by_component_name(
            component_name(instance_path),
            all_occurrences,
            cad_ref,
        )
        if occurrence is None:
            raise AssemblyCompositionError(
                f"{cad_ref} assembly topology is missing occurrence {component_name(instance_path)!r}"
            )
        occurrence_id = str(occurrence.get("id") or "").strip()
        return _linked_native_assembly_node(
            topology_path=topology_path,
            source_topology_path=existing_part_glb_path(Path(source_step_path)) or part_glb_path(Path(source_step_path)),
            source_assembly=native_source_assembly,
            source_path=source_path,
            occurrence=occurrence,
            occurrence_id=occurrence_id,
            all_occurrences=all_occurrences,
            display_name=display_name,
            instance_path=".".join(instance_path),
            use_source_colors=use_source_colors,
            local_transform=instance_transform,
            world_transform=tuple(float(value) for value in occurrence.get("transform") or world_transform),
        )

    if child_kind == "part":
        occurrence = target_occurrence or _find_occurrence_by_component_name(
            component_name(instance_path),
            component_occurrences,
            cad_ref,
        )
        if occurrence is None:
            raise AssemblyCompositionError(
                f"{cad_ref} assembly topology is missing occurrence {component_name(instance_path)!r}"
            )
        if source_step_path is None:
            raise AssemblyCompositionError(f"{cad_ref} component {component_name(instance_path)} is missing STEP source")
        source_counts = _source_topology_counts(existing_part_glb_path(Path(source_step_path)) or part_glb_path(Path(source_step_path)))
        occurrence_counts = _occurrence_topology_counts(occurrence)
        if source_counts != occurrence_counts:
            raise AssemblyCompositionError(
                f"{cad_ref} assembly occurrence {occurrence.get('id')!r} count mismatch for "
                f"{source_path}: source={source_counts} assembly={occurrence_counts}"
            )
        occurrence_id = str(occurrence.get("id") or "").strip()
        return _part_node(
            id=occurrence_id,
            occurrence_id=occurrence_id,
            display_name=display_name,
            source_kind="catalog",
            source_path=source_path,
            instance_path=".".join(instance_path),
            use_source_colors=use_source_colors,
            local_transform=instance_transform,
            world_transform=tuple(float(value) for value in occurrence.get("transform") or world_transform),
            bbox=occurrence.get("bbox"),
            topology_counts=_public_topology_counts(occurrence_counts),
        )

    raise AssemblyCompositionError(
        f"{cad_ref} component {component_name(instance_path)} must resolve to a STEP part or assembly source"
    )


def _source_assembly_payload(step_path: Path) -> dict[str, Any] | None:
    source_topology_path = existing_part_glb_path(step_path) or part_glb_path(step_path)
    cache_key = _file_cache_key(source_topology_path)
    if cache_key in _SOURCE_ASSEMBLY_PAYLOAD_CACHE:
        return _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key]
    payload = read_step_topology_manifest_from_glb(source_topology_path)
    if payload is None:
        _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key] = None
        return None
    assembly = payload.get("assembly")
    root = assembly.get("root") if isinstance(assembly, dict) else None
    if isinstance(root, dict) and root.get("children"):
        _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key] = assembly
        return assembly
    if not _manifest_has_native_assembly_structure(payload):
        _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key] = None
        return None
    native_payload = build_native_assembly_composition(
        cad_ref=relative_to_repo(step_path.with_suffix("")),
        topology_path=source_topology_path,
        topology_manifest=payload,
        mesh_path=source_topology_path,
    )
    native_root = native_payload.get("root")
    if not isinstance(native_root, dict) or not native_root.get("children"):
        _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key] = None
        return None
    _SOURCE_ASSEMBLY_PAYLOAD_CACHE[cache_key] = native_payload
    return native_payload


def _manifest_has_native_assembly_structure(payload: Mapping[str, Any]) -> bool:
    occurrences = _rows(dict(payload), "occurrences", "occurrenceColumns")
    if len(occurrences) <= 1:
        return False
    occurrence_ids = {
        str(row.get("id") or "").strip()
        for row in occurrences
        if str(row.get("id") or "").strip()
    }
    for row in occurrences:
        parent_id = str(row.get("parentId") or "").strip()
        if parent_id and parent_id in occurrence_ids:
            return True
    return False


def _linked_native_assembly_node(
    *,
    topology_path: Path,
    source_topology_path: Path,
    source_assembly: Mapping[str, Any],
    source_path: str,
    occurrence: Mapping[str, Any],
    occurrence_id: str,
    all_occurrences: Sequence[dict[str, Any]],
    display_name: str,
    instance_path: str,
    use_source_colors: bool,
    local_transform: Sequence[float],
    world_transform: tuple[float, ...],
) -> dict[str, Any]:
    source_root = source_assembly.get("root")
    if not isinstance(source_root, Mapping):
        raise AssemblyCompositionError(f"Native source assembly is missing root: {relative_to_repo(source_topology_path)}")
    source_root_occurrence_id = str(source_root.get("occurrenceId") or source_root.get("id") or "").strip()
    source_children = source_root.get("children")
    if not isinstance(source_children, list) or not source_children:
        raise AssemblyCompositionError(f"Native source assembly has no children: {relative_to_repo(source_topology_path)}")
    target_children_by_parent = _children_by_parent(all_occurrences)
    target_children = target_children_by_parent.get(occurrence_id, [])
    child_nodes = [
        _clone_native_source_node(
            source_node=_source_node_for_native_target_child(
                source_root,
                source_children,
                target_children,
                target_children_by_parent,
                child,
                index,
            ),
            target_row=child,
            target_children_by_parent=target_children_by_parent,
            topology_path=topology_path,
            source_topology_path=source_topology_path,
            source_root_occurrence_id=source_root_occurrence_id,
            source_path=source_path,
            source_root_target_occurrence_id=occurrence_id,
            target_parent_occurrence_id=occurrence_id,
            parent_world_transform=world_transform,
            parent_instance_path=instance_path,
            parent_use_source_colors=use_source_colors,
        )
        for index, child in enumerate(target_children)
    ]
    if not child_nodes:
        child_nodes = [
            _clone_native_source_node(
                source_node=child,
                target_row=None,
                target_children_by_parent={},
                topology_path=topology_path,
                source_topology_path=source_topology_path,
                source_root_occurrence_id=source_root_occurrence_id,
                target_parent_occurrence_id=occurrence_id,
                parent_world_transform=world_transform,
                parent_instance_path=instance_path,
                parent_use_source_colors=use_source_colors,
            )
            for child in source_children
            if isinstance(child, Mapping)
        ]
    source_root_target_occurrence_id = occurrence_id
    for child in child_nodes:
        if str(child.get("sourceOccurrenceId") or "").strip() == source_root_occurrence_id:
            source_root_target_occurrence_id = str(child.get("occurrenceId") or child.get("id") or occurrence_id).strip()
            break
    child_counts = _sum_public_counts(child_nodes)
    node = _assembly_node(
        id=occurrence_id,
        occurrence_id=occurrence_id,
        display_name=display_name,
        source_kind="native",
        source_path=source_path,
        instance_path=instance_path,
        use_source_colors=use_source_colors,
        local_transform=local_transform,
        world_transform=world_transform,
        bbox=occurrence.get("bbox") or _merge_bbox([child.get("bbox") for child in child_nodes]),
        topology_counts=child_counts if _counts_have_values(child_counts) else _public_topology_counts(_occurrence_topology_counts(occurrence)),
        children=child_nodes,
    )
    _attach_native_source_metadata(
        node,
        source_occurrence_id=source_root_occurrence_id,
        source_root_occurrence_id=source_root_occurrence_id,
        source_root_target_occurrence_id=source_root_target_occurrence_id,
    )
    return node


def _clone_native_source_node(
    *,
    source_node: Mapping[str, Any],
    target_row: Mapping[str, Any] | None,
    target_children_by_parent: Mapping[str, list[dict[str, Any]]],
    topology_path: Path,
    source_topology_path: Path,
    source_root_occurrence_id: str,
    source_path: str,
    source_root_target_occurrence_id: str,
    target_parent_occurrence_id: str,
    parent_world_transform: tuple[float, ...],
    parent_instance_path: str,
    parent_use_source_colors: bool,
) -> dict[str, Any]:
    source_occurrence_id = str(source_node.get("occurrenceId") or source_node.get("id") or "").strip()
    occurrence_id = (
        str(target_row.get("id") or "").strip()
        if target_row is not None
        else _prefix_native_occurrence_id(
            source_occurrence_id,
            source_root_occurrence_id=source_root_occurrence_id,
            target_parent_occurrence_id=target_parent_occurrence_id,
        )
    )
    next_source_root_target_occurrence_id = (
        occurrence_id
        if source_root_occurrence_id and source_occurrence_id == source_root_occurrence_id
        else source_root_target_occurrence_id
    )
    source_world_transform = _transform_tuple(source_node.get("worldTransform"), IDENTITY_TRANSFORM)
    source_local_transform = _transform_tuple(source_node.get("localTransform"), source_world_transform)
    world_transform = _row_transform(target_row) if target_row is not None else multiply_transforms(parent_world_transform, source_local_transform)
    local_transform = (
        _relative_transform(parent_world_transform, world_transform)
        if target_row is not None
        else source_local_transform
    )
    source_children = source_node.get("children")
    target_children = target_children_by_parent.get(occurrence_id, []) if occurrence_id else []
    if target_children and not _source_node_has_children(source_node):
        return _native_source_part_node(
            source_node=source_node,
            target_row=target_row,
            occurrence_id=occurrence_id,
            source_path=source_path,
            source_occurrence_id=source_occurrence_id,
            source_root_occurrence_id=source_root_occurrence_id,
            source_root_target_occurrence_id=next_source_root_target_occurrence_id,
            display_name=_occurrence_display_name(target_row) if target_row is not None else "",
            instance_path=".".join(
                part
                for part in (
                    parent_instance_path,
                    str(target_row.get("path") or "") if target_row is not None else str(source_node.get("instancePath") or source_occurrence_id),
                )
                if part
            ),
            use_source_colors=parent_use_source_colors and source_node.get("useSourceColors") is not False,
            local_transform=local_transform,
            world_transform=world_transform,
            bbox=target_row.get("bbox") if target_row is not None else _transform_bbox(parent_world_transform, source_node.get("bbox")),
        )
    node_children = [
        _clone_native_source_node(
            source_node=_match_source_child_for_target(source_children, child, index),
            target_row=child,
            target_children_by_parent=target_children_by_parent,
            topology_path=topology_path,
            source_topology_path=source_topology_path,
            source_root_occurrence_id=source_root_occurrence_id,
            source_path=source_path,
            source_root_target_occurrence_id=next_source_root_target_occurrence_id,
            target_parent_occurrence_id=target_parent_occurrence_id,
            parent_world_transform=world_transform,
            parent_instance_path=parent_instance_path,
            parent_use_source_colors=parent_use_source_colors,
        )
        for index, child in enumerate(target_children)
    ]
    if not node_children and isinstance(source_children, list) and source_children:
        node_children = [
            _clone_native_source_node(
                source_node=child,
                target_row=None,
                target_children_by_parent={},
                topology_path=topology_path,
                source_topology_path=source_topology_path,
                source_root_occurrence_id=source_root_occurrence_id,
                source_path=source_path,
                source_root_target_occurrence_id=next_source_root_target_occurrence_id,
                target_parent_occurrence_id=occurrence_id,
                parent_world_transform=world_transform,
                parent_instance_path=parent_instance_path,
                parent_use_source_colors=parent_use_source_colors,
            )
            for child in source_children
            if isinstance(child, Mapping)
        ]
    display_name = str(
        _occurrence_display_name(target_row)
        if target_row is not None
        else source_node.get("displayName") or source_node.get("name") or occurrence_id
    ).strip()
    instance_path = ".".join(
        part
        for part in (
            parent_instance_path,
            str(target_row.get("path") or "") if target_row is not None else str(source_node.get("instancePath") or source_occurrence_id),
        )
        if part
    )
    topology_counts = source_node.get("topologyCounts") if isinstance(source_node.get("topologyCounts"), Mapping) else {}
    bbox = target_row.get("bbox") if target_row is not None else _transform_bbox(parent_world_transform, source_node.get("bbox"))
    use_source_colors = parent_use_source_colors and source_node.get("useSourceColors") is not False
    if node_children:
        child_counts = _sum_public_counts(node_children)
        node = _assembly_node(
            id=occurrence_id,
            occurrence_id=occurrence_id,
            display_name=display_name,
            source_kind="native",
            source_path=source_path,
            instance_path=instance_path,
            use_source_colors=use_source_colors,
            local_transform=local_transform,
            world_transform=world_transform,
            bbox=bbox or _merge_bbox([child.get("bbox") for child in node_children]),
            topology_counts=child_counts if _counts_have_values(child_counts) else _public_topology_counts(topology_counts),
            children=node_children,
        )
        _attach_native_source_metadata(
            node,
            source_occurrence_id=source_occurrence_id,
            source_root_occurrence_id=source_root_occurrence_id,
            source_root_target_occurrence_id=next_source_root_target_occurrence_id,
        )
        return node

    return _native_source_part_node(
        source_node=source_node,
        target_row=target_row,
        occurrence_id=occurrence_id,
        source_path=source_path,
        source_occurrence_id=source_occurrence_id,
        source_root_occurrence_id=source_root_occurrence_id,
        source_root_target_occurrence_id=next_source_root_target_occurrence_id,
        display_name=display_name,
        instance_path=instance_path,
        use_source_colors=use_source_colors,
        local_transform=local_transform,
        world_transform=world_transform,
        bbox=bbox,
    )


def _source_node_has_children(source_node: Mapping[str, Any]) -> bool:
    source_children = source_node.get("children")
    return isinstance(source_children, list) and bool(source_children)


def _native_source_part_node(
    *,
    source_node: Mapping[str, Any],
    target_row: Mapping[str, Any] | None,
    occurrence_id: str,
    display_name: str,
    instance_path: str,
    use_source_colors: bool,
    local_transform: Sequence[float],
    world_transform: Sequence[float],
    bbox: Any,
    source_path: str = "",
    source_occurrence_id: str = "",
    source_root_occurrence_id: str = "",
    source_root_target_occurrence_id: str = "",
) -> dict[str, Any]:
    topology_counts = (
        _occurrence_topology_counts(target_row)
        if target_row is not None
        else source_node.get("topologyCounts") if isinstance(source_node.get("topologyCounts"), Mapping) else {}
    )
    node = _part_node(
        id=occurrence_id,
        occurrence_id=occurrence_id,
        display_name=display_name,
        source_kind="native",
        source_path=source_path,
        instance_path=instance_path,
        use_source_colors=use_source_colors,
        local_transform=local_transform,
        world_transform=world_transform,
        bbox=bbox,
        topology_counts=_public_topology_counts(topology_counts),
    )
    _attach_native_source_metadata(
        node,
        source_occurrence_id=source_occurrence_id,
        source_root_occurrence_id=source_root_occurrence_id,
        source_root_target_occurrence_id=source_root_target_occurrence_id,
    )
    return node


def _attach_native_source_metadata(
    node: dict[str, Any],
    *,
    source_occurrence_id: str,
    source_root_occurrence_id: str,
    source_root_target_occurrence_id: str,
) -> None:
    if source_occurrence_id:
        node["sourceOccurrenceId"] = source_occurrence_id
    if source_root_occurrence_id:
        node["sourceRootOccurrenceId"] = source_root_occurrence_id
    if source_root_target_occurrence_id:
        node["sourceRootTargetOccurrenceId"] = source_root_target_occurrence_id


def _prefix_native_occurrence_id(
    source_occurrence_id: str,
    *,
    source_root_occurrence_id: str,
    target_parent_occurrence_id: str,
) -> str:
    if source_root_occurrence_id and source_occurrence_id == source_root_occurrence_id:
        return target_parent_occurrence_id
    prefix = f"{source_root_occurrence_id}."
    if source_root_occurrence_id and source_occurrence_id.startswith(prefix):
        return f"{target_parent_occurrence_id}.{source_occurrence_id[len(prefix):]}"
    suffix = source_occurrence_id[1:] if source_occurrence_id.startswith("o") else source_occurrence_id
    return f"{target_parent_occurrence_id}.{suffix}" if suffix else target_parent_occurrence_id


def _children_by_parent(occurrences: Sequence[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
    children_by_parent: dict[str, list[dict[str, Any]]] = {}
    for row in occurrences:
        parent_id = str(row.get("parentId") or "").strip()
        if parent_id:
            children_by_parent.setdefault(parent_id, []).append(row)
    return children_by_parent


def _match_source_child_for_target(
    source_children: object,
    target_row: Mapping[str, Any],
    index: int,
) -> Mapping[str, Any]:
    candidates = [child for child in source_children if isinstance(child, Mapping)] if isinstance(source_children, list) else []
    if not candidates:
        return {}
    target_names = {
        str(target_row.get("name") or "").strip(),
        str(target_row.get("sourceName") or "").strip(),
        str(target_row.get("id") or "").strip(),
    }
    target_names.discard("")
    for candidate in candidates:
        candidate_names = {
            str(candidate.get("displayName") or "").strip(),
            str(candidate.get("name") or "").strip(),
            str(candidate.get("occurrenceId") or "").strip(),
            str(candidate.get("id") or "").strip(),
        }
        if target_names.intersection(candidate_names):
            return candidate
    if 0 <= index < len(candidates):
        return candidates[index]
    return candidates[-1]


def _source_node_for_native_target_child(
    source_root: Mapping[str, Any],
    source_children: object,
    target_siblings: Sequence[Mapping[str, Any]],
    target_children_by_parent: Mapping[str, list[dict[str, Any]]],
    target_row: Mapping[str, Any],
    index: int,
) -> Mapping[str, Any]:
    target_row_id = str(target_row.get("id") or "").strip()
    target_child_rows = target_children_by_parent.get(target_row_id, []) if target_row_id else []
    source_child_count = len(source_children) if isinstance(source_children, list) else 0
    if (
        len(target_siblings) == 1
        and len(target_child_rows) > 1
        and source_child_count > 1
        and isinstance(source_root, Mapping)
    ):
        return source_root
    return _match_source_child_for_target(source_children, target_row, index)


def _match_occurrence_child_for_instance(
    target_children: Sequence[Mapping[str, Any]],
    instance: object,
    index: int,
    instance_path: tuple[str, ...],
) -> Mapping[str, Any] | None:
    if not target_children:
        return None
    expected_names = {
        str(getattr(instance, "instance_id", "") or "").strip(),
        str(getattr(instance, "name", "") or "").strip(),
        component_name(instance_path),
    }
    expected_names.discard("")
    for child in target_children:
        child_names = {
            str(child.get("name") or "").strip(),
            str(child.get("sourceName") or "").strip(),
            str(child.get("id") or "").strip(),
        }
        if expected_names.intersection(child_names):
            return child
    if 0 <= index < len(target_children):
        return target_children[index]
    return target_children[-1]


def _native_occurrence_node(
    row: dict[str, Any],
    *,
    children_by_parent: Mapping[str, list[dict[str, Any]]],
    topology_path: Path,
    parent_world_transform: tuple[float, ...],
) -> dict[str, Any]:
    row_id = str(row.get("id") or "").strip()
    children = children_by_parent.get(row_id, [])
    if not children:
        return _native_part_node(
            row,
            topology_path=topology_path,
            parent_world_transform=parent_world_transform,
        )
    world_transform = _row_transform(row)
    child_nodes = [
        _native_occurrence_node(
            child,
            children_by_parent=children_by_parent,
            topology_path=topology_path,
            parent_world_transform=world_transform,
        )
        for child in children
    ]
    return _assembly_node(
        id=row_id,
        occurrence_id=row_id,
        display_name=_occurrence_display_name(row),
        source_kind="native",
        source_path="",
        instance_path=str(row.get("path") or row_id),
        use_source_colors=True,
        local_transform=_relative_transform(parent_world_transform, world_transform),
        world_transform=world_transform,
        bbox=row.get("bbox") or _merge_bbox([child.get("bbox") for child in child_nodes]),
        topology_counts=_public_topology_counts(_occurrence_topology_counts(row)),
        children=child_nodes,
    )


def _native_part_node(
    row: dict[str, Any],
    *,
    topology_path: Path,
    parent_world_transform: tuple[float, ...],
) -> dict[str, Any]:
    occurrence_id = str(row.get("id") or "").strip()
    if not occurrence_id:
        raise AssemblyCompositionError("Native assembly occurrence is missing an id")
    world_transform = _row_transform(row)
    return _part_node(
        id=occurrence_id,
        occurrence_id=occurrence_id,
        display_name=_occurrence_display_name(row),
        source_kind="native",
        source_path="",
        instance_path=str(row.get("path") or occurrence_id),
        use_source_colors=True,
        local_transform=_relative_transform(parent_world_transform, world_transform),
        world_transform=world_transform,
        bbox=row.get("bbox"),
        topology_counts=_public_topology_counts(_occurrence_topology_counts(row)),
    )


def _part_node(
    *,
    id: str,
    occurrence_id: str,
    display_name: str,
    source_kind: str,
    source_path: str,
    instance_path: str,
    use_source_colors: bool,
    local_transform: Sequence[float],
    world_transform: Sequence[float],
    bbox: Any,
    topology_counts: Mapping[str, int],
    asset_url: str = "",
    asset_hash: str = "",
) -> dict[str, Any]:
    node = {
        "id": id,
        "occurrenceId": occurrence_id,
        "nodeType": "part",
        "displayName": display_name,
        "sourceKind": source_kind,
        "sourcePath": source_path,
        "instancePath": instance_path,
        "useSourceColors": use_source_colors,
        "localTransform": _transform_list(local_transform),
        "worldTransform": _transform_list(world_transform),
        "bbox": bbox,
        "topologyCounts": _public_topology_counts(topology_counts),
        "leafPartIds": [id],
        "children": [],
    }
    if asset_url:
        node["assets"] = {
            "glb": {
                "url": asset_url,
                "hash": asset_hash,
            }
        }
    return node


def _assembly_node(
    *,
    id: str,
    occurrence_id: str,
    display_name: str,
    source_kind: str,
    source_path: str,
    instance_path: str,
    use_source_colors: bool,
    local_transform: Sequence[float],
    world_transform: Sequence[float],
    bbox: Any,
    topology_counts: Mapping[str, int],
    children: Sequence[dict[str, Any]],
) -> dict[str, Any]:
    leaf_part_ids = _leaf_part_ids(children)
    return {
        "id": id,
        "occurrenceId": occurrence_id,
        "nodeType": "assembly",
        "displayName": display_name,
        "sourceKind": source_kind,
        "sourcePath": source_path,
        "instancePath": instance_path,
        "useSourceColors": use_source_colors,
        "localTransform": _transform_list(local_transform),
        "worldTransform": _transform_list(world_transform),
        "bbox": bbox,
        "topologyCounts": _public_topology_counts(topology_counts),
        "leafPartIds": leaf_part_ids,
        "children": list(children),
    }


def _leaf_part_ids(children: Sequence[Mapping[str, Any]]) -> list[str]:
    ids: list[str] = []
    for child in children:
        child_ids = child.get("leafPartIds")
        if isinstance(child_ids, list):
            ids.extend(str(value) for value in child_ids if str(value or "").strip())
            continue
        child_id = str(child.get("id") or "").strip()
        if child_id:
            ids.append(child_id)
    return ids


def _assembly_root_node(cad_ref: str, root_occurrence: dict[str, Any], children: Sequence[dict[str, Any]]) -> dict[str, Any]:
    counts = _sum_public_counts(children)
    if not _counts_have_values(counts):
        counts = _public_topology_counts(_occurrence_topology_counts(root_occurrence))
    return _assembly_node(
        id="root",
        occurrence_id=str(root_occurrence.get("id") or "root").strip() or "root",
        display_name=_root_display_name(cad_ref, root_occurrence),
        source_kind="catalog",
        source_path="",
        instance_path="",
        use_source_colors=True,
        local_transform=IDENTITY_TRANSFORM,
        world_transform=IDENTITY_TRANSFORM,
        bbox=root_occurrence.get("bbox") or _merge_bbox([child.get("bbox") for child in children]),
        topology_counts=counts,
        children=children,
    )


def _root_display_name(cad_ref: str, root_occurrence: Mapping[str, Any]) -> str:
    display_name = str(root_occurrence.get("name") or "").strip()
    if not display_name or display_name.lower() == "root":
        return cad_ref.rsplit("/", 1)[-1]
    return display_name


def _rows(manifest: dict[str, Any], row_key: str, columns_key: str) -> list[dict[str, Any]]:
    columns = manifest.get("tables", {}).get(columns_key)
    rows = manifest.get(row_key)
    if not isinstance(columns, list) or not isinstance(rows, list):
        return []
    output: list[dict[str, Any]] = []
    for row in rows:
        if isinstance(row, list):
            output.append({str(column): row[index] if index < len(row) else None for index, column in enumerate(columns)})
    return output


def _component_occurrences(manifest: dict[str, Any]) -> list[dict[str, Any]]:
    occurrences = _rows(manifest, "occurrences", "occurrenceColumns")
    parent_ids = {
        str(row.get("parentId") or "").strip()
        for row in occurrences
        if str(row.get("parentId") or "").strip()
    }
    candidate_occurrences = [
        row
        for row in occurrences
        if str(row.get("parentId") or "").strip() and int(row.get("shapeCount") or 0) > 0
    ]
    if candidate_occurrences:
        return candidate_occurrences

    leaf_occurrences = [
        row
        for row in occurrences
        if str(row.get("id") or "").strip() not in parent_ids and int(row.get("shapeCount") or 0) > 0
    ]
    if not leaf_occurrences:
        leaf_occurrences = [
            row
            for row in occurrences
            if int(row.get("shapeCount") or 0) > 0
        ]
    if not leaf_occurrences:
        raise AssemblyCompositionError("Assembly topology has no component occurrences")
    return leaf_occurrences


def _find_occurrence_by_component_name(
    component: str,
    occurrences: Sequence[dict[str, Any]],
    cad_ref: str,
) -> dict[str, Any] | None:
    matches = []
    for occurrence in occurrences:
        names = {
            str(occurrence.get("name") or "").strip(),
            str(occurrence.get("sourceName") or "").strip(),
        }
        if component in names:
            matches.append(occurrence)
    if len(matches) > 1:
        raise AssemblyCompositionError(
            f"Assembly topology has duplicate component occurrence name for {cad_ref}: {component}"
        )
    return matches[0] if matches else None


def _read_json(path: Path) -> dict[str, Any]:
    import json

    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise AssemblyCompositionError(f"Failed to read JSON: {relative_to_repo(path)}") from exc
    if not isinstance(payload, dict):
        raise AssemblyCompositionError(f"Expected JSON object: {relative_to_repo(path)}")
    return payload


def _file_cache_key(path: Path) -> tuple[str, int, int]:
    resolved = path.resolve()
    try:
        stat = resolved.stat()
    except OSError:
        return (resolved.as_posix(), -1, -1)
    return (resolved.as_posix(), int(stat.st_size), int(stat.st_mtime_ns))


def _source_topology_counts(topology_manifest_path: Path) -> dict[str, int]:
    cache_key = _file_cache_key(topology_manifest_path)
    cached = _SOURCE_TOPOLOGY_COUNTS_CACHE.get(cache_key)
    if cached is not None:
        return dict(cached)
    manifest = read_step_topology_manifest_from_glb(topology_manifest_path)
    if manifest is None:
        raise AssemblyCompositionError(
            f"Source GLB topology is missing: {relative_to_repo(topology_manifest_path)}"
        )
    stats = manifest.get("stats")
    if not isinstance(stats, dict):
        raise AssemblyCompositionError(
            f"Source topology is missing stats: {relative_to_repo(topology_manifest_path)}"
        )
    counts = {
        "shapes": int(stats.get("shapeCount") or 0),
        "faces": int(stats.get("faceCount") or 0),
        "edges": int(stats.get("edgeCount") or 0),
    }
    if any(value <= 0 for value in counts.values()):
        raise AssemblyCompositionError(
            f"Source topology has invalid counts in {relative_to_repo(topology_manifest_path)}: {counts}"
        )
    _SOURCE_TOPOLOGY_COUNTS_CACHE[cache_key] = counts
    return counts


def _occurrence_topology_counts(occurrence: Mapping[str, Any]) -> dict[str, int]:
    return {
        "shapes": int(occurrence.get("shapeCount") or 0),
        "faces": int(occurrence.get("faceCount") or 0),
        "edges": int(occurrence.get("edgeCount") or 0),
    }


def _public_topology_counts(counts: Mapping[str, int]) -> dict[str, int]:
    return {
        "shapes": int(counts.get("shapes") or 0),
        "faces": int(counts.get("faces") or 0),
        "edges": int(counts.get("edges") or 0),
    }


def _sum_public_counts(children: Sequence[Mapping[str, Any]]) -> dict[str, int]:
    total = {"shapes": 0, "faces": 0, "edges": 0}
    for child in children:
        counts = child.get("topologyCounts")
        if not isinstance(counts, Mapping):
            continue
        for key in total:
            total[key] += int(counts.get(key) or 0)
    return total


def _counts_have_values(counts: Mapping[str, int]) -> bool:
    return any(int(counts.get(key) or 0) > 0 for key in ("shapes", "faces", "edges"))


def _occurrence_display_name(row: Mapping[str, Any]) -> str:
    name = str(row.get("name") or "").strip()
    source_name = str(row.get("sourceName") or "").strip()
    if name and not (name.startswith("=>[") and name.endswith("]")):
        return name
    return source_name or name or str(row.get("path") or row.get("id") or "component").strip()


def _row_transform(row: Mapping[str, Any]) -> tuple[float, ...]:
    raw_transform = row.get("transform")
    if not isinstance(raw_transform, list) or len(raw_transform) != 16:
        return IDENTITY_TRANSFORM
    return tuple(float(value) for value in raw_transform)


def _transform_tuple(raw_transform: Any, fallback: Sequence[float]) -> tuple[float, ...]:
    if not isinstance(raw_transform, list) or len(raw_transform) != 16:
        return tuple(float(value) for value in fallback)
    return tuple(float(value) for value in raw_transform)


def _transform_list(transform: Sequence[float]) -> list[float]:
    return [float(value) for value in transform]


def _transform_point(transform: Sequence[float], point: Sequence[float]) -> list[float]:
    x = float(point[0])
    y = float(point[1])
    z = float(point[2])
    return [
        (float(transform[0]) * x) + (float(transform[1]) * y) + (float(transform[2]) * z) + float(transform[3]),
        (float(transform[4]) * x) + (float(transform[5]) * y) + (float(transform[6]) * z) + float(transform[7]),
        (float(transform[8]) * x) + (float(transform[9]) * y) + (float(transform[10]) * z) + float(transform[11]),
    ]


def _transform_bbox(transform: Sequence[float], bbox: Any) -> dict[str, Any] | None:
    if not isinstance(bbox, Mapping) or not isinstance(bbox.get("min"), list) or not isinstance(bbox.get("max"), list):
        return None
    mins = bbox["min"]
    maxs = bbox["max"]
    corners = [
        [mins[0], mins[1], mins[2]],
        [mins[0], mins[1], maxs[2]],
        [mins[0], maxs[1], mins[2]],
        [mins[0], maxs[1], maxs[2]],
        [maxs[0], mins[1], mins[2]],
        [maxs[0], mins[1], maxs[2]],
        [maxs[0], maxs[1], mins[2]],
        [maxs[0], maxs[1], maxs[2]],
    ]
    transformed = [_transform_point(transform, corner) for corner in corners]
    return {
        "min": [min(point[index] for point in transformed) for index in range(3)],
        "max": [max(point[index] for point in transformed) for index in range(3)],
    }


def _merge_bbox(boxes: Sequence[Any]) -> dict[str, Any]:
    valid_boxes = [
        box
        for box in boxes
        if isinstance(box, Mapping) and isinstance(box.get("min"), list) and isinstance(box.get("max"), list)
    ]
    if not valid_boxes:
        return {"min": [0.0, 0.0, 0.0], "max": [0.0, 0.0, 0.0]}
    mins = [list(box["min"]) for box in valid_boxes]
    maxs = [list(box["max"]) for box in valid_boxes]
    return {
        "min": [min(float(point[index]) for point in mins) for index in range(3)],
        "max": [max(float(point[index]) for point in maxs) for index in range(3)],
    }


def _relative_transform(parent_world_transform: tuple[float, ...], world_transform: tuple[float, ...]) -> tuple[float, ...]:
    return multiply_transforms(_invert_affine_transform(parent_world_transform), world_transform)


def _invert_affine_transform(transform: tuple[float, ...]) -> tuple[float, ...]:
    a, b, c, tx, d, e, f, ty, g, h, i, tz = transform[:12]
    det = (
        a * (e * i - f * h)
        - b * (d * i - f * g)
        + c * (d * h - e * g)
    )
    if abs(det) <= 1e-12:
        return IDENTITY_TRANSFORM
    inv_det = 1.0 / det
    r00 = (e * i - f * h) * inv_det
    r01 = (c * h - b * i) * inv_det
    r02 = (b * f - c * e) * inv_det
    r10 = (f * g - d * i) * inv_det
    r11 = (a * i - c * g) * inv_det
    r12 = (c * d - a * f) * inv_det
    r20 = (d * h - e * g) * inv_det
    r21 = (b * g - a * h) * inv_det
    r22 = (a * e - b * d) * inv_det
    return (
        r00,
        r01,
        r02,
        -((r00 * tx) + (r01 * ty) + (r02 * tz)),
        r10,
        r11,
        r12,
        -((r10 * tx) + (r11 * ty) + (r12 * tz)),
        r20,
        r21,
        r22,
        -((r20 * tx) + (r21 * ty) + (r22 * tz)),
        0.0,
        0.0,
        0.0,
        1.0,
    )