| """Derive a minimal rectangular-room bbox task from sealed SpatialLM geometry. |
| |
| This builder qualifies geometric box axes, not asset semantic fronts. Parent |
| splits and labels are preserved; a recorded room-frame translation is the only |
| allowed numeric target transformation. Old datasets remain immutable. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| from collections import Counter |
| from contextlib import ExitStack |
| from copy import deepcopy |
| import json |
| import math |
| from pathlib import Path |
|
|
| from shapely.geometry import Polygon |
|
|
| from .direct_layout import request_to_condition |
| from .io import fingerprint, safe_output |
| from .schema import validate_layout |
|
|
| SPATIALLM_IR_SHA256 = "f582b46d45b38e59b7b593d045f946f889d08942c10147bf4f077b7f33ef4410" |
| SPLITS = ("train", "validation", "test") |
| UNKNOWN_ROOM_TYPES = {"misc", "other", "other room", "unknown", "undefined", "none", ""} |
| YAW_SEMANTICS = "local_bbox_axes_not_certified_semantic_front" |
|
|
|
|
| def _finite(value): |
| return not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(value) |
|
|
|
|
| def _room_frame(room): |
| if not isinstance(room.get("room_type"), str) or room["room_type"].strip().lower() in UNKNOWN_ROOM_TYPES: |
| raise ValueError("room_type_unknown") |
| if room.get("boundary_known") is not True: |
| raise ValueError("boundary_unknown") |
| if room.get("floor_known") is not True or not _finite(room.get("floor_z_m")): |
| raise ValueError("floor_unknown") |
| points = room.get("floor_polygon_xy_m", []) |
| if len(points) < 3 or not all(len(p) == 2 and all(_finite(v) for v in p) for p in points): |
| raise ValueError("rectangle_invalid_polygon") |
| polygon = Polygon(points) |
| lo = [min(p[q] for p in points) for q in (0, 1)] |
| hi = [max(p[q] for p in points) for q in (0, 1)] |
| area = (hi[0] - lo[0]) * (hi[1] - lo[1]) |
| if not polygon.is_valid or area <= 0 or abs(polygon.area - area) > max(1e-8, area * 1e-6): |
| raise ValueError("rectangle_not_axis_aligned") |
| height = room.get("height_m") |
| if height is not None and (not _finite(height) or height <= 0): |
| raise ValueError("height_invalid") |
| return [*lo, float(room["floor_z_m"])], [hi[q] - lo[q] for q in (0, 1)], height |
|
|
|
|
| def _source_room_metadata(sample, source_room): |
| room, provenance = sample["condition"]["room"], sample["provenance"] |
| if source_room.get("room_type") != room.get("room_type"): |
| raise ValueError("source_room_type_mismatch") |
| parent_height, source_height = room.get("height_m"), source_room.get("height") |
| recorded_drop = parent_height is None and provenance.get("legacy_height_dropped") is True |
| if parent_height != source_height and not recorded_drop: |
| raise ValueError("source_room_height_mismatch") |
| if source_height is not None and (not _finite(source_height) or source_height <= 0): |
| raise ValueError("source_room_height_invalid") |
| |
| |
| source_floor = source_room.get("floor_z_m", 0.) |
| if not _finite(source_floor) or source_floor != room.get("floor_z_m"): |
| raise ValueError("source_floor_frame_mismatch") |
| if source_room.get("boundary_type") != "polygon": |
| raise ValueError("source_boundary_semantics_mismatch") |
| if room.get("frame") != "right_handed_z_up": |
| raise ValueError("source_floor_frame_axis_mismatch") |
| return source_height |
|
|
|
|
| def _source_geometry(sample, source_room): |
| provenance = sample["provenance"] |
| if (source_room.get("source") != "SpatialLM" or source_room.get("uid") != provenance.get("scene_id") |
| or source_room.get("group") != provenance.get("group") |
| or source_room.get("group") != provenance.get("house_id")): |
| raise ValueError("source_identity_mismatch") |
| source_height = _source_room_metadata(sample, source_room) |
| if source_room.get("boundary") != sample["condition"]["room"]["floor_polygon_xy_m"]: |
| raise ValueError("source_boundary_mismatch") |
| source_ids = provenance.get("target_source_ids", []) |
| targets = sample["target"]["objects"] |
| evidence = provenance.get("field_evidence", []) |
| if len(source_ids) != len(targets) or len(set(source_ids)) != len(source_ids) or len(evidence) != len(targets): |
| raise ValueError("source_identity_missing_or_duplicate") |
| source_objects = {obj["id"]: obj for obj in source_room.get("objects", [])} |
| for target, source_id, fields in zip(targets, source_ids, evidence): |
| obj = source_objects.get(source_id) |
| if obj is None: |
| raise ValueError("source_identity_missing_object") |
| if obj.get("tilted") is not False or fields.get("tilted") is not False: |
| raise ValueError("tilted_object") |
| if fields.get("size_semantics") != "canonical_source_IR": |
| raise ValueError("source_geometry_proxy_size") |
| if not _finite(obj.get("yaw")): |
| raise ValueError("source_geometry_invalid_yaw") |
| yaw = (obj["yaw"] + math.pi) % (2 * math.pi) - math.pi |
| if (obj.get("pos") != target.get("bottom_center_m") or obj.get("size") != target.get("target_size_local_m") |
| or yaw != target.get("yaw_rad")): |
| raise ValueError("source_geometry_mismatch") |
| return source_height |
|
|
|
|
| def _complete_geometry(sample): |
| targets = sample["target"]["objects"] |
| masks = sample.get("validity", {}) |
| for key in ("position", "size"): |
| rows = masks.get(key, []) |
| if len(rows) != len(targets) or not all(len(row) == 3 and all(flag is True for flag in row) for row in rows): |
| raise ValueError("incomplete_geometry_labels") |
| for target in targets: |
| p, s, yaw = target.get("bottom_center_m", []), target.get("target_size_local_m", []), target.get("yaw_rad") |
| if len(p) != 3 or len(s) != 3 or not all(_finite(v) for v in p + s) or min(s) <= 0 or not _finite(yaw): |
| raise ValueError("invalid_geometry_labels") |
|
|
|
|
| def _inside_room(targets, origin, dimensions, height, tolerance): |
| xmin, ymin, floor = origin |
| xmax, ymax = xmin + dimensions[0], ymin + dimensions[1] |
| for obj in targets: |
| p, s, yaw = obj["bottom_center_m"], obj["target_size_local_m"], obj["yaw_rad"] |
| for sx, sy in ((-1, -1), (-1, 1), (1, -1), (1, 1)): |
| x = p[0] + (sx * s[0] * math.cos(yaw) - sy * s[1] * math.sin(yaw)) / 2 |
| y = p[1] + (sx * s[0] * math.sin(yaw) + sy * s[1] * math.cos(yaw)) / 2 |
| if not xmin - tolerance <= x <= xmax + tolerance or not ymin - tolerance <= y <= ymax + tolerance: |
| raise ValueError("xy_boundary_conflict") |
| if p[2] < floor - tolerance: |
| raise ValueError("floor_boundary_conflict") |
| if height is not None and p[2] + s[2] > floor + height + tolerance: |
| raise ValueError("ceiling_boundary_conflict") |
|
|
|
|
| def _validate_options(max_objects, tolerance_m, room_dimension_mode): |
| if isinstance(max_objects, bool) or not isinstance(max_objects, int) or max_objects < 1: |
| raise ValueError("max_objects must be a positive integer") |
| if not _finite(tolerance_m) or tolerance_m < 0: |
| raise ValueError("tolerance_m must be finite and nonnegative") |
| if room_dimension_mode not in {"xy", "xyz"}: |
| raise ValueError("room_dimension_mode must be xy or xyz") |
|
|
|
|
| def project_sample(sample, source_room, *, split, max_objects=128, tolerance_m=.001, |
| room_dimension_mode="xy"): |
| """Reject uncertain source fields; use the exact public minimal input adapter.""" |
| _validate_options(max_objects, tolerance_m, room_dimension_mode) |
| provenance = sample["provenance"] |
| if provenance.get("source") != "SpatialLM": |
| raise ValueError("source_not_qualified") |
| if provenance.get("split") != split or split not in SPLITS: |
| raise ValueError("source_split_mismatch") |
| room = sample["condition"]["room"] |
| origin, dimensions, height = _room_frame(room) |
| requests, targets = sample["condition"]["objects"], sample["target"]["objects"] |
| if not requests or len(requests) > max_objects: |
| raise ValueError("max_objects_exceeded_or_empty") |
| _complete_geometry(sample) |
| qualification_height = _source_geometry(sample, source_room) |
| _inside_room(targets, origin, dimensions, qualification_height, tolerance_m) |
| inventory = [{"id": obj["id"], "category": obj["category"], "description": obj["category"], "count": 1} |
| for obj in requests] |
| request_dimensions = dimensions + ([height] if room_dimension_mode == "xyz" and height is not None else []) |
| request = {"room_type": room["room_type"], "room_size_m": request_dimensions, |
| "furniture_list": inventory} |
| condition = request_to_condition(request, max_objects=max_objects) |
| translated = [{**deepcopy(obj), "bottom_center_m": [v - offset for v, offset in zip(obj["bottom_center_m"], origin)]} |
| for obj in targets] |
| target = {"schema_version": "fastfill.v2", "objects": translated} |
| validate_layout(target, condition) |
| n = len(translated) |
| return {"schema_version": "fastfill.v2", "condition": condition, "target": target, |
| "validity": {"position": [[True] * 3 for _ in range(n)], "size": [[True] * 3 for _ in range(n)], |
| "yaw": [True] * n, "yaw_symmetry_order": [2] * n}, |
| "provenance": {**deepcopy(provenance), "minimal_request": request, |
| "condition_projection": "room_type_size_furniture_only-v1", |
| "room_dimension_mode": room_dimension_mode, |
| "qualification_source_height_m": qualification_height, |
| "frame_translation_m": [-value for value in origin], |
| "parent_yaw_validity": deepcopy(sample["validity"].get("yaw", [])), |
| "yaw_label_semantics": YAW_SEMANTICS, |
| "geometry_yaw_evidence": "sealed SpatialLM source IR: full local extents + yaw-only Bbox", |
| "correspondence": "fixed_request_identity", |
| "qualification_tolerance_m": tolerance_m, |
| "geometry_checks_scope": "OBB XY/floor/known ceiling only; collision/support/mesh/physics not certified"}} |
|
|
|
|
| def _parent_rows(parent): |
| for split in SPLITS: |
| with (parent / (split + ".jsonl")).open() as stream: |
| for number, line in enumerate(stream, 1): |
| if line.strip(): |
| yield split, number, json.loads(line) |
|
|
|
|
| def _split_integrity(parent): |
| uids, groups = set(), {} |
| for split, number, row in _parent_rows(parent): |
| provenance = row["provenance"] |
| uid = provenance.get("scene_id") |
| if not uid or uid in uids or provenance.get("split") != split: |
| raise ValueError(f"parent UID/split integrity failure at {split}:{number}") |
| uids.add(uid) |
| for alias in (provenance.get("group"), provenance.get("house_id")): |
| if not isinstance(alias, str) or not alias: |
| raise ValueError("parent underlying group identity missing") |
| if alias in groups and groups[alias] != split: |
| raise ValueError("parent group alias has cross-split conflict") |
| groups[alias] = split |
| return {"parent_uids_checked": len(uids), "parent_group_aliases_checked": len(groups), |
| "duplicate_uids": 0, "group_cross_split_conflicts": 0, "split_rule": "inherit parent without re-splitting"} |
|
|
|
|
| def _source_index(source): |
| result = {} |
| with source.open() as stream: |
| for line in stream: |
| if not line.strip(): |
| continue |
| row = json.loads(line) |
| if row.get("source") != "SpatialLM" or row.get("uid") in result: |
| raise ValueError("sealed SpatialLM source IR has invalid source or duplicate UID") |
| result[row["uid"]] = row |
| return result |
|
|
|
|
| def build_minimal_dataset(parent_root, source_ir, output, *, max_objects=128, tolerance_m=.001, |
| expected_source_sha256=SPATIALLM_IR_SHA256, room_dimension_mode="xy"): |
| """Build a new, hash-pinned derivative. No source files are normalized in place.""" |
| _validate_options(max_objects, tolerance_m, room_dimension_mode) |
| parent, source = Path(parent_root).resolve(), Path(source_ir).resolve() |
| target = safe_output(output) |
| if parent == target or parent in target.parents or target in parent.parents or source == target or target in source.parents: |
| raise ValueError("output must be new and outside parent/source data and ancestors") |
| inputs = [parent / (split + ".jsonl") for split in SPLITS] + [parent / "manifest.json", source] |
| before = {str(path): fingerprint(path) for path in inputs} |
| if before[str(source)] != expected_source_sha256: |
| raise ValueError("source IR SHA256 does not match the explicitly audited version") |
| integrity = _split_integrity(parent) |
| source_rows = _source_index(source) |
| counts, object_counts, rejected, inspected = Counter(), Counter(), Counter(), Counter() |
| target.mkdir(parents=True, exist_ok=False) |
| with ExitStack() as stack: |
| streams = {split: stack.enter_context((target / (split + ".jsonl")).open("x")) for split in SPLITS} |
| exclusions = stack.enter_context((target / "exclusions.jsonl").open("x")) |
| for split, number, row in _parent_rows(parent): |
| inspected[split] += 1 |
| uid, source_name = row["provenance"]["scene_id"], row["provenance"]["source"] |
| try: |
| derived = project_sample(row, source_rows.get(uid, {}), split=split, |
| max_objects=max_objects, tolerance_m=tolerance_m, |
| room_dimension_mode=room_dimension_mode) |
| except ValueError as error: |
| reason = str(error) |
| rejected[reason] += 1 |
| exclusions.write(json.dumps({"uid": uid, "split": split, "source": source_name, |
| "parent_line": number, "reason": reason}, allow_nan=False) + "\n") |
| continue |
| streams[split].write(json.dumps(derived, separators=(",", ":"), allow_nan=False) + "\n") |
| counts[split] += 1 |
| object_counts[split] += len(derived["target"]["objects"]) |
| if any(fingerprint(path) != before[str(path)] for path in inputs): |
| raise RuntimeError("input changed during immutable dataset build; output is not a valid release") |
| if not counts: |
| raise ValueError("no qualified minimal-condition examples; output is not a valid release") |
| manifest = {"schema_version": "fastfill.v2", "builder": "spatiallm-direct-bbox-v1", |
| "parent_root": str(parent), "source_ir": str(source), "source_ir_sha256": before[str(source)], |
| "parent_sha256": {path.name: before[str(path)] for path in inputs[:-1]}, |
| "samples_written": sum(counts.values()), "objects_written": sum(object_counts.values()), |
| "split_samples": {s: counts[s] for s in SPLITS}, "split_objects": {s: object_counts[s] for s in SPLITS}, |
| "split_input_samples": dict(inspected), "exclusion_counts": dict(rejected), "split_integrity": integrity, |
| "input_contract": "room_type + room_size_m" + ("[W,D]" if room_dimension_mode == "xy" else "[W,D,(H)]") |
| + " + furniture_list; no assets/fixed/constraints/support", |
| "room_dimension_mode": room_dimension_mode, |
| "height_policy": "xy excludes source height from model input; any source height remains qualification evidence only; " |
| "xyz explicitly opts into known exported room height; missing height stays unknown", |
| "source_metadata_agreement": "room type, source/published height, canonical floor frame, polygon boundary " |
| "semantics and house/group must agree with pinned IR; recorded height drops never bypass source ceiling checks", |
| "yaw_label_semantics": YAW_SEMANTICS, "yaw_symmetry_order": 2, |
| "correspondence": "fixed_request_identity", "max_objects": max_objects, "tolerance_m": tolerance_m, |
| "qualification": "SpatialLM; known type/floor/boundary; valid polygon area=AABB within 1e-6; " |
| "complete position/size; upright; finite source-matched geometric yaw; all OBB corners in XY, " |
| "bottom above floor and top below any known ceiling within tolerance", |
| "source_data_modified": False, "parent_data_modified": False, |
| "target_changes": "recorded common translation only; local sizes and yaw unchanged; no snapping/scaling", |
| "geometry_scope": "Boundary-qualified OBBs; collision/support completeness and mesh/physics not certified", |
| "old_pilot_scope": "Earlier two-scene MultiScan pilot used richer conditions and did not train this contract", |
| "implementation_sha256": fingerprint(Path(__file__)), |
| "output_sha256": {path.name: fingerprint(path) for path in sorted(target.glob("*.jsonl"))}} |
| (target / "manifest.json").write_text(json.dumps(manifest, indent=2, allow_nan=False) + "\n") |
| return manifest |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--parent-root", required=True) |
| parser.add_argument("--source-ir", required=True) |
| parser.add_argument("--output", required=True) |
| parser.add_argument("--max-objects", type=int, default=128) |
| parser.add_argument("--tolerance-m", type=float, default=.001) |
| parser.add_argument("--room-dimensions", choices=("xy", "xyz"), default="xy") |
| args = parser.parse_args() |
| result = build_minimal_dataset(args.parent_root, args.source_ir, args.output, |
| max_objects=args.max_objects, tolerance_m=args.tolerance_m, |
| room_dimension_mode=args.room_dimensions) |
| print(json.dumps(result, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
