"""
CodonAdaptPy Sequence and Metadata Parsers
This module provides batch-friendly readers for FASTA, GenBank, delimited
tables, ZIP archives, open text streams, and directly supplied nucleotide
strings. FASTA parsing is implemented in the standard library; GenBank parsing
uses Biopython when requested.
Features:
- Duplicate identifier detection across every input source
- Transparent DNA/RNA sequence preservation for later validation
- Metadata attachment from CSV and TSV files
- Safe ZIP member inspection without extracting archive contents
Classes:
- SequenceParser: Unified high-level parser for supported input formats.
:Created: July 20, 2026
:Updated: July 20, 2026
:Author: Naveen Duhan
:Version: 1.0.2
"""
from __future__ import annotations
import csv
import io
import shlex
import zipfile
from collections.abc import Iterable, Iterator
from pathlib import Path
from typing import TextIO, cast
from .exceptions import OptionalDependencyError, SequenceValidationError
from .models import SequenceRecord
[docs]
class SequenceParser:
"""Read supported sequence inputs into consistent sequence records."""
[docs]
def parse(self, source: str | Path | TextIO, *, fmt: str | None = None) -> list[SequenceRecord]:
"""Parse a path, text stream, or direct nucleotide sequence.
Format is inferred from file extensions when omitted. Plain strings
that do not identify existing paths are treated as pasted sequences
unless they begin with ``>`` and therefore represent FASTA text.
"""
if hasattr(source, "read"):
return list(self.parse_fasta(cast(TextIO, source)))
if isinstance(source, str):
text = source.strip()
if text.startswith(">"):
return list(self.parse_fasta(io.StringIO(text)))
compact = "".join(text.split()).upper()
if compact and not (set(compact) - set("ACGTURYSWKMBDHVN")):
return [SequenceRecord(identifier="sequence_1", sequence=text)]
path = Path(source)
if path.exists():
inferred = (fmt or path.suffix.lstrip(".")).lower()
if inferred in {"fa", "fna", "fas", "fasta"}:
with path.open(encoding="utf-8") as handle:
return list(self.parse_fasta(handle))
if inferred in {"gb", "gbk", "genbank"}:
return list(self.parse_genbank(path))
if inferred in {"csv", "tsv", "txt"}:
return list(self.parse_table(path))
if inferred == "zip":
return list(self.parse_zip(path))
raise ValueError(f"Unsupported sequence input format: {path.suffix or fmt}")
return [SequenceRecord(identifier="sequence_1", sequence=str(source))]
[docs]
def parse_fasta(self, handle: Iterable[str]) -> Iterator[SequenceRecord]:
"""Yield records from FASTA-formatted text with duplicate checks."""
identifier: str | None = None
description = ""
metadata: dict[str, str] = {}
sequence: list[str] = []
seen: set[str] = set()
for line_number, raw_line in enumerate(handle, start=1):
line = raw_line.strip()
if not line:
continue
if line.startswith(">"):
if identifier is not None:
yield SequenceRecord(identifier, "".join(sequence), description, metadata)
header = line[1:].strip()
if not header:
raise SequenceValidationError(f"Empty FASTA header on line {line_number}.")
fields = header.split(maxsplit=1)
identifier = fields[0]
description = fields[1] if len(fields) == 2 else ""
if identifier in seen:
raise SequenceValidationError(f"Duplicate FASTA identifier: {identifier}")
seen.add(identifier)
sequence = []
metadata = {
key: value
for token in shlex.split(description)
if "=" in token
for key, value in [token.split("=", 1)]
}
elif identifier is None:
raise SequenceValidationError(f"FASTA sequence data precede the first header on line {line_number}.")
else:
sequence.append(line)
if identifier is not None:
yield SequenceRecord(identifier, "".join(sequence), description, metadata)
[docs]
def parse_table(self, path: str | Path) -> Iterator[SequenceRecord]:
"""Yield records from CSV/TSV containing identifier and sequence columns."""
path = Path(path)
delimiter = "," if path.suffix.lower() == ".csv" else "\t"
with path.open(encoding="utf-8", newline="") as handle:
reader = csv.DictReader(handle, delimiter=delimiter)
normalized = {name.lower(): name for name in (reader.fieldnames or [])}
id_column = normalized.get("identifier") or normalized.get("id") or normalized.get("name")
sequence_column = normalized.get("sequence") or normalized.get("seq")
if not id_column or not sequence_column:
raise SequenceValidationError("Delimited input requires identifier/id/name and sequence/seq columns.")
seen: set[str] = set()
for row in reader:
identifier = (row.get(id_column) or "").strip()
if not identifier:
raise SequenceValidationError("Delimited input contains an empty identifier.")
if identifier in seen:
raise SequenceValidationError(f"Duplicate table identifier: {identifier}")
seen.add(identifier)
metadata = {key: value for key, value in row.items() if key not in {id_column, sequence_column}}
yield SequenceRecord(identifier, row.get(sequence_column) or "", metadata=metadata)
[docs]
def parse_genbank(self, path: str | Path) -> Iterator[SequenceRecord]:
"""Yield CDS features, or whole records without CDS, from GenBank input."""
try:
from Bio import SeqIO
except ImportError as exc:
raise OptionalDependencyError("GenBank parsing requires CodonAdaptPy[io].") from exc
for source in SeqIO.parse(str(path), "genbank"):
cds_features = [feature for feature in source.features if feature.type == "CDS"]
if not cds_features:
yield SequenceRecord(source.id, str(source.seq), source.description, dict(source.annotations))
continue
for index, feature in enumerate(cds_features, start=1):
qualifiers = feature.qualifiers
product = (qualifiers.get("product") or [""])[0]
protein_id = (qualifiers.get("protein_id") or [""])[0]
identifier = (
qualifiers.get("locus_tag") or qualifiers.get("protein_id") or [f"{source.id}_CDS{index}"]
)[0]
gene = (qualifiers.get("gene") or qualifiers.get("locus_tag") or [identifier])[0]
yield SequenceRecord(
identifier,
str(feature.extract(source.seq)),
product,
{
"source_record": source.id,
"isolate": source.id,
"gene": gene,
"product": product,
"protein_id": protein_id,
"feature_index": index,
"location": str(feature.location),
},
)
[docs]
def parse_zip(self, path: str | Path) -> Iterator[SequenceRecord]:
"""Yield FASTA and delimited records directly from a ZIP archive."""
seen: set[str] = set()
with zipfile.ZipFile(path) as archive:
for member in sorted(archive.namelist()):
suffix = Path(member).suffix.lower()
if member.endswith("/") or suffix not in {".fa", ".fna", ".fas", ".fasta"}:
continue
with archive.open(member) as binary:
text = io.TextIOWrapper(binary, encoding="utf-8")
for record in self.parse_fasta(text):
if record.identifier in seen:
raise SequenceValidationError(
f"Duplicate identifier across ZIP members: {record.identifier}"
)
seen.add(record.identifier)
record.metadata["archive_member"] = member
yield record