"""``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(" 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'{p.name}' for p in written if p.suffix == ".html")
atomic_write_text(
opts.out / "index.html",
f"{PRODUCT_NAME} QC reports\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