kemal
e518ea03d5
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.
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
"""IDAT reader: synthetic files + malformed-input matrix (spec IN-023, EDGE-004, EDGE-033)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gzip
|
|
import struct
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from simurg.core.errors import FormatError
|
|
from simurg.formats.idat import check_pair, parse_idat_filename, parse_run_time, read_idat
|
|
from simurg.synth.files import SynthIdat, encode_idat
|
|
|
|
|
|
def _synth(n: int = 50, **kw) -> SynthIdat: # type: ignore[no-untyped-def]
|
|
rng = np.random.default_rng(1)
|
|
ids = np.sort(rng.choice(10_000_000, n, replace=False)).astype(np.int32)
|
|
return SynthIdat(
|
|
address_ids=ids,
|
|
mean=rng.integers(0, 60000, n).astype(np.uint16),
|
|
n_beads=rng.integers(0, 30, n).astype(np.uint8),
|
|
**kw,
|
|
)
|
|
|
|
|
|
def test_roundtrip(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
s = _synth(sample_id="NA12878", plate="P1", well="A01")
|
|
p = tmp_path / "200000000001_R01C01_Grn.idat"
|
|
p.write_bytes(encode_idat(s))
|
|
i = read_idat(p)
|
|
assert i.n_addresses == 50
|
|
assert np.array_equal(i.address_ids, s.address_ids)
|
|
assert np.array_equal(i.mean, s.mean)
|
|
assert np.array_equal(i.n_beads, s.n_beads)
|
|
assert (i.barcode, i.position, i.sample_id, i.plate, i.well) == ("200000000001", "R01C01", "NA12878", "P1", "A01")
|
|
assert i.scan_time_iso == "2023-12-27T17:52:21"
|
|
assert i.scanner_software == "4.2.1.729"
|
|
assert i.mid_block_itemsize == 4
|
|
|
|
|
|
def test_midblock_uint8_inferred(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
p = tmp_path / "x.idat"
|
|
p.write_bytes(encode_idat(_synth(n=40, mid_block_itemsize=1)))
|
|
assert read_idat(p).mid_block_itemsize == 1
|
|
|
|
|
|
def test_gzip(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
p = tmp_path / "200000000001_R01C01_Red.idat.gz"
|
|
p.write_bytes(gzip.compress(encode_idat(_synth())))
|
|
assert read_idat(p).n_addresses == 50
|
|
|
|
|
|
def test_metadata_only(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
p = tmp_path / "x.idat"
|
|
p.write_bytes(encode_idat(_synth()))
|
|
i = read_idat(p, load_arrays=False)
|
|
assert i.n_addresses == 50 and len(i.mean) == 0 and i.barcode == "200000000001"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutate,code",
|
|
[
|
|
(lambda b: b"XDAT" + b[4:], "E_IDAT_FORMAT"), # wrong magic
|
|
(lambda b: b[:4] + struct.pack("<q", 4) + b[12:], "E_IDAT_FORMAT"), # unsupported version
|
|
(lambda b: b[:4] + struct.pack("<q", 1 | (7 << 32)) + b[12:], "E_IDAT_ENCRYPTED"), # encrypted v1
|
|
(lambda b: b[: len(b) // 2], "E_IDAT_FORMAT"), # truncated
|
|
(lambda b: b[:18] + struct.pack("<q", 10**9) + b[26:], "E_IDAT_FORMAT"), # TOC offset outside file
|
|
(lambda b: b[:8], "E_IDAT_FORMAT"), # tiny
|
|
],
|
|
)
|
|
def test_malformed(tmp_path, mutate, code) -> None: # type: ignore[no-untyped-def]
|
|
p = tmp_path / "bad.idat"
|
|
p.write_bytes(mutate(encode_idat(_synth())))
|
|
with pytest.raises(FormatError) as e:
|
|
read_idat(p)
|
|
assert e.value.code == code
|
|
|
|
|
|
def test_pair_mismatch(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
a, b = tmp_path / "a.idat", tmp_path / "b.idat"
|
|
a.write_bytes(encode_idat(_synth(n=50)))
|
|
b.write_bytes(encode_idat(_synth(n=49)))
|
|
with pytest.raises(FormatError) as e:
|
|
check_pair(read_idat(a), read_idat(b))
|
|
assert e.value.code == "E_IDAT_PAIR_MISMATCH"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name,expected",
|
|
[
|
|
("207895980019_R03C01_Grn.idat", ("207895980019", "R03C01", "Grn")),
|
|
("207895980019_r03c01_red.idat.gz", ("207895980019", "R03C01", "Red")),
|
|
("5723646052_R02C02_GRN.IDAT", ("5723646052", "R02C02", "Grn")),
|
|
("4019585376_B_Red.idat", None),
|
|
("notes.txt", None),
|
|
],
|
|
)
|
|
def test_filename(name, expected) -> None: # type: ignore[no-untyped-def]
|
|
assert parse_idat_filename(name) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw,iso",
|
|
[
|
|
("12/27/2023 5:53:34 PM", "2023-12-27T17:53:34"),
|
|
("1/2/2020 12:00:01 AM", "2020-01-02T00:00:01"),
|
|
("2020-01-02 13:14:15", "2020-01-02T13:14:15"),
|
|
("", ""),
|
|
("garbage", ""),
|
|
],
|
|
)
|
|
def test_run_time(raw, iso) -> None: # type: ignore[no-untyped-def]
|
|
assert parse_run_time(raw) == iso
|
|
|
|
|
|
def test_duplicate_addresses_rejected(tmp_path) -> None: # type: ignore[no-untyped-def]
|
|
from simurg.formats.idat import has_duplicates
|
|
|
|
assert not has_duplicates(np.array([1, 5, 9], np.int32))
|
|
assert not has_duplicates(np.array([9, 1, 5], np.int32))
|
|
assert has_duplicates(np.array([9, 1, 9], np.int32))
|
|
s = _synth(n=5)
|
|
s.address_ids = np.array([5, 1, 5, 7, 8], np.int32)
|
|
p = tmp_path / "dup.idat"
|
|
p.write_bytes(encode_idat(s))
|
|
with pytest.raises(FormatError):
|
|
read_idat(p)
|