diff --git a/copykat_py/__init__.py b/copykat_py/__init__.py index 499c1f4..2b58bc5 100644 --- a/copykat_py/__init__.py +++ b/copykat_py/__init__.py @@ -1,9 +1,44 @@ """CopyKAT-Py: Python implementation of CopyKAT for scRNA-seq copy number inference.""" +from copykat_py._types import ( + AnchorPath, + AnchorStrategy, + BaselineWarning, + CellLineMode, + DataQualityStatus, + DistanceMetric, + ExecutionBackend, + FinalCallStrategy, + GeneIdType, + Genome, + KSMethod, + PredictionLabel, + ReferenceMode, + ReportFormat, +) from copykat_py.cli import copykat_anndata from copykat_py.copykat import copykat from copykat_py.data_loader import load_example_data from copykat_py.plotting import plot_heatmap_annotated __version__ = "1.0.0" -__all__ = ["copykat", "copykat_anndata", "load_example_data", "plot_heatmap_annotated"] +__all__ = [ + "AnchorPath", + "AnchorStrategy", + "BaselineWarning", + "CellLineMode", + "DataQualityStatus", + "DistanceMetric", + "ExecutionBackend", + "FinalCallStrategy", + "GeneIdType", + "Genome", + "KSMethod", + "PredictionLabel", + "ReferenceMode", + "ReportFormat", + "copykat", + "copykat_anndata", + "load_example_data", + "plot_heatmap_annotated", +] diff --git a/copykat_py/_types.py b/copykat_py/_types.py index 56548dd..1fffd27 100644 --- a/copykat_py/_types.py +++ b/copykat_py/_types.py @@ -8,8 +8,11 @@ - Clustering works on the transpose, ``CellByFeature`` (cells x features). """ +from __future__ import annotations + from collections.abc import Sequence -from typing import Any, Literal, NotRequired, TypeAlias, TypedDict +from enum import StrEnum +from typing import Any, NotRequired, TypeAlias, TypedDict import numpy as np import numpy.typing as npt @@ -43,6 +46,124 @@ SparseMatrix: TypeAlias = sparse.coo_matrix | sparse.csr_matrix | sparse.csc_matrix +class DistanceMetric(StrEnum): + """Supported distance metrics for ordering cells in CNA heatmaps.""" + + EUCLIDEAN = "euclidean" + PEARSON = "pearson" + SPEARMAN = "spearman" + + +class Genome(StrEnum): + """Supported reference genome assemblies.""" + + HG20 = "hg20" + MM10 = "mm10" + + +class ExecutionBackend(StrEnum): + """Supported execution backends.""" + + CPU = "cpu" + GPU = "gpu" + GPU_COMPAT = "gpu-compat" + + +class AnchorStrategy(StrEnum): + """How CopyKAT chooses the normal-cell reference.""" + + SIGMA = "sigma" + MARKERS = "markers" + + +class FinalCallStrategy(StrEnum): + """Strategy used to label final copy-number calls.""" + + CLUSTERS = "clusters" + ARM_CORRELATION = "arm_correlation" + + +class KSMethod(StrEnum): + """Breakpoint statistic used during segmentation.""" + + MONTE_CARLO = "mc" + EXACT = "exact" + + +class GeneIdType(StrEnum): + """Gene identifier representation accepted by annotation.""" + + SYMBOL = "S" + ENSEMBL = "E" + + @classmethod + def normalize(cls, value: GeneIdType | str) -> GeneIdType: + """Normalize the ID-type argument: E-prefixed values select Ensembl; others select symbols.""" + if isinstance(value, cls): + return value + return cls.ENSEMBL if str(value).upper().startswith(cls.ENSEMBL.value) else cls.SYMBOL + + +class CellLineMode(StrEnum): + """Whether input contains a pure cell line sample.""" + + YES = "yes" + NO = "no" + + +class ReportFormat(StrEnum): + """Supported output formats for run reports.""" + + TEXT = "txt" + MARKDOWN = "markdown" + HTML = "html" + JSON = "json" + + +class ReferenceMode(StrEnum): + """Source used to establish the normal reference population.""" + + SYNTHETIC = "synthetic" + KNOWN_NORMAL = "known_normal" + AUTOMATIC = "automatic" + + +class AnchorPath(StrEnum): + """Evidence used to choose the normal-cell reference.""" + + SIGMA = "sigma" + IMMUNE = "immune" + ENDOTHELIAL = "endothelial" + ENRICHMENT = "enrichment" + + +class PredictionLabel(StrEnum): + """Copy-number classification labels emitted by CopyKAT.""" + + DIPLOID = "diploid" + ANEUPLOID = "aneuploid" + DIPLOID_LOW_CONFIDENCE = "c1:diploid:low.conf" + ANEUPLOID_LOW_CONFIDENCE = "c2:aneuploid:low.conf" + NOT_DEFINED = "not.defined" + UNKNOWN = "unknown" + + +class BaselineWarning(StrEnum): + """Baseline classification status reported by the pipeline.""" + + NONE = "" + UNCLASSIFIED = "unclassified.prediction" + CELL_LINE = "run with cell line mode" + KNOWN_NORMAL = "run with known normal" + + +class DataQualityStatus(StrEnum): + """Data-quality status used to choose baseline fallback behavior.""" + + OK = "data quality is ok" + LOW = "low data quality" + + class RawInput(TypedDict): """Count matrix with its gene and cell names (genes x cells).""" @@ -54,9 +175,6 @@ class RawInput(TypedDict): RawMatrix: TypeAlias = pd.DataFrame | npt.NDArray[Any] | SparseMatrix | RawInput | str """Inputs accepted by ``copykat()``: genes x cells, or a path to a .mtx/.csv/.tsv file.""" -BaselineWarning: TypeAlias = Literal["", "unclassified.prediction", "run with cell line mode", "run with known normal"] -"""Classification note carried through the pipeline (``WNS`` in the R code).""" - class BaselineResult(TypedDict): """Reference normal cells and the baseline profile derived from them.""" @@ -66,7 +184,7 @@ class BaselineResult(TypedDict): preN: list[str] | IntArray """Reference normal cells: names when cell names are known, otherwise column indices.""" cl: ClusterLabels | None - anchor_path: NotRequired[str] + anchor_path: NotRequired[AnchorPath] class SyntheticBaselineResult(TypedDict): diff --git a/copykat_py/allele/orient.py b/copykat_py/allele/orient.py index ca236d7..e802f0e 100644 --- a/copykat_py/allele/orient.py +++ b/copykat_py/allele/orient.py @@ -13,6 +13,7 @@ import numpy as np import pandas as pd +from copykat_py._types import PredictionLabel from copykat_py.allele import counts as _counts from copykat_py.allele import hmm as _hmm @@ -21,8 +22,13 @@ def _swap_labels(pred): - swap = {"aneuploid": "diploid", "diploid": "aneuploid"} - return pred.str.replace(r"aneuploid|diploid", lambda m: swap[m.group(0)], regex=True) + swap = { + PredictionLabel.ANEUPLOID.value: PredictionLabel.DIPLOID.value, + PredictionLabel.DIPLOID.value: PredictionLabel.ANEUPLOID.value, + PredictionLabel.ANEUPLOID_LOW_CONFIDENCE.value: PredictionLabel.DIPLOID_LOW_CONFIDENCE.value, + PredictionLabel.DIPLOID_LOW_CONFIDENCE.value: PredictionLabel.ANEUPLOID_LOW_CONFIDENCE.value, + } + return pred.replace(swap) def orient_prediction(prediction, counts_dir, phase_csv, min_f_diff=MIN_F_DIFF): @@ -42,21 +48,27 @@ def orient_prediction(prediction, counts_dir, phase_csv, min_f_diff=MIN_F_DIFF): pred = prediction.iloc[:, 1].astype(str) idx = prediction.iloc[:, 0].map(_counts.barcode_key).map(col) has = idx.notna().to_numpy() - groups = {"aneuploid": (pred.str.contains("aneuploid").to_numpy() & has), - "diploid": (pred.str.contains("diploid").to_numpy() & ~pred.str.contains("aneuploid").to_numpy() & has)} - report = {"snps_phased": int(len(snps)), "flipped": False} + is_aneuploid = pred.str.contains(PredictionLabel.ANEUPLOID.value).to_numpy() + is_diploid = pred.str.contains(PredictionLabel.DIPLOID.value).to_numpy() & ~is_aneuploid + groups = { + PredictionLabel.ANEUPLOID: is_aneuploid & has, + PredictionLabel.DIPLOID: is_diploid & has, + } + report = {"snps_phased": len(snps), "flipped": False} for g, m in groups.items(): cols = idx[m].astype(int).to_numpy() - report[f"cells_{g}"] = int(len(cols)) + report[f"cells_{g.value}"] = len(cols) if len(cols) < MIN_GROUP_CELLS: - report["note"] = f"fewer than {MIN_GROUP_CELLS} {g} cells with allele data; calls left as they are" + report["note"] = f"fewer than {MIN_GROUP_CELLS} {g.value} cells with allele data; calls left as they are" return prediction.copy(), report k = np.asarray(HA[:, cols].sum(axis=1)).ravel() n = np.asarray(DP[:, cols].sum(axis=1)).ravel() seg, frac = _hmm.segments(snps, k, n) - report[f"F_{g}"] = frac - report[f"segments_{g}"] = seg.to_dict(orient="records") - report["flipped"] = bool(report["F_diploid"] - report["F_aneuploid"] >= min_f_diff) + report[f"F_{g.value}"] = frac + report[f"segments_{g.value}"] = seg.to_dict(orient="records") + report["flipped"] = bool( + report[f"F_{PredictionLabel.DIPLOID.value}"] - report[f"F_{PredictionLabel.ANEUPLOID.value}"] >= min_f_diff + ) out = prediction.copy() if report["flipped"]: out.iloc[:, 1] = _swap_labels(pred).to_numpy() @@ -68,8 +80,13 @@ def orient_prediction_file(prediction_txt, counts_dir, phase_csv, out_prefix=Non (out_prefix defaults to the prediction file's path without 'prediction.txt').""" pred = pd.read_csv(prediction_txt, sep="\t") out, report = orient_prediction(pred, counts_dir, phase_csv, min_f_diff) - prefix = out_prefix if out_prefix is not None else prediction_txt[: -len("prediction.txt")] \ - if prediction_txt.endswith("prediction.txt") else prediction_txt + "." + prefix = ( + out_prefix + if out_prefix is not None + else prediction_txt[: -len("prediction.txt")] + if prediction_txt.endswith("prediction.txt") + else prediction_txt + "." + ) out.to_csv(f"{prefix}prediction.allele_oriented.txt", sep="\t", index=False) with open(f"{prefix}allele_orientation.json", "w") as f: json.dump(report, f, indent=1) diff --git a/copykat_py/anchor.py b/copykat_py/anchor.py index 0c95e6c..0d47dde 100644 --- a/copykat_py/anchor.py +++ b/copykat_py/anchor.py @@ -26,16 +26,43 @@ import pandas as pd from scipy import sparse +from copykat_py._types import AnchorPath + IMMUNE_MARKERS = ("PTPRC", "LAPTM5", "CORO1A", "CD53", "LCP1", "CD52", "ARHGDIB") ENDOTHELIAL_MARKERS = ("PECAM1", "VWF", "CDH5", "CLDN5", "ESAM", "EMCN", "PLVAP") -MIN_MARKERS = 3 # markers detected for a cell to count as immune/endothelial -MIN_FRACTION = 0.5 # marked fraction for a cluster to qualify -MIN_CELLS = 20 # and at least max(MIN_CELLS, 1% of cells) -ENRICHMENT = 3.0 # fallback: mean immune-marker count vs the rest of the cells +MIN_MARKERS = 3 # markers detected for a cell to count as immune/endothelial +MIN_FRACTION = 0.5 # marked fraction for a cluster to qualify +MIN_CELLS = 20 # and at least max(MIN_CELLS, 1% of cells) +ENRICHMENT = 3.0 # fallback: mean immune-marker count vs the rest of the cells # hg38 centromere midpoints (Mb), chr1..22, X (=23) -_CENTROMERES_MB = np.array([123.4, 93.9, 90.9, 50.0, 48.8, 59.8, 60.1, 45.2, 43.0, 39.8, 53.4, 35.5, - 17.7, 17.2, 19.0, 36.8, 25.1, 18.5, 26.2, 28.1, 12.0, 15.0, 60.6]) +_CENTROMERES_MB = np.array( + [ + 123.4, + 93.9, + 90.9, + 50.0, + 48.8, + 59.8, + 60.1, + 45.2, + 43.0, + 39.8, + 53.4, + 35.5, + 17.7, + 17.2, + 19.0, + 36.8, + 25.1, + 18.5, + 26.2, + 28.1, + 12.0, + 15.0, + 60.6, + ] +) def count_markers(rawmat, markers): @@ -69,14 +96,16 @@ def _largest_population(labels, marked, n_cells): return best -def choose_anchor_cluster(labels, immune_counts, endothelial_counts, sigma_cluster): +def choose_anchor_cluster( + labels: np.ndarray, immune_counts: np.ndarray, endothelial_counts: np.ndarray, sigma_cluster: int +) -> tuple[int, AnchorPath]: """Return (cluster id, path) for the normal reference; path is "immune", "endothelial", "enrichment" or "sigma" and doubles as a coarse confidence indicator.""" labels = np.asarray(labels) imm = np.asarray(immune_counts, dtype=float) endo = np.asarray(endothelial_counts, dtype=float) n = len(labels) - for path, marked in (("immune", imm >= MIN_MARKERS), ("endothelial", endo >= MIN_MARKERS)): + for path, marked in ((AnchorPath.IMMUNE, imm >= MIN_MARKERS), (AnchorPath.ENDOTHELIAL, endo >= MIN_MARKERS)): k = _largest_population(labels, marked, n) if k is not None: return k, path @@ -86,8 +115,8 @@ def choose_anchor_cluster(labels, immune_counts, endothelial_counts, sigma_clust m = labels == ids[j] rest = imm[~m].mean() if (~m).any() else 0.0 if m.sum() >= MIN_CELLS and means[j] >= 1.0 and means[j] >= ENRICHMENT * max(rest, 1e-9): - return ids[j], "enrichment" - return sigma_cluster, "sigma" + return int(ids[j]), AnchorPath.ENRICHMENT + return sigma_cluster, AnchorPath.SIGMA def _arm_ids(chrom, pos_bp): @@ -104,10 +133,10 @@ def arm_correlation_calls(values, chrom, pos_bp, anchor_mask, top_fraction=0.10, A = np.asarray(agg @ np.asarray(values, dtype=np.float32), dtype=np.float64) # arms x cells A = A.astype(np.float16).astype(np.float64) # same precision as the evaluated harness w = sizes.astype(np.float64) - energy = np.sqrt((w[:, None] * A ** 2).sum(0) / w.sum()) + energy = np.sqrt((w[:, None] * A**2).sum(0) / w.sum()) consensus = A[:, energy >= np.percentile(energy, 100 * (1 - top_fraction))].mean(1) Ac = A - (w[:, None] * A).sum(0) / w.sum() cc = consensus - (w * consensus).sum() / w.sum() - r = (w[:, None] * Ac * cc[:, None]).sum(0) / np.sqrt((w[:, None] * Ac ** 2).sum(0) * (w * cc ** 2).sum() + 1e-30) + r = (w[:, None] * Ac * cc[:, None]).sum(0) / np.sqrt((w[:, None] * Ac**2).sum(0) * (w * cc**2).sum() + 1e-30) anchor_mask = np.asarray(anchor_mask, dtype=bool) return r > np.percentile(r[anchor_mask], percentile) diff --git a/copykat_py/annotation.py b/copykat_py/annotation.py index 2411bee..eb71fe8 100644 --- a/copykat_py/annotation.py +++ b/copykat_py/annotation.py @@ -8,6 +8,7 @@ import numpy.typing as npt import pandas as pd +from copykat_py._types import GeneIdType, Genome from copykat_py.data_loader import load_full_anno from copykat_py.genomic_coordinates import annotation_order @@ -15,7 +16,9 @@ def annotate_gene_rows( - genes: Sequence[Hashable] | pd.Index | npt.NDArray[Any], id_type: str = "S", genome: str = "hg20" + genes: Sequence[Hashable] | pd.Index | npt.NDArray[Any], + id_type: GeneIdType = GeneIdType.SYMBOL, + genome: Genome = Genome.HG20, ) -> tuple[pd.DataFrame, npt.NDArray[np.intp]]: """Annotate gene identifiers with genomic coordinates, without touching expression values. @@ -25,8 +28,8 @@ def annotate_gene_rows( Gene identifiers, one per expression-matrix row. id_type : str "S" for gene Symbol, "E" for Ensembl ID. - genome : str - "hg20" or "mm10". + genome : Genome + Genome.HG20 or Genome.MM10. Returns ------- @@ -36,15 +39,17 @@ def annotate_gene_rows( rows : np.ndarray For each annotation row, the index of the matching expression-matrix row. """ + genome = Genome(genome) + id_type = GeneIdType.normalize(id_type) logger.info(" start annotation ...") full_anno = load_full_anno(genome) - if genome == "mm10": + if genome is Genome.MM10: symbol_col = "mgi_symbol" else: symbol_col = "hgnc_symbol" - if id_type.upper().startswith("E"): + if id_type is GeneIdType.ENSEMBL: id_col = "ensembl_gene_id" else: id_col = symbol_col @@ -75,7 +80,9 @@ def annotate_gene_rows( return anno, rows -def annotate_genes(mat: pd.DataFrame, id_type: str = "S", genome: str = "hg20") -> pd.DataFrame: +def annotate_genes( + mat: pd.DataFrame, id_type: GeneIdType = GeneIdType.SYMBOL, genome: Genome = Genome.HG20 +) -> pd.DataFrame: """Annotate gene expression matrix with genomic coordinates. Parameters @@ -84,8 +91,8 @@ def annotate_genes(mat: pd.DataFrame, id_type: str = "S", genome: str = "hg20") Gene expression matrix, genes in rows, cells in columns. id_type : str "S" for gene Symbol, "E" for Ensembl ID. - genome : str - "hg20" or "mm10". + genome : Genome + Genome.HG20 or Genome.MM10. Returns ------- diff --git a/copykat_py/backend.py b/copykat_py/backend.py index ce43fb9..eba7af4 100644 --- a/copykat_py/backend.py +++ b/copykat_py/backend.py @@ -5,19 +5,20 @@ silhouette, FP64 segment accumulation, and full-tree heatmap ordering. """ -_STATE = {"name": "cpu"} +from copykat_py._types import ExecutionBackend -BACKENDS = ("cpu", "gpu", "gpu-compat") +_STATE: dict[str, ExecutionBackend] = {"name": ExecutionBackend.CPU} +BACKENDS = tuple(ExecutionBackend) -def set_backend(name): - name = str(name or "cpu").lower() +def set_backend(name: ExecutionBackend | str) -> None: + name = ExecutionBackend(str(name or ExecutionBackend.CPU).lower()) if name not in BACKENDS: raise ValueError(f"Unknown backend {name!r}; expected one of {BACKENDS}") - if name != "cpu": + if name is not ExecutionBackend.CPU: try: - import torch import cupy # noqa: F401 + import torch except ImportError as exc: # pragma: no cover - depends on install raise RuntimeError(f"backend {name!r} needs torch and cupy with CUDA: {exc}") from exc if not torch.cuda.is_available(): @@ -25,14 +26,14 @@ def set_backend(name): _STATE["name"] = name -def get_backend(): +def get_backend() -> ExecutionBackend: return _STATE["name"] -def use_gpu(): - return _STATE["name"] != "cpu" +def use_gpu() -> bool: + return _STATE["name"] is not ExecutionBackend.CPU -def exact_algorithms(): +def exact_algorithms() -> bool: """Whether the separately reviewed precision policy is enabled.""" - return _STATE["name"] == "gpu" + return _STATE["name"] is ExecutionBackend.GPU diff --git a/copykat_py/baseline.py b/copykat_py/baseline.py index 93a56c5..f15b191 100644 --- a/copykat_py/baseline.py +++ b/copykat_py/baseline.py @@ -17,7 +17,9 @@ from sklearn.decomposition import PCA from sklearn.metrics import silhouette_score +from copykat_py import backend from copykat_py._types import ( + AnchorPath, BaselineResult, BaselineWarning, CellByFeature, @@ -25,12 +27,13 @@ FloatArray, GeneByCell, GeneProfile, + Genome, IntArray, LinkageMatrix, ParallelInfo, + PredictionLabel, SyntheticBaselineResult, ) -from copykat_py import backend logger = logging.getLogger(__name__) @@ -76,7 +79,11 @@ def get_last_cluster_info() -> ParallelInfo: return _LAST_CLUSTER_INFO.copy() -def resolve_adaptive_pca_components(n_cells: int, pca_components: int | None = None, genome: str = "hg20") -> int: +def resolve_adaptive_pca_components( + n_cells: int, + pca_components: int | None = None, + genome: Genome = Genome.HG20, +) -> int: """Choose the PCA component cap for large-cell clustering. When ``pca_components`` is provided, that explicit value is used. @@ -89,9 +96,9 @@ def resolve_adaptive_pca_components(n_cells: int, pca_components: int | None = N return int(pca_components) n_cells = int(n_cells) - genome = str(genome).strip().lower() + genome = Genome(str(genome).strip().lower()) - if genome == "mm10": + if genome is Genome.MM10: if n_cells < MOUSE_AUTO_PCA_SMALL_CELL_COUNT_CUTOFF: return MOUSE_AUTO_PCA_SMALL_SAMPLE if n_cells < MOUSE_AUTO_PCA_MEDIUM_CELL_COUNT_CUTOFF: @@ -397,8 +404,8 @@ def baseline_norm_cl( n_cores: int = 1, cell_names: Sequence[str] | None = None, pca_components: int | None = None, - genome: str = "hg20", - anchor_selector: Callable[[ClusterLabels, int], tuple[int, str]] | None = None, + genome: Genome = Genome.HG20, + anchor_selector: Callable[[ClusterLabels, int], tuple[int, AnchorPath]] | None = None, ) -> BaselineResult: """Find a cluster of diploid cells using integrative clustering + GMM variance test. @@ -515,15 +522,15 @@ def fit_gmm_sigma(cluster_consensus: GeneProfile) -> float: PDt = f_dist.sf(f_stat, n_genes, n_genes) if wn <= 0.15 or not np.all(_cluster_sizes(labels) > min_cells) or PDt > 0.05: - WNS: BaselineWarning = "unclassified.prediction" + WNS: BaselineWarning = BaselineWarning.UNCLASSIFIED logger.warning(" low confidence in classification") else: - WNS = "" + WNS = BaselineWarning.NONE # Cluster with minimum sigma is the 'confident normal' cluster min_sigma_idx = np.argmin(SDM) normal_cluster_id = unique_clusters[min_sigma_idx] - anchor_path = "sigma" + anchor_path = AnchorPath.SIGMA if anchor_selector is not None: normal_cluster_id, anchor_path = anchor_selector(labels, int(normal_cluster_id)) @@ -554,7 +561,7 @@ def baseline_gmm( RE_before: BaselineResult | None = None, n_cores: int = 1, pca_components: int | None = None, - genome: str = "hg20", + genome: Genome = Genome.HG20, cluster: bool = True, ) -> BaselineResult: """Identify diploid cells one-by-one using GMM (fallback when clustering is uncertain). @@ -598,20 +605,17 @@ def baseline_gmm( # Check if any component mean is near zero (neutral) neutral_mask = np.abs(means) <= mu_cut - s = np.sum(neutral_mask) - - if s >= 1: - frq = np.sum(weights[neutral_mask]) - if frq > Nfraq_cut: - pred = "diploid" - else: - pred = "aneuploid" - else: - pred = "aneuploid" + has_neutral_component = np.any(neutral_mask) + neutral_fraction = np.sum(weights[neutral_mask]) + pred = ( + PredictionLabel.DIPLOID + if has_neutral_component and neutral_fraction > Nfraq_cut + else PredictionLabel.ANEUPLOID + ) N_normal_labels.append(pred) - if pred == "diploid": + if pred is PredictionLabel.DIPLOID: N_normal.append(cell_names[m]) if len(N_normal) >= max_normal: @@ -636,7 +640,7 @@ def baseline_gmm( ) if len(N_normal) > 2: - WNS: BaselineWarning = "" + WNS: BaselineWarning = BaselineWarning.NONE preN = N_normal normal_mask = np.isin(np.asarray(cell_names, dtype=object), np.asarray(preN, dtype=object)) basel = np.mean(CNA_mat[:, normal_mask], axis=1) @@ -646,7 +650,7 @@ def baseline_gmm( return RE_before else: # Fallback: use the full dataset median as baseline - WNS = "unclassified.prediction" + WNS = BaselineWarning.UNCLASSIFIED return {"basel": np.median(CNA_mat, axis=1), "WNS": WNS, "preN": N_normal, "cl": labels} @@ -655,7 +659,7 @@ def baseline_synthetic( min_cells: int = 10, n_cores: int = 1, pca_components: int | None = None, - genome: str = "hg20", + genome: Genome = Genome.HG20, ) -> SyntheticBaselineResult: """Estimate baseline using synthetic normal profiles (for cell line data). diff --git a/copykat_py/cli.py b/copykat_py/cli.py index 69b9533..5fd31c8 100644 --- a/copykat_py/cli.py +++ b/copykat_py/cli.py @@ -16,7 +16,20 @@ from scipy.io import mmread from copykat_py._logging import default_progress_output, log_progress_to, with_default_progress_output -from copykat_py._types import CopyKATResult, RawInput, RawMatrix, SparseMatrix +from copykat_py._types import ( + AnchorStrategy, + CellLineMode, + CopyKATResult, + DistanceMetric, + ExecutionBackend, + FinalCallStrategy, + GeneIdType, + Genome, + KSMethod, + RawInput, + RawMatrix, + SparseMatrix, +) from copykat_py.genomic_coordinates import split_cna_table logger = logging.getLogger(__name__) @@ -44,14 +57,16 @@ def _add_common_copykat_args(parser: argparse.ArgumentParser) -> None: """Attach CopyKAT runtime arguments shared by matrix and Python wrappers.""" parser.add_argument( "--id-type", - default="S", - choices=["S", "E"], + type=GeneIdType, + default=GeneIdType.SYMBOL, + choices=list(GeneIdType), help="[optional] Gene ID type: S=Symbol, E=Ensembl (default: S)", ) parser.add_argument( "--cell-line", - default="no", - choices=["yes", "no"], + type=CellLineMode, + default=CellLineMode.NO, + choices=list(CellLineMode), help="[optional] Pure cell line mode (default: no)", ) parser.add_argument( @@ -102,8 +117,9 @@ def _add_common_copykat_args(parser: argparse.ArgumentParser) -> None: ) parser.add_argument( "--distance", - default="euclidean", - choices=["euclidean", "pearson", "spearman"], + type=DistanceMetric, + default=DistanceMetric.EUCLIDEAN, + choices=list(DistanceMetric), help="[optional] Distance metric for clustering (default: euclidean)", ) parser.add_argument( @@ -126,8 +142,9 @@ def _add_common_copykat_args(parser: argparse.ArgumentParser) -> None: ) parser.add_argument( "--genome", - default="hg20", - choices=["hg20", "mm10"], + type=Genome, + default=Genome.HG20, + choices=list(Genome), help="[optional] Genome assembly (default: hg20)", ) parser.add_argument( @@ -144,26 +161,32 @@ def _add_common_copykat_args(parser: argparse.ArgumentParser) -> None: ) parser.add_argument( "--backend", - default="cpu", - choices=["cpu", "gpu-compat", "gpu"], + type=ExecutionBackend, + default=ExecutionBackend.CPU, + choices=list(ExecutionBackend), help="[experimental] Execution backend; GPU modes require CUDA-enabled torch and CuPy.", ) parser.add_argument( "--anchor", - default="sigma", - choices=["sigma", "markers"], - help="[experimental] Normal reference: smallest GMM sigma (default) or marker-defined immune/endothelial cells.", + type=AnchorStrategy, + default=AnchorStrategy.SIGMA, + choices=list(AnchorStrategy), + help=( + "[experimental] Normal reference: smallest GMM sigma (default) or marker-defined immune/endothelial cells." + ), ) parser.add_argument( "--final-call", - default="clusters", - choices=["clusters", "arm_correlation"], + type=FinalCallStrategy, + default=FinalCallStrategy.CLUSTERS, + choices=list(FinalCallStrategy), help="[experimental] Final calls from the Ward split (default) or arm-level correlation.", ) parser.add_argument( "--ks-method", - default="mc", - choices=["mc", "exact"], + type=KSMethod, + default=KSMethod.MONTE_CARLO, + choices=list(KSMethod), help="Breakpoint statistic: Monte Carlo posterior KS (default) or exact posterior-Gamma KS.", ) parser.add_argument( @@ -489,9 +512,7 @@ def _run_copykat_analysis( from copykat_py.allele.orient import orient_prediction - oriented, allele_report = orient_prediction( - result["prediction"], allele_counts_path, allele_phase_path - ) + oriented, allele_report = orient_prediction(result["prediction"], allele_counts_path, allele_phase_path) prefix = f"{args.sample_name}_copykat_" with open(f"{prefix}allele_orientation.json", "w", encoding="utf-8") as report_file: json.dump(allele_report, report_file, indent=2) @@ -573,14 +594,14 @@ def copykat_anndata( selecting_meta: str | Iterable[str] | None = None, row_split: str | None = None, sample_name: str = "", - distance: str = "euclidean", - genome: str = "hg20", + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, + genome: Genome = Genome.HG20, n_cores: int = 1, output_dir: str | os.PathLike[str] = ".", layer: str | None = None, use_raw: bool = False, - id_type: str = "S", - cell_line: str = "no", + id_type: GeneIdType = GeneIdType.SYMBOL, + cell_line: CellLineMode = CellLineMode.NO, ngene_chr: int = 5, min_genes: int = 200, low_dr: float = 0.05, @@ -593,6 +614,10 @@ def copykat_anndata( pca_components: int | None = None, ) -> CopyKATResult: """Python-friendly AnnData wrapper that accepts an in-memory AnnData object.""" + genome = Genome(genome) + distance = DistanceMetric(distance) + id_type = GeneIdType.normalize(id_type) + cell_line = CellLineMode(cell_line) _, rawmat, matrix_label = _anndata_to_rawmat( adata, layer=layer, @@ -719,8 +744,9 @@ def plot_main() -> None: ) parser.add_argument( "--distance", - default="euclidean", - choices=["euclidean", "pearson", "spearman"], + type=DistanceMetric, + default=DistanceMetric.EUCLIDEAN, + choices=list(DistanceMetric), help="[optional] Distance metric for within-group cell clustering (default: euclidean).", ) parser.add_argument( @@ -737,11 +763,15 @@ def plot_main() -> None: help="[optional] Output PNG path. Defaults to {sample_name}_copykat_annotated_heatmap.png.", ) parser.add_argument( - "--continuous-meta", nargs="+", default=None, metavar="COLUMN", + "--continuous-meta", + nargs="+", + default=None, + metavar="COLUMN", help="Force selected numeric metadata columns to use continuous colors (e.g. n_umi).", ) parser.add_argument( - "--no-row-split", action="store_true", + "--no-row-split", + action="store_true", help="Cluster all cells together without categorical row groups.", ) diff --git a/copykat_py/convert_bins.py b/copykat_py/convert_bins.py index 856cb27..0147a7b 100644 --- a/copykat_py/convert_bins.py +++ b/copykat_py/convert_bins.py @@ -11,7 +11,7 @@ import pandas as pd from joblib import Parallel, delayed -from copykat_py._types import BinConversion, GeneByCell, ParallelInfo +from copykat_py._types import BinConversion, GeneByCell, Genome, ParallelInfo from copykat_py.data_loader import load_dna_bins, load_full_anno _LAST_PAR_INFO: ParallelInfo = { @@ -30,7 +30,7 @@ def get_last_convert_bins_info() -> ParallelInfo: def convert_to_bins( RNA_mat: pd.DataFrame, - genome: str = "hg20", + genome: Genome = Genome.HG20, n_cores: int = 1, values: GeneByCell | None = None, cell_names: Sequence[str] | None = None, @@ -42,7 +42,7 @@ def convert_to_bins( RNA_mat : pd.DataFrame CNA results with columns: abspos, chromosome_name, start_position, end_position, ensembl_gene_id, hgnc_symbol (or mgi_symbol), band, cell1, cell2, ... - genome : str + genome : Genome "hg20" (bins conversion only supported for hg20). n_cores : int Number of parallel workers. @@ -60,12 +60,12 @@ def convert_to_bins( 'RNA_adj': pd.DataFrame - CNA values at genomic bins (chrom, chrompos, abspos, cell1, ...) 'RNA_adj_values': np.ndarray - the (n_bins, n_cells) values backing 'RNA_adj' """ - if genome != "hg20": + if genome != Genome.HG20: # For mm10, return gene-level results (no bin conversion, same as R) return None - full_anno = load_full_anno("hg20") - DNA_mat = load_dna_bins("hg20") + full_anno = load_full_anno(Genome.HG20) + DNA_mat = load_dna_bins(Genome.HG20) # Remove chromosome 24 (Y) DNA = DNA_mat[DNA_mat["chrom"] != 24].copy().reset_index(drop=True) diff --git a/copykat_py/copykat.py b/copykat_py/copykat.py index 8fb1414..c1859a1 100644 --- a/copykat_py/copykat.py +++ b/copykat_py/copykat.py @@ -31,14 +31,27 @@ from copykat_py import backend from copykat_py._logging import with_default_progress_output from copykat_py._types import ( + AnchorPath, + AnchorStrategy, + BaselineWarning, BoolArray, + CellLineMode, ClusteringResult, ClusterLabels, CopyKATResult, + DataQualityStatus, + DistanceMetric, + ExecutionBackend, + FinalCallStrategy, FloatArray, + GeneIdType, GeneProfile, + Genome, + KSMethod, ParallelInfo, + PredictionLabel, RawMatrix, + ReferenceMode, RuntimeInfo, SparseMatrix, ) @@ -115,7 +128,7 @@ def _meta_with_pred(meta_csv: str, pred_dict: dict[str, str] | None, sample_name """Read *meta_csv*, append copykat-py predictions as the last column. Returns the path to a new CSV written alongside the original outputs. - Cells absent from *pred_dict* receive ``"not.defined"``. + Cells absent from *pred_dict* receive ``PredictionLabel.NOT_DEFINED``. """ import pandas as pd @@ -123,9 +136,9 @@ def _meta_with_pred(meta_csv: str, pred_dict: dict[str, str] | None, sample_name cell_col = meta.columns[0] meta = meta.set_index(cell_col) if pred_dict is not None: - meta["copykat_pred_py"] = meta.index.map(pred_dict).fillna("not.defined") + meta["copykat_pred_py"] = meta.index.map(pred_dict).fillna(PredictionLabel.NOT_DEFINED) else: - meta["copykat_pred_py"] = "not.defined" + meta["copykat_pred_py"] = PredictionLabel.NOT_DEFINED out_path = f"{sample_name}meta_with_pred.csv" meta.reset_index().to_csv(out_path, index=False) return out_path @@ -136,10 +149,10 @@ def _run_plot_heatmap( chrom_info: npt.NDArray[Any], predictions: dict[str, str] | None, sample_name: str, - distance: str, + distance: DistanceMetric, n_cores: int, - WNS1: str, - WNS: str, + WNS1: DataQualityStatus, + WNS: BaselineWarning, output_path: str, ) -> None: from copykat_py.plotting import plot_heatmap @@ -212,7 +225,10 @@ def _preN_to_names(preN: str | bytes | Sequence[Any] | npt.NDArray[Any] | None, def _assign_binary_labels( - cluster_labels: ClusterLabels, scores: Sequence[float] | FloatArray, high_label: str, low_label: str + cluster_labels: ClusterLabels, + scores: Sequence[float] | FloatArray, + high_label: PredictionLabel, + low_label: PredictionLabel, ) -> npt.NDArray[np.object_]: labels = np.empty(len(cluster_labels), dtype=object) labels[:] = "" @@ -231,8 +247,8 @@ def _assign_binary_labels( @with_default_progress_output def copykat( rawmat: RawMatrix, - id_type: str = "S", - cell_line: str = "no", + id_type: GeneIdType = GeneIdType.SYMBOL, + cell_line: CellLineMode = CellLineMode.NO, ngene_chr: int = 5, min_gene_per_cell: int = 200, LOW_DR: float = 0.05, @@ -241,18 +257,18 @@ def copykat( norm_cell_names: str | list[str] = "", KS_cut: float = 0.1, sam_name: str = "", - distance: str = "euclidean", + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, output_seg: bool = False, plot_genes: bool = True, - genome: str = "hg20", + genome: Genome = Genome.HG20, n_cores: int = 1, pca_components: int | None = None, meta_csv: str | None = None, row_split_col: str | None = None, - backend_name: str = "cpu", - ks_method: str = "mc", - anchor: str = "sigma", - final_call: str = "clusters", + backend_name: ExecutionBackend = ExecutionBackend.CPU, + ks_method: KSMethod = KSMethod.MONTE_CARLO, + anchor: AnchorStrategy = AnchorStrategy.SIGMA, + final_call: FinalCallStrategy = FinalCallStrategy.CLUSTERS, ) -> CopyKATResult: """Run CopyKAT analysis: infer copy number profiles from scRNA-seq data. @@ -261,10 +277,10 @@ def copykat( rawmat : pd.DataFrame, np.ndarray, scipy.sparse, or str UMI count matrix (genes in rows, cells in columns). If str, path to .mtx, .csv, or .tsv file. - id_type : str - Gene ID type: "S" for Symbol, "E" for Ensembl. - cell_line : str - "yes" for pure cell line data, "no" for tumor/normal mixture. + id_type : GeneIdType + Gene ID type: ``GeneIdType.SYMBOL`` or ``GeneIdType.ENSEMBL``. + cell_line : CellLineMode + ``CellLineMode.YES`` for pure cell line data, ``CellLineMode.NO`` for tumor/normal mixture. ngene_chr : int Minimum number of genes per chromosome for cell filtering. min_gene_per_cell : int @@ -281,14 +297,14 @@ def copykat( KS test cutoff for breakpoint detection (0 to 1). sam_name : str Sample name prefix for output files. - distance : str - Distance metric: "euclidean", "pearson", or "spearman". + distance : DistanceMetric + Cell-ordering distance metric. output_seg : bool Whether to output .seg file for IGV. plot_genes : bool Whether to plot gene-level heatmap. - genome : str - "hg20" or "mm10". + genome : Genome + Genome.HG20 or Genome.MM10. n_cores : int Number of CPU cores for parallel computation. pca_components : int or None @@ -314,12 +330,16 @@ def copykat( 'CNAmat': pd.DataFrame with CNA results 'hclustering': linkage matrix or cluster labels """ + distance = DistanceMetric(distance) + backend_name = ExecutionBackend(backend_name) + id_type = GeneIdType.normalize(id_type) + cell_line = CellLineMode(cell_line) + ks_method = KSMethod(ks_method) + anchor = AnchorStrategy(anchor) + final_call = FinalCallStrategy(final_call) backend.set_backend(backend_name) - if anchor not in {"sigma", "markers"}: - raise ValueError("anchor must be 'sigma' or 'markers'") - if final_call not in {"clusters", "arm_correlation"}: - raise ValueError("final_call must be 'clusters' or 'arm_correlation'") - if final_call == "arm_correlation" and str(genome).lower() == "mm10": + genome = Genome(genome) + if final_call is FinalCallStrategy.ARM_CORRELATION and genome is Genome.MM10: raise ValueError("arm_correlation currently uses hg38 centromere coordinates and is only supported for hg20") start_time = time.perf_counter() # Global seed kept for reproducibility with earlier versions; moving to a @@ -371,7 +391,7 @@ def copykat( logger.info("step 1: read and filter data ...") step_start = time.perf_counter() marker_counts = None - if anchor == "markers": + if anchor is AnchorStrategy.MARKERS: marker_counts = ( _anchor.count_markers(rawmat, _anchor.IMMUNE_MARKERS), _anchor.count_markers(rawmat, _anchor.ENDOTHELIAL_MARKERS), @@ -389,7 +409,7 @@ def copykat( runtime_info["pca_selection_mode"] = "manual" if pca_components is not None else "auto_by_input_cell_count" runtime_info["pca_selection_genome"] = str(genome) runtime_info["pca_selection_input_cells"] = input_cell_count - if str(genome).strip().lower() == "mm10": + if genome is Genome.MM10: runtime_info["pca_selection_rule"] = ( f"<{MOUSE_AUTO_PCA_SMALL_CELL_COUNT_CUTOFF}->{MOUSE_AUTO_PCA_SMALL_SAMPLE}," f"<{MOUSE_AUTO_PCA_MEDIUM_CELL_COUNT_CUTOFF}->{MOUSE_AUTO_PCA_MEDIUM_SAMPLE}," @@ -412,9 +432,9 @@ def copykat( ) logger.info(f" {rawmat.shape[0]} genes past LOW_DR filtering") - WNS1 = "data quality is ok" + WNS1 = DataQualityStatus.OK if rawmat.shape[0] < 7000: - WNS1 = "low data quality" + WNS1 = DataQualityStatus.LOW UP_DR = LOW_DR logger.warning(" WARNING: low data quality; assigned LOW_DR to UP_DR...") runtime_info["warnings"].append("Low data quality; effective UP_DR was set to LOW_DR") @@ -428,12 +448,12 @@ def copykat( logger.info("step 2: annotating gene coordinates ...") step_start = time.perf_counter() anno_mat, anno_rows = annotate_gene_rows(gene_names, id_type=id_type, genome=genome) - runtime_info["parameters"]["gene_order"] = "chromosome,start_position" if genome == "mm10" else "abspos" + runtime_info["parameters"]["gene_order"] = "chromosome,start_position" if genome is Genome.MM10 else "abspos" # ========================================================================= # Step 3: Remove cell cycle genes and HLA genes (hg20 only) # ========================================================================= - if genome == "hg20": + if genome is Genome.HG20: symbol_col = "hgnc_symbol" cyclegenes = load_cyclegenes() hla_genes = anno_mat[symbol_col][anno_mat[symbol_col].str.startswith("HLA-", na=False)].tolist() @@ -517,7 +537,7 @@ def copykat( cell_name_list = cell_cols - if cell_line == "yes": + if cell_line is CellLineMode.YES: logger.info(" running pure cell line mode") relt = baseline_synthetic( norm_mat_smooth, @@ -528,7 +548,7 @@ def copykat( ) norm_mat_relat = relt["expr_relat"] CL = relt["cl"] - WNS = "run with cell line mode" + WNS = BaselineWarning.CELL_LINE preN = None elif isinstance(norm_cell_names, list) and len(norm_cell_names) > 1: # Known normal cells provided @@ -574,17 +594,17 @@ def copykat( if km == 2: break - WNS = "run with known normal" + WNS = BaselineWarning.KNOWN_NORMAL preN = np.asarray(cell_name_list, dtype=object)[known_normal_mask].tolist() norm_mat_relat = norm_mat_smooth - basel[:, np.newaxis] else: # Auto-detect normal cells anchor_selector = None - if anchor == "markers" and marker_counts is not None: + if anchor is AnchorStrategy.MARKERS and marker_counts is not None: immune_counts = marker_counts[0].groupby(level=0).first().reindex(cell_name_list).fillna(0).to_numpy() endothelial_counts = marker_counts[1].groupby(level=0).first().reindex(cell_name_list).fillna(0).to_numpy() - def anchor_selector(labels: ClusterLabels, sigma_cluster: int) -> tuple[int, str]: + def anchor_selector(labels: ClusterLabels, sigma_cluster: int) -> tuple[int, AnchorPath]: selected, path = _anchor.choose_anchor_cluster(labels, immune_counts, endothelial_counts, sigma_cluster) return int(selected), path @@ -603,14 +623,18 @@ def anchor_selector(labels: ClusterLabels, sigma_cluster: int) -> tuple[int, str clustered = basa["cl"] assert clustered is not None # baseline_norm_cl always clusters CL = clustered - runtime_info["anchor_path"] = basa.get("anchor_path", "sigma") - if anchor == "markers": - WNS = "" if runtime_info["anchor_path"] in {"immune", "endothelial"} else "unclassified.prediction" + runtime_info["anchor_path"] = basa.get("anchor_path", AnchorPath.SIGMA) + if anchor is AnchorStrategy.MARKERS: + WNS = ( + BaselineWarning.NONE + if runtime_info["anchor_path"] in {AnchorPath.IMMUNE, AnchorPath.ENDOTHELIAL} + else BaselineWarning.UNCLASSIFIED + ) logger.info(f" normal reference from markers: path={runtime_info['anchor_path']}, cells={len(preN)}") - if WNS == "unclassified.prediction" and anchor != "markers": + if WNS is BaselineWarning.UNCLASSIFIED and anchor is not AnchorStrategy.MARKERS: cluster_preN = list(preN) if preN is not None else [] - keep_cluster_anchor = WNS1 == "low data quality" and len(cluster_preN) >= max( + keep_cluster_anchor = WNS1 is DataQualityStatus.LOW and len(cluster_preN) >= max( 50, int(0.05 * len(cell_name_list)) ) if keep_cluster_anchor: @@ -671,15 +695,19 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: ) basel = basa["basel"] preN = basa["preN"] - WNS = "unclassified.prediction" + WNS = BaselineWarning.UNCLASSIFIED norm_mat_relat = norm_mat_smooth - basel[:, np.newaxis] del norm_mat_smooth baseline_cluster_info = get_last_cluster_info() + if cell_line is CellLineMode.YES: + reference_mode = ReferenceMode.SYNTHETIC + elif WNS is BaselineWarning.KNOWN_NORMAL: + reference_mode = ReferenceMode.KNOWN_NORMAL + else: + reference_mode = ReferenceMode.AUTOMATIC runtime_info["reference"] = { - "mode": ( - "synthetic" if cell_line == "yes" else "known_normal" if WNS == "run with known normal" else "automatic" - ), + "mode": reference_mode, "supplied_count": len(set(norm_cell_names)) if isinstance(norm_cell_names, list) else 0, "matched_supplied_count": len(set(norm_cell_names).intersection(cell_name_list)) if isinstance(norm_cell_names, list) @@ -807,7 +835,7 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: # ========================================================================= # Step 6: Convert to genomic bins (hg20 only) # ========================================================================= - if genome == "hg20": + if genome is Genome.HG20: logger.info("step 6: convert to genomic bins ...") step_start = time.perf_counter() Aj = convert_to_bins(gene_anno, genome=genome, n_cores=n_cores, values=results_com, cell_names=cell_cols_seg) @@ -830,11 +858,11 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: step_start = time.perf_counter() step7_reduce = uber_mat_adj.shape[1] > FULL_CLUSTER_MAX_CELLS - if cell_line == "yes": + if cell_line is CellLineMode.YES: mat_adj = uber_mat_adj else: arm_calls = None - if final_call == "arm_correlation" and preN is not None and len(preN) > 0: + if final_call is FinalCallStrategy.ARM_CORRELATION and preN is not None and len(preN) > 0: preN_names = _preN_to_names(preN, cell_name_list) anchor_mask = np.array([cell in preN_names for cell in cell_cols_seg], dtype=bool) if anchor_mask.sum() >= 5: @@ -844,7 +872,7 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: bin_coords["chrompos"].to_numpy(), anchor_mask, ) - runtime_info["final_call_path"] = "arm_correlation" + runtime_info["final_call_path"] = FinalCallStrategy.ARM_CORRELATION else: runtime_info["final_call_path"] = "clusters_insufficient_anchor" # First hierarchical clustering for initial prediction @@ -867,20 +895,22 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: cli_names = [cell_cols_seg[j] for j in range(len(cell_cols_seg)) if hc_umap[j] == cl_val] pid = len(set(cli_names) & preN_names) / max(len(cli_names), 1) cl_ID.append(pid) - com_pred = _assign_binary_labels(hc_umap, cl_ID, "diploid", "aneuploid") + com_pred = _assign_binary_labels(hc_umap, cl_ID, PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID) else: # If no preN, assign based on total CNA magnitude cl_mag = [] for cl_val in sorted(set(hc_umap)): mask = hc_umap == cl_val cl_mag.append(np.mean(np.abs(uber_mat_adj[:, mask]))) - com_pred = _assign_binary_labels(hc_umap, -np.asarray(cl_mag, dtype=float), "diploid", "aneuploid") + com_pred = _assign_binary_labels( + hc_umap, -np.asarray(cl_mag, dtype=float), PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID + ) if arm_calls is not None: - com_pred = np.where(arm_calls, "aneuploid", "diploid") + com_pred = np.where(arm_calls, PredictionLabel.ANEUPLOID, PredictionLabel.DIPLOID) # Baseline adjustment: subtract diploid mean, then denoise - diploid_mask = com_pred == "diploid" + diploid_mask = com_pred == PredictionLabel.DIPLOID if diploid_mask.sum() > 0: mat_adj = _adjust_baseline_inplace(uber_mat_adj, diploid_mask) else: @@ -902,7 +932,7 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: logger.info("step 8: final prediction ...") step_start = time.perf_counter() step8_reduce = mat_adj.shape[1] > FULL_CLUSTER_MAX_CELLS - if cell_line != "yes": + if cell_line is not CellLineMode.YES: labels_final, Z_final = _hierarchical_cluster( mat_adj.T, 2, @@ -921,21 +951,29 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: cli_names = [cell_cols_seg[j] for j in range(len(cell_cols_seg)) if hc_final[j] == cl_val] pid = len(set(cli_names) & preN_names) / max(len(cli_names), 1) cl_ID_final.append(pid) - com_preN = _assign_binary_labels(hc_final, cl_ID_final, "diploid", "aneuploid") + com_preN = _assign_binary_labels( + hc_final, cl_ID_final, PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID + ) else: cl_mag = [] for cl_val in sorted(set(hc_final)): mask = hc_final == cl_val cl_mag.append(np.mean(np.abs(mat_adj[:, mask]))) - com_preN = _assign_binary_labels(hc_final, -np.asarray(cl_mag, dtype=float), "diploid", "aneuploid") + com_preN = _assign_binary_labels( + hc_final, -np.asarray(cl_mag, dtype=float), PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID + ) if arm_calls is not None: # The Ward tree still orders the heatmap; this option supplies the final call. - com_preN = np.where(arm_calls, "aneuploid", "diploid") + com_preN = np.where(arm_calls, PredictionLabel.ANEUPLOID, PredictionLabel.DIPLOID) - if WNS == "unclassified.prediction": - com_preN = np.where(com_preN == "diploid", "c1:diploid:low.conf", com_preN) - com_preN = np.where(com_preN == "aneuploid", "c2:aneuploid:low.conf", com_preN) + if WNS is BaselineWarning.UNCLASSIFIED: + com_preN = np.where( + com_preN == PredictionLabel.DIPLOID, PredictionLabel.DIPLOID_LOW_CONFIDENCE, com_preN + ) + com_preN = np.where( + com_preN == PredictionLabel.ANEUPLOID, PredictionLabel.ANEUPLOID_LOW_CONFIDENCE, com_preN + ) else: labels, Z = _hierarchical_cluster( mat_adj.T, @@ -962,11 +1000,11 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: # ========================================================================= pred_dict = None res = None - if cell_line != "yes": + if cell_line is not CellLineMode.YES: pred_dict = {cell_cols_seg[i]: com_preN[i] for i in range(len(cell_cols_seg))} for cell in original_cell_names: if cell not in pred_dict: - pred_dict[cell] = "not.defined" + pred_dict[cell] = PredictionLabel.NOT_DEFINED res = pd.DataFrame( { "cell.names": list(pred_dict.keys()), @@ -986,8 +1024,8 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: # Save clustering clustering_data: ClusteringResult = { - "labels": labels_final if cell_line != "yes" else labels, - "Z": Z_final if cell_line != "yes" else Z, + "labels": labels_final if cell_line is not CellLineMode.YES else labels, + "Z": Z_final if cell_line is not CellLineMode.YES else Z, } with open(f"{sample_name}clustering_results.pkl", "wb") as f: pickle.dump(clustering_data, f) @@ -1005,7 +1043,7 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: if plot_genes: logger.info("step 10: plotting heatmap ...") plot_step_start = time.perf_counter() - predictions = pred_dict if cell_line != "yes" else None + predictions = pred_dict if cell_line is not CellLineMode.YES else None _run_plot_heatmap( mat_adj, chrom_info, @@ -1101,12 +1139,14 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: cl_ID.append(float(np.mean(np.abs(uber_mat_adj[:, mask])))) if preN is not None and len(preN) > 0: - com_pred = _assign_binary_labels(hc_umap, cl_ID, "diploid", "aneuploid") + com_pred = _assign_binary_labels(hc_umap, cl_ID, PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID) else: - com_pred = _assign_binary_labels(hc_umap, -np.asarray(cl_ID, dtype=float), "diploid", "aneuploid") + com_pred = _assign_binary_labels( + hc_umap, -np.asarray(cl_ID, dtype=float), PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID + ) # Baseline adjustment - diploid_mask = com_pred == "diploid" + diploid_mask = com_pred == PredictionLabel.DIPLOID if diploid_mask.sum() > 0: mat_adj = _adjust_baseline_inplace(uber_mat_adj, diploid_mask) else: @@ -1144,17 +1184,21 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: cli_names = [cell_cols_seg[j] for j in range(len(cell_cols_seg)) if hc_final[j] == cl_val] pid = len(set(cli_names) & preN_names) / max(len(cli_names), 1) cl_ID_final.append(pid) - com_preN = _assign_binary_labels(hc_final, cl_ID_final, "diploid", "aneuploid") + com_preN = _assign_binary_labels(hc_final, cl_ID_final, PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID) else: cl_mag = [] for cl_val in sorted(set(hc_final)): mask = hc_final == cl_val cl_mag.append(np.mean(np.abs(mat_adj[:, mask]))) - com_preN = _assign_binary_labels(hc_final, -np.asarray(cl_mag, dtype=float), "diploid", "aneuploid") + com_preN = _assign_binary_labels( + hc_final, -np.asarray(cl_mag, dtype=float), PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID + ) - if WNS == "unclassified.prediction": - com_preN = np.where(com_preN == "diploid", "c1:diploid:low.conf", com_preN) - com_preN = np.where(com_preN == "aneuploid", "c2:aneuploid:low.conf", com_preN) + if WNS is BaselineWarning.UNCLASSIFIED: + com_preN = np.where(com_preN == PredictionLabel.DIPLOID, PredictionLabel.DIPLOID_LOW_CONFIDENCE, com_preN) + com_preN = np.where( + com_preN == PredictionLabel.ANEUPLOID, PredictionLabel.ANEUPLOID_LOW_CONFIDENCE, com_preN + ) cluster_info = get_last_cluster_info() elapsed = _record_step( runtime_info, "final_prediction", step_start, parallel_info=cluster_info, extra={"warning": WNS} @@ -1171,7 +1215,7 @@ def _basel_sigma(basel_vec: GeneProfile) -> float: pred_dict = {cell_cols_seg[i]: com_preN[i] for i in range(len(cell_cols_seg))} for cell in original_cell_names: if cell not in pred_dict: - pred_dict[cell] = "not.defined" + pred_dict[cell] = PredictionLabel.NOT_DEFINED res = pd.DataFrame( { diff --git a/copykat_py/data_loader.py b/copykat_py/data_loader.py index a468964..36c3003 100644 --- a/copykat_py/data_loader.py +++ b/copykat_py/data_loader.py @@ -6,6 +6,8 @@ from scipy import sparse from scipy.io import mmread +from copykat_py._types import Genome + _DATA_DIR = os.path.join(os.path.dirname(__file__), "data") @@ -13,23 +15,25 @@ def _load_csv(filename: str) -> pd.DataFrame: return pd.read_csv(os.path.join(_DATA_DIR, filename)) -def load_full_anno(genome: str = "hg20") -> pd.DataFrame: +def load_full_anno(genome: Genome = Genome.HG20) -> pd.DataFrame: """Load the gene annotation table. Columns: abspos, chromosome_name, start_position, end_position, ensembl_gene_id, hgnc_symbol (or mgi_symbol for mm10), band. """ - if genome == "hg20": + genome = Genome(genome) + if genome is Genome.HG20: return _load_csv("full_anno_hg20.csv") - elif genome == "mm10": + elif genome is Genome.MM10: return _load_csv("full_anno_mm10.csv") else: raise ValueError(f"Unsupported genome: {genome}") -def load_dna_bins(genome: str = "hg20") -> pd.DataFrame: +def load_dna_bins(genome: Genome = Genome.HG20) -> pd.DataFrame: """Load 220KB variable genomic bins (chrom, chrompos, abspos).""" - if genome == "hg20": + genome = Genome(genome) + if genome is Genome.HG20: return _load_csv("DNA_hg20.csv") else: raise ValueError(f"DNA bins only available for hg20, got {genome}") diff --git a/copykat_py/genomic_coordinates.py b/copykat_py/genomic_coordinates.py index 9fe2a12..47d069c 100644 --- a/copykat_py/genomic_coordinates.py +++ b/copykat_py/genomic_coordinates.py @@ -6,14 +6,16 @@ import numpy.typing as npt import pandas as pd +from copykat_py._types import Genome -def annotation_order(annotation: pd.DataFrame, genome: str) -> npt.NDArray[np.intp]: + +def annotation_order(annotation: pd.DataFrame, genome: Genome) -> npt.NDArray[np.intp]: """Order mm10 by chromosome and gene start; retain hg20 absolute-position ordering. Mouse annotation abspos values are chromosome offsets, not individual gene positions. Stable sorting keeps equal-position genes in their original order. """ - if genome != "mm10": + if genome != Genome.MM10: return np.argsort(annotation["abspos"].to_numpy(), kind="mergesort") chrom = pd.to_numeric(annotation["chromosome_name"], errors="raise").to_numpy(dtype=float) start = pd.to_numeric(annotation["start_position"], errors="raise").to_numpy(dtype=float) @@ -22,25 +24,25 @@ def annotation_order(annotation: pd.DataFrame, genome: str) -> npt.NDArray[np.in return np.lexsort((start, chrom)).astype(np.intp) -def chromosome_label(chromosome: Any, genome: str) -> str: +def chromosome_label(chromosome: Any, genome: Genome) -> str: """Decode numeric sex chromosomes according to the supplied genome.""" try: numeric = float(str(chromosome)) if not numeric.is_integer(): return str(chromosome) code = int(numeric) - sex = {20: "X", 21: "Y"} if genome == "mm10" else {23: "X", 24: "Y"} + sex = {20: "X", 21: "Y"} if genome == Genome.MM10 else {23: "X", 24: "Y"} return sex.get(code, str(code)) except (ValueError, OverflowError): return str(chromosome) -def split_cna_table(frame: pd.DataFrame) -> tuple[npt.NDArray[np.float32], list[str], npt.NDArray[Any], str]: +def split_cna_table(frame: pd.DataFrame) -> tuple[npt.NDArray[np.float32], list[str], npt.NDArray[Any], Genome]: """Split saved human-bin or mouse-gene CNA results without treating annotation as cells.""" if "mgi_symbol" in frame.columns and "chromosome_name" in frame.columns: - leading, genome, chromosome = 7, "mm10", "chromosome_name" + leading, genome, chromosome = 7, Genome.MM10, "chromosome_name" elif list(frame.columns[:3]) == ["chrom", "chrompos", "abspos"]: - leading, genome, chromosome = 3, "hg20", "chrom" + leading, genome, chromosome = 3, Genome.HG20, "chrom" else: raise ValueError("Unrecognized CNA annotation columns; expected hg20 bins or mm10 gene results") return ( diff --git a/copykat_py/metadata_labels.py b/copykat_py/metadata_labels.py index 4f394c2..83d91a6 100644 --- a/copykat_py/metadata_labels.py +++ b/copykat_py/metadata_labels.py @@ -2,15 +2,17 @@ from collections.abc import Iterable +from copykat_py._types import BaselineWarning, DataQualityStatus, PredictionLabel + _PREDICTION_LABELS = { - "diploid": "Diploid", - "aneuploid": "Aneuploid", - "c1:diploid:low.conf": "Diploid (low confidence)", - "c2:aneuploid:low.conf": "Aneuploid (low confidence)", + PredictionLabel.DIPLOID: "Diploid", + PredictionLabel.ANEUPLOID: "Aneuploid", + PredictionLabel.DIPLOID_LOW_CONFIDENCE: "Diploid (low confidence)", + PredictionLabel.ANEUPLOID_LOW_CONFIDENCE: "Aneuploid (low confidence)", # copykat.py adds this for original cells without a final prediction. - "not.defined": "Not classified", + PredictionLabel.NOT_DEFINED: "Not classified", # plotting.py creates this when aligning absent/empty cell metadata. - "unknown": "Missing annotation", + PredictionLabel.UNKNOWN: "Missing annotation", } @@ -28,11 +30,11 @@ def annotation_title(column: str) -> str: def warning_caption(*warnings: str) -> str: """Readable plot subtitles for runtime warning/status strings.""" labels = { - "data quality is ok": "Data quality: OK", - "low data quality": "Low data quality", - "unclassified.prediction": "Low-confidence classification", - "run with known normal": "Supplied normal-cell reference", - "run with cell line mode": "Cell-line mode", + DataQualityStatus.OK: "Data quality: OK", + DataQualityStatus.LOW: "Low data quality", + BaselineWarning.UNCLASSIFIED: "Low-confidence classification", + BaselineWarning.KNOWN_NORMAL: "Supplied normal-cell reference", + BaselineWarning.CELL_LINE: "Cell-line mode", } return " · ".join(labels.get(value, value) for value in warnings if value) @@ -42,7 +44,7 @@ def is_prediction_column(column: str, values: Iterable[object]) -> bool: if column.lower() in {"copykat.pred", "copykat_pred", "copykat_pred_py", "copykat_pred_r"}: return True categories = {str(value) for value in values} - calls = set(_PREDICTION_LABELS) - {"not.defined", "unknown"} + calls = set(_PREDICTION_LABELS) - {PredictionLabel.NOT_DEFINED, PredictionLabel.UNKNOWN} return categories <= set(_PREDICTION_LABELS) and bool(categories & calls) diff --git a/copykat_py/plotting.py b/copykat_py/plotting.py index 4ab9038..d33c622 100644 --- a/copykat_py/plotting.py +++ b/copykat_py/plotting.py @@ -27,7 +27,16 @@ from threadpoolctl import threadpool_limits from copykat_py._logging import with_default_progress_output -from copykat_py._types import FloatArray, IntArray, LinkageMatrix +from copykat_py._types import ( + BaselineWarning, + DataQualityStatus, + DistanceMetric, + FloatArray, + Genome, + IntArray, + LinkageMatrix, + PredictionLabel, +) from copykat_py.baseline import _collapse_repeated_features, _ward_linkage from copykat_py.genomic_coordinates import chromosome_label from copykat_py.metadata_colors import ContinuousAnnotation, continuous_annotation, is_continuous @@ -73,10 +82,13 @@ def _simple_cell_order(mat: FloatArray, predictions: Mapping[str, str] | None = """Cheap fallback ordering used when clustering is unavailable or unsafe.""" if predictions is not None: pred_list = list(predictions.values()) - pred_rank = np.array( - [0 if "aneuploid" in str(p) else 1 if "diploid" in str(p) else 2 for p in pred_list], - dtype=np.int16, - ) + rank_by_label = { + PredictionLabel.ANEUPLOID: 0, + PredictionLabel.ANEUPLOID_LOW_CONFIDENCE: 0, + PredictionLabel.DIPLOID: 1, + PredictionLabel.DIPLOID_LOW_CONFIDENCE: 1, + } + pred_rank = np.array([rank_by_label.get(str(pred), 2) for pred in pred_list], dtype=np.int16) cna_magnitude = np.sum(np.abs(mat), axis=0) return np.lexsort((-cna_magnitude, pred_rank)) @@ -84,14 +96,18 @@ def _simple_cell_order(mat: FloatArray, predictions: Mapping[str, str] | None = return np.argsort(cna_magnitude)[::-1] -def _compute_distance(mat: FloatArray, distance: str = "euclidean", n_cores: int = 1) -> FloatArray: +def _compute_distance( + mat: FloatArray, + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, + n_cores: int = 1, +) -> FloatArray: """Compute distance matrix for cells. Parameters ---------- mat : np.ndarray, shape (n_bins, n_cells) CNA matrix. - distance : str + distance : DistanceMetric "euclidean", "pearson", or "spearman". Returns @@ -99,13 +115,14 @@ def _compute_distance(mat: FloatArray, distance: str = "euclidean", n_cores: int dist : np.ndarray Condensed distance matrix. """ - if distance == "euclidean": + distance = DistanceMetric(distance) + if distance == DistanceMetric.EUCLIDEAN: return pdist(mat.T, metric="euclidean") - elif distance == "pearson": + elif distance == DistanceMetric.PEARSON: corr = np.corrcoef(mat.T) corr = np.clip(corr, -1, 1) return pdist(1 - corr) - elif distance == "spearman": + elif distance == DistanceMetric.SPEARMAN: from scipy.stats import spearmanr corr, _ = spearmanr(mat, axis=0) @@ -116,7 +133,11 @@ def _compute_distance(mat: FloatArray, distance: str = "euclidean", n_cores: int def _safe_linkage( - mat: FloatArray, distance: str = "euclidean", method: str = "ward", n_cores: int = 1, max_cells: int = 65536 + mat: FloatArray, + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, + method: str = "ward", + n_cores: int = 1, + max_cells: int = 65536, ) -> LinkageMatrix: """Compute linkage with fastcluster-first execution. @@ -124,7 +145,7 @@ def _safe_linkage( regardless of cell count so plotting matches the main clustering path. The ``max_cells`` argument is retained for compatibility. """ - if distance == "euclidean" and method.startswith("ward"): + if distance == DistanceMetric.EUCLIDEAN and method.startswith("ward"): data = mat.T collapsed = _collapse_repeated_features(data) return _ward_linkage(data if collapsed is None else collapsed, n_cores=n_cores)[0] @@ -293,7 +314,7 @@ def _safe_dendrogram_with_recursion_management(Z: LinkageMatrix, ax: Axes, n_cel sys.setrecursionlimit(old_limit) -def _add_chr_labels(ax: Axes, chrom_info: npt.NDArray[Any], below: bool = False, genome: str = "hg20") -> None: +def _add_chr_labels(ax: Axes, chrom_info: npt.NDArray[Any], below: bool = False, genome: Genome = Genome.HG20) -> None: """Place chromosome name labels beside the chromosome-bar axes. Uses a mixed-coordinate transform (x in data coordinates, y in axes @@ -333,12 +354,12 @@ def plot_heatmap( chrom_info: npt.NDArray[Any], predictions: Mapping[str, str] | None = None, sample_name: str = "", - distance: str = "euclidean", + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, n_cores: int = 1, - WNS1: str = "", - WNS: str = "", + WNS1: DataQualityStatus = DataQualityStatus.OK, + WNS: BaselineWarning = BaselineWarning.NONE, output_path: str | None = None, - genome: str = "hg20", + genome: Genome = Genome.HG20, ) -> None: """Plot CNA heatmap with hierarchical clustering dendrogram. @@ -358,7 +379,7 @@ def plot_heatmap( Cell name -> "aneuploid"/"diploid" predictions. sample_name : str Sample name for title. - distance : str + distance : DistanceMetric Distance metric. n_cores : int Number of cores. @@ -369,6 +390,8 @@ def plot_heatmap( output_path : str or None Path to save figure. """ + genome = Genome(genome) + distance = DistanceMetric(distance) if output_path is None: output_path = f"{sample_name}_copykat_heatmap.png" @@ -524,18 +547,25 @@ def plot_heatmap( if predictions is not None: ax_pred = fig.add_subplot(gs[1, col_pred], sharey=ax_heat) pred_list = list(predictions.values()) - pred_ordered = [pred_list[i] if i < len(pred_list) else "not.defined" for i in cell_order] + pred_ordered = [pred_list[i] if i < len(pred_list) else PredictionLabel.NOT_DEFINED for i in cell_order] pred_palette = _assign_cat_colors(pred_ordered) pred_colors = np.array([mcolors.to_rgb(pred_palette[str(p)]) for p in pred_ordered]).reshape(n_cells, 1, 3) - ax_pred.imshow( - pred_colors, aspect="auto", interpolation="nearest" - ) + ax_pred.imshow(pred_colors, aspect="auto", interpolation="nearest") ax_pred.set_xticks([]) ax_pred.set_yticks([]) ax_pred_top = fig.add_subplot(gs[0, col_pred]) ax_pred_top.axis("off") - ax_pred_top.text(0.5, 0.02, "CopyKAT Python", ha="left", va="bottom", fontsize=10, - rotation=45, rotation_mode="anchor", transform=ax_pred_top.transAxes) + ax_pred_top.text( + 0.5, + 0.02, + "CopyKAT Python", + ha="left", + va="bottom", + fontsize=10, + rotation=45, + rotation_mode="anchor", + transform=ax_pred_top.transAxes, + ) for spine in ax_pred.spines.values(): spine.set_linewidth(0.6) @@ -558,8 +588,17 @@ def plot_heatmap( cbar = fig.colorbar(im, cax=ax_cbar, orientation="vertical") ax_cbar_top = fig.add_subplot(gs[0, col_cbar]) ax_cbar_top.axis("off") - ax_cbar_top.text(0.5, 0.02, "Relative CNA", ha="left", va="bottom", fontsize=10, - rotation=45, rotation_mode="anchor", transform=ax_cbar_top.transAxes) + ax_cbar_top.text( + 0.5, + 0.02, + "Relative CNA", + ha="left", + va="bottom", + fontsize=10, + rotation=45, + rotation_mode="anchor", + transform=ax_cbar_top.transAxes, + ) cbar.ax.tick_params(labelsize=10) for ax in [ax_heat, ax_chr, ax_cbar]: for spine in ax.spines.values(): @@ -584,12 +623,12 @@ def _natural_sort_key(s: object) -> list[int | str]: # Fixed colors for copykat prediction values (aneuploid=orange, diploid=blue) _COPYKAT_PRED_COLORS = { - "aneuploid": "#E8601C", # orange - "c2:aneuploid:low.conf": "#F4A86A", # light orange - "diploid": "#3A87C8", # blue - "c1:diploid:low.conf": "#9EC8E8", # light blue - "not.defined": "#B0B0B0", # grey - "unknown": "#D4D4D4", # light grey + PredictionLabel.ANEUPLOID: "#E8601C", # orange + PredictionLabel.ANEUPLOID_LOW_CONFIDENCE: "#F4A86A", # light orange + PredictionLabel.DIPLOID: "#3A87C8", # blue + PredictionLabel.DIPLOID_LOW_CONFIDENCE: "#9EC8E8", # light blue + PredictionLabel.NOT_DEFINED: "#B0B0B0", # grey + PredictionLabel.UNKNOWN: "#D4D4D4", # light grey } @@ -604,7 +643,8 @@ def _assign_cat_colors(values: Iterable[object]) -> dict[str, str]: cats = sorted({str(v) for v in values}, key=_natural_sort_key) # Detect copykat prediction columns by value content - if any("aneuploid" in c or "diploid" in c for c in cats): + prediction_labels = set(PredictionLabel) - {PredictionLabel.UNKNOWN, PredictionLabel.NOT_DEFINED} + if set(cats) & prediction_labels: return {cat: _COPYKAT_PRED_COLORS.get(cat, "#B0B0B0") for cat in cats} n = len(cats) @@ -615,8 +655,8 @@ def _assign_cat_colors(values: Iterable[object]) -> dict[str, str]: else: palette = [mcolors.to_hex(plt.cm.hsv(i / n)) for i in range(n)] cmap = {cat: palette[i % len(palette)] for i, cat in enumerate(cats)} - if "unknown" in cmap: - cmap["unknown"] = "#cccccc" + if PredictionLabel.UNKNOWN in cmap: + cmap[PredictionLabel.UNKNOWN] = "#cccccc" return cmap @@ -652,7 +692,11 @@ def _read_meta_csv(path: str) -> pd.DataFrame: return df -def _order_group(mat_grp: FloatArray, distance: str = "euclidean", n_cores: int = 1) -> IntArray: +def _order_group( + mat_grp: FloatArray, + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, + n_cores: int = 1, +) -> IntArray: """Return a cell-ordering index array for one CNA sub-matrix. Uses full Ward linkage for groups ≤ 3 000 cells; K-means block ordering @@ -689,11 +733,11 @@ def plot_heatmap_annotated( meta_csv: str, row_split_col: str | None = None, sample_name: str = "", - distance: str = "euclidean", + distance: DistanceMetric = DistanceMetric.EUCLIDEAN, n_cores: int = 1, output_path: str | None = None, continuous_meta: Sequence[str] | None = None, - genome: str = "hg20", + genome: Genome = Genome.HG20, ) -> None: """Plot CNA heatmap with per-cell metadata annotation bars and row splitting. @@ -722,7 +766,7 @@ def plot_heatmap_annotated( and clusters all cells together. sample_name : str Label shown in the figure title and used for the default filename. - distance : str + distance : DistanceMetric Distance metric for within-group clustering (``"euclidean"``, ``"pearson"``, or ``"spearman"``). n_cores : int @@ -735,6 +779,8 @@ def plot_heatmap_annotated( distinct values. By default numeric measurements are detected; small integer-coded categories and the row-split column remain categorical. """ + genome = Genome(genome) + distance = DistanceMetric(distance) if output_path is None: output_path = f"{sample_name}_copykat_annotated_heatmap.png" @@ -761,7 +807,7 @@ def plot_heatmap_annotated( ann_cols = [row_split_col] + [c for c in ann_cols if c != row_split_col] cell_names_str = [str(c) for c in cell_names] - meta_aligned = meta_df.reindex(cell_names_str).fillna("unknown") + meta_aligned = meta_df.reindex(cell_names_str).fillna(PredictionLabel.UNKNOWN) # ── 2. Per-group ordering: sort groups, then cluster cells within ───── split_vals = meta_aligned[row_split_col].astype(str).to_numpy() if row_split_col else np.full(n_cells, "") @@ -897,8 +943,17 @@ def plot_heatmap_annotated( cbar = fig.colorbar(im, cax=ax_cbar, orientation="vertical") ax_cbar_top = fig.add_subplot(gs[0, col_cbar]) ax_cbar_top.axis("off") - ax_cbar_top.text(0.5, 0.02, "Relative CNA", ha="left", va="bottom", fontsize=10, - rotation=45, rotation_mode="anchor", transform=ax_cbar_top.transAxes) + ax_cbar_top.text( + 0.5, + 0.02, + "Relative CNA", + ha="left", + va="bottom", + fontsize=10, + rotation=45, + rotation_mode="anchor", + transform=ax_cbar_top.transAxes, + ) cbar.ax.tick_params(labelsize=10) # ── 10. Categorical legend ──────────────────────────────────────────── diff --git a/copykat_py/reporting/charts.py b/copykat_py/reporting/charts.py index 0e1fcd8..447e780 100644 --- a/copykat_py/reporting/charts.py +++ b/copykat_py/reporting/charts.py @@ -5,30 +5,34 @@ from collections.abc import Mapping from typing import Any +from copykat_py._types import BaselineWarning, PredictionLabel from copykat_py.metadata_labels import legend_label COLORS = { - "diploid": "#3A87C8", - "aneuploid": "#E8601C", - "c1:diploid:low.conf": "#9EC8E8", - "c2:aneuploid:low.conf": "#F4A86A", - "not.defined": "#B0B0B0", - "unknown": "#D4D4D4", + PredictionLabel.DIPLOID: "#3A87C8", + PredictionLabel.ANEUPLOID: "#E8601C", + PredictionLabel.DIPLOID_LOW_CONFIDENCE: "#9EC8E8", + PredictionLabel.ANEUPLOID_LOW_CONFIDENCE: "#F4A86A", + PredictionLabel.NOT_DEFINED: "#B0B0B0", + PredictionLabel.UNKNOWN: "#D4D4D4", } def confidence_status(predictions: Mapping[str, int] | None, runtime: Mapping[str, Any]) -> str: """Summarize saved classification flags, never invent confidence for missing calls.""" if any( - step.get("warning") == "unclassified.prediction" + step.get("warning") == BaselineWarning.UNCLASSIFIED for step in runtime.get("steps", []) if step.get("step") == "final_prediction" ): return "Low" calls = predictions or {} - if any(calls.get(state, 0) > 0 for state in ("c1:diploid:low.conf", "c2:aneuploid:low.conf")): + if any( + calls.get(state, 0) > 0 + for state in (PredictionLabel.DIPLOID_LOW_CONFIDENCE, PredictionLabel.ANEUPLOID_LOW_CONFIDENCE) + ): return "Low" - if any(calls.get(state, 0) > 0 for state in ("diploid", "aneuploid")): + if any(calls.get(state, 0) > 0 for state in (PredictionLabel.DIPLOID, PredictionLabel.ANEUPLOID)): return "High" return "Unavailable" diff --git a/copykat_py/reporting/render.py b/copykat_py/reporting/render.py index 9d75330..602cfe4 100644 --- a/copykat_py/reporting/render.py +++ b/copykat_py/reporting/render.py @@ -7,11 +7,17 @@ from typing import Any from urllib.parse import quote +from copykat_py._types import ReportFormat from copykat_py.reporting.html import render_html from copykat_py.reporting.images import preview_issue from copykat_py.reporting.model import RunReport -FORMATS = {"txt": ".txt", "markdown": ".md", "html": ".html", "json": ".json"} +FORMATS: dict[ReportFormat, str] = { + ReportFormat.TEXT: ".txt", + ReportFormat.MARKDOWN: ".md", + ReportFormat.HTML: ".html", + ReportFormat.JSON: ".json", +} def report_sections(report: RunReport) -> list[tuple[str, list[tuple[str, Any]]]]: @@ -98,12 +104,14 @@ def _href(path: Path, output_dir: Path) -> str: return quote(path.name, safe="") if path.parent == output_dir else path.as_uri() -def render_report(report: RunReport, fmt: str, output_dir: Path) -> str: +def render_report(report: RunReport, fmt: ReportFormat | str, output_dir: Path) -> str: """Render without executing code or loading large matrix files.""" - if fmt not in FORMATS: - raise ValueError(f"Unknown report format: {fmt}") + try: + fmt = ReportFormat(fmt) + except ValueError as error: + raise ValueError(f"Unknown report format: {fmt}") from error sections = report_sections(report) - if fmt == "json": + if fmt is ReportFormat.JSON: return ( json.dumps( { @@ -121,13 +129,13 @@ def render_report(report: RunReport, fmt: str, output_dir: Path) -> str: ) + "\n" ) - if fmt == "txt": + if fmt is ReportFormat.TEXT: lines = ["CopyKAT-Python run report"] for title, rows in sections: lines.extend(["", title, *(f" {key}: {_value(value)}" for key, value in rows)]) lines.extend(["", "Output files", *(f" {path}" for path in report.files)]) return "\n".join(lines) + "\n" - if fmt == "markdown": + if fmt is ReportFormat.MARKDOWN: blocks = ["# CopyKAT-Python run report"] for title, rows in sections: blocks.append(f"## {title}\n\n" + "\n".join(f"- {_md(key)}: {_md(value)}" for key, value in rows)) @@ -151,10 +159,13 @@ def render_report(report: RunReport, fmt: str, output_dir: Path) -> str: return render_html(report, sections, output_dir) -def write_reports(report: RunReport, formats: Iterable[str], output_dir: str | Path) -> list[Path]: +def write_reports(report: RunReport, formats: Iterable[ReportFormat | str], output_dir: str | Path) -> list[Path]: """Write selected formats; only report files are created or replaced.""" - selected = list(dict.fromkeys(formats)) - if not selected or any(fmt not in FORMATS for fmt in selected): + try: + selected = list(dict.fromkeys(ReportFormat(fmt) for fmt in formats)) + except ValueError as error: + raise ValueError(f"Select one or more formats: {', '.join(FORMATS)}") from error + if not selected: raise ValueError(f"Select one or more formats: {', '.join(FORMATS)}") directory = Path(output_dir).resolve() # Render everything before writing, so a missing image does not leave partial reports. diff --git a/copykat_py/segmentation.py b/copykat_py/segmentation.py index 1d00bd4..b222fcb 100644 --- a/copykat_py/segmentation.py +++ b/copykat_py/segmentation.py @@ -20,7 +20,7 @@ from scipy.special import gammainc, gammaincinv, gammaln from scipy.stats import ks_2samp -from copykat_py._types import ClusterLabels, FloatArray, GeneByCell, ParallelInfo, SegmentationResult +from copykat_py._types import ClusterLabels, FloatArray, GeneByCell, KSMethod, ParallelInfo, SegmentationResult _LAST_PAR_INFO: ParallelInfo = { "step": "cna_mcmc", @@ -145,7 +145,7 @@ def log_density_ratio(t: float) -> float: edges = [lo, turning_point, hi] distance = 0.0 - for left, right in zip(edges[:-1], edges[1:]): + for left, right in zip(edges[:-1], edges[1:], strict=False): f_left, f_right = log_density_ratio(left), log_density_ratio(right) if f_left == 0: root = left @@ -183,7 +183,7 @@ def cna_mcmc( cut_cor: float = 0.1, n_cores: int = 1, mc_samples: int | None = None, - ks_method: str = "mc", + ks_method: KSMethod = KSMethod.MONTE_CARLO, ) -> SegmentationResult: """MCMC segmentation of copy number data. @@ -209,8 +209,7 @@ def cna_mcmc( 'breaks': list - breakpoint positions """ n_genes, n_cells = fttmat.shape - if ks_method not in {"mc", "exact"}: - raise ValueError("ks_method must be 'mc' or 'exact'") + ks_method = KSMethod(ks_method) # Adaptive MC sample size: fewer samples for small datasets (speed optimization) if mc_samples is None: @@ -226,7 +225,7 @@ def cna_mcmc( # Step 2: Find breakpoints for each cluster consensus breakpoints: set[int] = set() for c in range(norm_mat_sm.shape[1]): - if ks_method == "exact": + if ks_method is KSMethod.EXACT: bre = _find_breakpoints_exact(norm_mat_sm[:, c], bins, cut_cor) else: bre = _find_breakpoints_for_cluster(