kemalandClaude Opus 5.5
febdd121ef
- 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
88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
"""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]
|