diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9a50b6a..6161182 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -20,7 +20,7 @@ repos: hooks: - id: pytest name: pytest - entry: .venv/bin/python -m pytest -m "not integration" --no-header -q + entry: .venv/bin/python -m pytest -m "not integration" --no-header -q --no-cov language: system pass_filenames: false always_run: true diff --git a/CHANGELOG.md b/CHANGELOG.md index 99e1173..1e6c5da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.1.0] - 13-03-2026 + +### Changed + +- **Vectorize off-target amplicon finding** (`blast/offtarget_finder.py`): Replace O(F × R) nested `iterrows()` + `query()` loop with sorted arrays and `np.searchsorted` for O(F log R) per-chromosome lookups. Eliminates heavy per-row Pandas overhead for large multiplex panels. +- **Vectorize BLAST annotation** (`blast/annotator.py`): Replace 4× `apply(axis=1)` Python lambdas with direct vectorized column operations for `from_3prime`, `length_pass_3prime`, `evalue_pass_3prime`, and `predicted_bound`. + +### Added + +- **HTML Visual QC Report (REPT-02)**: New self-contained HTML report (`panel_report.html`) generated alongside `panel_qc.json` after each pipeline run. Includes interactive Plotly.js charts for Tm distribution, amplicon size, GC content, and a cross-reactivity heatmap, plus tables for sequence flags, off-targets, failed junctions, and solution comparison. Reports inline Plotly.js (~1MB) for fully offline viewing. Added `plexus report` CLI command for standalone re-generation from existing pipeline output. +- **Jinja2 dependency** added for HTML template rendering. + ## [1.0.2] - 04-03-2026 ### Changed @@ -12,6 +24,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Rich progress bars replace log output during pipeline runs**: The CLI now shows clean Rich progress bars on stderr (step-level + per-junction detail for primer design and SNP check) instead of a wall of log messages. All detailed logs still go to the file. Warnings and errors are printed above the progress bar. Multi-panel parallel mode shows a panel-level progress bar. Progress bars are only active when stderr is a TTY; non-interactive runs behave as before. - **SNP and off-target filters now retain all tied least-affected pairs** (`snpcheck/checker.py`, `blast/specificity.py`): When all primer pairs for a junction overlap SNPs or have off-target products, the filters now keep every pair tied at the minimum count instead of arbitrarily picking one. This lets the downstream selector evaluate tied candidates on other properties (Tm, GC%, pair penalty, etc.). - **Normalise cross-dimer penalty in multiplex cost function** (`selector/cost.py`): The cross-dimer penalty was a raw sum over all C(2n, 2) pairwise primer interactions, scaling quadratically with multiplex size. This caused it to dominate the cost function at higher plexities, effectively drowning out off-target and SNP penalties during selection. The penalty is now divided by the number of interactions, making it a per-interaction average. Weights are now directly comparable regardless of multiplex size. +- **Increase default cross-dimer weight** (`data/designer_default_config.json`): `wt_cross_dimer` 1.0 → 20.0 to better penalise adapter-driven cross-dimer formation in multiplex selection, particularly for panels with custom adapter tails. +- **Off-target CSV now includes alignment details** (`blast/offtarget_finder.py`, `designer/multiplexpanel.py`): `off_targets.csv` now reports per-primer alignment quality for each off-target product: percent identity, number of mismatches, alignment length, and E-value for both forward and reverse primer hits. Helps assess the likelihood of off-target amplification. - **Separate warnings from errors in pipeline output** (`pipeline.py`, `cli.py`): Off-target and SNP fallback messages (where all pairs had issues but the least-affected were kept) were incorrectly reported as errors, causing the CLI to display "Some panels had errors" for panels that completed successfully. These are now reported as warnings. Errors are reserved for actual failures (e.g. design exceptions, BLAST unavailable). The CLI now shows a distinct warnings section below the success summary. - **Update fallback message wording**: "least-affected pair kept" now reads "all least-affected pairs kept" to reflect the v1.0.2 change that retains all tied pairs. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 148ac86..269455f 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -589,6 +589,55 @@ the BLAST database may be incomplete. --- +### SPEC-01 · Assign cross-pair off-target amplicons to primer pairs + +**Severity: Important · Files: `src/plexus/blast/specificity.py`, `src/plexus/selector/cost.py`** + +`AmpliconFinder.find_amplicons()` correctly discovers amplicons formed by primers from different +pairs (e.g., forward from pair A + reverse from pair B on the same chromosome). However, the +mapping step in `run_specificity_check()` (step 6, line 112+) only looks up amplicons keyed by +a single pair's own `(f_id, r_id)`. Cross-pair amplicons exist in `amplicon_map` but are never +retrieved — they silently fall through because no single `PrimerPair` owns both primer IDs. + +**Impact:** For multiplex panels, cross-pair off-target products are the most likely source of +spurious amplification (many more primer combinations than same-pair). These products are +computed by BLAST and identified by `AmpliconFinder`, but never assigned to any pair. They do +not appear in `off_targets.csv`, do not affect `filter_offtarget_pairs()`, and do not contribute +to the `wt_off_target` cost term. The specificity check is therefore blind to the dominant class +of off-target risk in multiplex reactions. + +**Required changes:** + +1. After the per-pair mapping loop in `run_specificity_check()`, perform a second pass that + identifies cross-pair amplicons — entries in `amplicon_map` where `F_primer` belongs to one + pair and `R_primer` belongs to a different pair (or different junction). +2. Assign each cross-pair amplicon to **both** contributing pairs (the pair that owns the forward + primer and the pair that owns the reverse primer), so that both are penalised. Add a + `cross_pair: bool` flag to the product dict to distinguish these from same-pair off-targets. +3. Ensure cross-pair off-targets are surfaced in `off_targets.csv` with both contributing pair + IDs visible. +4. Verify that `filter_offtarget_pairs()` and `wt_off_target` in the cost function correctly + account for the increased off-target counts. + +**Design consideration:** A cross-pair amplicon penalises two pairs simultaneously. If pair A's +forward primer participates in cross-products with 5 different reverse primers from other pairs, +replacing pair A eliminates all 5 — but the current per-pair count treats them independently. +Consider whether cross-pair off-targets should carry a reduced weight or whether the existing +`wt_off_target` is sufficient given that the selector can swap individual pairs. + +**Relationship to other items:** +- ISPCR-04 (cross-target interaction matrix) is a reporting view of the same underlying data; + SPEC-01 makes the data actionable in the selector. +- ISPCR-02 (ΔG-weighted off-target cost) would naturally extend to cross-pair products once + they are tracked. + +**Tests to add:** `tests/test_specificity.py` — construct a mock `bound_df` with primers from +two different pairs that form a cross-pair amplicon; assert that both pairs receive the +off-target product. Assert that the off-target CSV includes the cross-pair product with both +pair IDs. + +--- + ### ISPCR-04 · Improve AmpliconFinder: cross-target interaction matrix **File: `src/plexus/blast/offtarget_finder.py`** @@ -660,13 +709,18 @@ consistently to junction coordinates, VCF queries, and BLAST sequence IDs. --- -### REPT-02 · Visual QC Report (HTML) +### ~~REPT-02 · Visual QC Report (HTML)~~ ✅ Implemented in v1.1.0 **Severity: Low** -Transform the `panel_qc.json` (REPT-01) into a visual, standalone HTML report. This should -include Plotly or Seaborn charts for distributions and a searchable heatmap for the -cross-reactivity matrix, facilitating rapid review by clinical lab staff. +Transform the `panel_qc.json` (REPT-01) into a visual, standalone HTML report with Plotly.js +charts for Tm distribution, amplicon size, GC content, and a cross-reactivity heatmap. + +**Implementation:** Added `src/plexus/reporting/html_report.py` with `generate_html_report()` +and a Jinja2 template (`panel_report.html.j2`) that inlines Plotly.js basic for fully +offline-capable reports. Integrated into the pipeline after `panel_qc.json` write. Added +`plexus report` CLI command for standalone re-generation. 11 tests in +`tests/test_reporting_html.py`. --- @@ -724,11 +778,12 @@ project. | ISPCR-01 | ntthal ΔG scoring for BLAST binding sites | v1.1 | Important | | | ISPCR-02 | ΔG-weighted off-target cost in selector | v1.1 | Important | | | ISPCR-03 | Template mispriming check | v1.1 | Low | | +| SPEC-01 | Assign cross-pair off-target amplicons to primer pairs | v1.1 | Important | | | ISPCR-04 | Cross-target interaction matrix output | v1.1 | Low | | | PERF-01 | Parallel thermodynamics by default | v1.1 | Low | | | PERF-02 | Pre-filter candidates per junction before optimisation | v1.1 | Low | | | EXT-01 | Additional genome presets | v1.1 | Low | | | EXT-02 | Chromosome naming normalisation | v1.1 | Low | | -| REPT-02 | Visual QC Report (HTML) | v1.1 | Low | | +| ~~REPT-02~~ | ~~Visual QC Report (HTML)~~ | ~~v1.1~~ | ~~Low~~ | ✅ v1.1.0 | | SPLIT-01 | Automated Panel Splitting | Future | Future | | | ~~TEST-01~~ | ~~End-to-end integration test with real BLAST~~ | ~~v1.0~~ | ~~Important~~ | ✅ v0.4.0 | diff --git a/pyproject.toml b/pyproject.toml index eb5abeb..5b576ab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", ] dependencies = [ + "Jinja2", "loguru", "pandas", "primer3-py>=2.3.0", diff --git a/src/plexus/blast/annotator.py b/src/plexus/blast/annotator.py index 562872f..2bfff4b 100644 --- a/src/plexus/blast/annotator.py +++ b/src/plexus/blast/annotator.py @@ -38,33 +38,38 @@ def build_annotation_dict( at or near the 3' end will have ``qend < qlen``. A tolerance of 2-3 catches these cases without accepting 5'-only hits. """ + self.annotation_params = { + "length_threshold": length_threshold, + "evalue_threshold": evalue_threshold, + "max_mismatches": max_mismatches, + "three_prime_tolerance": three_prime_tolerance, + } + # Keep annotation names for summarise_by_primer self.annotations = { - "from_3prime": lambda row: (row["qlen"] - row["qend"]) - <= three_prime_tolerance, - "length_pass_3prime": lambda row: ( - row["length"] >= length_threshold - and row["mismatch"] <= max_mismatches - and row["from_3prime"] - ), - "evalue_pass_3prime": lambda row: ( - row["evalue"] < evalue_threshold and row["from_3prime"] - ), - "predicted_bound": lambda row: ( - row["length_pass_3prime"] or row["evalue_pass_3prime"] - ), + "from_3prime": None, + "length_pass_3prime": None, + "evalue_pass_3prime": None, + "predicted_bound": None, } def add_annotations(self): """ - Add annotations to `blast_df` + Add annotations to `blast_df` using vectorized column operations. """ - for annot_name, annot_func in self.annotations.items(): - self.blast_df.insert( - self.blast_df.shape[1], - annot_name, - self.blast_df.apply(func=annot_func, axis=1), - ) + p = self.annotation_params + df = self.blast_df + + df["from_3prime"] = (df["qlen"] - df["qend"]) <= p["three_prime_tolerance"] + df["length_pass_3prime"] = ( + (df["length"] >= p["length_threshold"]) + & (df["mismatch"] <= p["max_mismatches"]) + & df["from_3prime"] + ) + df["evalue_pass_3prime"] = (df["evalue"] < p["evalue_threshold"]) & df[ + "from_3prime" + ] + df["predicted_bound"] = df["length_pass_3prime"] | df["evalue_pass_3prime"] def get_predicted_bound(self): """ diff --git a/src/plexus/blast/offtarget_finder.py b/src/plexus/blast/offtarget_finder.py index 45892e2..7c08827 100644 --- a/src/plexus/blast/offtarget_finder.py +++ b/src/plexus/blast/offtarget_finder.py @@ -1,5 +1,4 @@ -from collections import namedtuple - +import numpy as np import pandas as pd @@ -30,6 +29,14 @@ class AmpliconFinder: - F_start: 5' genomic position of the forward primer - R_start: 5' genomic position of the reverse primer (on minus strand) - product_bp: Predicted size of the amplicon in base pairs (inclusive) + - F_pident: Percent identity of the forward primer alignment + - R_pident: Percent identity of the reverse primer alignment + - F_mismatch: Number of mismatches in the forward primer alignment + - R_mismatch: Number of mismatches in the reverse primer alignment + - F_align_len: Alignment length of the forward primer (bp) + - R_align_len: Alignment length of the reverse primer (bp) + - F_evalue: E-value of the forward primer alignment + - R_evalue: E-value of the reverse primer alignment """ def __init__(self, bound_df, target_map=None): @@ -48,63 +55,129 @@ def __init__(self, bound_df, target_map=None): # Generated self.amplicon_df = None + def _get_col_or_none(self, df, col): + """Return column values as array, or None if column is missing.""" + if col in df.columns: + return df[col].values + return None + def find_amplicons(self, max_size_bp=6000): """ Identify all potential amplicons within max_size_bp. + Uses sorted arrays and np.searchsorted for O(F log R) per chromosome + instead of O(F × R) nested iteration. + Args: max_size_bp (int): Maximum predicted product size to consider. """ - # Storage -- can make flexible - Amplicon = namedtuple( - "Amplicon", - [ - "chrom", - "F_target", - "R_target", - "F_primer", - "R_primer", - "F_start", - "R_start", - "product_bp", - ], - ) - - # Iterate and find amplicons - amplicons = [] + amplicon_columns = [ + "chrom", + "F_target", + "R_target", + "F_primer", + "R_primer", + "F_start", + "R_start", + "product_bp", + "F_pident", + "R_pident", + "F_mismatch", + "R_mismatch", + "F_align_len", + "R_align_len", + "F_evalue", + "R_evalue", + ] + result_chunks = [] + target_map = self.target_map + for chrom, chrom_df in self.bound_df.groupby("sseqid"): - for _, F_df in chrom_df.query("sstrand == 'plus'").iterrows(): - # Get start position of the forward primer - # 5' position - F_start = int(F_df["sstart"]) - - # Get proximal reverse primers, if they exist - # - Very important to get directionality of query correct - R_df = chrom_df.query( - "sstrand == 'minus' and 0 < (sstart - @F_start) < @max_size_bp" - ) - - # Check for any pairs - if R_df.shape[0] > 0: - F_amplicons = [ - Amplicon( - chrom=chrom, - F_target=self.target_map.get( - F_df["qseqid"], F_df["qseqid"].split("_")[0] - ), - R_target=self.target_map.get( - row["qseqid"], row["qseqid"].split("_")[0] - ), - F_primer=F_df["qseqid"], - R_primer=row["qseqid"], - F_start=F_start, - R_start=row["sstart"], - product_bp=row["sstart"] - F_start + 1, - ) - for _, row in R_df.iterrows() - ] - amplicons.extend(F_amplicons) + fwd = chrom_df[chrom_df["sstrand"] == "plus"] + rev = chrom_df[chrom_df["sstrand"] == "minus"] + + if fwd.empty or rev.empty: + continue + + # Sort reverse hits by sstart for searchsorted + rev = rev.sort_values("sstart") + r_starts = rev["sstart"].values + r_qseqids = rev["qseqid"].values + r_pident = self._get_col_or_none(rev, "pident") + r_mismatch = self._get_col_or_none(rev, "mismatch") + r_length = self._get_col_or_none(rev, "length") + r_evalue = self._get_col_or_none(rev, "evalue") + + f_starts = fwd["sstart"].values + f_qseqids = fwd["qseqid"].values + f_pident = self._get_col_or_none(fwd, "pident") + f_mismatch = self._get_col_or_none(fwd, "mismatch") + f_length = self._get_col_or_none(fwd, "length") + f_evalue = self._get_col_or_none(fwd, "evalue") + + for i in range(len(f_starts)): + f_start = f_starts[i] + # Find reverse hits where 0 < (r_start - f_start) < max_size_bp + lo = np.searchsorted( + r_starts, f_start, side="right" + ) # r_start > f_start + hi = np.searchsorted( + r_starts, f_start + max_size_bp, side="left" + ) # r_start < f_start + max_size_bp + + if lo >= hi: + continue + + n_matches = hi - lo + f_qseqid = f_qseqids[i] + f_target = target_map.get(f_qseqid, f_qseqid.split("_")[0]) + + matched_r_starts = r_starts[lo:hi] + matched_r_qseqids = r_qseqids[lo:hi] + + r_targets = [ + target_map.get(rq, rq.split("_")[0]) for rq in matched_r_qseqids + ] + + chunk = { + "chrom": np.full(n_matches, chrom), + "F_target": np.full(n_matches, f_target), + "R_target": r_targets, + "F_primer": np.full(n_matches, f_qseqid), + "R_primer": matched_r_qseqids, + "F_start": np.full(n_matches, f_start, dtype=int), + "R_start": matched_r_starts, + "product_bp": matched_r_starts - f_start + 1, + "F_pident": np.full( + n_matches, f_pident[i] if f_pident is not None else None + ), + "R_pident": r_pident[lo:hi] + if r_pident is not None + else np.full(n_matches, None), + "F_mismatch": np.full( + n_matches, f_mismatch[i] if f_mismatch is not None else None + ), + "R_mismatch": r_mismatch[lo:hi] + if r_mismatch is not None + else np.full(n_matches, None), + "F_align_len": np.full( + n_matches, f_length[i] if f_length is not None else None + ), + "R_align_len": r_length[lo:hi] + if r_length is not None + else np.full(n_matches, None), + "F_evalue": np.full( + n_matches, f_evalue[i] if f_evalue is not None else None + ), + "R_evalue": r_evalue[lo:hi] + if r_evalue is not None + else np.full(n_matches, None), + } + result_chunks.append(pd.DataFrame(chunk, columns=amplicon_columns)) # Store - self.amplicon_df = pd.DataFrame(amplicons) + if result_chunks: + self.amplicon_df = pd.concat(result_chunks, ignore_index=True) + else: + self.amplicon_df = pd.DataFrame(columns=amplicon_columns) diff --git a/src/plexus/cli.py b/src/plexus/cli.py index 339b477..0e51db7 100644 --- a/src/plexus/cli.py +++ b/src/plexus/cli.py @@ -673,6 +673,48 @@ def template( ) +@app.command() +def report( + output_dir: Annotated[ + Path, + typer.Argument(help="Pipeline output directory containing panel_qc.json."), + ], + output_file: Annotated[ + Path | None, + typer.Option( + "--output", + "-o", + help="Output HTML path (default: /panel_report.html).", + ), + ] = None, +) -> None: + """Generate an HTML QC report from existing pipeline output.""" + from plexus.reporting.html_report import generate_html_report + + if not output_dir.is_dir(): + console.print(f"[bold red]Error: {output_dir} is not a directory[/bold red]") + raise typer.Exit(code=1) + + qc_path = output_dir / "panel_qc.json" + if not qc_path.is_file(): + console.print( + f"[bold red]Error: panel_qc.json not found in {output_dir}[/bold red]" + ) + raise typer.Exit(code=1) + + try: + html_path = generate_html_report(output_dir) + if output_file is not None: + import shutil + + shutil.move(str(html_path), str(output_file)) + html_path = output_file + console.print(f"[bold green]Report written to:[/bold green] {html_path}") + except Exception as e: + console.print(f"[bold red]Error generating report: {e}[/bold red]") + raise typer.Exit(code=1) from e + + import plexus.cli_docker # noqa: E402, F401 — registers docker command if __name__ == "__main__": diff --git a/src/plexus/data/designer_default_config.json b/src/plexus/data/designer_default_config.json index 7ebc898..f0cd00a 100644 --- a/src/plexus/data/designer_default_config.json +++ b/src/plexus/data/designer_default_config.json @@ -20,7 +20,7 @@ "primer_min_gc": 30, "primer_max_gc": 70, "primer_gc_clamp": 1, - "primer_max_poly_x": 5, + "primer_max_poly_x": 4, "primer_max_poly_gc": 3, "primer_max_n": 0, "PRIMER_MAX_SELF_ANY_TH": 45.0, @@ -45,7 +45,7 @@ "PRIMER_PAIR_MAX_DIFF_TM": 3.0, "PRIMER_PRODUCT_OPT_SIZE": 60, "PRIMER_PRODUCT_MIN_INSERT_SIZE": 20, - "PRIMER_PRODUCT_MAX_INSERT_SIZE": 60, + "PRIMER_PRODUCT_MAX_INSERT_SIZE": 80, "PRIMER_PRODUCT_MAX_SIZE": 120, "PRIMER_PAIR_WT_PR_PENALTY": 1.0, "PRIMER_PAIR_WT_DIFF_TM": 0.0, diff --git a/src/plexus/designer/multiplexpanel.py b/src/plexus/designer/multiplexpanel.py index cda427b..e951b63 100644 --- a/src/plexus/designer/multiplexpanel.py +++ b/src/plexus/designer/multiplexpanel.py @@ -991,6 +991,14 @@ def save_off_targets_csv(self, file_path: str, selected_pairs: list) -> None: "OT_Product_Size": prod.get("product_bp", ""), "OT_F_Start": prod.get("F_start", ""), "OT_R_Start": prod.get("R_start", ""), + "OT_F_Pident": prod.get("F_pident", ""), + "OT_R_Pident": prod.get("R_pident", ""), + "OT_F_Mismatch": prod.get("F_mismatch", ""), + "OT_R_Mismatch": prod.get("R_mismatch", ""), + "OT_F_Align_Len": prod.get("F_align_len", ""), + "OT_R_Align_Len": prod.get("R_align_len", ""), + "OT_F_Evalue": prod.get("F_evalue", ""), + "OT_R_Evalue": prod.get("R_evalue", ""), } ) @@ -1007,6 +1015,14 @@ def save_off_targets_csv(self, file_path: str, selected_pairs: list) -> None: "OT_Product_Size", "OT_F_Start", "OT_R_Start", + "OT_F_Pident", + "OT_R_Pident", + "OT_F_Mismatch", + "OT_R_Mismatch", + "OT_F_Align_Len", + "OT_R_Align_Len", + "OT_F_Evalue", + "OT_R_Evalue", ] ).to_csv(file_path, index=False) return diff --git a/src/plexus/pipeline.py b/src/plexus/pipeline.py index 5430fce..609433e 100644 --- a/src/plexus/pipeline.py +++ b/src/plexus/pipeline.py @@ -829,6 +829,18 @@ def advance_step(label=None): logger.warning(f"Could not write panel QC report: {e}") result.errors.append(f"Panel QC report failed: {e}") + # HTML QC report (REPT-02) + try: + from plexus.reporting.html_report import generate_html_report + + html_path = generate_html_report( + output_dir, panel_name=panel.panel_name + ) + logger.info(f"Wrote HTML QC report to {html_path.name}") + except Exception as e: + logger.warning(f"Could not write HTML QC report: {e}") + result.errors.append(f"HTML QC report failed: {e}") + # Failed junctions report if result.failed_junctions: import pandas as pd diff --git a/src/plexus/reporting/html_report.py b/src/plexus/reporting/html_report.py new file mode 100644 index 0000000..df00cba --- /dev/null +++ b/src/plexus/reporting/html_report.py @@ -0,0 +1,201 @@ +"""HTML QC report generation (REPT-02). + +Transforms panel_qc.json and related pipeline outputs into a self-contained, +interactive HTML report with Plotly.js charts. +""" + +from __future__ import annotations + +import csv +import gzip +import json +from datetime import datetime, timezone +from pathlib import Path + +from jinja2 import Environment, FileSystemLoader + +from plexus.version import __version__ + +_TEMPLATES_DIR = Path(__file__).parent / "templates" +_PLOTLY_JS_GZ_PATH = _TEMPLATES_DIR / "plotly.min.js.gz" + + +def _read_json(path: Path) -> dict | None: + if path.is_file(): + with path.open() as f: + return json.load(f) + return None + + +def _read_csv(path: Path) -> list[dict] | None: + if path.is_file(): + with path.open(newline="") as f: + reader = csv.DictReader(f) + rows = [] + for row in reader: + # Convert numeric fields + for key in row: + val = row[key] + if val == "": + continue + try: + row[key] = int(val) + except (ValueError, TypeError): + try: + row[key] = float(val) + except (ValueError, TypeError): + pass + rows.append(row) + return rows if rows else None + return None + + +def _load_plotly_js() -> str: + with gzip.open(_PLOTLY_JS_GZ_PATH, "rt", encoding="utf-8") as f: + return f.read() + + +def _render_report( + qc_data: dict, + *, + panel_name: str = "Panel", + summary_data: dict | None = None, + selected_pairs: list[dict] | None = None, + off_targets: list[dict] | None = None, + failed_junctions: list[dict] | None = None, + provenance_data: dict | None = None, + top_panels: list[dict] | None = None, +) -> str: + """Render the HTML report string from data.""" + env = Environment( + loader=FileSystemLoader(str(_TEMPLATES_DIR)), + autoescape=False, + ) + template = env.get_template("panel_report.html.j2") + + num_failed = len(failed_junctions) if failed_junctions else 0 + + return template.render( + panel_name=panel_name, + report_date=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), + qc=qc_data, + qc_json=json.dumps(qc_data), + summary=summary_data, + num_failed=num_failed, + provenance=provenance_data, + selected_pairs_data=selected_pairs, + selected_pairs_json=json.dumps(selected_pairs) if selected_pairs else "[]", + off_targets_data=off_targets, + failed_junctions_data=failed_junctions, + top_panels_data=top_panels, + plexus_version=__version__, + plotly_js=_load_plotly_js(), + ) + + +def generate_html_report( + output_dir: Path, + *, + panel_name: str = "Panel", +) -> Path: + """Generate an HTML QC report from pipeline output files. + + Reads from output_dir: panel_qc.json (required), panel_summary.json, + selected_multiplex.csv, off_targets.csv, failed_junctions.csv, + provenance.json, top_panels.csv — all optional except panel_qc.json. + + Returns the path to the generated panel_report.html. + """ + output_dir = Path(output_dir) + + qc_data = _read_json(output_dir / "panel_qc.json") + if qc_data is None: + raise FileNotFoundError(f"panel_qc.json not found in {output_dir}") + + summary_data = _read_json(output_dir / "panel_summary.json") + provenance_data = _read_json(output_dir / "provenance.json") + selected_pairs = _read_csv(output_dir / "selected_multiplex.csv") + off_targets = _read_csv(output_dir / "off_targets.csv") + failed_junctions = _read_csv(output_dir / "failed_junctions.csv") + top_panels = _read_csv(output_dir / "top_panels.csv") + + # Use panel name from summary if available + if summary_data and summary_data.get("panel_name"): + panel_name = summary_data["panel_name"] + + html = _render_report( + qc_data, + panel_name=panel_name, + summary_data=summary_data, + selected_pairs=selected_pairs, + off_targets=off_targets, + failed_junctions=failed_junctions, + provenance_data=provenance_data, + top_panels=top_panels, + ) + + html_path = output_dir / "panel_report.html" + html_path.write_text(html, encoding="utf-8") + return html_path + + +def generate_html_report_from_data( + qc_data: dict, + summary_data: dict | None, + selected_pairs_csv: str | None, + off_targets_csv: str | None, + failed_junctions_csv: str | None, + output_path: Path, + *, + panel_name: str = "Panel", + provenance_data: dict | None = None, + top_panels_csv: str | None = None, +) -> Path: + """Generate HTML report from in-memory data (for pipeline integration). + + CSV arguments are raw CSV strings; they are parsed internally. + """ + import io + + def _parse_csv_string(csv_str: str | None) -> list[dict] | None: + if not csv_str: + return None + reader = csv.DictReader(io.StringIO(csv_str)) + rows = [] + for row in reader: + for key in row: + val = row[key] + if val == "": + continue + try: + row[key] = int(val) + except (ValueError, TypeError): + try: + row[key] = float(val) + except (ValueError, TypeError): + pass + rows.append(row) + return rows if rows else None + + selected_pairs = _parse_csv_string(selected_pairs_csv) + off_targets = _parse_csv_string(off_targets_csv) + failed_junctions = _parse_csv_string(failed_junctions_csv) + top_panels = _parse_csv_string(top_panels_csv) + + if summary_data and summary_data.get("panel_name"): + panel_name = summary_data["panel_name"] + + html = _render_report( + qc_data, + panel_name=panel_name, + summary_data=summary_data, + selected_pairs=selected_pairs, + off_targets=off_targets, + failed_junctions=failed_junctions, + provenance_data=provenance_data, + top_panels=top_panels, + ) + + output_path = Path(output_path) + output_path.write_text(html, encoding="utf-8") + return output_path diff --git a/src/plexus/reporting/templates/panel_report.html.j2 b/src/plexus/reporting/templates/panel_report.html.j2 new file mode 100644 index 0000000..17775e2 --- /dev/null +++ b/src/plexus/reporting/templates/panel_report.html.j2 @@ -0,0 +1,458 @@ + + + + + +{{ panel_name }} — QC Report + + + + + + +

{{ panel_name }}

+

Plexus QC Report — generated {{ report_date }}

+ +
+
+
+ {% if summary %} +
Genome
{{ summary.get("genome", "—") }}
+
Junctions
+
{{ summary.get("num_junctions", "—") }} input → + {{ summary.get("num_selected_pairs", "—") }} designed + {% if num_failed > 0 %} → {{ num_failed }} failed{% endif %} +
+
Best Multiplex Cost
{{ "%.2f"|format(summary.best_multiplex_cost) if summary.get("best_multiplex_cost") is not none else "—" }}
+ {% else %} +
Panel
{{ panel_name }}
+ {% endif %} +
+
+ +
+
+ {% if provenance %} +
Plexus Version
{{ provenance.get("plexus_version", "—") }}
+
Mode
{{ provenance.get("operational_mode", "—") }}
+
Status
+
+ {% set status = provenance.get("status", "unknown") %} + {% if status == "completed" %}COMPLETED + {% elif status == "failed" %}FAILED + {% else %}{{ status|upper }}{% endif %} +
+ {% if provenance.get("compliance_environment") %} +
Compliance
+
+ {% set ce = provenance.compliance_environment %} + {% set verdicts = [] %} + {% for k, v in ce.items() if k != "manifest_version" and v is mapping %} + {% if v.get("verdict") == "pass" %}{% set _ = verdicts.append(true) %} + {% else %}{% set _ = verdicts.append(false) %}{% endif %} + {% endfor %} + {% if verdicts and verdicts|select("equalto", false)|list|length == 0 %} + ALL PASS + {% else %} + CHECK FAILURES + {% endif %} +
+ {% endif %} + {% else %} +
Version
—
+ {% endif %} +
+
+
+ + +

Tm Distribution

+
+ + +{% if selected_pairs_data %} +

Amplicon Size

+
+ + +

GC Content

+
+{% endif %} + + +{% if qc.cross_reactivity_matrix and qc.cross_reactivity_matrix.matrix %} +

Cross-Reactivity Heatmap

+
+{% endif %} + + +{% if qc.sequence_flags.flagged_primers %} +

Sequence Flags

+

+ High GC: {{ qc.sequence_flags.high_gc_count }} • + Low GC: {{ qc.sequence_flags.low_gc_count }} • + Homopolymer: {{ qc.sequence_flags.homopolymer_count }} +

+ + + + {% for p in qc.sequence_flags.flagged_primers %} + + + + + + + + + {% endfor %} + +
JunctionDirPrimerGC%SequenceFlags
{{ p.junction }}{{ p.direction }}{{ p.name }}{{ p.gc }}{{ p.sequence }}{% for f in p.flags %}{{ f }}{% endfor %}
+{% endif %} + + +{% if off_targets_data %} +

Off-Target Products

+ + + + {% for ot in off_targets_data %} + + + + + + + + + {% endfor %} + +
PairJunctionChromF PrimerR PrimerSize (bp)
{{ ot.get("Pair_ID", "—") }}{{ ot.get("Junction", "—") }}{{ ot.get("OT_Chrom", "—") }}{{ ot.get("OT_F_Primer", "—") }}{{ ot.get("OT_R_Primer", "—") }}{{ ot.get("OT_Product_Size", "—") }}
+{% endif %} + + +{% if failed_junctions_data %} +

Failed Junctions

+ + + + {% for fj in failed_junctions_data %} + + + + + + + + {% endfor %} + +
JunctionChromStartEndError
{{ fj.get("Junction", "—") }}{{ fj.get("Chrom", "—") }}{{ fj.get("Start", "—") }}{{ fj.get("End", "—") }}{{ fj.get("Error", "—") }}
+{% endif %} + + +{% if top_panels_data and top_panels_data|length > 1 %} +

Solution Comparison

+

+ Top {{ top_panels_data|map(attribute="Solution_Rank")|unique|list|length }} solutions shown. +

+ + + + {% set ns = namespace(prev_rank="") %} + {% for row in top_panels_data %} + + {% if row.Solution_Rank|string != ns.prev_rank %} + + + {% set ns.prev_rank = row.Solution_Rank|string %} + {% else %} + + {% endif %} + + + + {% endfor %} + +
RankCostJunctionPair
{{ row.Solution_Rank }}{{ "%.2f"|format(row.Solution_Cost) }}{{ row.Junction }}{{ row.Pair_ID }}
+{% endif %} + + +{% if provenance %} +
+

Provenance

+
+
Run:
{{ provenance.get("run_timestamp", "—") }}
+ {% if provenance.get("completed_at") %}
Completed:
{{ provenance.completed_at }}
{% endif %} +
FASTA:
{{ provenance.get("fasta_sha256", "—")[:16] }}…
+ {% if provenance.get("snp_vcf_sha256") %}
VCF:
{{ provenance.snp_vcf_sha256[:16] }}…
{% endif %} + {% if provenance.get("tool_versions") %} + {% for tool, ver in provenance.tool_versions.items() %} +
{{ tool }}:
{{ ver }}
+ {% endfor %} + {% endif %} +
+
+{% endif %} + +

+ Generated by Plexus {{ plexus_version }} +

+ + + + + + diff --git a/src/plexus/reporting/templates/plotly.min.js.gz b/src/plexus/reporting/templates/plotly.min.js.gz new file mode 100644 index 0000000..af85ffa Binary files /dev/null and b/src/plexus/reporting/templates/plotly.min.js.gz differ diff --git a/src/plexus/version.py b/src/plexus/version.py index 7863915..6849410 100644 --- a/src/plexus/version.py +++ b/src/plexus/version.py @@ -1 +1 @@ -__version__ = "1.0.2" +__version__ = "1.1.0" diff --git a/tests/test_reporting_html.py b/tests/test_reporting_html.py new file mode 100644 index 0000000..e63bc5f --- /dev/null +++ b/tests/test_reporting_html.py @@ -0,0 +1,252 @@ +"""Tests for HTML QC report generation (REPT-02).""" + +from __future__ import annotations + +import json + +import pytest + +from plexus.reporting.html_report import ( + generate_html_report, + generate_html_report_from_data, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +MINIMAL_QC = { + "tm_distribution": { + "mean": 60.0, + "std": 1.5, + "min": 58.0, + "max": 62.0, + "per_primer": [ + { + "junction": "GENE_A", + "direction": "forward", + "name": "GENE_A_fwd", + "tm": 59.5, + }, + { + "junction": "GENE_A", + "direction": "reverse", + "name": "GENE_A_rev", + "tm": 60.5, + }, + { + "junction": "GENE_B", + "direction": "forward", + "name": "GENE_B_fwd", + "tm": 58.0, + }, + { + "junction": "GENE_B", + "direction": "reverse", + "name": "GENE_B_rev", + "tm": 62.0, + }, + ], + }, + "sequence_flags": { + "gc_high_threshold": 70.0, + "gc_low_threshold": 30.0, + "homopolymer_min_run": 4, + "high_gc_count": 0, + "low_gc_count": 0, + "homopolymer_count": 1, + "flagged_primers": [ + { + "junction": "GENE_B", + "direction": "forward", + "name": "GENE_B_fwd", + "gc": 35.0, + "sequence": "AAAATGCATGCATGCATGC", + "flags": ["homopolymer"], + } + ], + }, + "cross_reactivity_matrix": { + "dimer_threshold": 0.0, + "matrix": { + "GENE_A": { + "GENE_B": {"min_dimer_score": -5.2, "interaction_count": 4}, + }, + "GENE_B": { + "GENE_A": {"min_dimer_score": -5.2, "interaction_count": 4}, + }, + }, + }, +} + +MINIMAL_SUMMARY = { + "panel_name": "TestPanel", + "genome": "hg38", + "num_junctions": 3, + "num_selected_pairs": 2, + "best_multiplex_cost": 1234.56, +} + + +@pytest.fixture +def qc_output_dir(tmp_path): + """Create a minimal pipeline output directory.""" + (tmp_path / "panel_qc.json").write_text(json.dumps(MINIMAL_QC)) + (tmp_path / "panel_summary.json").write_text(json.dumps(MINIMAL_SUMMARY)) + return tmp_path + + +@pytest.fixture +def full_output_dir(qc_output_dir): + """Output dir with all optional files.""" + csv = ( + "Junction,Chrom,Junction_Start,Junction_End,Pair_ID,Forward_Seq,Reverse_Seq," + "Forward_Full_Seq,Reverse_Full_Seq,Forward_Tm,Reverse_Tm,Tm_Diff," + "Forward_Bound,Reverse_Bound,Forward_GC,Reverse_GC,Forward_Length,Reverse_Length," + "Forward_Genomic_Start,Forward_Genomic_End,Reverse_Genomic_Start,Reverse_Genomic_End," + "Amplicon_Length,Insert_Size,Pair_Penalty,Dimer_Score,Off_Target_Count," + "Specificity_Checked,On_Target_Detected,SNP_Count,SNP_Penalty," + "Forward_SNP_Count,Reverse_SNP_Count\n" + "GENE_A,chr1,100,100,GENE_A_fwd_rev,ATCG,GCTA,ATCG,GCTA," + "59.5,60.5,1.0,80,70,50.0,55.0,20,20,80,99,101,120,80,40," + "100.0,-1.5,0,True,True,0,0.0,0,0\n" + "GENE_B,chr2,200,200,GENE_B_fwd_rev,TTTT,CCCC,TTTT,CCCC," + "58.0,62.0,4.0,75,65,35.0,65.0,22,22,180,201,201,222,90,46," + "120.0,-2.0,1,True,True,0,0.0,0,0\n" + ) + (qc_output_dir / "selected_multiplex.csv").write_text(csv) + + ot_csv = ( + "Pair_ID,Junction,OT_Chrom,OT_F_Primer,OT_R_Primer,OT_Product_Size,OT_F_Start,OT_R_Start\n" + "GENE_B_fwd_rev,GENE_B,chr5,SEQ_1,SEQ_2,150,5000,5150\n" + ) + (qc_output_dir / "off_targets.csv").write_text(ot_csv) + + fj_csv = ( + "Junction,Chrom,Start,End,Error\nGENE_C,chr3,300,300,no valid primer pairs\n" + ) + (qc_output_dir / "failed_junctions.csv").write_text(fj_csv) + + return qc_output_dir + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestGenerateHtmlReport: + def test_generates_html_file(self, qc_output_dir): + path = generate_html_report(qc_output_dir) + assert path.exists() + assert path.name == "panel_report.html" + assert path.stat().st_size > 1000 + + def test_html_contains_panel_name(self, qc_output_dir): + path = generate_html_report(qc_output_dir, panel_name="MyTestPanel") + html = path.read_text() + # panel_name from summary takes precedence + assert "TestPanel" in html + + def test_html_contains_plotly(self, qc_output_dir): + path = generate_html_report(qc_output_dir) + html = path.read_text() + assert "plotly.js" in html.lower() or "Plotly.newPlot" in html + + def test_html_contains_tm_data(self, qc_output_dir): + path = generate_html_report(qc_output_dir) + html = path.read_text() + assert "59.5" in html + assert "GENE_A" in html + + def test_html_contains_heatmap_data(self, qc_output_dir): + path = generate_html_report(qc_output_dir) + html = path.read_text() + assert "-5.2" in html + assert "cross_reactivity_matrix" in html or "heatmap" in html.lower() + + def test_html_handles_missing_optional_files(self, tmp_path): + (tmp_path / "panel_qc.json").write_text(json.dumps(MINIMAL_QC)) + path = generate_html_report(tmp_path, panel_name="Bare") + html = path.read_text() + assert "Bare" in html + assert path.exists() + + def test_html_handles_empty_panel(self, tmp_path): + empty_qc = { + "tm_distribution": { + "mean": None, + "std": None, + "min": None, + "max": None, + "per_primer": [], + }, + "sequence_flags": { + "gc_high_threshold": 70.0, + "gc_low_threshold": 30.0, + "homopolymer_min_run": 4, + "high_gc_count": 0, + "low_gc_count": 0, + "homopolymer_count": 0, + "flagged_primers": [], + }, + "cross_reactivity_matrix": {"dimer_threshold": 0.0, "matrix": {}}, + } + (tmp_path / "panel_qc.json").write_text(json.dumps(empty_qc)) + path = generate_html_report(tmp_path, panel_name="Empty") + assert path.exists() + html = path.read_text() + assert "Empty" in html + + def test_raises_without_qc_json(self, tmp_path): + with pytest.raises(FileNotFoundError, match="panel_qc.json"): + generate_html_report(tmp_path) + + def test_full_report_with_all_files(self, full_output_dir): + path = generate_html_report(full_output_dir) + html = path.read_text() + # Amplicon chart should be present + assert "amplicon-chart" in html + # Off-targets table + assert "GENE_B" in html + assert "chr5" in html + # Failed junctions table + assert "GENE_C" in html + assert "no valid primer pairs" in html + # GC chart + assert "gc-chart" in html + + +class TestGenerateHtmlReportFromData: + def test_from_data_creates_file(self, tmp_path): + output = tmp_path / "report.html" + path = generate_html_report_from_data( + qc_data=MINIMAL_QC, + summary_data=MINIMAL_SUMMARY, + selected_pairs_csv=None, + off_targets_csv=None, + failed_junctions_csv=None, + output_path=output, + panel_name="DataTest", + ) + assert path.exists() + html = path.read_text() + assert "TestPanel" in html # from summary_data + + def test_from_data_with_csv_strings(self, tmp_path): + csv_str = ( + "Junction,Forward_GC,Reverse_GC,Amplicon_Length\nGENE_A,50.0,55.0,80\n" + ) + output = tmp_path / "report.html" + path = generate_html_report_from_data( + qc_data=MINIMAL_QC, + summary_data=None, + selected_pairs_csv=csv_str, + off_targets_csv=None, + failed_junctions_csv=None, + output_path=output, + panel_name="CSVTest", + ) + assert path.exists() + html = path.read_text() + assert "CSVTest" in html