P5b: self-contained HTML QC report and QC table (simurg qc report)
- pipeline/qc_report: dataset discovery/labels/types (QC-041..043), robust TSV/CSV reading (QC-061), join + unmatched warning (QC-040), privacy options (QC-046), xlsx (4 sheets, conditional fills) / csv table (QC-045), byte-deterministic HTML with gzip+base64 embedded data (QC-044, QC-047) - qc/report: template, CSS and JS app — Control Dashboard, Automated QC (table with sort/filter/search/ paging/column menu/export, metric scatters with threshold lines), Sample Heatmaps (chip/plate, continuous/binary), Trend Analysis (grouped summary, Tukey box plots, counts, sex, PCA), thresholds modal (apply / download / upload YAML), linked selection across plots and table (QC-048..060) - bundled plotly.js 2.35.2 (MIT); openpyxl dependency (MIT) - tests: discovery/labels/errors/determinism/xlsx + headless-browser self-test - known_unknowns K-020: remaining report deviations
This commit is contained in:
kemal committed
1 parent
cea710f7b7
commit
c9a6173fbd
13 files changed
+1601
No files matched your search
@@ -0,0 +1,623 @@
|
||||
"""``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_.-]")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------------ 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
|
||||
return None if math.isnan(f) or math.isinf(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 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__*/", _asset("vendor/plotly.min.js").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):
|
||||
return None if math.isnan(o) or math.isinf(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()),
|
||||
)
|
||||
if "methylation" in types:
|
||||
raise SimurgError("E_QC_METHYLATION", "methylation datasets in the report arrive with v2 (QC-062)")
|
||||
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
|
||||
Reference in new issue
Block a user