"""
CodonAdaptPy Per-Isolate CDS Aggregation
This module groups independently framed coding sequences by isolate, produces
per-gene results, creates boundary-aware genome-wide coding-region summaries,
and selects any user-nominated focus genes such as F, G, or L. It never treats
a raw multi-ORF genome as one continuous reading frame.
Classes:
- IsolateAnalysis: Per-gene, genome-wide, and focus-gene results.
- CDSAggregator: Metadata-aware analysis of isolate collections.
:Created: July 20, 2026
:Updated: July 20, 2026
:Author: Naveen Duhan
:Version: 1.0.2
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import asdict, dataclass, field
from typing import Any
from .analyzer import CodonAnalyzer
from .exceptions import AnalysisError
from .models import AnalysisResult, SequenceRecord
[docs]
@dataclass(slots=True)
class IsolateAnalysis:
"""Store all analysis scopes generated for one biological isolate."""
isolate_id: str
genome_result: AnalysisResult
gene_results: list[AnalysisResult]
focus_gene_results: dict[str, list[AnalysisResult]] = field(default_factory=dict)
[docs]
def to_dict(self) -> dict[str, Any]:
"""Return nested isolate output in JSON-compatible form."""
return asdict(self)
[docs]
class CDSAggregator:
"""Group CDS records and calculate gene-wise plus genome-wide results."""
def __init__(self, analyzer: CodonAnalyzer) -> None:
"""Initialize aggregation around an existing configured analyzer."""
self.analyzer = analyzer
[docs]
def analyze(
self,
records: list[SequenceRecord],
*,
isolate_key: str = "isolate",
gene_key: str = "gene",
focus_genes: tuple[str, ...] = (),
) -> list[IsolateAnalysis]:
"""Analyze CDSs grouped by metadata-defined biological isolate."""
grouped: dict[str, list[SequenceRecord]] = defaultdict(list)
for record in records:
isolate = str(record.metadata.get(isolate_key, "")).strip()
gene = str(record.metadata.get(gene_key, "")).strip()
if not isolate:
raise AnalysisError(f"Record {record.identifier!r} lacks metadata field {isolate_key!r}.")
if not gene:
raise AnalysisError(f"Record {record.identifier!r} lacks metadata field {gene_key!r}.")
grouped[isolate].append(record)
requested = {gene.casefold(): gene for gene in focus_genes}
outputs: list[IsolateAnalysis] = []
for isolate, members in sorted(grouped.items()):
gene_results: list[AnalysisResult] = []
focus_results: dict[str, list[AnalysisResult]] = {gene: [] for gene in focus_genes}
for record in members:
result = self.analyzer.analyze(record)
gene = str(record.metadata[gene_key])
result.metadata.update({"analysis_scope": "gene", "isolate": isolate, "gene": gene})
gene_results.append(result)
if gene.casefold() in requested:
focus_results[requested[gene.casefold()]].append(result)
missing_focus = [gene for gene, results in focus_results.items() if not results]
if missing_focus:
raise AnalysisError(f"Isolate {isolate!r} is missing focus gene(s): {', '.join(missing_focus)}")
genome_result = self.analyzer.analyze_cds_collection(
isolate,
members,
metadata={"isolate": isolate, "genes": [record.metadata[gene_key] for record in members]},
)
outputs.append(IsolateAnalysis(isolate, genome_result, gene_results, focus_results))
return outputs