"""
CodonAdaptPy Reference-Data Management
This module creates, validates, loads, saves, versions, and catalogs codon
reference profiles. A reference may be derived from validated CDS records or
loaded from a JSON/CSV/TSV table. The package intentionally labels its bundled
uniform profile as synthetic; organism-specific biological profiles must carry
source and version metadata supplied by the user or distributor.
Classes:
- ReferenceProfile: Named codon counts with provenance metadata.
- ReferenceManager: Registry and persistence interface for profiles.
:Created: July 20, 2026
:Updated: July 20, 2026
:Author: Naveen Duhan
:Version: 1.0.2
"""
from __future__ import annotations
import csv
import hashlib
import json
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from .exceptions import ReferenceDataError
from .genetic_code import GeneticCode
from .models import SequenceRecord, ValidationConfig
from .validation import SequenceValidator
[docs]
@dataclass(slots=True)
class ReferenceProfile:
"""Represent versioned codon counts and their scientific provenance."""
name: str
counts: dict[str, float]
genetic_code: int = 1
version: str = "1"
source: str = "user-supplied"
description: str = ""
metadata: dict[str, Any] = field(default_factory=dict)
[docs]
def to_dict(self) -> dict[str, Any]:
"""Return profile content in JSON-compatible form."""
return asdict(self)
[docs]
class ReferenceManager:
"""Manage named codon profiles and reproducible reference persistence."""
def __init__(self) -> None:
"""Initialize a registry containing an explicitly synthetic baseline."""
code = GeneticCode.from_ncbi(1)
self._profiles: dict[str, ReferenceProfile] = {
"uniform_standard": ReferenceProfile(
name="uniform_standard",
counts={codon: 1.0 for codon, amino_acid in code.forward_table.items() if amino_acid != "*"},
source="CodonAdaptPy synthetic baseline",
description="Non-biological equal-count profile for testing and normalization.",
metadata={"biological": False},
)
}
[docs]
def register(self, profile: ReferenceProfile, *, replace: bool = False) -> None:
"""Validate and register a profile under its unique name."""
if profile.name in self._profiles and not replace:
existing = self._profiles[profile.name]
if existing.to_dict() == profile.to_dict():
return
raise ReferenceDataError(f"Reference profile already exists: {profile.name}")
code = GeneticCode.from_ncbi(profile.genetic_code)
sense = {codon for codon, amino_acid in code.forward_table.items() if amino_acid != "*"}
invalid = set(profile.counts) - set(code.forward_table)
if invalid:
raise ReferenceDataError(f"Reference contains unsupported codons: {', '.join(sorted(invalid))}")
if any(float(value) < 0 for value in profile.counts.values()):
raise ReferenceDataError("Reference counts must be non-negative.")
profile.counts = {codon: float(profile.counts.get(codon, 0.0)) for codon in sorted(sense)}
if not sum(profile.counts.values()):
raise ReferenceDataError("Reference contains no positive sense-codon counts.")
self._profiles[profile.name] = profile
[docs]
def from_sequences(
self,
name: str,
records: list[SequenceRecord],
*,
genetic_code: int = 1,
version: str = "1",
source: str = "user CDS set",
) -> ReferenceProfile:
"""Build a reference by pooling strictly validated coding sequences."""
validator = SequenceValidator(ValidationConfig(genetic_code=genetic_code))
code = validator.genetic_code
counts = {codon: 0.0 for codon, amino_acid in code.forward_table.items() if amino_acid != "*"}
checksums: list[str] = []
for record in records:
report = validator.validate(record, raise_on_error=True)
framed = report.normalized_sequence[validator.config.reading_frame :]
for index in range(0, len(framed) - 2, 3):
codon = framed[index : index + 3]
if codon in counts:
counts[codon] += 1
checksums.append(hashlib.sha256(report.normalized_sequence.encode()).hexdigest())
profile = ReferenceProfile(
name,
counts,
genetic_code,
version,
source,
metadata={"sequence_count": len(records), "sequence_sha256": checksums},
)
self.register(profile)
return profile
[docs]
def get(self, name: str) -> ReferenceProfile:
"""Return a registered profile or raise a domain-specific error."""
try:
return self._profiles[name]
except KeyError as exc:
raise ReferenceDataError(f"Unknown reference profile: {name}") from exc
[docs]
def list(self) -> list[dict[str, Any]]:
"""Return catalog metadata without duplicating full count tables."""
return [
{
"name": profile.name,
"version": profile.version,
"source": profile.source,
"genetic_code": profile.genetic_code,
"description": profile.description,
}
for profile in self._profiles.values()
]
[docs]
def save(self, name: str, path: str | Path) -> Path:
"""Serialize a registered profile to deterministic JSON."""
destination = Path(path)
destination.write_text(json.dumps(self.get(name).to_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8")
return destination
[docs]
def load(self, path: str | Path, *, replace: bool = False) -> ReferenceProfile:
"""Load JSON or two-column CSV/TSV reference data and register it."""
source_path = Path(path)
if source_path.suffix.lower() == ".json":
payload = json.loads(source_path.read_text(encoding="utf-8"))
profile = ReferenceProfile(**payload)
else:
delimiter = "," if source_path.suffix.lower() == ".csv" else "\t"
with source_path.open(encoding="utf-8", newline="") as handle:
rows = list(csv.DictReader(handle, delimiter=delimiter))
counts = {
row["codon"].upper().replace("U", "T"): float(row.get("count", row.get("frequency", 0))) for row in rows
}
profile = ReferenceProfile(source_path.stem, counts, source=str(source_path))
self.register(profile, replace=replace)
return profile