"""
CodonAdaptPy Analysis Interface
This module exposes the primary high-level Python API. :class:`CodonAnalyzer`
coordinates normalization, validation, intrinsic metrics, optional reference-
dependent adaptation statistics, multi-host comparisons, simulation, and
sliding-window profiles while keeping each calculation module independently
testable.
Classes:
- AnalysisConfig: Controls optional analyses and profile resolution.
- CodonAnalyzer: Analyze one record, many records, or a supported file.
:Created: July 20, 2026
:Updated: July 20, 2026
:Author: Naveen Duhan
:Version: 1.0.2
"""
from __future__ import annotations
import hashlib
import platform
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from .__version__ import __version__
from .exceptions import AnalysisError
from .metrics import (
AdaptationMetrics,
CodonUsageMetrics,
CompositionMetrics,
EvolutionaryMetrics,
HostComparator,
PairMetrics,
)
from .models import AnalysisResult, SequenceRecord, ValidationConfig
from .parsers import SequenceParser
from .references import ReferenceManager, ReferenceProfile
from .validation import SequenceValidator
[docs]
@dataclass(slots=True)
class AnalysisConfig:
"""Configure validation, windows, simulation, and reproducibility settings."""
validation: ValidationConfig = field(default_factory=ValidationConfig)
sliding_window: int = 30
sliding_step: int = 1
simulate_cai: int = 0
random_seed: int = 1
strict: bool = True
[docs]
class CodonAnalyzer:
"""Coordinate codon-usage and adaptation analyses through one interface."""
def __init__(
self,
config: AnalysisConfig | None = None,
*,
reference: ReferenceProfile | None = None,
hosts: dict[str, ReferenceProfile] | None = None,
) -> None:
"""Initialize an analyzer with optional target and host references."""
self.config = config or AnalysisConfig()
self.validator = SequenceValidator(self.config.validation)
self.code = self.validator.genetic_code
self.usage = CodonUsageMetrics(self.code)
self.composition = CompositionMetrics()
self.adaptation = AdaptationMetrics(self.code)
self.pairs = PairMetrics(self.code)
self.evolutionary = EvolutionaryMetrics()
self.host_comparator = HostComparator(self.code)
self.reference = reference
self.hosts = hosts or {}
for profile in [reference, *self.hosts.values()]:
if profile and profile.genetic_code != self.code.table_id:
raise AnalysisError(
f"Reference {profile.name!r} uses genetic code {profile.genetic_code}; "
f"analyzer uses {self.code.table_id}."
)
[docs]
def analyze(self, record: SequenceRecord) -> AnalysisResult:
"""Analyze one sequence and return a structured result."""
validation = self.validator.validate(record, raise_on_error=self.config.strict)
sequence = validation.normalized_sequence
frame = self.config.validation.reading_frame
framed = sequence[frame:]
all_codons = [framed[index : index + 3] for index in range(0, len(framed) - 2, 3)]
codons = [codon for codon in all_codons if self.code.translate(codon) not in {None, "*"}]
if not codons:
raise AnalysisError(f"Sequence {record.identifier!r} has no analyzable sense codons.")
coding_sequence = "".join(codons)
composition = self.composition.gc_metrics(coding_sequence, genetic_code=self.code)
codon_counts = self.usage.counts(codons)
rscu = self.usage.rscu(codons)
pair_scores = self.pairs.scores(codons)
metrics: dict[str, Any] = {
"length_nt": len(sequence),
"sense_codons": len(codons),
"enc": self.usage.enc(codons),
"icdi": self.usage.icdi(codons),
"codon_pair_bias": self.pairs.bias(codons),
"coding_length_nt": len(coding_sequence),
**composition,
**self.evolutionary.parity_rule_2(
composition["a3"], composition["t3"], composition["g3"], composition["c3"]
),
}
sliding_windows: dict[str, list[dict[str, float]]] = {}
if self.reference:
weights = self.adaptation.weights_from_counts(self.reference.counts)
optimal = self.usage.optimal_codons(self.reference.counts)
reference_total = sum(self.reference.counts.values())
reference_frequencies = {
codon: count / reference_total if reference_total else 0.0
for codon, count in self.reference.counts.items()
}
metrics.update(
{
"cai": self.adaptation.cai(codons, weights),
"rcdi": self.adaptation.rcdi(codons, reference_frequencies),
"fop": self.usage.fop(codons, optimal),
"cbi": self.usage.cbi(codons, optimal),
}
)
sliding_windows["cai"] = self.adaptation.sliding_cai(
codons, weights, window=self.config.sliding_window, step=self.config.sliding_step
)
if self.config.simulate_cai:
metrics["simulated_cai"] = self.adaptation.simulated_cai(
codons, weights, simulations=self.config.simulate_cai, seed=self.config.random_seed
)
host_results = self.host_comparator.compare_many(
codons, {name: profile.counts for name, profile in self.hosts.items()}
)
dinucleotide_odds = self.composition.dinucleotide_odds(coding_sequence)
metrics["cpg_representation"] = dinucleotide_odds["CG"]
metrics["upa_representation"] = dinucleotide_odds["TA"]
return AnalysisResult(
identifier=record.identifier,
validation=validation,
metrics=metrics,
codon_counts=codon_counts,
rscu=rscu,
amino_acid_composition=self.usage.amino_acid_composition(codons),
dinucleotide_frequencies=self.composition.oligonucleotide_frequencies(coding_sequence, 2),
dinucleotide_odds=dinucleotide_odds,
dinucleotide_bias=self.composition.classify_dinucleotide_odds(dinucleotide_odds),
trinucleotide_frequencies=self.composition.oligonucleotide_frequencies(coding_sequence, 3),
codon_pair_scores=pair_scores,
sliding_windows=sliding_windows,
host_comparisons=host_results,
metadata=self._metadata(record, sequence),
)
[docs]
def analyze_many(self, records: list[SequenceRecord]) -> list[AnalysisResult]:
"""Analyze records in deterministic input order."""
identifiers = [record.identifier for record in records]
duplicates = sorted({identifier for identifier in identifiers if identifiers.count(identifier) > 1})
if duplicates:
raise AnalysisError(f"Duplicate identifiers: {', '.join(duplicates)}")
return [self.analyze(record) for record in records]
[docs]
def analyze_cds_collection(
self,
identifier: str,
records: list[SequenceRecord],
*,
metadata: dict[str, Any] | None = None,
) -> AnalysisResult:
"""Analyze pooled coding regions while preserving every CDS boundary.
Each member is validated and framed independently. Terminal stops are
excluded from pooled codon and composition metrics, and sequence-word
plus codon-pair calculations never span adjacent CDSs.
"""
if not records:
raise AnalysisError("CDS aggregation requires at least one sequence record.")
segment_codons: list[list[str]] = []
segment_sequences: list[str] = []
member_checksums: dict[str, str] = {}
for record in records:
report = self.validator.validate(record, raise_on_error=self.config.strict)
framed = report.normalized_sequence[self.config.validation.reading_frame :]
raw_codons = [framed[index : index + 3] for index in range(0, len(framed) - 2, 3)]
codons = [codon for codon in raw_codons if self.code.translate(codon) not in {None, "*"}]
if not codons:
raise AnalysisError(f"CDS {record.identifier!r} has no analyzable sense codons.")
segment_codons.append(codons)
segment_sequences.append("".join(codons))
member_checksums[record.identifier] = hashlib.sha256(report.normalized_sequence.encode()).hexdigest()
terminal_stop = sorted(self.code.stop_codons)[0]
aggregate_record = SequenceRecord(
identifier=identifier,
sequence="".join(segment_sequences) + terminal_stop,
description=f"Boundary-aware aggregate of {len(records)} CDSs",
metadata=metadata or {},
)
result = self.analyze(aggregate_record)
pair_scores = self.pairs.scores_many(segment_codons)
dinucleotide_odds = self.composition.dinucleotide_odds_many(segment_sequences)
result.codon_pair_scores = pair_scores
result.metrics["codon_pair_bias"] = self.pairs.bias_many(segment_codons, pair_scores)
result.dinucleotide_frequencies = self.composition.oligonucleotide_frequencies_many(segment_sequences, 2)
result.dinucleotide_odds = dinucleotide_odds
result.dinucleotide_bias = self.composition.classify_dinucleotide_odds(dinucleotide_odds)
result.trinucleotide_frequencies = self.composition.oligonucleotide_frequencies_many(segment_sequences, 3)
result.metrics["cpg_representation"] = dinucleotide_odds["CG"]
result.metrics["upa_representation"] = dinucleotide_odds["TA"]
result.metrics["length_nt"] = sum(len("".join(record.sequence.split())) for record in records)
result.metrics["coding_length_nt"] = sum(len(sequence) for sequence in segment_sequences)
result.sliding_windows = {}
result.metadata.update(
{
"analysis_scope": "genome_wide_cds",
"cds_count": len(records),
"member_identifiers": [record.identifier for record in records],
"member_sha256": member_checksums,
"boundary_policy": "reading frames reset and terminal stops removed per CDS",
}
)
return result
[docs]
def analyze_file(self, source: str | Path, *, fmt: str | None = None) -> list[AnalysisResult]:
"""Parse and analyze every record in a supported input file."""
return self.analyze_many(SequenceParser().parse(source, fmt=fmt))
def _metadata(self, record: SequenceRecord, sequence: str) -> dict[str, Any]:
"""Record analysis provenance required for reproducible reporting."""
return {
**record.metadata,
"description": record.description,
"software_version": __version__,
"python_version": platform.python_version(),
"analysis_date_utc": datetime.now(timezone.utc).isoformat(),
"input_sha256": hashlib.sha256(sequence.encode()).hexdigest(),
"genetic_code": self.code.table_id,
"reading_frame": self.config.validation.reading_frame,
"random_seed": self.config.random_seed,
"reference": self.reference.name if self.reference else None,
"reference_version": self.reference.version if self.reference else None,
}
[docs]
def analyzer_from_reference_files(
reference_path: str | Path | None = None,
host_paths: list[str | Path] | None = None,
*,
config: AnalysisConfig | None = None,
) -> CodonAnalyzer:
"""Build an analyzer by loading reference profiles from disk."""
manager = ReferenceManager()
reference = manager.load(reference_path) if reference_path else None
hosts = {}
for path in host_paths or []:
profile = manager.load(path)
hosts[profile.name] = profile
return CodonAnalyzer(config, reference=reference, hosts=hosts)