v2 CN: cn annotate (JSON report, ISCN 2020, cytobands, genes) and the annotation bundle builder

cn call now records per-chromosome and sample statistics (percent het, LRR stats, BAF
deviations, robust LogR dev, call rate) in the sidecar; cn annotate writes
<sample>.cyto.json (CYTO-033) with ISCN microarray/traditional strings (CYTO-034), bands
and RefSeq gene lists (CYTO-035), LOH summaries, weighted median CN (CYTO-026); annotation
bundle from UCSC cytoBand + ncbiRefSeq (CYTO-036, IN-054). Tests for ISCN strings, gene
overlap, sidecar-missing error and an end-to-end call+annotate.
This commit is contained in:
kemal committed 2026-09-29 01:37:56 +01:00
1 parent 2f01d43269
commit 6d1877da6a
6 files changed
+428 -2

No files matched your search

+130
View File
@@ -0,0 +1,130 @@
"""CN event annotation: cytobands, genes, ISCN 2020 strings — spec CYTO-033 … CYTO-035. Pure functions, no I/O.
[DOC: ISCN 2020 (An International System for Human Cytogenomic Nomenclature), microarray nomenclature section;
UCSC cytoBand tables] Microarray form ``arr[<build>] <chr><band_start>[<band_end>](<S>_<E>)x<CN>`` per event, LOH as
``x2 hmz`` (gain-LOH ``x3 hmz``), mosaic as ``x<base>~<CN>``; normal results ``arr[<build>](1-22,X)x2`` (female),
``(1-22)x2,(XY)x1`` (male), ``(1-22)x2`` (sex unresolved). Traditional form ``del(<chr>)(<bands>)`` /
``dup(<chr>)(<bands>)`` for CNVs only.
"""
from __future__ import annotations
import bisect
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
CHROM_ORDER = {str(i): i for i in range(1, 23)} | {"X": 23, "Y": 24}
@dataclass
class Bands:
"""1-based inclusive band intervals per chromosome, sorted."""
starts: dict[str, list[int]]
ends: dict[str, list[int]]
names: dict[str, list[str]]
@classmethod
def from_ucsc(cls, text: str) -> Bands:
rows: dict[str, list[tuple[int, int, str]]] = {}
for line in text.splitlines():
f = line.split("\t")
if len(f) < 4:
continue
c = f[0][3:] if f[0].lower().startswith("chr") else f[0]
if "_" in c:
continue
rows.setdefault(c, []).append((int(f[1]) + 1, int(f[2]), f[3]))
s, e, n = {}, {}, {}
for c, v in rows.items():
v.sort()
s[c], e[c], n[c] = [x[0] for x in v], [x[1] for x in v], [x[2] for x in v]
return cls(s, e, n)
def band(self, chrom: str, pos: int) -> str:
st = self.starts.get(chrom)
if not st:
return ""
i = max(0, bisect.bisect_right(st, pos) - 1)
return self.names[chrom][i]
@dataclass
class Genes:
"""Transcripts: per chromosome sorted by start, with the gene label (HGNC symbol or Ensembl/RefSeq ID)."""
starts: dict[str, list[int]]
ends: dict[str, list[int]]
labels: dict[str, list[str]]
max_len: dict[str, int]
@classmethod
def from_rows(cls, rows: Sequence[tuple[str, int, int, str]]) -> Genes:
by: dict[str, list[tuple[int, int, str]]] = {}
for c, s, e, lab in rows:
by.setdefault(c, []).append((s, e, lab))
st, en, lb, ml = {}, {}, {}, {}
for c, v in by.items():
v.sort()
st[c], en[c], lb[c] = [x[0] for x in v], [x[1] for x in v], [x[2] for x in v]
ml[c] = max(x[1] - x[0] for x in v)
return cls(st, en, lb, ml)
def overlapping(self, chrom: str, start: int, end: int) -> list[str]:
"""CYTO-035: every transcript overlapping by ≥ 1 bp; labels deduplicated in transcript-start order."""
st = self.starts.get(chrom)
if not st:
return []
lo = bisect.bisect_left(st, start - self.max_len[chrom])
hi = bisect.bisect_right(st, end)
out: list[str] = []
seen: set[str] = set()
for i in range(lo, hi):
if self.ends[chrom][i] >= start and self.labels[chrom][i] not in seen:
seen.add(self.labels[chrom][i])
out.append(self.labels[chrom][i])
return out
def _band_span(bands: Bands, ev: dict[str, Any]) -> tuple[str, str]:
return bands.band(ev["chrom"], ev["start"]), bands.band(ev["chrom"], ev["end"])
def _chrom_label(c: str) -> str:
return c
def iscn_microarray_event(ev: dict[str, Any], bands: Bands) -> str:
b1, b2 = _band_span(bands, ev)
span = b1 if b1 == b2 else f"{b1}{b2}"
cn = int(ev["copy_number"])
if ev.get("mosaic"):
base = int(ev.get("base_cn", 2))
copies = f"x{base}~{cn if ev['call'] != 'MLOH' else base}"
if ev["call"] == "MLOH":
copies += " hmz"
elif ev["call_type"] == "LOH":
copies = f"x{cn} hmz"
else:
copies = f"x{cn}"
return f"{_chrom_label(ev['chrom'])}{span}({ev['start']}_{ev['end']}){copies}"
def iscn_traditional_event(ev: dict[str, Any], bands: Bands) -> str | None:
if ev["call_type"] not in ("DEL", "DUP"):
return None
b1, b2 = _band_span(bands, ev)
kind = "del" if ev["call_type"] == "DEL" else "dup"
return f"{kind}({ev['chrom']})({b1}{b2})"
def iscn_sample(events: list[dict[str, Any]], bands: Bands, build: str, sex: str) -> tuple[str, str]:
"""(traditional, microarray) strings for the reported events (already filtered)."""
evs = sorted(events, key=lambda e: (CHROM_ORDER.get(e["chrom"], 99), e["start"]))
if not evs:
normal = {"F": "(1-22,X)x2", "M": "(1-22)x2,(XY)x1"}.get(sex, "(1-22)x2")
return "", f"arr[{build}]{normal}"
micro = f"arr[{build}] " + ",".join(iscn_microarray_event(e, bands) for e in evs)
trad = ",".join(t for t in (iscn_traditional_event(e, bands) for e in evs) if t)
return trad, micro