kemal
3eb51639cc
Methylation CSV manifest reader (MET-002/003), raw M/U assembly with bead masking (MET-005), pOOBAH detection p-values, noob background correction, linear/nonlinear dye bias, beta and M-values (MET-010..012), the 21 methylation control metrics (MET-006, bs-conv red-bg set option), per-sample ctrl/cgs tables, summary tsv/xlsx/html with the detection histogram and a beta PCA, pcs.tsv.gz, controls_raw/controls_qc.tsv; methylation datasets load in qc report (MET-014); qc controls --array-type methylation; minfi-style Sentrix_ID/Sentrix_Position sheet keys. Vendor EPIC manifests carry a 5th field on some control lines: >= 4 fields accepted. Runs on the public minfiDataEPIC set (3 x NA12878).
633 lines
27 KiB
Python
633 lines
27 KiB
Python
"""``simurg qc report`` — QC table (xlsx/csv) and a self-contained HTML report (spec QC-040 … QC-063, CLI-026).
|
|
|
|
The report reads only dataset folders (control tables, optionally the sample summary), never raw data. Data is
|
|
embedded as gzip-compressed JSON (decoded in the page with the browser's DecompressionStream) next to the bundled
|
|
Plotly.js (MIT), so the file works offline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import csv
|
|
import datetime as _dt
|
|
import gzip
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import math
|
|
import re
|
|
from collections import Counter
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass, field
|
|
from importlib import resources
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from simurg import PRODUCT_NAME, __version__
|
|
from simurg.core.atomic import atomic_write_bytes, atomic_write_text
|
|
from simurg.core.errors import ExitCode, SimurgError
|
|
from simurg.core.log import Reporter
|
|
from simurg.qc.controls import COMPAT_METRIC_NAMES, METRICS, beads_from_definitions, excluded_from_report, metrics_for
|
|
from simurg.qc.status import control_status, functional_status
|
|
from simurg.qc.thresholds import ABOVE_KEYS, FUNCTIONAL_KEYS, KEY_TO_COLUMN, KNOWN_KEYS, Thresholds, resolve_thresholds
|
|
|
|
REQUIRED = (("controls_raw.tsv", "controls.raw_metrics.csv"), ("controls_qc.tsv", "controls.qc_metrics.csv"))
|
|
SUMMARY_NAMES = ("sample_summary.tsv", "sample_summary.csv", "gt_sample_summary.csv")
|
|
QC_META = ("imaging_date", "sample_plate", "sample_well", "scanner_id", "scanner_version")
|
|
SUMMARY_CORE = {
|
|
"sample_id", "sample_name", "sample_folder", "status", "fail_reason", "autosomal_call_rate", "call_rate",
|
|
"logr_dev", "sex_est", "sex_reported", "sex_check", "gc10", "gc50", "n_calls", "n_nocalls", "n_intensity_only",
|
|
"p05_x", "p50_x", "p95_x", "p05_y", "p50_y", "p95_y", "het_rate", "contamination", "functional_status",
|
|
"control_status", "control_flag_codes", "flags",
|
|
} # fmt: skip
|
|
COMPAT_QC = {
|
|
"SampleId": "sample_id", "ImagingDate": "imaging_date", "SamplePlate": "sample_plate",
|
|
"SampleWell": "sample_well", "ScannerId": "scanner_id", "ScannerVersion": "scanner_version",
|
|
} # fmt: skip
|
|
COMPAT_RAW = {
|
|
"SampleId": "sample_id", "Category": "category", "Beadtype": "bead_address", "Control": "control_name",
|
|
"Color": "color_label", "GreenIntensity": "green_intensity", "RedIntensity": "red_intensity",
|
|
} # fmt: skip
|
|
COMPAT_SUMMARY = {
|
|
"Sample ID": "sample_id", "Sample Name": "sample_name", "Sample Folder": "sample_folder",
|
|
"Autosomal Call Rate": "autosomal_call_rate", "Call Rate": "call_rate", "Log R Ratio Std Dev": "logr_dev",
|
|
"Sex Estimate": "sex_est",
|
|
} # fmt: skip
|
|
FORMULAS = {
|
|
"restoration": "Restoration Green / (Bkg Green + offset)",
|
|
"staining_green": "Biotin (High) Green / Biotin (Bkg) Green",
|
|
"staining_red": "DNP (High) Red / DNP (Bkg) Red",
|
|
"extension_green": "lowest Extension (C, G) Green / highest Extension (A, T) Green",
|
|
"extension_red": "lowest Extension (A, T) Red / highest Extension (C, G) Red",
|
|
"hyb_high_med": "Hyb (High) Green / Hyb (Medium) Green",
|
|
"hyb_med_low": "Hyb (Medium) Green / Hyb (Low) Green",
|
|
"target_removal_1": "(Bkg Green + offset) / Target Removal Green",
|
|
"target_removal_2": "(Bkg Green + offset) / Target Removal 2 Green",
|
|
"stringency": "String (PM) Red / String (MM) Red",
|
|
"nsb_green": "(Bkg Green + offset) / highest NSB Green",
|
|
"nsb_red": "(Bkg Red + offset) / highest NSB Red",
|
|
"np_green": "lowest NP (C, G) Green / highest NP (A, T) Green",
|
|
"np_red": "lowest NP (A, T) Red / highest NP (C, G) Red",
|
|
"assay_control_norm_r": "normalized R of the assay-control locus (lowest over listed probes)",
|
|
"autosomal_call_rate": "calls / loci on chromosomes 1-22",
|
|
"logr_dev": "standard deviation of autosomal Log R ratio",
|
|
}
|
|
# QC-052 chip formats: samples per chip -> (rows, columns)
|
|
CHIP_FORMATS = {4: (4, 1), 8: (8, 1), 12: (12, 1), 16: (8, 2), 24: (12, 2), 48: (12, 4), 96: (12, 8)}
|
|
POS_RE = re.compile(r"^(\d+)_R(\d{2})C(\d{2})$")
|
|
LABEL_RE = re.compile(r"[^A-Za-z0-9_.-]")
|
|
JSON_INF = 1e308 # JSON has no Infinity; the page shows values >= 1e307 as "Inf"
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------ reading
|
|
def read_table(path: Path) -> list[dict[str, str]]:
|
|
"""QC-061: UTF-8 (BOM allowed), delimiter by extension, RFC-4180 quoting; errors name file and line."""
|
|
delim = "\t" if path.suffix.lower() == ".tsv" else ","
|
|
try:
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
except UnicodeDecodeError as e:
|
|
raise SimurgError("E_QC_DATASET_INCOMPLETE", f"{path}: not UTF-8 ({e})", path=str(path)) from None
|
|
reader = csv.reader(io.StringIO(text), delimiter=delim)
|
|
try:
|
|
header = next(reader)
|
|
except StopIteration:
|
|
return []
|
|
rows = []
|
|
for i, rec in enumerate(reader, start=2):
|
|
if not rec:
|
|
continue
|
|
if len(rec) != len(header):
|
|
raise SimurgError(
|
|
"E_QC_DATASET_INCOMPLETE",
|
|
f"{path}: line {i} has {len(rec)} columns, header has {len(header)}",
|
|
path=str(path),
|
|
)
|
|
rows.append(dict(zip(header, rec, strict=True)))
|
|
return rows
|
|
|
|
|
|
def _rename(rows: list[dict[str, str]], mapping: Mapping[str, str]) -> list[dict[str, str]]:
|
|
return [{mapping.get(k, k): v for k, v in r.items()} for r in rows]
|
|
|
|
|
|
def _find(folder: Path, names: Sequence[str]) -> Path | None:
|
|
for n in names:
|
|
if (folder / n).exists():
|
|
return folder / n
|
|
return None
|
|
|
|
|
|
def _is_dataset(folder: Path) -> bool:
|
|
return all(_find(folder, pair) for pair in REQUIRED)
|
|
|
|
|
|
def resolve_datasets(data: Sequence[str]) -> list[Path]:
|
|
"""QC-041: one folder, a comma list, or one parent folder searched recursively (sorted by path)."""
|
|
items = [Path(p.strip()) for d in data for p in str(d).split(",") if p.strip()]
|
|
if not items:
|
|
raise SimurgError("E_QC_NO_DATASETS", "No valid dataset folders detected.")
|
|
for p in items:
|
|
if not p.exists():
|
|
raise SimurgError("E_QC_NO_DATASETS", f"Input data folder does not exist: {p}", path=str(p))
|
|
if len(items) == 1 and not _is_dataset(items[0]):
|
|
found = sorted({f.parent for f in items[0].rglob("controls_qc.tsv")} | {
|
|
f.parent for f in items[0].rglob("controls.qc_metrics.csv")}) # fmt: skip
|
|
found = [f for f in found if _is_dataset(f)]
|
|
if not found:
|
|
raise SimurgError(
|
|
"E_QC_NO_DATASETS",
|
|
"No valid dataset folders detected. Each folder must include at least: controls_raw.tsv, "
|
|
"controls_qc.tsv",
|
|
)
|
|
return found
|
|
for p in items:
|
|
missing = [pair[0] for pair in REQUIRED if not _find(p, pair)]
|
|
if missing:
|
|
raise SimurgError(
|
|
"E_QC_DATASET_INCOMPLETE", f"Dataset {p.name} is missing required file(s): {', '.join(missing)}"
|
|
)
|
|
return items
|
|
|
|
|
|
def assign_labels(folders: Sequence[Path], labels: Sequence[str] | None, rep: Reporter) -> list[str]:
|
|
"""QC-042."""
|
|
if labels:
|
|
labels = [x for lab in labels for x in str(lab).split(",") if x != ""]
|
|
if len(labels) != len(folders):
|
|
raise SimurgError(
|
|
"E_QC_LABEL_COUNT",
|
|
f"Number of labels ({len(labels)}) does not match number of datasets ({len(folders)}).",
|
|
)
|
|
raw = list(labels)
|
|
else:
|
|
seen: Counter[str] = Counter()
|
|
raw = []
|
|
for f in folders:
|
|
seen[f.name] += 1
|
|
raw.append(f.name if seen[f.name] == 1 else f"{f.name}_{seen[f.name]}")
|
|
out = []
|
|
for lab in raw:
|
|
clean = LABEL_RE.sub("_", lab)
|
|
if clean != lab:
|
|
rep.warn("W_QC_LABEL_SANITIZED", f"Label sanitized: {lab} → {clean}")
|
|
out.append(clean)
|
|
dups = [k for k, c in Counter(out).items() if c > 1]
|
|
if dups:
|
|
raise SimurgError(
|
|
"E_QC_LABEL_DUPLICATE", f"Dataset labels are not unique after sanitisation: {', '.join(dups)}"
|
|
)
|
|
return out
|
|
|
|
|
|
def _num(v: Any) -> float | None:
|
|
if v is None or v == "":
|
|
return None
|
|
try:
|
|
f = float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if math.isinf(f):
|
|
return JSON_INF if f > 0 else -JSON_INF
|
|
return None if math.isnan(f) else f
|
|
|
|
|
|
@dataclass
|
|
class Dataset:
|
|
label: str
|
|
path: Path
|
|
type: str
|
|
samples: list[dict[str, Any]]
|
|
probes: list[dict[str, str]] # raw probe columns: key, name, category, channel, color
|
|
metrics: list[str] # derived metric columns present
|
|
metadata: list[str]
|
|
has_summary: bool
|
|
chip_format: tuple[int, int] | None
|
|
profile: str = ""
|
|
files: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
def _chip_format(folder: Path) -> tuple[int, int] | None:
|
|
for mf in sorted(folder.glob("run_manifest*.json")):
|
|
try:
|
|
doc = json.loads(mf.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
continue
|
|
for inp in doc.get("inputs", []):
|
|
m = re.search(r"-(\d+)v\d", str(inp.get("name", "")))
|
|
if inp.get("role") in ("bpm", "csv_manifest") and m and int(m.group(1)) in CHIP_FORMATS:
|
|
return CHIP_FORMATS[int(m.group(1))]
|
|
return None
|
|
|
|
|
|
def load_dataset(folder: Path, label: str, rep: Reporter) -> Dataset:
|
|
raw_p = _find(folder, REQUIRED[0])
|
|
qc_p = _find(folder, REQUIRED[1])
|
|
assert raw_p is not None and qc_p is not None
|
|
raw = _rename(read_table(raw_p), COMPAT_RAW)
|
|
qc = _rename(read_table(qc_p), COMPAT_QC)
|
|
sum_p = _find(folder, SUMMARY_NAMES)
|
|
summary = _rename(read_table(sum_p), COMPAT_SUMMARY) if sum_p else []
|
|
for rows, name in ((raw, raw_p), (qc, qc_p)):
|
|
if rows and "sample_id" not in rows[0]:
|
|
raise SimurgError("E_QC_DATASET_INCOMPLETE", f"{name}: no sample_id column", path=str(name))
|
|
cats = {r.get("category", "").casefold() for r in raw}
|
|
array_type = (qc[0].get("array_type") if qc else "") or (
|
|
"methylation" if cats & {"bisulfite conversion i", "bisulfite conversion ii"} else "genotyping"
|
|
)
|
|
compat_metric = {v: k for k, v in COMPAT_METRIC_NAMES.items()}
|
|
qc = [{compat_metric.get(k, k): v for k, v in r.items()} for r in qc]
|
|
metrics = [m.column for m in metrics_for(array_type) if qc and m.column in qc[0]]
|
|
if qc and "assay_control_norm_r" in qc[0]:
|
|
metrics.append("assay_control_norm_r")
|
|
# raw probes, wide
|
|
defs: dict[str, dict[str, str]] = {}
|
|
name_count = (
|
|
Counter((r["control_name"], r["color_label"]) for r in raw if r["sample_id"] == raw[0]["sample_id"])
|
|
if raw
|
|
else Counter()
|
|
)
|
|
names_dup = Counter(n for n, _ in name_count)
|
|
wide: dict[str, dict[str, float | None]] = {}
|
|
for r in raw:
|
|
bead = beads_from_definitions(
|
|
[
|
|
{
|
|
"address": r.get("bead_address") or "0",
|
|
"category": r["category"],
|
|
"color": r["color_label"],
|
|
"name": r["control_name"],
|
|
}
|
|
]
|
|
)[0]
|
|
if excluded_from_report(bead):
|
|
continue
|
|
base = r["control_name"] if names_dup[r["control_name"]] <= 1 else f"{r['control_name']} {r['color_label']}"
|
|
for ch, col in (("Green", "green_intensity"), ("Red", "red_intensity")):
|
|
key = f"{base}_{ch.lower()}"
|
|
defs.setdefault(key, {"key": key, "name": base, "category": r["category"], "channel": ch,
|
|
"color": r["color_label"]}) # fmt: skip
|
|
wide.setdefault(r["sample_id"], {})[key] = _num(r[col])
|
|
by_sum = {r["sample_id"]: r for r in summary}
|
|
by_qc = {r["sample_id"]: r for r in qc}
|
|
ids = list(by_qc)
|
|
unmatched = sorted(set(by_sum) ^ set(by_qc)) if summary else []
|
|
if unmatched:
|
|
rep.warn(
|
|
"W_QC_UNMATCHED_SAMPLES",
|
|
f"{label}: {len(unmatched)} sample ID(s) in only one of the QC files, e.g. {', '.join(unmatched[:20])}",
|
|
)
|
|
ids += [s for s in by_sum if s not in by_qc]
|
|
meta_cols: list[str] = []
|
|
if summary:
|
|
cols = list(summary[0])
|
|
tail = cols[cols.index("flags") + 1 :] if "flags" in cols else [c for c in cols if c not in SUMMARY_CORE]
|
|
meta_cols = [c for c in tail if not c.endswith("_norm_r")]
|
|
meta_cols.append("sample_folder")
|
|
meta_cols += [c for c in QC_META]
|
|
probe_cols = [c for c in (summary[0] if summary else {}) if c.endswith("_norm_r")]
|
|
samples = []
|
|
for sid in ids:
|
|
s_row, q_row = by_sum.get(sid, {}), by_qc.get(sid, {})
|
|
m = POS_RE.match(sid)
|
|
rec: dict[str, Any] = {
|
|
"sample_id": sid,
|
|
"sample_name": s_row.get("sample_name", sid),
|
|
"barcode": m.group(1) if m else "",
|
|
"row": int(m.group(2)) if m else None,
|
|
"col": int(m.group(3)) if m else None,
|
|
"status": s_row.get("status", "OK" if summary else ""),
|
|
"autosomal_call_rate": _num(s_row.get("autosomal_call_rate")),
|
|
"call_rate": _num(s_row.get("call_rate")),
|
|
"logr_dev": _num(s_row.get("logr_dev")),
|
|
"sex_est": s_row.get("sex_est", ""),
|
|
"sex_check": s_row.get("sex_check", ""),
|
|
"het_rate": _num(s_row.get("het_rate")),
|
|
"contamination": _num(s_row.get("contamination")),
|
|
"flags": s_row.get("flags", ""),
|
|
"na_metrics": q_row.get("na_metrics", ""),
|
|
"has_summary": bool(s_row),
|
|
"m": {c: _num(q_row.get(c)) for c in metrics},
|
|
"raw": wide.get(sid, {}),
|
|
"meta": {c: (s_row.get(c) if c in s_row else q_row.get(c, "")) or "" for c in meta_cols},
|
|
}
|
|
if "assay_control_norm_r" not in q_row and probe_cols:
|
|
vals = [v for v in (_num(s_row.get(c)) for c in probe_cols) if v is not None]
|
|
rec["m"]["assay_control_norm_r"] = min(vals) if vals else None
|
|
if "assay_control_norm_r" not in metrics:
|
|
metrics.append("assay_control_norm_r")
|
|
samples.append(rec)
|
|
files = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in (raw_p, qc_p, sum_p) if p is not None}
|
|
return Dataset(
|
|
label=label,
|
|
path=folder,
|
|
type=array_type,
|
|
samples=samples,
|
|
probes=list(defs.values()),
|
|
metrics=metrics,
|
|
metadata=meta_cols,
|
|
has_summary=bool(summary),
|
|
chip_format=_chip_format(folder),
|
|
files=files,
|
|
)
|
|
|
|
|
|
def apply_privacy(
|
|
datasets: Sequence[Dataset], drop: Sequence[str], keep: Sequence[str] | None, no_metadata: bool, redact_paths: bool
|
|
) -> bool:
|
|
"""QC-046; returns True when any metadata remains."""
|
|
drop_f = {d.casefold() for x in drop for d in x.split(",") if d}
|
|
keep_f = {k.casefold() for x in keep for k in x.split(",") if k} if keep else None
|
|
any_left = False
|
|
for ds in datasets:
|
|
cols = []
|
|
for c in ds.metadata:
|
|
if no_metadata or c.casefold() in drop_f or (keep_f is not None and c.casefold() not in keep_f):
|
|
continue
|
|
cols.append(c)
|
|
if redact_paths:
|
|
for s in ds.samples:
|
|
s["meta"]["sample_folder"] = ""
|
|
for s in ds.samples:
|
|
s["meta"] = {c: s["meta"].get(c, "") for c in cols}
|
|
ds.metadata = sorted(cols, key=str.casefold)
|
|
any_left |= any(any(s["meta"].values()) for s in ds.samples)
|
|
return any_left
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------ status
|
|
def evaluate(ds: Dataset, th: Thresholds) -> None:
|
|
for s in ds.samples:
|
|
row = {"status": s["status"], "autosomal_call_rate": s["autosomal_call_rate"], "logr_dev": s["logr_dev"]}
|
|
row = {k: (math.nan if v is None else v) for k, v in row.items()}
|
|
s["functional_status"] = functional_status(row if s["has_summary"] else None, th)
|
|
vals = {k: (math.nan if v is None else v) for k, v in s["m"].items()}
|
|
cs, codes = control_status(vals, [c for c in s["na_metrics"].split(";") if c], th)
|
|
s["control_status"], s["control_flag_codes"] = cs, ";".join(codes)
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------ table
|
|
def table_rows(datasets: Sequence[Dataset]) -> tuple[list[str], list[dict[str, Any]]]:
|
|
"""QC-045 Samples sheet: fixed columns, functional metrics, derived metrics, raw intensities, metadata."""
|
|
metrics: list[str] = []
|
|
probes: list[str] = []
|
|
meta: list[str] = []
|
|
for ds in datasets:
|
|
metrics += [m for m in ds.metrics if m not in metrics and m != "assay_control_norm_r"]
|
|
probes += [p["key"] for p in ds.probes if p["key"] not in probes]
|
|
meta += [c for c in ds.metadata if c not in meta]
|
|
meta = sorted(meta, key=str.casefold)
|
|
cols = ["dataset", "sample_id", "sample_name", "functional_status", "control_status", "control_flag_codes",
|
|
"na_metrics", "autosomal_call_rate", "call_rate", "logr_dev", "sex_est", "assay_control_norm_r",
|
|
*metrics, *probes, *meta] # fmt: skip
|
|
rows = []
|
|
for ds in datasets:
|
|
for s in ds.samples:
|
|
r: dict[str, Any] = {k: s.get(k) for k in cols[:11]}
|
|
r["dataset"] = ds.label
|
|
r["assay_control_norm_r"] = s["m"].get("assay_control_norm_r")
|
|
r.update({m: s["m"].get(m) for m in metrics})
|
|
r.update({p: s["raw"].get(p) for p in probes})
|
|
r.update({c: s["meta"].get(c, "") for c in meta})
|
|
rows.append(r)
|
|
return cols, rows
|
|
|
|
|
|
def write_csv_table(path: Path, cols: Sequence[str], rows: Sequence[Mapping[str, Any]]) -> None:
|
|
buf = io.StringIO()
|
|
w = csv.writer(buf, lineterminator="\n")
|
|
w.writerow(cols)
|
|
for r in rows:
|
|
w.writerow(["" if r.get(c) is None else r.get(c) for c in cols])
|
|
atomic_write_text(path, buf.getvalue())
|
|
|
|
|
|
def write_xlsx_table(
|
|
path: Path,
|
|
cols: Sequence[str],
|
|
rows: Sequence[Mapping[str, Any]],
|
|
th: Thresholds,
|
|
datasets: Sequence[Dataset],
|
|
generated: str,
|
|
) -> None:
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill
|
|
|
|
fills = {
|
|
"PASS": PatternFill("solid", fgColor="BBD7F0"),
|
|
"FAIL": PatternFill("solid", fgColor="F5C28E"),
|
|
"FLAGGED": PatternFill("solid", fgColor="FFE08A"),
|
|
"CLEAR": PatternFill("solid", fgColor="BBD7F0"),
|
|
"func_out": PatternFill("solid", fgColor="F5C28E"),
|
|
"ctrl_out": PatternFill("solid", fgColor="FFF1C2"),
|
|
"code": PatternFill("solid", fgColor="FFE08A"),
|
|
}
|
|
col_key = {v: k for k, v in KEY_TO_COLUMN.items()} | {"autosomal_call_rate": "callRate", "logr_dev": "logRDev"}
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Samples"
|
|
ws.append(list(cols))
|
|
for c in ws[1]:
|
|
c.font = Font(bold=True)
|
|
for r in rows:
|
|
ws.append(["" if r.get(c) is None else r.get(c) for c in cols])
|
|
cur = ws.max_row
|
|
for j, c in enumerate(cols, start=1):
|
|
v = r.get(c)
|
|
cell = ws.cell(cur, j)
|
|
if c in ("functional_status", "control_status") and v in fills:
|
|
cell.fill = fills[str(v)]
|
|
elif c == "control_flag_codes" and v:
|
|
cell.fill = fills["code"]
|
|
elif c in col_key and isinstance(v, int | float):
|
|
t = th.get(col_key[c])
|
|
if t.enabled and t.fails(float(v)):
|
|
cell.fill = fills["func_out" if col_key[c] in FUNCTIONAL_KEYS else "ctrl_out"]
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = ws.dimensions
|
|
t2 = wb.create_sheet("Thresholds")
|
|
t2.append(["key", "description", "threshold", "direction", "enabled", "source"])
|
|
for k in KNOWN_KEYS:
|
|
t = th.get(k)
|
|
desc = FORMULAS.get(
|
|
KEY_TO_COLUMN.get(k, {"callRate": "autosomal_call_rate", "logRDev": "logr_dev"}.get(k, "")), ""
|
|
)
|
|
t2.append(
|
|
[k, desc, t.value, t.direction, t.enabled, "default" if t.source in ("builtin", "unset") else t.source]
|
|
)
|
|
t3 = wb.create_sheet("Datasets")
|
|
t3.append(["label", "path", "type", "n_samples", "profile"])
|
|
for ds in datasets:
|
|
t3.append([ds.label, str(ds.path), ds.type, len(ds.samples), th.profile])
|
|
t4 = wb.create_sheet("About")
|
|
t4.append(["software", f"{PRODUCT_NAME} {__version__}"])
|
|
t4.append(["generated", generated])
|
|
for ds in datasets:
|
|
for name, h in sorted(ds.files.items()):
|
|
t4.append([f"{ds.label}/{name}", h])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
atomic_write_bytes(path, buf.getvalue())
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------ html
|
|
def _asset(name: str) -> str:
|
|
return resources.files("simurg.qc.report").joinpath(name).read_text(encoding="utf-8")
|
|
|
|
|
|
def _plotly() -> str:
|
|
"""Bundled plotly.js 2.35.2 (MIT), stored gzip-compressed in the package."""
|
|
data = resources.files("simurg.qc.report").joinpath("vendor", "plotly.min.js.gz").read_bytes()
|
|
return gzip.decompress(data).decode("utf-8")
|
|
|
|
|
|
def payload(datasets: Sequence[Dataset], th: Thresholds, generated: str, metadata_notice: bool) -> dict[str, Any]:
|
|
metric_defs = [
|
|
{
|
|
"column": m.column,
|
|
"key": m.key,
|
|
"formula": FORMULAS.get(m.column, ""),
|
|
"direction": "above" if m.key in ABOVE_KEYS else "below",
|
|
}
|
|
for m in METRICS
|
|
] + [
|
|
{
|
|
"column": "assay_control_norm_r",
|
|
"key": "assayControl",
|
|
"formula": FORMULAS["assay_control_norm_r"],
|
|
"direction": "below",
|
|
}
|
|
]
|
|
return {
|
|
"product": PRODUCT_NAME,
|
|
"version": __version__,
|
|
"generated": generated,
|
|
"metadataNotice": metadata_notice,
|
|
"profile": th.profile,
|
|
"thresholds": {
|
|
k: {"value": th.get(k).value, "direction": th.get(k).direction, "source": th.get(k).source}
|
|
for k in KNOWN_KEYS
|
|
},
|
|
"functionalKeys": [
|
|
{
|
|
"key": "callRate",
|
|
"column": "autosomal_call_rate",
|
|
"label": "Autosomal Call Rate",
|
|
"formula": FORMULAS["autosomal_call_rate"],
|
|
},
|
|
{"key": "logRDev", "column": "logr_dev", "label": "LRR SD", "formula": FORMULAS["logr_dev"]},
|
|
],
|
|
"metricDefs": metric_defs,
|
|
"datasets": [
|
|
{
|
|
"label": ds.label,
|
|
"type": ds.type,
|
|
"hasSummary": ds.has_summary,
|
|
"chipFormat": list(ds.chip_format) if ds.chip_format else None,
|
|
"metrics": ds.metrics,
|
|
"probes": ds.probes,
|
|
"metadata": ds.metadata,
|
|
"samples": ds.samples,
|
|
}
|
|
for ds in datasets
|
|
],
|
|
}
|
|
|
|
|
|
def render_html(data: Mapping[str, Any]) -> str:
|
|
blob = json.dumps(data, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
|
|
packed = base64.b64encode(gzip.compress(blob.encode("utf-8"), compresslevel=9, mtime=0)).decode("ascii")
|
|
page = _asset("template.html").replace("__TITLE__", f"{PRODUCT_NAME} QC Report").replace("__DATA__", packed)
|
|
for marker, text in (
|
|
("/*__STYLE__*/", _asset("report.css")),
|
|
("/*__APP__*/", _asset("report.js")),
|
|
("/*__PLOTLY__*/", _plotly().replace("</script", "<\\/script")),
|
|
):
|
|
head, _, tail = page.partition(marker) # insert verbatim once; never rescan inserted code
|
|
page = head + text + tail
|
|
return page
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------ command
|
|
@dataclass
|
|
class QcReportOptions:
|
|
data: list[str]
|
|
out: Path
|
|
labels: list[str] | None = None
|
|
output_format: str = "xlsx"
|
|
report_name: str | None = None
|
|
drop_metadata: list[str] = field(default_factory=list)
|
|
keep_metadata: list[str] | None = None
|
|
no_metadata: bool = False
|
|
redact_paths: bool = False
|
|
split_by: str = "none"
|
|
thresholds: Thresholds | None = None
|
|
now: _dt.datetime | None = None
|
|
|
|
|
|
def _sanitize_json(o: Any) -> Any:
|
|
if isinstance(o, float):
|
|
if math.isinf(o):
|
|
return JSON_INF if o > 0 else -JSON_INF
|
|
return None if math.isnan(o) else o
|
|
if isinstance(o, dict):
|
|
return {k: _sanitize_json(v) for k, v in o.items()}
|
|
if isinstance(o, list):
|
|
return [_sanitize_json(v) for v in o]
|
|
return o
|
|
|
|
|
|
def run_qc_report(opts: QcReportOptions, rep: Reporter) -> int:
|
|
if opts.output_format not in ("xlsx", "csv", "both"):
|
|
raise SimurgError("E_QC_FORMAT", "Invalid --output-format value. Only csv, xlsx or both are accepted.")
|
|
if opts.split_by not in ("none", "dataset"):
|
|
raise SimurgError("E_QC_FORMAT", "--split-by supports none and dataset in this version")
|
|
th = opts.thresholds or resolve_thresholds()
|
|
folders = resolve_datasets(opts.data)
|
|
labels = assign_labels(folders, opts.labels, rep)
|
|
datasets = [load_dataset(f, lab, rep) for f, lab in zip(folders, labels, strict=True)]
|
|
types = {ds.type for ds in datasets}
|
|
if len(types) > 1:
|
|
by_type = {t: [ds.label for ds in datasets if ds.type == t] for t in sorted(types)}
|
|
raise SimurgError(
|
|
"E_QC_MIXED_TYPES",
|
|
"Genotyping and methylation datasets cannot be mixed: "
|
|
+ "; ".join(f"{t}: {', '.join(v)}" for t, v in by_type.items()),
|
|
)
|
|
total = sum(len(ds.samples) for ds in datasets)
|
|
if total > 20000:
|
|
rep.warn("W_QC_TOO_FEW", f"{total} samples in total: the report may be slow to open (QC-057)")
|
|
notice = apply_privacy(datasets, opts.drop_metadata, opts.keep_metadata, opts.no_metadata, opts.redact_paths)
|
|
for ds in datasets:
|
|
evaluate(ds, th)
|
|
now = opts.now or _dt.datetime.now(_dt.UTC)
|
|
generated = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
day = now.strftime("%Y_%m_%d")
|
|
opts.out.mkdir(parents=True, exist_ok=True)
|
|
groups = [(None, datasets)] if opts.split_by == "none" else [(ds.label, [ds]) for ds in datasets]
|
|
written = []
|
|
for suffix, group in groups:
|
|
stem_r = opts.report_name or f"qc_report_{day}"
|
|
stem_t = opts.report_name or f"qc_table_{day}"
|
|
if opts.report_name:
|
|
stem_r, stem_t = f"{opts.report_name}", f"{opts.report_name}_table"
|
|
if suffix:
|
|
stem_r, stem_t = f"{stem_r}_{suffix}", f"{stem_t}_{suffix}"
|
|
cols, rows = table_rows(group)
|
|
if opts.output_format in ("xlsx", "both"):
|
|
write_xlsx_table(opts.out / f"{stem_t}.xlsx", cols, rows, th, group, generated)
|
|
written.append(opts.out / f"{stem_t}.xlsx")
|
|
if opts.output_format in ("csv", "both"):
|
|
write_csv_table(opts.out / f"{stem_t}.csv", cols, rows)
|
|
written.append(opts.out / f"{stem_t}.csv")
|
|
html = render_html(_sanitize_json(payload(group, th, generated, notice)))
|
|
atomic_write_text(opts.out / f"{stem_r}.html", html)
|
|
written.append(opts.out / f"{stem_r}.html")
|
|
if opts.split_by != "none":
|
|
links = "".join(f'<li><a href="{p.name}">{p.name}</a></li>' for p in written if p.suffix == ".html")
|
|
atomic_write_text(
|
|
opts.out / "index.html",
|
|
f"<!doctype html><meta charset=utf-8><title>{PRODUCT_NAME} QC reports</title><ul>{links}</ul>\n",
|
|
)
|
|
rep.info(f"QC report for {total} sample(s) in {len(datasets)} dataset(s): " + ", ".join(p.name for p in written))
|
|
return ExitCode.OK
|