| """Audit fixes 2026-10-06 (owner B): group bookkeeping, rendering order, augmentation, flag filter.""" |
| import json |
| import math |
| from pathlib import Path |
|
|
| import pytest |
| from shapely.geometry import Polygon |
| import torch |
|
|
| from fastfill.v2.batch import TinyTokenizer, augment_sample, collate_samples, condition_segments |
| from fastfill.v2.data import filter_rows_by_flags |
| from fastfill.v2.direct_layout import request_to_condition |
| from fastfill.v2.geometry import wrap_yaw |
| from fastfill.v2.matching import certify_group |
| from fastfill.v2.validation import _geometry, footprint |
|
|
| REAL_ROWS = Path("/Volumes/harddisk/FastFill_v2_multisource_20261006/eligible-local/validation.jsonl") |
| NO_AUGMENT = {"rotate90": False, "mirror": False, "shuffle_objects": False, |
| "drop_constraints_p": 0., "drop_support_p": 0., "category_only_description_p": 0., "minimal_form_p": 0.} |
| L_SHAPE = [[0, 0], [6, 0], [6, 3], [3, 3], [3, 5], [0, 5]] |
|
|
|
|
| def sample(): |
| """L-shaped room, support chain, every constraint reference kind, one exchangeable pair.""" |
| objects = [{"id": "chair_a", "category": "chair", "description": "chair"}, |
| {"id": "chair_b", "category": "chair", "description": "chair"}, |
| {"id": "table", "category": "table", "description": "oak table", "support_parent": "floor"}, |
| {"id": "lamp", "category": "lamp", "description": "lamp", "support_parent": "table"}, |
| {"id": "rug", "category": "rug", "description": "rug", "support_parent": "floor"}] |
| targets = {"chair_a": ([1., 1., 0.], .3), "chair_b": ([5., 1., 0.], 2.), "table": ([2., 2., 0.], 1.), |
| "lamp": ([2., 2., .7], -2.5), "rug": ([1.5, 4., 0.], 0.)} |
| return { |
| "schema_version": "fastfill.v2", |
| "condition": {"schema_version": "fastfill.v2", "room": { |
| "frame": "right_handed_z_up", "floor_polygon_xy_m": L_SHAPE, "floor_z_m": 0., "height_m": 2.8, |
| "floor_known": True, "boundary_known": True, "boundary_quality": "polygon", "room_type": "office", |
| "fixed_objects": [{"id": "fixed_0000", "category": "door", "size_local_m": [.1, .9, 2.1], |
| "bottom_center_m": [4.5, 0.05, 0.], "yaw_rad": 1.2}]}, |
| "objects": objects, "constraints": [ |
| {"type": "near", "object_id": "lamp", "target_id": "table", "max_distance_m": .5}, |
| {"type": "between", "object_id": "rug", "target_ids": ["table", "fixed_0000"]}, |
| {"type": "on", "object_id": "lamp", "parent_id": "table"}, |
| {"type": "faces_direction", "object_id": "table", "direction_xy": [1, 0]}, |
| {"type": "keepout", "polygon_xy_m": [[.1, .1], [.4, .1], [.4, .4], [.1, .4]]}]}, |
| "target": {"objects": [{"id": o["id"], "target_size_local_m": [.6, .5, .7], "bottom_center_m": targets[o["id"]][0], |
| "yaw_rad": targets[o["id"]][1]} for o in objects]}, |
| "validity": {"position": [[True] * 3 for _ in objects], "size": [[True] * 3 for _ in objects], |
| "yaw": [True for _ in objects], "yaw_symmetry_order": [2, 2, 2, 1, 2], |
| "size_axis_swap_allowed": [False] * len(objects), |
| "exchangeable_group": ["anonymous_0", "anonymous_0", None, None, None]}, |
| "provenance": {"source": "offline_test", "house_id": "h", "split": "train", "legacy_flags": {}}} |
|
|
|
|
| def rendered(condition): |
| text = "".join(segment for segment, _ in condition_segments(condition)) |
| body = text[text.index("{"):text.rindex("}") + 1] |
| return text, json.loads(body) |
|
|
|
|
| def boxes(s): |
| return [_geometry(obj, "target") for obj in s["target"]["objects"]] |
|
|
|
|
| def real_row(): |
| if not REAL_ROWS.is_file(): |
| pytest.skip("external data disk is not mounted") |
| with REAL_ROWS.open() as stream: |
| for line in stream: |
| row = json.loads(line) |
| if sum(obj.get("exchangeable_group") is not None for obj in row["condition"]["objects"]) >= 2: |
| return row |
| pytest.skip("no legacy-group row in the first rows") |
|
|
|
|
| |
|
|
| def test_collate_exposes_groups_per_request_slot_from_validity(): |
| s = sample() |
| s["target"]["objects"].reverse() |
| s["validity"]["exchangeable_group"] = [None, None, None, "anonymous_0", "anonymous_0"] |
| b = collate_samples([s], TinyTokenizer()) |
| assert b["exchangeable_group"] == [["anonymous_0", "anonymous_0", None, None, None]] |
| assert all("exchangeable_group" not in obj for obj in b["objects"][0]) |
|
|
|
|
| def test_legacy_condition_groups_migrate_into_validity_and_leave_the_text(): |
| s = sample() |
| del s["validity"]["exchangeable_group"] |
| for obj in s["condition"]["objects"][:2]: |
| obj["exchangeable_group"] = "anonymous_7" |
| s["target"]["objects"].reverse() |
| b = collate_samples([s], TinyTokenizer()) |
| assert b["exchangeable_group"] == [["anonymous_7", "anonymous_7", None, None, None]] |
| assert all("exchangeable_group" not in obj for obj in b["objects"][0] + b["conditions"][0]["objects"]) |
| assert "exchangeable_group" not in TinyTokenizer().decode(b["input_ids"][0]) |
| s["validity"]["exchangeable_group"] = [None] * 5 |
| with pytest.raises(ValueError, match="both condition objects and validity"): |
| collate_samples([s], TinyTokenizer()) |
| |
| inference = collate_samples([{"condition": s["condition"]}], TinyTokenizer()) |
| assert inference["exchangeable_group"] == [[None] * 5] and not inference["validity"]["position"].any() |
|
|
|
|
| def test_group_labels_must_be_nonempty_strings_or_null(): |
| s = sample() |
| s["validity"]["exchangeable_group"][0] = "" |
| with pytest.raises(ValueError, match="exchangeable_group labels"): |
| collate_samples([s], TinyTokenizer()) |
|
|
|
|
| |
|
|
| def test_rendered_condition_order_and_stripped_keys(): |
| s = sample() |
| s["condition"]["objects"][0]["exchangeable_group"] = "anonymous_0" |
| s["condition"]["room"]["_debug"] = {"x": 1} |
| s["condition"]["objects"][1]["attributes"] = {"_hidden": 1, "colour": "red"} |
| text, body = rendered(s["condition"]) |
| assert list(body) == ["schema_version", "room", "constraints", "objects"] |
| assert "boundary_quality" not in text and "exchangeable_group" not in text and "_debug" not in text and "_hidden" not in text |
| assert body["room"]["boundary_known"] is True and body["room"]["floor_known"] is True |
| assert body["objects"][1]["attributes"] == {"colour": "red"} and body["room"]["fixed_objects"] |
| assert body["objects"][0] == {"id": "chair_a", "category": "chair", "description": "chair"} |
| with pytest.raises(ValueError, match="unknown fields"): |
| condition_segments({**s["condition"], "target": {}}) |
|
|
|
|
| def test_direct_request_renders_the_same_room_fields_as_a_training_rectangle(): |
| direct = request_to_condition({"room_type": "office", "room_size_m": [4., 3., 2.7], "furniture_list": ["chair"]}) |
| training = {**sample()["condition"]} |
| training["room"] = {k: v for k, v in training["room"].items() if k != "fixed_objects"} |
| _, direct_body = rendered(direct) |
| _, training_body = rendered(training) |
| assert set(direct_body["room"]) == set(training_body["room"]) |
| assert "boundary_quality" not in direct_body["room"] |
| assert "boundary_quality" in direct["room"] |
|
|
|
|
| |
|
|
| def test_eval_path_and_zero_augmentation_are_identity(): |
| s = sample() |
| plain = collate_samples([s], TinyTokenizer()) |
| zero = collate_samples([s], TinyTokenizer(), augment=NO_AUGMENT, generator=torch.Generator().manual_seed(0)) |
| assert torch.equal(plain["input_ids"], zero["input_ids"]) |
| assert torch.equal(plain["targets"]["position_normalized"], zero["targets"]["position_normalized"]) |
| assert plain["conditions"] == zero["conditions"] == [sample()["condition"]] |
| with pytest.raises(ValueError, match="unknown augmentation"): |
| augment_sample(s, {"rotate": True}) |
| with pytest.raises(ValueError, match="probability"): |
| augment_sample(s, {"drop_support_p": 1.5}) |
|
|
|
|
| def test_augmentation_is_deterministic_given_a_generator(): |
| s = sample() |
| runs = [augment_sample(s, {}, torch.Generator().manual_seed(3)) for _ in range(2)] |
| assert runs[0] == runs[1] |
| assert runs[0] != augment_sample(s, {}, torch.Generator().manual_seed(4)) |
| assert s == sample() |
|
|
|
|
| @pytest.mark.parametrize("mirror", [False, True]) |
| def test_rigid_augmentation_preserves_containment_distances_and_relative_yaw(mirror): |
| s = sample() |
| before = boxes(s) |
| seen, mirrored_runs = set(), set() |
| for seed in range(12): |
| a = augment_sample(s, {**NO_AUGMENT, "rotate90": True, "mirror": mirror}, torch.Generator().manual_seed(seed)) |
| polygon = a["condition"]["room"]["floor_polygon_xy_m"] |
| seen.add((tuple(map(tuple, polygon)))) |
| room = Polygon(polygon) |
| mirrored = room.exterior.is_ccw != Polygon(L_SHAPE).exterior.is_ccw |
| mirrored_runs.add(mirrored) |
| assert room.is_valid and math.isclose(room.area, Polygon(L_SHAPE).area) |
| assert [min(p[q] for p in polygon) for q in range(2)] == [0, 0] |
| after = boxes(a) |
| assert all(room.buffer(1e-9).covers(footprint(obj)) for obj in after) |
| door = [_geometry(x["condition"]["room"]["fixed_objects"][0], "target") for x in (s, a)] |
| for i in range(len(after)): |
| for j in range(i + 1, len(after)): |
| assert math.isclose(math.dist(before[i]["_pos"], before[j]["_pos"]), math.dist(after[i]["_pos"], after[j]["_pos"]), abs_tol=1e-9) |
| relative = [rows[i]["_yaw"] - rows[j]["_yaw"] for rows in (before, after)] |
| assert abs(wrap_yaw(relative[1] - (-relative[0] if mirrored else relative[0]))) < 1e-9 |
| assert math.isclose(math.dist(door[0]["_pos"], before[0]["_pos"]), math.dist(door[1]["_pos"], after[0]["_pos"]), abs_tol=1e-9) |
| assert abs(wrap_yaw((door[1]["_yaw"] - after[0]["_yaw"]) - (-1 if mirrored else 1) * (door[0]["_yaw"] - before[0]["_yaw"]))) < 1e-9 |
| assert room.buffer(1e-9).covers(Polygon(a["condition"]["constraints"][4]["polygon_xy_m"])) |
| direction = a["condition"]["constraints"][3]["direction_xy"] |
| facing = after[2]["_yaw"] |
| assert math.isclose(math.cos(facing) * direction[0] + math.sin(facing) * direction[1], math.cos(1.), abs_tol=1e-9) |
| assert all(obj["target_size_local_m"] == [.6, .5, .7] for obj in a["target"]["objects"]) |
| collate_samples([a], TinyTokenizer()) |
| assert len(seen) >= 3 and mirrored_runs == ({True, False} if mirror else {False}) |
|
|
|
|
| def test_rotation_swaps_null_coordinates_with_their_validity_flags(): |
| s = sample() |
| s["target"]["objects"][2]["bottom_center_m"] = [2., None, 0.] |
| s["validity"]["position"][2] = [True, False, True] |
| s["target"]["objects"][3]["yaw_rad"] = None |
| s["validity"]["yaw"][3] = False |
| for seed in range(8): |
| a = augment_sample(s, {**NO_AUGMENT, "rotate90": True}, torch.Generator().manual_seed(seed)) |
| position = a["target"]["objects"][2]["bottom_center_m"] |
| assert [v is not None for v in position] == a["validity"]["position"][2] |
| assert a["target"]["objects"][3]["yaw_rad"] is None |
| b = collate_samples([a], TinyTokenizer()) |
| assert b["validity"]["position"][0, 2].tolist() == a["validity"]["position"][2] |
|
|
|
|
| def test_shuffle_keeps_alignment_references_and_certifiable_groups(): |
| s = sample() |
| s["validity"]["position"][1] = [False, False, False] |
| certify_group(s["condition"]["objects"], s["condition"]["constraints"], [0, 1]) |
| original = {o["description"]: (t["bottom_center_m"], t["yaw_rad"]) for o, t in zip(s["condition"]["objects"], s["target"]["objects"])} |
| orders = set() |
| for seed in range(8): |
| a = augment_sample(s, {**NO_AUGMENT, "shuffle_objects": True}, torch.Generator().manual_seed(seed)) |
| objects, targets = a["condition"]["objects"], a["target"]["objects"] |
| ids = [o["id"] for o in objects] |
| orders.add(tuple(o["description"] for o in objects)) |
| assert ids == [f"obj_{i:04d}" for i in range(5)] and [t["id"] for t in targets] == ids |
| by_id = {o["id"]: o for o in objects} |
| for obj, target in zip(objects, targets): |
| if obj["category"] != "chair": |
| assert (target["bottom_center_m"], target["yaw_rad"]) == original[obj["description"]] |
| lamp = next(o for o in objects if o["category"] == "lamp") |
| assert by_id[lamp["support_parent"]]["category"] == "table" |
| assert next(o for o in objects if o["category"] == "table")["support_parent"] == "floor" |
| kinds = {c["type"]: c for c in a["condition"]["constraints"]} |
| assert by_id[kinds["near"]["object_id"]]["category"] == "lamp" and by_id[kinds["near"]["target_id"]]["category"] == "table" |
| assert by_id[kinds["on"]["parent_id"]]["category"] == "table" |
| assert by_id[kinds["between"]["object_id"]]["category"] == "rug" |
| assert by_id[kinds["between"]["target_ids"][0]]["category"] == "table" and kinds["between"]["target_ids"][1] == "fixed_0000" |
| assert by_id[kinds["faces_direction"]["object_id"]]["category"] == "table" |
| b = collate_samples([a], TinyTokenizer()) |
| groups = b["exchangeable_group"][0] |
| members = [i for i, g in enumerate(groups) if g == "anonymous_0"] |
| assert [objects[i]["category"] for i in members] == ["chair", "chair"] and len(members) == 2 |
| certify_group(b["objects"][0], b["conditions"][0]["constraints"], members) |
| invalid = [i for i, row in enumerate(a["validity"]["position"]) if not any(row)] |
| assert len(invalid) == 1 and objects[invalid[0]]["category"] == "chair" |
| assert a["validity"]["yaw_symmetry_order"] == [1 if o["category"] == "lamp" else 2 for o in objects] |
| assert b["yaw_symmetry_order"][0].tolist() == a["validity"]["yaw_symmetry_order"] |
| assert not b["validity"]["position"][0, invalid[0]].any() |
| assert len(orders) > 1 |
|
|
|
|
| def test_shuffle_rejects_misaligned_validity_and_colliding_ids(): |
| s = sample() |
| s["validity"]["yaw"] = [True] |
| with pytest.raises(ValueError, match="aligned with target objects"): |
| augment_sample(s, {**NO_AUGMENT, "shuffle_objects": True}, torch.Generator().manual_seed(0)) |
| s = sample() |
| s["condition"]["room"]["fixed_objects"][0]["id"] = "obj_0002" |
| with pytest.raises(ValueError, match="collide"): |
| augment_sample(s, {**NO_AUGMENT, "shuffle_objects": True}, torch.Generator().manual_seed(0)) |
|
|
|
|
| def test_drop_augmentations_follow_probabilities(): |
| s = sample() |
| dropped = augment_sample(s, {**NO_AUGMENT, "drop_constraints_p": 1., "drop_support_p": 1., "category_only_description_p": 1.}) |
| assert dropped["condition"]["constraints"] == [] |
| assert all("support_parent" not in o and o["description"] == o["category"] for o in dropped["condition"]["objects"]) |
| assert dropped["target"] == s["target"] and dropped["validity"] == s["validity"] |
| collate_samples([dropped], TinyTokenizer()) |
| kept = augment_sample(s, NO_AUGMENT) |
| assert kept["condition"] == s["condition"] |
|
|
|
|
| def test_mirror_refuses_asset_local_metadata_and_rotate90_leaves_openings_unturned(): |
| s = sample() |
| s["condition"]["room"]["fixed_objects"][0]["semantic_front_local"] = [1., 0., 0.] |
| with pytest.raises(ValueError, match="mirror cannot transform"): |
| for seed in range(8): |
| augment_sample(s, {**NO_AUGMENT, "mirror": True}, torch.Generator().manual_seed(seed)) |
| s = sample() |
| s["condition"]["room"]["openings"] = [{"kind": "door"}] |
| for seed in range(8): |
| out = augment_sample(s, {**NO_AUGMENT, "rotate90": True}, torch.Generator().manual_seed(seed)) |
| assert out["condition"]["room"] == s["condition"]["room"] and out["target"] == s["target"] |
|
|
|
|
| |
|
|
| def test_filter_rows_by_flags_uses_legacy_flag_truthiness(): |
| rows = [{"provenance": {"legacy_flags": {"oob_objects": 5}}}, {"provenance": {"legacy_flags": {"oob_objects": 0}}}, |
| {"provenance": {"legacy_flags": {"fixed_collision": 1}}}, {"provenance": {}}] |
| assert filter_rows_by_flags(rows, ["oob_objects", "fixed_collision", "overlapping_furniture"]) == rows[1:2] + rows[3:] |
| assert filter_rows_by_flags(rows, []) == rows |
| with pytest.raises(ValueError, match="exclude_flags"): |
| filter_rows_by_flags(rows, "oob_objects") |
|
|
|
|
| |
|
|
| def test_real_legacy_row_collates_renders_and_augments(): |
| row = real_row() |
| tokenizer = TinyTokenizer() |
| b = collate_samples([row], tokenizer, max_length=10**7, max_objects=10**4) |
| text = tokenizer.decode(b["input_ids"][0]) |
| assert "exchangeable_group" not in text and "boundary_quality" not in text |
| assert text.index('"room":') < text.index('"constraints":') < text.index('"objects":') |
| assert text.rstrip().endswith("]}\nAssistant:") |
| expected = [obj.get("exchangeable_group") for obj in row["condition"]["objects"]] |
| assert b["exchangeable_group"] == [expected] and sum(g is not None for g in expected) >= 2 |
| assert all("exchangeable_group" not in obj for obj in b["objects"][0]) |
| before = boxes(row) |
| for seed in range(3): |
| a = augment_sample(row, {}, torch.Generator().manual_seed(seed)) |
| ab = collate_samples([a], tokenizer, max_length=10**7, max_objects=10**4) |
| groups = ab["exchangeable_group"][0] |
| |
| assert sum(g is not None for g in groups) >= sum(g is not None for g in expected) |
| for label in {g for g in groups if g}: |
| members = [i for i, g in enumerate(groups) if g == label] |
| certify_group(ab["objects"][0], ab["conditions"][0]["constraints"], members) |
| assert ab["validity"]["position"].sum() == b["validity"]["position"].sum() |
| assert ab["validity"]["size"].sum() == b["validity"]["size"].sum() |
| rigid = augment_sample(row, {"shuffle_objects": False, "mirror": False}, torch.Generator().manual_seed(seed)) |
| after = boxes(rigid) |
| assert all(math.isclose(math.dist(before[i]["_pos"], before[j]["_pos"]), math.dist(after[i]["_pos"], after[j]["_pos"]), abs_tol=1e-6) |
| for i in range(len(after)) for j in range(i + 1, len(after))) |
| assert all(abs(wrap_yaw((after[i]["_yaw"] - after[0]["_yaw"]) - (before[i]["_yaw"] - before[0]["_yaw"]))) < 1e-9 for i in range(len(after))) |
| assert all(x["target_size_local_m"] == y["target_size_local_m"] for x, y in zip(row["target"]["objects"], rigid["target"]["objects"])) |
|
|
