265 lines
12 KiB
Python
265 lines
12 KiB
Python
"""Structured multi-view image observations used by the CAD agent.
|
|
|
|
The observation contract intentionally keeps uncertain image evidence separate
|
|
from executable CDSL. It can therefore retain free-form/polyline candidates
|
|
without pretending that the local CAD runtime supports them directly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
|
|
OBSERVATION_SCHEMA_VERSION = "cad.image-observation.v2"
|
|
TEXT_LIMIT = 300
|
|
LIMITS = {
|
|
"views": 12,
|
|
"surfaces": 24,
|
|
"profiles": 32,
|
|
"segments": 256,
|
|
"holes": 64,
|
|
"bends": 16,
|
|
"measurements": 128,
|
|
"uncertainties": 64,
|
|
}
|
|
|
|
|
|
def _text(value: Any, name: str, limit: int = TEXT_LIMIT, *, required: bool = False) -> str:
|
|
result = str(value or "").strip()
|
|
if required and not result:
|
|
raise ValueError(f"{name} must be a non-empty string")
|
|
return result[:limit]
|
|
|
|
|
|
def _text_list(value: Any, name: str, limit: int) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if not isinstance(value, list):
|
|
raise ValueError(f"{name} must be an array")
|
|
return [_text(item, name, required=True) for item in value[:limit]]
|
|
|
|
|
|
def _number(value: Any, name: str) -> float | None:
|
|
if value is None or value == "":
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError) as error:
|
|
raise ValueError(f"{name} must be numeric") from error
|
|
|
|
|
|
def _point(value: Any, name: str, dimensions: int = 2) -> list[float] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, list) or len(value) < dimensions:
|
|
raise ValueError(f"{name} must contain at least {dimensions} numbers")
|
|
output: list[float] = []
|
|
for index, component in enumerate(value[:dimensions]):
|
|
parsed = _number(component, f"{name}[{index}]")
|
|
if parsed is None:
|
|
raise ValueError(f"{name}[{index}] must be numeric")
|
|
output.append(parsed)
|
|
return output
|
|
|
|
|
|
def _confidence(value: Any) -> float | None:
|
|
parsed = _number(value, "confidence")
|
|
if parsed is None:
|
|
return None
|
|
return max(0.0, min(1.0, parsed))
|
|
|
|
|
|
def _source_images(value: Any) -> list[str]:
|
|
return _text_list(value, "source_images", LIMITS["views"])
|
|
|
|
|
|
def _normalize_segment(segment: Any) -> dict[str, Any]:
|
|
if not isinstance(segment, dict):
|
|
raise ValueError("profile segments must contain objects")
|
|
kind = _text(segment.get("type"), "segment.type", 32, required=True)
|
|
if kind not in {"line", "arc", "circle", "polyline", "unknown_curve"}:
|
|
raise ValueError(f"unsupported image segment type: {kind}")
|
|
result: dict[str, Any] = {"type": kind}
|
|
if kind in {"line", "arc"}:
|
|
result["start"] = _point(segment.get("start"), "segment.start")
|
|
result["end"] = _point(segment.get("end"), "segment.end")
|
|
if result["start"] is None or result["end"] is None:
|
|
raise ValueError(f"{kind} segments require start and end")
|
|
if kind == "arc":
|
|
result["center"] = _point(segment.get("center"), "segment.center")
|
|
result["radius_mm"] = _number(segment.get("radius_mm"), "segment.radius_mm")
|
|
result["clockwise"] = bool(segment.get("clockwise"))
|
|
if kind == "circle":
|
|
result["center"] = _point(segment.get("center"), "segment.center")
|
|
result["radius_mm"] = _number(segment.get("radius_mm"), "segment.radius_mm")
|
|
if result["center"] is None or result["radius_mm"] is None:
|
|
raise ValueError("circle segments require center and radius_mm")
|
|
if kind in {"polyline", "unknown_curve"}:
|
|
points = segment.get("points")
|
|
if not isinstance(points, list) or not points:
|
|
raise ValueError(f"{kind} segments require points")
|
|
result["points"] = [_point(point, "segment.points") for point in points[:LIMITS["segments"]]]
|
|
if any(point is None for point in result["points"]):
|
|
raise ValueError(f"{kind} segment contains an invalid point")
|
|
result["image_uv"] = {
|
|
"start": _point(segment.get("image_start"), "segment.image_start"),
|
|
"end": _point(segment.get("image_end"), "segment.image_end"),
|
|
}
|
|
result["confidence"] = _confidence(segment.get("confidence"))
|
|
result["notes"] = _text(segment.get("notes"), "segment.notes")
|
|
return result
|
|
|
|
|
|
def _normalize_profile(profile: Any) -> dict[str, Any]:
|
|
if not isinstance(profile, dict):
|
|
raise ValueError("profiles must contain objects")
|
|
segments = profile.get("segments") or []
|
|
if not isinstance(segments, list):
|
|
raise ValueError("profile.segments must be an array")
|
|
return {
|
|
"id": _text(profile.get("id"), "profile.id", 80, required=True),
|
|
"role": _text(profile.get("role"), "profile.role", 40),
|
|
"plane_hint": _text(profile.get("plane_hint"), "profile.plane_hint"),
|
|
"closed": bool(profile.get("closed")),
|
|
"coordinate_space": _text(profile.get("coordinate_space"), "profile.coordinate_space") or "image_uv",
|
|
"segments": [_normalize_segment(item) for item in segments[:LIMITS["segments"]]],
|
|
"source_images": _source_images(profile.get("source_images")),
|
|
"confidence": _confidence(profile.get("confidence")),
|
|
"uncertain": _text_list(profile.get("uncertain"), "profile.uncertain", 16),
|
|
"notes": _text(profile.get("notes"), "profile.notes"),
|
|
}
|
|
|
|
|
|
def _normalize_measurement(measurement: Any) -> dict[str, Any]:
|
|
if not isinstance(measurement, dict):
|
|
raise ValueError("measurements must contain objects")
|
|
source = _text(measurement.get("source"), "measurement.source") or "image"
|
|
if source not in {"user", "image", "cv", "assumption"}:
|
|
raise ValueError("measurement.source must be user, image, cv, or assumption")
|
|
value = _number(measurement.get("value_mm"), "measurement.value_mm")
|
|
minimum = _number(measurement.get("min_mm"), "measurement.min_mm")
|
|
maximum = _number(measurement.get("max_mm"), "measurement.max_mm")
|
|
return {
|
|
"name": _text(measurement.get("name"), "measurement.name", 120, required=True),
|
|
"value_mm": value,
|
|
"min_mm": minimum,
|
|
"max_mm": maximum,
|
|
"source": source,
|
|
"confidence": _confidence(measurement.get("confidence")),
|
|
"evidence": _text(measurement.get("evidence"), "measurement.evidence"),
|
|
"source_images": _source_images(measurement.get("source_images")),
|
|
}
|
|
|
|
|
|
def normalize_image_observation(arguments: dict[str, Any], *, attachment_ids: list[str]) -> dict[str, Any]:
|
|
"""Normalize the survey tool output while preserving uncertain geometry."""
|
|
if not isinstance(arguments, dict):
|
|
raise ValueError("image observation arguments must be an object")
|
|
raw_ids = [str(item) for item in arguments.get("attachment_ids") or attachment_ids if str(item)]
|
|
normalized_ids = list(dict.fromkeys(raw_ids or attachment_ids))
|
|
if not normalized_ids:
|
|
raise ValueError("image observation requires at least one attachment")
|
|
views = arguments.get("views") or []
|
|
profiles = arguments.get("profiles") or []
|
|
measurements = arguments.get("measurements") or []
|
|
result: dict[str, Any] = {
|
|
"schema_version": OBSERVATION_SCHEMA_VERSION,
|
|
"attachment_ids": normalized_ids,
|
|
"part_type": _text(arguments.get("part_type"), "part_type", required=True),
|
|
"visible_features": _text_list(arguments.get("visible_features"), "visible_features", 32),
|
|
"uncertain_features": _text_list(arguments.get("uncertain_features"), "uncertain_features", LIMITS["uncertainties"]),
|
|
"views": [],
|
|
"scale_references": deepcopy(arguments.get("scale_references") or [])[:LIMITS["views"]],
|
|
"overall_geometry": arguments.get("overall_geometry") if isinstance(arguments.get("overall_geometry"), dict) else {},
|
|
"surfaces": deepcopy(arguments.get("surfaces") or [])[:LIMITS["surfaces"]],
|
|
"profiles": [_normalize_profile(item) for item in profiles[:LIMITS["profiles"]]],
|
|
"holes": deepcopy(arguments.get("holes") or [])[:LIMITS["holes"]],
|
|
"bends": deepcopy(arguments.get("bends") or [])[:LIMITS["bends"]],
|
|
"measurements": [_normalize_measurement(item) for item in measurements[:LIMITS["measurements"]]],
|
|
"uncertainties": _text_list(arguments.get("uncertainties"), "uncertainties", LIMITS["uncertainties"]),
|
|
"assumptions": _text_list(arguments.get("assumptions"), "assumptions", LIMITS["uncertainties"]),
|
|
"cv_hints": deepcopy(arguments.get("cv_hints") or [])[:LIMITS["profiles"]],
|
|
}
|
|
for item in views[:LIMITS["views"]]:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("views must contain objects")
|
|
result["views"].append({
|
|
"attachment_id": _text(item.get("attachment_id"), "view.attachment_id", 120, required=True),
|
|
"view_role": _text(item.get("view_role"), "view.view_role"),
|
|
"orientation": _text(item.get("orientation"), "view.orientation"),
|
|
"visible_regions": _text_list(item.get("visible_regions"), "view.visible_regions", 24),
|
|
"occluded_regions": _text_list(item.get("occluded_regions"), "view.occluded_regions", 24),
|
|
"quality": _text(item.get("quality"), "view.quality"),
|
|
"scale_reference_id": _text(item.get("scale_reference_id"), "view.scale_reference_id", 120),
|
|
"confidence": _confidence(item.get("confidence")),
|
|
})
|
|
return result
|
|
|
|
|
|
def normalize_sketch_candidates(arguments: dict[str, Any], *, attachment_ids: list[str]) -> dict[str, Any]:
|
|
"""Normalize the second-stage sketch extraction result."""
|
|
if not isinstance(arguments, dict):
|
|
raise ValueError("sketch candidate arguments must be an object")
|
|
profiles = arguments.get("profiles") or arguments.get("sketches") or []
|
|
base = normalize_image_observation({
|
|
"attachment_ids": attachment_ids,
|
|
"part_type": arguments.get("part_type") or "image reference",
|
|
"visible_features": arguments.get("visible_features") or ["profile candidates"],
|
|
"uncertain_features": arguments.get("uncertain_features") or [],
|
|
"profiles": profiles,
|
|
"measurements": arguments.get("measurements") or [],
|
|
"uncertainties": arguments.get("uncertainties") or [],
|
|
"assumptions": arguments.get("assumptions") or [],
|
|
"cv_hints": arguments.get("cv_hints") or [],
|
|
}, attachment_ids=attachment_ids)
|
|
return {
|
|
"profiles": base["profiles"],
|
|
"measurements": base["measurements"],
|
|
"uncertainties": base["uncertainties"],
|
|
"assumptions": base["assumptions"],
|
|
"cv_hints": base["cv_hints"],
|
|
}
|
|
|
|
|
|
def merge_image_observations(survey: dict[str, Any], sketches: dict[str, Any]) -> dict[str, Any]:
|
|
"""Merge the two stages and keep user-sourced measurements authoritative."""
|
|
result = deepcopy(survey)
|
|
result["schema_version"] = OBSERVATION_SCHEMA_VERSION
|
|
result["profiles"] = sketches.get("profiles") or result.get("profiles") or []
|
|
existing = {str(item.get("name")): item for item in result.get("measurements") or () if isinstance(item, dict)}
|
|
for item in sketches.get("measurements") or ():
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key = str(item.get("name") or "")
|
|
prior = existing.get(key)
|
|
if prior and prior.get("source") == "user" and item.get("source") != "user":
|
|
continue
|
|
existing[key] = item
|
|
result["measurements"] = list(existing.values())
|
|
result["uncertainties"] = list(dict.fromkeys([
|
|
*(result.get("uncertainties") or []),
|
|
*(sketches.get("uncertainties") or []),
|
|
]))[:LIMITS["uncertainties"]]
|
|
result["assumptions"] = list(dict.fromkeys([
|
|
*(result.get("assumptions") or []),
|
|
*(sketches.get("assumptions") or []),
|
|
]))[:LIMITS["uncertainties"]]
|
|
result["cv_hints"] = sketches.get("cv_hints") or result.get("cv_hints") or []
|
|
return result
|
|
|
|
|
|
def render_image_observation_context(observation: dict[str, Any] | None) -> str:
|
|
if not isinstance(observation, dict):
|
|
return ""
|
|
compact = {
|
|
key: observation.get(key)
|
|
for key in (
|
|
"schema_version", "attachment_ids", "part_type", "views", "overall_geometry",
|
|
"surfaces", "profiles", "holes", "bends", "measurements", "uncertainties", "assumptions",
|
|
)
|
|
if observation.get(key) not in (None, [], {})
|
|
}
|
|
return json.dumps(compact, ensure_ascii=False, separators=(",", ":"))
|