Buckets:
| import hashlib | |
| import io | |
| import numpy as np | |
| from PIL import Image | |
| import pytest | |
| from thinkmorph_sft.split_validation import ( | |
| TASKS, audit_split, group_rows, image_fingerprint, near_duplicate, | |
| select_groups, source_identity, split_records, | |
| ) | |
| def _record(index, task="Chart_Refocus", subtype="default"): | |
| key = hashlib.sha256(f"image-{index}".encode()).hexdigest() | |
| image = {"sha256": key, "size": (64, 64), "dhash": int(key[:32], 16), | |
| "thumbnail": bytes([index % 256]) * (32 * 32 * 3)} | |
| return {"sample_id": str(index), "task": task, "subtype": subtype, | |
| "source_keys": [f"{task}/source-{index}"], "images": [image]} | |
| def test_decoded_image_hash_ignores_file_encoding_and_detects_resizing(): | |
| y, x = np.mgrid[:64, :64] | |
| pixels = np.stack([x * 4, y * 4, (x + y) * 2], axis=-1).astype(np.uint8) | |
| image = Image.fromarray(pixels) | |
| def encode(image, kind): | |
| stream = io.BytesIO() | |
| image.save(stream, format=kind) | |
| return image_fingerprint(stream.getvalue()) | |
| png, bmp = encode(image, "PNG"), encode(image, "BMP") | |
| assert png["sha256"] == bmp["sha256"] | |
| resized = encode(image.resize((128, 128), Image.Resampling.BICUBIC), "PNG") | |
| assert png["sha256"] != resized["sha256"] | |
| assert near_duplicate(png, resized) | |
| assert not near_duplicate(png, encode(Image.fromarray(255 - pixels), "PNG")) | |
| def test_image_families_group_jigsaw_layouts_and_maze_trace_variants(): | |
| assert source_identity("Jigsaw_Assembly", "sat_seg13_image_12_border.jpg") == source_identity("Jigsaw_Assembly", "sat_seg13_image_12_answer_2x2.png") | |
| assert source_identity("Spatial_Navigation", "frozenlake_s3_42_cot.png") == source_identity("Spatial_Navigation", "frozenlake_s3_42_visualization.png") | |
| assert source_identity("Visual_Search", "1234.png") != source_identity("Chart_Refocus", "1234.png") | |
| assert source_identity("Visual_Search", None) is None | |
| def test_transitive_grouping_includes_input_target_and_cross_task_matches(): | |
| first, second, third, unrelated = [_record(i) for i in range(4)] | |
| second["source_keys"] = first["source_keys"] | |
| second["images"].append(third["images"][0]) | |
| third["task"] = "Spatial_Navigation" | |
| groups, _ = group_rows([first, second, third, unrelated]) | |
| assert groups["0"] == groups["1"] == groups["2"] | |
| assert groups["3"] != groups["0"] | |
| def test_split_is_order_independent_stratified_and_group_safe(): | |
| records = [_record(i + task_index * 40, task) for task_index, task in enumerate(TASKS) for i in range(40)] | |
| records[1]["images"] = records[0]["images"] | |
| one, report = split_records(records) | |
| two, _ = split_records(list(reversed(records))) | |
| assert one == two | |
| for task in TASKS: | |
| assert report["task_counts"][task] == {"validation": 2, "train": 38} | |
| assert one["0"] == one["1"] | |
| audit_split(records, one) | |
| two["1"] = (one["1"][0], "train" if one["0"][1] == "validation" else "validation") | |
| with pytest.raises(AssertionError, match="leakage"): | |
| audit_split(records, two) | |
| def test_whole_groups_take_priority_over_exact_fraction(): | |
| groups = {"a": [0, 1], "b": [2, 3]} | |
| selected = select_groups(groups, target=3, seed=17) | |
| assert sum(len(groups[key]) for key in selected) == 2 | |
| assert select_groups(groups, target=0, seed=17) == set() | |
| with pytest.raises(ValueError, match="fraction"): | |
| split_records([], fraction=0) | |
| def test_audit_catches_pixel_leakage_even_with_inconsistent_group_ids(): | |
| first, second = _record(0), _record(1) | |
| second["images"] = first["images"] | |
| with pytest.raises(AssertionError, match="leakage"): | |
| audit_split([first, second], {"0": ("a", "train"), "1": ("b", "validation")}) | |
| def test_cache_view_uses_existing_shards_and_keeps_all_train_source(tmp_path, monkeypatch): | |
| import json | |
| import pyarrow as pa | |
| import pyarrow.parquet as pq | |
| import torch | |
| from sft.data import InterleavedCache | |
| import thinkmorph_sft.split_validation as split | |
| raw_root, cache = tmp_path / "raw", tmp_path / "full" | |
| cache.mkdir() | |
| (cache / "shards").mkdir() | |
| torch.save({"schema": "zebra_uniar_cache_v1", "items": []}, cache / "shards/data.pt") | |
| rows = [] | |
| for task_index, task in enumerate(TASKS): | |
| raw = [] | |
| directory = raw_root / task / "data" | |
| directory.mkdir(parents=True) | |
| relative = f"{task}/data/train-00000.parquet" | |
| for index in range(20): | |
| sample = f"{task}-{index}" | |
| pixels = np.random.default_rng(task_index * 100 + index).integers(0, 256, (32, 32, 3), dtype=np.uint8) | |
| stream = io.BytesIO() | |
| Image.fromarray(pixels).save(stream, format="PNG") | |
| cell = {"bytes": stream.getvalue(), "path": sample + ".png"} | |
| raw.append({"pid": sample, "problem_image_0": cell, "reasoning_image_0": cell}) | |
| rows.append({"sample_id": sample, "config": task, "group_id": sample, | |
| "source": {"shard": relative, "row": index, "pid": sample}, | |
| "split": "train", "split_score": .2, "cache_shard": "shards/data.pt", "cache_index": index}) | |
| pq.write_table(pa.Table.from_pylist(raw), directory / "train-00000.parquet") | |
| original = "".join(json.dumps(row) + "\n" for row in rows) | |
| (cache / "manifest.jsonl").write_text(original) | |
| (cache / "READY").write_text("complete\n") | |
| (cache / "config.json").write_text('{"schema":"fixture"}') | |
| monkeypatch.setattr(split, "EXPECTED_COUNTS", {task: 20 for task in TASKS}) | |
| output = tmp_path / "dev" | |
| report = split.build_view(raw_root, cache, output, workers=1) | |
| assert report["counts"] == {"train": 76, "validation": 4} | |
| assert report["group_leakage"] == report["input_or_target_pixel_leakage"] == 0 | |
| full, dev = InterleavedCache(cache), InterleavedCache(output) | |
| assert len(full.split_rows("train")) == 80 | |
| assert len(dev.split_rows("train")) == 76 | |
| assert len(dev.validation_rows(["validation"], 100)) == 4 | |
| assert full._items("shards/data.pt") == dev._items(dev.rows[0]["cache_shard"]) | |
| assert (cache / "manifest.jsonl").read_text() == original | |
| with pytest.raises(FileExistsError): | |
| split.build_view(raw_root, cache, output, workers=1) | |
Xet Storage Details
- Size:
- 6.24 kB
- Xet hash:
- b6dc32b0f562ab72b8bd440d67ce9c3e5ad38c2d19b145adb5d8081e3a079008
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.