File size: 18,193 Bytes
e6f80f3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 | """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")
# Sealed SpatialLM IR uses source meters/Z-up and a canonical zero floor.
# An explicit source floor supports separately audited test/version frames.
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()
|