#!/usr/bin/env python3 """Development-only E40 feature and split audit. The script intentionally rejects validation rows before feature extraction. It compares route-coordinate-free candidates under two leakage-resistant protocols: * contiguous source-time folds; * whole detector tracks plus 50-frame geometry scene windows. It never writes a model or a sealed result. A production E40 profile can only be frozen after one fixed candidate passes the development gate here. """ from __future__ import annotations import argparse import hashlib import json import math from collections import Counter from pathlib import Path from typing import Any import numpy as np from PIL import Image from k1link.compute.e38_perception_baseline import _predict_tree, _train_tree LABELS = ( "background-or-noise", "object-present", "occupied-environment", ) TARGET = 0.9 CATEGORIES = { "stratum": ("agree", "camera-only", "conflict", "geometry-only", "unknown"), "range": ("near", "middle", "far", "unavailable"), "geometry": ( "agree", "conflict", "single-source-camera", "single-source-geometry", "unknown", "unavailable", ), "label": ("car", "person", "truck", "bicycle", "motorcycle", "bus", "none"), "reason": ( "camera-semantic-without-qualified-occupied-lidar-support", "camera-semantic-with-connected-occupied-lidar-support", "semantic-observation-not-current", "camera-object-region-observed-as-local-surface", "none", ), "association": ("vehicle", "person", "bicycle", "motorcycle", "none"), "motion": ("unknown", "static", "dynamic", "none"), "camera_motion": ("unknown", "static", "dynamic", "none"), "semantic_current": ("true", "false", "none"), } def main() -> int: args = _parse_args() acceptance_rows = _read_jsonl(args.acceptance_root / "acceptance-items.jsonl") materialization_rows = _read_jsonl(args.materialization_root / "materialized-items.jsonl") materialization_by_id = {str(row["item_id"]): row for row in materialization_rows} development = [row for row in acceptance_rows if row.get("split") == "development"] if len(development) != 340: raise RuntimeError("E40 development denominator must remain 340") if any(row.get("split") != "development" for row in development): raise RuntimeError("E40 analysis received a non-development row") feature_rows = [ _features( acceptance=row, materialization=materialization_by_id[str(row["item_id"])], materialization_root=args.materialization_root, ) for row in development ] feature_names = sorted({name for row in feature_rows for name in row}) matrix = np.asarray( [[row.get(name, 0.0) for name in feature_names] for row in feature_rows], dtype=np.float64, ) labels = np.asarray( [LABELS.index(str(row["reference"]["presence"])) for row in development], dtype=np.int64, ) if not np.isfinite(matrix).all(): raise RuntimeError("E40 development features contain non-finite values") assignments = { "contiguous-source-time-five-fold": _contiguous_folds(development), "whole-track-or-scene-window-five-fold": _track_scene_folds( development, materialization_by_id, ), } print( json.dumps( { "development_items": len(development), "feature_dimensions": len(feature_names), "validation_rows_loaded_for_features": 0, "protocols": { name: {str(fold): count for fold, count in sorted(Counter(values).items())} for name, values in assignments.items() }, }, indent=2, sort_keys=True, ) ) feature_sets = { "structured": [ index for index, name in enumerate(feature_names) if not _source_array_feature(name) and not name.startswith("image_") ], "structured-shape": [ index for index, name in enumerate(feature_names) if not name.startswith("image_") ], "structured-image": [ index for index, name in enumerate(feature_names) if not _source_array_feature(name) or name.startswith("candidate_class_fraction_") or name.startswith("image_") ], "all": list(range(len(feature_names))), } candidates = [ ("structured-knn-k3", "knn", 3, "structured"), ("structured-softmax-l2-0.01", "softmax", 0.01, "structured"), ( "structured-shape-softmax-l2-0.01", "softmax", 0.01, "structured-shape", ), ( "structured-image-softmax-l2-0.01", "softmax", 0.01, "structured-image", ), ("all-softmax-l2-0.01", "softmax", 0.01, "all"), ("all-softmax-l2-0.03", "softmax", 0.03, "all"), ("all-softmax-l2-0.1", "softmax", 0.1, "all"), ( "hierarchical-structured-softmax-l2-0.01", "hierarchical-softmax", 0.01, "structured", ), ( "hierarchical-structured-image-softmax-l2-0.01", "hierarchical-softmax", 0.01, "structured-image", ), ( "hierarchical-all-softmax-l2-0.01", "hierarchical-softmax", 0.01, "all", ), ] candidate_results: list[dict[str, Any]] = [] for candidate_name, model_type, model_parameter, feature_set in candidates: columns = feature_sets[feature_set] result = { protocol: _cross_validate( matrix=matrix[:, columns], labels=labels, feature_names=[feature_names[index] for index in columns], feature_rows=feature_rows, source_strata=[str(row["source_stratum"]) for row in development], assignments=folds, model_type=model_type, model_parameter=model_parameter, ) for protocol, folds in assignments.items() } candidate_results.append( { "candidate": candidate_name, "feature_set": feature_set, "feature_dimensions": len(columns), "accuracy": {protocol: row["accuracy"] for protocol, row in result.items()}, "by_stratum": {protocol: row["by_stratum"] for protocol, row in result.items()}, "by_fold": {protocol: row["by_fold"] for protocol, row in result.items()}, "passed_both": all(row["accuracy"] >= TARGET for row in result.values()), } ) print(json.dumps(candidate_results, indent=2, sort_keys=True)) return 0 def _features( *, acceptance: dict[str, Any], materialization: dict[str, Any], materialization_root: Path, ) -> dict[str, float]: snapshot = _object(materialization.get("e29_snapshot")) evidence = _object(materialization.get("materialization")) values: dict[str, float] = {} observed = { "stratum": materialization.get("stratum"), "range": materialization.get("range_bucket"), "geometry": snapshot.get("geometry_status"), "label": snapshot.get("label") or "none", "reason": snapshot.get("geometry_reason") or "none", "association": snapshot.get("association_group") or "none", "motion": snapshot.get("motion_state") or "none", "camera_motion": snapshot.get("camera_motion_state") or "none", "semantic_current": ( str(snapshot.get("semantic_current")).lower() if snapshot.get("semantic_current") is not None else "none" ), } for prefix, categories in CATEGORIES.items(): for category in categories: values[f"{prefix}={category}"] = float(observed[prefix] == category) for name in ( "selected_point_count", "candidate_point_count", "rejected_candidate_point_count", "projected_point_count", "frame_point_count", ): values[name] = math.log1p(max(0.0, _number(evidence.get(name), 0.0))) values["detector_score"] = _number(evidence.get("detector_score"), -1.0) for name in ( "nearest_range_m", "point_count", "voxel_count", "score", "camera_motion_confidence", "range_m", ): raw = _number(snapshot.get(name), -1.0) values[name] = ( math.log1p(raw) if name in {"point_count", "voxel_count"} and raw >= 0.0 else raw ) bbox = snapshot.get("bbox_xyxy") if isinstance(bbox, list) and len(bbox) == 4: x1, y1, x2, y2 = (_number(value, 0.0) for value in bbox) width = max(0.0, x2 - x1) / 800.0 height = max(0.0, y2 - y1) / 600.0 values.update( { "bbox_present": 1.0, "bbox_center_x": (x1 + x2) / 1600.0, "bbox_center_y": (y1 + y2) / 1200.0, "bbox_width": width, "bbox_height": height, "bbox_area": width * height, "bbox_aspect": width / (height + 1e-6), } ) else: values.update( { "bbox_present": 0.0, "bbox_center_x": -1.0, "bbox_center_y": -1.0, "bbox_width": -1.0, "bbox_height": -1.0, "bbox_area": -1.0, "bbox_aspect": -1.0, } ) support = snapshot.get("support") support = support if isinstance(support, dict) else {} support_names = ( "below_surface_points_in_bbox", "classified_points_in_bbox", "connected_occupied_points", "connected_occupied_voxels", "occupied_points_in_bbox", "projected_points_in_bbox", "surface_points_in_bbox", ) support_values = {name: _number(support.get(name), 0.0) for name in support_names} values.update( {f"support_{name}": math.log1p(max(0.0, value)) for name, value in support_values.items()} ) projected = max(1.0, support_values["projected_points_in_bbox"]) occupied = max(1.0, support_values["occupied_points_in_bbox"]) values.update( { "support_occupied_fraction": (support_values["occupied_points_in_bbox"] / projected), "support_classified_fraction": ( support_values["classified_points_in_bbox"] / projected ), "support_surface_fraction": (support_values["surface_points_in_bbox"] / projected), "support_connected_fraction": (support_values["connected_occupied_points"] / occupied), } ) _span_features(values, "bounds", snapshot.get("bounds_map_xyz_m")) _range_span(values, "height_span", snapshot.get("height_range_m")) _range_span( values, "occupied_height_span", snapshot.get("occupied_height_range_m"), ) artifact = _object(materialization.get("artifact")) artifact_path = materialization_root / str(artifact.get("path")) with np.load(artifact_path, allow_pickle=False) as arrays: pixels = arrays["projected_pixels_xy"] candidate_mask = arrays["projected_candidate_mask"].astype(bool) selected_mask = arrays["projected_selected_mask"].astype(bool) sensor_position = arrays["sensor_position_map_xyz_m"] sensor_orientation = arrays["sensor_orientation_map_from_lidar_xyzw"] _named_point_statistics( values, "candidate_lidar", arrays["candidate_points_map_xyz_m"], sensor_position, sensor_orientation, ) _named_point_statistics( values, "selected_lidar", arrays["selected_points_map_xyz_m"], sensor_position, sensor_orientation, ) projection_width = max(1, int(evidence.get("projection_width", 800))) projection_height = max(1, int(evidence.get("projection_height", 600))) _named_pixel_statistics( values, "candidate_pixel", pixels[candidate_mask], projection_width, projection_height, ) _named_pixel_statistics( values, "selected_pixel", pixels[selected_mask], projection_width, projection_height, ) _named_quantiles( values, "candidate_depth", arrays["projected_depth_m"][candidate_mask], ) _named_quantiles( values, "selected_depth", arrays["projected_depth_m"][selected_mask], ) _named_quantiles( values, "candidate_height", arrays["projected_point_height_m"][candidate_mask], ) _named_quantiles( values, "selected_height", arrays["projected_point_height_m"][selected_mask], ) point_classes = arrays["projected_point_class"][candidate_mask] for index in range(8): values[f"candidate_class_fraction_{index}"] = ( float(np.mean(point_classes == index)) if point_classes.size else 0.0 ) _image_features( values, materialization=materialization, materialization_root=materialization_root, candidate_pixels=pixels[candidate_mask], projection_width=projection_width, projection_height=projection_height, ) # Source frame, session time, review ordinal, track ID and absolute map # coordinates are deliberately absent. return values def _cross_validate( *, matrix: np.ndarray, labels: np.ndarray, feature_names: list[str], feature_rows: list[dict[str, float]], source_strata: list[str], assignments: np.ndarray, model_type: str, model_parameter: object, ) -> dict[str, Any]: predictions = np.full(len(labels), -1, dtype=np.int64) for fold in range(5): train_indices = np.flatnonzero(assignments != fold) test_indices = np.flatnonzero(assignments == fold) train, test = _robust_transform( matrix[train_indices], matrix[test_indices], ) if model_type == "knn": distances = np.mean( np.square(test[:, None, :] - train[None, :, :]), axis=2, ) nearest = np.argsort(distances, axis=1, kind="stable")[:, : int(model_parameter)] for local_index, neighbor_indices in enumerate(nearest): votes = Counter(labels[train_indices][neighbor_indices]) predictions[test_indices[local_index]] = sorted( votes.items(), key=lambda item: (-item[1], item[0]), )[0][0] elif model_type == "softmax": weights = _train_softmax( train, labels[train_indices], l2=float(model_parameter), ) predictions[test_indices] = np.argmax( np.column_stack((test, np.ones(len(test)))) @ weights, axis=1, ) elif model_type == "hierarchical-softmax": _predict_hierarchical_fold( predictions=predictions, train_indices=train_indices, test_indices=test_indices, training=train, testing=test, labels=labels, source_strata=source_strata, l2=float(model_parameter), ) elif model_type == "tree": depth, min_leaf = model_parameter # type: ignore[misc] tree = _train_tree( [ ( {name: feature_rows[index].get(name, 0.0) for name in feature_names}, LABELS[int(labels[index])], ) for index in train_indices ], feature_names=feature_names, max_depth=int(depth), min_leaf=int(min_leaf), ) for index in test_indices: predictions[index] = LABELS.index(_predict_tree(tree, feature_rows[int(index)])) else: raise RuntimeError(f"unsupported model: {model_type}") accuracy = float(np.mean(predictions == labels)) stratum_metrics = {} for stratum in sorted(set(source_strata)): indices = np.asarray( [index for index, value in enumerate(source_strata) if value == stratum], dtype=np.int64, ) stratum_metrics[stratum] = round( float(np.mean(predictions[indices] == labels[indices])), 6, ) fold_metrics = {} for fold in range(5): indices = np.flatnonzero(assignments == fold) fold_metrics[str(fold)] = round( float(np.mean(predictions[indices] == labels[indices])), 6, ) return { "accuracy": round(accuracy, 6), "correct": int(np.sum(predictions == labels)), "incorrect": int(np.sum(predictions != labels)), "passed": accuracy >= TARGET, "by_stratum": stratum_metrics, "by_fold": fold_metrics, "confusion": [ { "reference": LABELS[reference], "prediction": LABELS[prediction], "count": count, } for (reference, prediction), count in sorted( Counter( zip( labels.tolist(), predictions.tolist(), strict=True, ) ).items(), key=lambda item: (-item[1], item[0]), ) ], } def _predict_hierarchical_fold( *, predictions: np.ndarray, train_indices: np.ndarray, test_indices: np.ndarray, training: np.ndarray, testing: np.ndarray, labels: np.ndarray, source_strata: list[str], l2: float, ) -> None: fixed = { "conflict": LABELS.index("background-or-noise"), "agree": LABELS.index("object-present"), "unknown": LABELS.index("object-present"), # A geometry-only cluster is positive occupied evidence, but without # camera semantics it must not be promoted to a named object. Keeping # it occupied is the conservative product state. "geometry-only": LABELS.index("occupied-environment"), } for source_index in test_indices: stratum = source_strata[int(source_index)] if stratum in fixed: predictions[source_index] = fixed[stratum] for stratum, fallback in (("camera-only", LABELS.index("object-present")),): local_training = np.asarray( [ index for index, source_index in enumerate(train_indices) if source_strata[int(source_index)] == stratum ], dtype=np.int64, ) local_testing = np.asarray( [ index for index, source_index in enumerate(test_indices) if source_strata[int(source_index)] == stratum ], dtype=np.int64, ) if not len(local_testing): continue observed = sorted(set(labels[train_indices][local_training].tolist())) if len(local_training) < 10 or len(observed) < 2: predictions[test_indices[local_testing]] = fallback continue weights = _train_softmax( training[local_training], labels[train_indices][local_training], l2=l2, ) predictions[test_indices[local_testing]] = np.argmax( np.column_stack( ( testing[local_testing], np.ones(len(local_testing)), ) ) @ weights, axis=1, ) def _train_softmax( matrix: np.ndarray, labels: np.ndarray, *, l2: float, ) -> np.ndarray: rows, dimensions = matrix.shape design = np.column_stack((matrix, np.ones(rows))) targets = np.eye(len(LABELS), dtype=np.float64)[labels] weights = np.zeros((dimensions + 1, len(LABELS)), dtype=np.float64) first_moment = np.zeros_like(weights) second_moment = np.zeros_like(weights) for step in range(1, 1201): logits = design @ weights logits -= np.max(logits, axis=1, keepdims=True) probabilities = np.exp(logits) probabilities /= np.sum(probabilities, axis=1, keepdims=True) regularizer = np.vstack((weights[:-1], np.zeros((1, len(LABELS))))) gradient = design.T @ (probabilities - targets) / rows gradient += l2 * regularizer first_moment = 0.9 * first_moment + 0.1 * gradient second_moment = 0.999 * second_moment + 0.001 * np.square(gradient) corrected_first = first_moment / (1.0 - 0.9**step) corrected_second = second_moment / (1.0 - 0.999**step) weights -= 0.03 * corrected_first / (np.sqrt(corrected_second) + 1e-8) return weights def _robust_transform( training: np.ndarray, testing: np.ndarray, ) -> tuple[np.ndarray, np.ndarray]: median = np.median(training, axis=0) scale = np.percentile(training, 75, axis=0) - np.percentile( training, 25, axis=0, ) scale[scale < 1e-8] = 1.0 return ( np.clip((training - median) / scale, -10.0, 10.0), np.clip((testing - median) / scale, -10.0, 10.0), ) def _source_array_feature(name: str) -> bool: return name.startswith( ( "candidate_lidar_", "selected_lidar_", "candidate_pixel_", "selected_pixel_", "candidate_depth_", "selected_depth_", "candidate_height_", "selected_height_", "candidate_class_fraction_", ) ) def _contiguous_folds(rows: list[dict[str, Any]]) -> np.ndarray: order = np.argsort( [int(row["source_frame_index"]) for row in rows], kind="stable", ) assignments = np.empty(len(rows), dtype=np.int64) for fold, indices in enumerate(np.array_split(order, 5)): assignments[indices] = fold return assignments def _track_scene_folds( rows: list[dict[str, Any]], materialization_by_id: dict[str, dict[str, Any]], ) -> np.ndarray: assignments: list[int] = [] for row in rows: snapshot = _object(materialization_by_id[str(row["item_id"])].get("e29_snapshot")) track_id = snapshot.get("track_id") group = ( f"track:{track_id}" if track_id is not None else f"scene:{int(row['source_frame_index']) // 50}" ) digest = hashlib.sha256(f"e40:{group}".encode()).hexdigest() assignments.append(int(digest[:8], 16) % 5) return np.asarray(assignments, dtype=np.int64) def _named_point_statistics( values: dict[str, float], prefix: str, points_map: np.ndarray, sensor_position: np.ndarray, sensor_orientation_xyzw: np.ndarray, ) -> None: points = np.asarray(points_map, dtype=np.float64) if points.ndim != 2 or points.shape[1] != 3 or not len(points): for axis in "xyz": _named_quantiles(values, f"{prefix}_{axis}", np.asarray([])) for index in range(3): values[f"{prefix}_covariance_ratio_{index}"] = 0.0 return rotation = _rotation_matrix(sensor_orientation_xyzw) points_local = (points - np.asarray(sensor_position)) @ rotation for axis, index in zip("xyz", range(3), strict=True): _named_quantiles(values, f"{prefix}_{axis}", points_local[:, index]) if len(points_local) >= 3: eigenvalues = np.maximum( np.linalg.eigvalsh(np.cov(points_local, rowvar=False)), 0.0, ) else: eigenvalues = np.zeros(3) ratios = eigenvalues / (float(np.sum(eigenvalues)) + 1e-9) for index, ratio in enumerate(ratios): values[f"{prefix}_covariance_ratio_{index}"] = float(ratio) def _rotation_matrix(quaternion_xyzw: np.ndarray) -> np.ndarray: x, y, z, w = (float(value) for value in quaternion_xyzw) return np.asarray( [ [ 1.0 - 2.0 * (y * y + z * z), 2.0 * (x * y - z * w), 2.0 * (x * z + y * w), ], [ 2.0 * (x * y + z * w), 1.0 - 2.0 * (x * x + z * z), 2.0 * (y * z - x * w), ], [ 2.0 * (x * z - y * w), 2.0 * (y * z + x * w), 1.0 - 2.0 * (x * x + y * y), ], ], dtype=np.float64, ) def _named_pixel_statistics( values: dict[str, float], prefix: str, pixels: np.ndarray, width: int, height: int, ) -> None: points = np.asarray(pixels, dtype=np.float64) if points.ndim != 2 or points.shape[1] != 2 or not len(points): _named_quantiles(values, f"{prefix}_x", np.asarray([])) _named_quantiles(values, f"{prefix}_y", np.asarray([])) values[f"{prefix}_span_x"] = -1.0 values[f"{prefix}_span_y"] = -1.0 values[f"{prefix}_density"] = -1.0 return x = points[:, 0] / width y = points[:, 1] / height _named_quantiles(values, f"{prefix}_x", x) _named_quantiles(values, f"{prefix}_y", y) span_x = max(float(np.max(x) - np.min(x)), 1.0 / width) span_y = max(float(np.max(y) - np.min(y)), 1.0 / height) values[f"{prefix}_span_x"] = span_x values[f"{prefix}_span_y"] = span_y values[f"{prefix}_density"] = len(points) / (span_x * span_y * width * height + 1.0) def _image_features( values: dict[str, float], *, materialization: dict[str, Any], materialization_root: Path, candidate_pixels: np.ndarray, projection_width: int, projection_height: int, ) -> None: snapshot = _object(materialization.get("e29_snapshot")) bbox = snapshot.get("bbox_xyxy") if not isinstance(bbox, list) or len(bbox) != 4: points = np.asarray(candidate_pixels, dtype=np.float64) if points.ndim != 2 or points.shape[1] != 2 or not len(points): _empty_image_features(values) return low = np.min(points, axis=0) high = np.max(points, axis=0) center = (low + high) / 2.0 support = np.maximum(high - low, 48.0) bbox = [ center[0] - support[0], center[1] - support[1], center[0] + support[0], center[1] + support[1], ] x1, y1, x2, y2 = (_number(value, 0.0) for value in bbox) box = ( max(0, int(x1)), max(0, int(y1)), min(projection_width, int(math.ceil(x2))), min(projection_height, int(math.ceil(y2))), ) if box[2] <= box[0] or box[3] <= box[1]: _empty_image_features(values) return frame = _object(materialization.get("camera_frame")) with Image.open(materialization_root / str(frame.get("path"))) as source: rgb = ( np.asarray( source.convert("RGB").crop(box).resize((32, 32), Image.Resampling.BILINEAR), dtype=np.float64, ) / 255.0 ) for channel, name in enumerate(("red", "green", "blue")): histogram, _ = np.histogram( rgb[:, :, channel], bins=4, range=(0.0, 1.0), ) histogram = histogram / max(1, int(np.sum(histogram))) for index, fraction in enumerate(histogram): values[f"image_{name}_histogram_{index}"] = float(fraction) luma = np.mean(rgb, axis=2) saturation = np.max(rgb, axis=2) - np.min(rgb, axis=2) edge = np.concatenate( ( np.abs(np.diff(luma, axis=1)).reshape(-1), np.abs(np.diff(luma, axis=0)).reshape(-1), ) ) values["image_luma_mean"] = float(np.mean(luma)) values["image_luma_std"] = float(np.std(luma)) for name, quantile in zip( ("p10", "p25", "p50", "p75", "p90"), np.quantile(luma, (0.1, 0.25, 0.5, 0.75, 0.9)), strict=True, ): values[f"image_luma_{name}"] = float(quantile) values["image_saturation_mean"] = float(np.mean(saturation)) values["image_saturation_std"] = float(np.std(saturation)) values["image_edge_mean"] = float(np.mean(edge)) values["image_edge_p90"] = float(np.quantile(edge, 0.9)) small_luma = ( np.asarray( Image.fromarray(np.uint8(np.clip(luma * 255.0, 0.0, 255.0))).resize( (4, 4), Image.Resampling.BILINEAR, ), dtype=np.float64, ) / 255.0 ) small_luma -= float(np.mean(small_luma)) for y in range(4): for x in range(4): values[f"image_luma_centered_{y}_{x}"] = float(small_luma[y, x]) def _empty_image_features(values: dict[str, float]) -> None: for channel in ("red", "green", "blue"): for index in range(4): values[f"image_{channel}_histogram_{index}"] = 0.0 for name in ( "mean", "std", "p10", "p25", "p50", "p75", "p90", ): values[f"image_luma_{name}"] = 0.0 values["image_saturation_mean"] = 0.0 values["image_saturation_std"] = 0.0 values["image_edge_mean"] = 0.0 values["image_edge_p90"] = 0.0 for y in range(4): for x in range(4): values[f"image_luma_centered_{y}_{x}"] = 0.0 def _named_quantiles( values: dict[str, float], prefix: str, raw: object, ) -> None: array = np.asarray(raw, dtype=np.float64).reshape(-1) finite = array[np.isfinite(array)] names = ("min", "p10", "p25", "p50", "p75", "p90", "max") quantiles = ( np.quantile(finite, (0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0)) if finite.size else np.full(7, -1.0) ) for name, value in zip(names, quantiles, strict=True): values[f"{prefix}_{name}"] = float(value) def _span_features( values: dict[str, float], prefix: str, raw: object, ) -> None: if ( isinstance(raw, list) and len(raw) == 2 and all(isinstance(item, list) and len(item) == 3 for item in raw) ): for index, axis in enumerate("xyz"): values[f"{prefix}_{axis}_span"] = _number( raw[1][index], -1.0, ) - _number(raw[0][index], -1.0) else: for axis in "xyz": values[f"{prefix}_{axis}_span"] = -1.0 def _range_span( values: dict[str, float], name: str, raw: object, ) -> None: values[name] = ( _number(raw[1], -1.0) - _number(raw[0], -1.0) if isinstance(raw, list) and len(raw) == 2 else -1.0 ) def _number(value: object, fallback: float) -> float: try: parsed = float(value) # type: ignore[arg-type] except (TypeError, ValueError): return fallback return parsed if math.isfinite(parsed) else fallback def _object(value: object) -> dict[str, Any]: if not isinstance(value, dict): raise RuntimeError("expected object") return value def _read_jsonl(path: Path) -> list[dict[str, Any]]: with path.open("r", encoding="utf-8-sig") as stream: return [_object(json.loads(line)) for line in stream] def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--acceptance-root", type=Path, required=True) parser.add_argument("--materialization-root", type=Path, required=True) return parser.parse_args() if __name__ == "__main__": raise SystemExit(main())