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:
1 parent
705138d70b
commit
e518ea03d5
40 files changed
+4742
No files matched your search
@@ -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)
|
||||
Reference in new issue
Block a user