P4 (WIP): GenCall-equivalent calling, BAF/LRR, sex, GTC v5, simurg genotype
- genotype/clusters.py: EGT aligned to manifest order (float32 mean R) - genotype/gencall.py: flair, six-case likelihood, priors/posteriors, fuzzy sub-scores and score map, vectorised with the reference's float/double evaluation order and libm transcendentals - genotype/baf_lrr.py, sex.py (rules v1/v2), sample.py (IDAT -> Gtc) - pipeline/inputs.py shared by ingest and genotype - pipeline/genotype.py + CLI: GTC per sample, sample_summary.tsv/json (OUT-060..063), run manifest - GTCs from IDATs are byte-identical to the oracle for 36 GSA samples x 2 normalization versions (NaN sign canonicalised to the x86 pattern) Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01814GKVzZVL3FuF3DeiZXhv
This commit is contained in:
1 parent
0cd7eb8aa0
commit
febdd121ef
12 files changed
+1270
-53
No files matched your search
@@ -0,0 +1,87 @@
|
||||
"""Shared sample-set resolution for IDAT-based commands (spec CLI-006, IN-001 … IN-022, IN-010/IN-012)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from simurg.core.errors import SimurgError
|
||||
from simurg.core.log import Reporter
|
||||
from simurg.core.provenance import RunManifest
|
||||
from simurg.formats.discovery import ResolvedSample, discover_idats, resolve_input_folder, resolve_samples
|
||||
from simurg.formats.idat import read_idat
|
||||
from simurg.formats.samplesheet import SampleSheet, output_stem, parse_samplesheet
|
||||
|
||||
|
||||
@dataclass
|
||||
class SampleSet:
|
||||
folder: Path
|
||||
sheet: SampleSheet | None
|
||||
samples: list[ResolvedSample]
|
||||
stems: dict[str, str] # sample_id -> output stem
|
||||
|
||||
|
||||
def check_paths(pairs: list[tuple[Path | None, str]]) -> None:
|
||||
for p, what in pairs:
|
||||
if p is not None and not Path(p).exists():
|
||||
raise SimurgError("E_FILE_NOT_FOUND", f"{what} not found: {p}", path=str(p))
|
||||
|
||||
|
||||
def resolve_sample_set(
|
||||
idats: Path | None, sheet_path: Path | None, on_duplicate_idat: str, rep: Reporter, rm: RunManifest
|
||||
) -> SampleSet:
|
||||
sheet = parse_samplesheet(sheet_path) if sheet_path else None
|
||||
if sheet:
|
||||
rm.sheet_header = sheet.header
|
||||
for code, msg in sheet.warnings:
|
||||
rep.warn(code, msg)
|
||||
folder, notes = resolve_input_folder(idats, sheet)
|
||||
for code, msg in notes:
|
||||
rep.warn(code, msg) if code.startswith("W_") else rep.info(msg, code)
|
||||
disc = discover_idats(folder, on_duplicate=on_duplicate_idat)
|
||||
for code, msg in disc.warnings:
|
||||
rep.warn(code, msg)
|
||||
samples, notes = resolve_samples(disc, sheet)
|
||||
for code, msg in notes:
|
||||
rep.info(msg, code)
|
||||
if not samples:
|
||||
raise SimurgError("E_NO_INPUT", f"no IDAT pairs found under {folder}")
|
||||
|
||||
use_idat_names = sheet is None or not sheet.has_sample_name
|
||||
stems: dict[str, str] = {}
|
||||
taken: dict[str, str] = {}
|
||||
for s in samples:
|
||||
if use_idat_names and s.pair is not None and s.pair.grn is not None:
|
||||
# IN-010: sheet Sample_Name > IDAT sample name (field 406) > sample_id
|
||||
try:
|
||||
idat_name = read_idat(s.pair.grn, load_arrays=False).sample_id.strip()
|
||||
except SimurgError:
|
||||
idat_name = ""
|
||||
if idat_name:
|
||||
s.sample_name = idat_name
|
||||
stem = output_stem(s.sample_name)
|
||||
if stem in taken and taken[stem] != s.sample_id:
|
||||
if not use_idat_names: # IN-012: sheet names that sanitise to the same stem
|
||||
raise SimurgError(
|
||||
"E_SHEET_NAME_COLLISION",
|
||||
f"samples {taken[stem]} and {s.sample_id} both map to output name {stem!r}",
|
||||
)
|
||||
rep.info(f"{s.sample_id}: IDAT sample name {s.sample_name!r} is not unique; using the sample ID")
|
||||
s.sample_name = s.sample_id
|
||||
stem = output_stem(s.sample_id)
|
||||
taken[stem] = s.sample_id
|
||||
stems[s.sample_id] = stem
|
||||
return SampleSet(folder=folder, sheet=sheet, samples=samples, stems=stems)
|
||||
|
||||
|
||||
def sample_dir(out: Path, stem: str, layout: str) -> Path:
|
||||
return out / "samples" / stem if layout == "nested" else out
|
||||
|
||||
|
||||
def sheet_values(s: ResolvedSample, sheet: SampleSheet | None) -> dict[str, Any]:
|
||||
"""Pass-through sheet columns for summaries (IN-013)."""
|
||||
if sheet is None or s.sheet_row is None:
|
||||
return {}
|
||||
row = s.sheet_row
|
||||
return {c: row.values.get(c, "") for c in sheet.passthrough_columns()} # type: ignore[attr-defined]
|
||||
Reference in new issue
Block a user