P1: file parsers, sample sheet, discovery, CLI inspect

Readers for IDAT v3, BPM (locus entries v4-v8), CSV manifest, EGT
(data blocks 5-9) and GTC v3-v5 (plus a layout-preserving GTC writer);
sample-sheet validation (IN-001..IN-014), IDAT discovery and pairing
(IN-020..IN-022), message catalogue and exit codes (CLI-014/016),
JSON Lines logging (CLI-007/008), atomic writes (CLI-011), run manifest,
and `simurg inspect`.

Validation (docs/validation/P1.md): GTC, BPM and EGT readers agree
exactly with Illumina BeadArrayFiles on GSA v4 / GDA v1 files; GTCs
round-trip byte-identically; BPM and CSV manifests are consistent.
This commit is contained in:
kemal committed 2026-09-27 16:56:39 +01:00
1 parent 368c97b9aa
commit 8cd092460f
40 files changed
+4742

No files matched your search

+152
View File
@@ -0,0 +1,152 @@
"""GTC codec: synthetic round-trips, property tests, corruption handling (spec IN-032, IN-033, EDGE-002)."""
from __future__ import annotations
import numpy as np
import pytest
from hypothesis import given, settings
from hypothesis import strategies as st
from simurg.core.errors import FormatError
from simurg.formats.gtc import (
CANONICAL_ORDER_V3,
CANONICAL_ORDER_V5,
XFORM_DTYPE,
Gtc,
ScannerData,
encode_gtc,
read_gtc,
write_gtc,
)
def make_gtc(n: int = 20, version: int = 5, seed: int = 0) -> Gtc:
rng = np.random.default_rng(seed)
xf = np.zeros(3, XFORM_DTYPE)
xf["version"] = 1
xf["offset_x"] = rng.normal(size=3)
xf["theta"] = rng.normal(size=3)
g = Gtc(
version=version,
num_snps=n,
sample_name="NA12878 ünicode",
cluster_file="c.egt",
snp_manifest="m.bpm",
imaging_date="12/27/2023 5:53:34 PM",
autocall_date="01/01/26 12:00 PM",
autocall_version="simurg-0.1",
normalization_transforms=xf,
controls_x=rng.integers(0, 60000, 8).astype(np.uint16),
controls_y=rng.integers(0, 60000, 8).astype(np.uint16),
raw_x=rng.integers(0, 60000, n).astype(np.uint16),
raw_y=rng.integers(0, 60000, n).astype(np.uint16),
genotypes=rng.integers(0, 4, n).astype(np.uint8),
base_calls=np.frombuffer(b"AG" * n, np.uint8).reshape(n, 2).copy(),
genotype_scores=rng.random(n).astype(np.float32),
scanner_data=ScannerData("N103", b"\x01\x00\x00\x00", b"\x02\x00\x00\x00", "4.2.1", "user"),
call_rate=0.99,
gender=b"F",
logr_dev=0.12,
gc10=0.4,
b_allele_freqs=rng.random(n).astype(np.float32),
logr_ratios=rng.normal(size=n).astype(np.float32),
percentiles_x=np.array([1, 2, 3], np.uint16),
percentiles_y=np.array([4, 5, 6], np.uint16),
slide_identifier="200000000001",
)
g.sample_data["p50gc"] = 0.7
g.sample_data["num_calls"] = n - 2
g.sample_data["num_no_calls"] = 2
return g
@pytest.mark.parametrize("version", [3, 5])
def test_roundtrip_canonical(tmp_path, version) -> None: # type: ignore[no-untyped-def]
g = make_gtc(version=version)
p = tmp_path / "s.gtc"
write_gtc(g, p)
h = read_gtc(p)
assert h.toc_order == (CANONICAL_ORDER_V5 if version == 5 else CANONICAL_ORDER_V3)
assert h.sample_name == g.sample_name
assert np.array_equal(h.raw_x, g.raw_x) and np.array_equal(h.genotypes, g.genotypes)
assert np.array_equal(h.genotype_scores, g.genotype_scores)
assert h.scanner_data == g.scanner_data
assert h.gc50 == pytest.approx(0.7) and h.num_calls == 18
assert h.gender_char == "F"
assert h.base_call_strings()[0] == "AG"
assert encode_gtc(h) == p.read_bytes() # byte-identical round-trip
if version == 5:
assert np.array_equal(h.logr_ratios, g.logr_ratios)
def test_noncanonical_layout_preserved(tmp_path) -> None: # type: ignore[no-untyped-def]
g = make_gtc()
g.toc_order = list(reversed(CANONICAL_ORDER_V5))
g.data_order = [t for t in CANONICAL_ORDER_V5 if t > 3][::-1]
data = encode_gtc(g)
p = tmp_path / "s.gtc"
p.write_bytes(data)
h = read_gtc(p)
assert h.toc_order == g.toc_order and h.data_order == g.data_order
assert encode_gtc(h) == data
@settings(max_examples=25, deadline=None)
@given(n=st.integers(0, 200), seed=st.integers(0, 10**6), name=st.text(max_size=300))
def test_property_roundtrip(tmp_path_factory, n, seed, name) -> None: # type: ignore[no-untyped-def]
g = make_gtc(n=n, seed=seed)
g.sample_name = name
data = encode_gtc(g)
p = tmp_path_factory.mktemp("g") / "x.gtc"
p.write_bytes(data)
h = read_gtc(p)
assert h.sample_name == name and h.num_snps == n
assert encode_gtc(h) == data
def test_truncated_and_bad_magic(tmp_path) -> None: # type: ignore[no-untyped-def]
data = encode_gtc(make_gtc())
for label, blob in {
"trunc": data[: len(data) // 3],
"magic": b"xtc" + data[3:],
"tiny": data[:5],
"version": data[:3] + b"\x09" + data[4:],
}.items():
p = tmp_path / f"{label}.gtc"
p.write_bytes(blob)
with pytest.raises(FormatError) as e:
read_gtc(p)
assert e.value.code == "E_GTC_CORRUPT", label
def test_bad_string_length_is_clean_error(tmp_path) -> None: # type: ignore[no-untyped-def] # EDGE-002
g = make_gtc()
data = bytearray(encode_gtc(g))
header = 8 + 6 * len(CANONICAL_ORDER_V5)
data[header] = 0xFF # sample-name length prefix now runs past EOF
data[header + 1] = 0xFF
data[header + 2] = 0xFF
data[header + 3] = 0x7F
p = tmp_path / "s.gtc"
p.write_bytes(bytes(data))
with pytest.raises(FormatError) as e:
read_gtc(p)
assert e.value.code == "E_GTC_CORRUPT"
def test_incomplete_write_detection(tmp_path) -> None: # type: ignore[no-untyped-def]
g = make_gtc()
g.toc_order = [t for t in CANONICAL_ORDER_V5 if t != 1016]
g.data_order = [t for t in g.toc_order if t > 3]
p = tmp_path / "s.gtc"
p.write_bytes(encode_gtc(g))
with pytest.raises(FormatError):
read_gtc(p)
assert read_gtc(p, check_write_complete=False).num_snps == 20
def test_length_mismatch_rejected_on_write() -> None:
g = make_gtc()
g.raw_x = g.raw_x[:-1]
with pytest.raises(ValueError):
encode_gtc(g)