dee/core/identify.py

"""What is this thing I just pasted? — answered locally first, and honestly.

The audit calls this "the single most common first move a scientist makes"
(#5, High). Someone has a sequence from a collaborator, a supplementary table
or a sequencing run, and the first question is not "improve it" but "what am I
even looking at?".

THE DISTINCTION THIS MODULE IS BUILT AROUND
-------------------------------------------
**Characterizing is not identifying.** Everything here is computed from the
residues themselves against catalogues already in this repo: it can say "this
is 1,341 bp of DNA, one clean reading frame, carrying a T7 promoter and a
C-terminal 6xHis, and its 3' junction matches pET-28a(+)". That is a real,
checkable, useful answer and it costs nothing.

What it can NOT say is "this is human MC1R". Naming a sequence requires
comparing it to every known sequence, which means BLAST — and BLAST is the one
operation in this engine that sends the user's actual residues outside the
Space. So `identified` is **always False** here, and the result names the tool
that would change that. An agent that says "this is insulin" because it found
a His-tag is exactly the failure this module exists to prevent.

That ordering is also the privacy-correct one. Local analysis answers most of
the question for free; BLAST is offered as a deliberate, consented step rather
than being the reflex.

REUSE, DON'T REINVENT
---------------------
The audit's single most common finding is capability that already exists in
the repo with nothing pointing at it. So this module computes almost nothing
of its own:

* `sequence.classify_sequence`  — the three-way DNA/protein/ambiguous call
* `plasmid.find_motifs`         — the curated motif catalogue (promoters,
                                  terminators, tags, loxP/FRT/att sites)
* `plasmid.find_orfs`           — linear-time ORF scan on both strands
* `edits.is_coding` / `translate` — the strict CDS test and the codon table
* `accession.classify`          — is the paste actually an identifier?
* `vectors`                     — the 61-backbone catalogue and its junctions

The protein tag table is **derived by translating the DNA motifs already in
`plasmid._MOTIFS`** rather than typed out here. One source of truth, and no
sequence in this file was written from recall.
"""
from __future__ import annotations

import re
from typing import Any, Dict, List, Optional

from dee.core import accession as _acc
from dee.core import edits as _edits
from dee.core import plasmid as _plasmid
from dee.core import sequence as _seq
from dee.core import vectors as _vectors

# Below this there is not enough sequence to say anything, and a 6-mer matches
# something by chance in every database on earth.
MIN_LEN = 12
# Shortest ORF worth reporting. Below ~30 aa an "ORF" in random DNA is noise:
# a stop codon appears roughly every 21 codons by chance.
MIN_ORF_AA = 30
# A vector junction shorter than this is usually just a restriction site, which
# is shared by hundreds of backbones and proves nothing. Most of the 61
# catalogue entries record only a 6-bp site, so only a minority can ever
# produce a junction call — which is the honest outcome, not a shortfall.
MIN_JUNCTION = 16
# Distinct residues below which a "protein" looks synthetic or mis-pasted.
LOW_COMPLEXITY_BELOW = 8
LOW_COMPLEXITY_MIN_LEN = 60

_FASTA_HEADER = re.compile(r"^\s*>([^\n\r]*)", re.M)
_GENBANK = re.compile(r"^\s*LOCUS\s+\S", re.M)


def _tag_peptides() -> List[Dict[str, str]]:
    """Protein-level tags, derived from the repo's own DNA motif catalogue.

    Translating `plasmid._MOTIFS` keeps a single source of truth and means no
    peptide here was typed from memory — if the DNA catalogue is corrected,
    this follows automatically.
    """
    out: List[Dict[str, str]] = []
    seen = set()
    for name, motif, ftype in _plasmid._MOTIFS:
        if ftype != "tag" or len(motif) % 3:
            continue
        pep = _edits.translate(motif)
        if not pep or "X" in pep or "*" in pep:
            continue
        if pep in seen:
            continue
        seen.add(pep)
        out.append({"name": name, "peptide": pep})
    return out


_TAGS = _tag_peptides()


def _clean_residues(text: str) -> str:
    """Strip FASTA headers, digits and whitespace; keep letters and stops."""
    body = re.sub(r"^\s*>[^\n]*$", "", text or "", flags=re.M)
    return re.sub(r"[^A-Za-z*]", "", body).upper()


def _read_paste(text: str) -> Dict[str, Any]:
    """Work out what SHAPE the paste is before worrying what it contains."""
    raw = (text or "").strip()
    if not raw:
        return {"format": "empty", "header": None, "residues": ""}
    if _GENBANK.search(raw):
        return {"format": "genbank", "header": None,
                "residues": _clean_residues(raw)}
    m = _FASTA_HEADER.search(raw)
    if m:
        return {"format": "fasta", "header": m.group(1).strip()[:200],
                "residues": _clean_residues(raw)}
    # A bare token with no residue-ish bulk may be an identifier, not a sequence.
    token = raw.split()[0] if raw.split() else ""
    if len(raw.split()) == 1 and _acc.looks_like_accession(token):
        return {"format": "accession", "header": None, "residues": "",
                "token": token}
    return {"format": "raw", "header": None, "residues": _clean_residues(raw)}


def _header_accession(header: str) -> Optional[Dict[str, str]]:
    """An accession quoted in a FASTA header is a CLAIM, not evidence."""
    for token in re.split(r"[|\s,;]+", header or ""):
        token = token.strip()
        if len(token) >= 4 and _acc.looks_like_accession(token):
            family, molecule = _acc.classify(token)
            return {"accession": token, "family": family, "molecule": molecule}
    return None


def _vector_junctions(dna: str) -> List[Dict[str, Any]]:
    """Catalogue backbones whose cloning junction appears in this sequence.

    Hits are grouped by end, because a junction identifies a FAMILY, not a
    vector: the whole pET series shares one C-terminal His junction, and the
    catalogue happens to record it at different lengths for different entries.
    A longer recorded junction is a longer catalogue string, not better
    evidence — naming the longest match alone would invent a distinction the
    sequence does not support.
    """
    found: Dict[str, List[Dict[str, Any]]] = {}
    try:
        cat = _vectors._load()
    except Exception:                                    # pragma: no cover
        return []
    up = dna.upper()
    for rec in cat.values():
        for field in ("flanking_5p_max", "flanking_3p_max"):
            junction = (rec.get(field) or "").upper()
            if len(junction) < MIN_JUNCTION:
                continue
            for strand, needle in ((1, junction), (-1, _plasmid.revcomp(junction))):
                idx = up.find(needle)
                if idx < 0:
                    continue
                end = "5'" if field.startswith("flanking_5p") else "3'"
                found.setdefault(end, []).append({
                    "vector": rec.get("name") or rec.get("id"),
                    "start": idx, "strand": strand, "matched_bp": len(junction),
                })
                break

    out: List[Dict[str, Any]] = []
    for end, hits in sorted(found.items()):
        hits.sort(key=lambda h: (-h["matched_bp"], str(h["vector"])))
        names = [h["vector"] for h in hits]
        out.append({
            "end": end,
            "vectors": names,
            "start": min(h["start"] for h in hits),
            "longest_match_bp": max(h["matched_bp"] for h in hits),
            "distinguishing": len(names) == 1,
            "note": None if len(names) == 1 else (
                f"{len(names)} catalogue backbones share this {end} junction. "
                f"The match says which FAMILY the construct came from, not "
                f"which vector — do not pick one."),
        })
    return out


def _analyse_dna(dna: str) -> Dict[str, Any]:
    ambiguous = sum(1 for c in dna if c not in "ACGT")
    out: Dict[str, Any] = {
        "gc_percent": _plasmid.gc_percent(dna),
        "ambiguity_codes": ambiguous,
        "multiple_of_three": len(dna) % 3 == 0,
        "is_clean_cds": _edits.is_coding(dna),
    }

    orfs = _plasmid.find_orfs(dna, min_aa=MIN_ORF_AA, max_orfs=200)
    if orfs:
        longest = max(orfs, key=lambda o: o["end"] - o["start"])
        span = longest["end"] - longest["start"]
        out["orfs"] = len(orfs)
        out["longest_orf"] = {
            "aa": span // 3 - 1,               # the terminal stop is not a residue
            "start": longest["start"], "end": longest["end"],
            "strand": longest["strand"],
            "covers_percent": round(100.0 * span / len(dna), 1),
        }
    else:
        out["orfs"] = 0
        out["longest_orf"] = None

    out["features"] = [
        {"name": f["name"], "type": f["type"], "start": f["start"],
         "end": f["end"], "strand": f["strand"]}
        for f in _plasmid.find_motifs(dna)
    ]
    out["vector_junctions"] = _vector_junctions(dna)
    return out


def _analyse_protein(protein: str) -> Dict[str, Any]:
    body = protein.rstrip("*")
    distinct = len(set(body))
    out: Dict[str, Any] = {
        "starts_with_methionine": body.startswith("M"),
        "internal_stop": "*" in body,
        "unknown_residues": body.count("X"),
        "distinct_residues": distinct,
    }
    out["tags"] = [
        {"name": t["name"], "peptide": t["peptide"], "start": body.find(t["peptide"])}
        for t in _TAGS if t["peptide"] in body
    ]
    # A real protein uses most of the alphabet. Very low diversity means either
    # a synthetic construct (poly-A linker, His run) or DNA pasted by mistake.
    out["low_complexity"] = (len(body) >= LOW_COMPLEXITY_MIN_LEN
                             and distinct < LOW_COMPLEXITY_BELOW)
    return out


def _summary(molecule: str, length: int, detail: Dict[str, Any]) -> str:
    """One line a bench scientist would actually say out loud."""
    if molecule == "protein":
        bits = [f"{length} aa protein"]
        if detail.get("tags"):
            bits.append(", ".join(t["name"] for t in detail["tags"]))
        # Worth surfacing unprompted: a premature stop means the translation
        # is truncated, and every downstream number would be about a protein
        # the cell never makes.
        if detail.get("internal_stop"):
            bits.append("contains an internal stop — this translation is truncated")
        if detail.get("unknown_residues"):
            bits.append(f"{detail['unknown_residues']} unknown (X) residues")
        if detail.get("low_complexity"):
            bits.append("unusually low residue diversity")
        return " · ".join(bits)
    unit = "nt RNA" if molecule == "rna" else "bp DNA"
    bits = [f"{length} {unit}", f"{detail.get('gc_percent')}% GC"]
    if detail.get("is_clean_cds"):
        bits.append("a clean CDS end to end")
    elif detail.get("longest_orf"):
        orf = detail["longest_orf"]
        bits.append(f"longest ORF {orf['aa']} aa ({orf['covers_percent']}% of it)")
    else:
        bits.append("no ORF of 30 aa or more")
    named = sorted({f["name"] for f in detail.get("features") or []})
    if named:
        bits.append(", ".join(named[:4]) + ("…" if len(named) > 4 else ""))
    for j in (detail.get("vector_junctions") or [])[:1]:
        names = j["vectors"]
        bits.append(f"{j['end']} junction matches {names[0]}" if j["distinguishing"]
                    else (f"{j['end']} junction shared by {len(names)} catalogue "
                          f"backbones ({names[0]}, {names[1]}…)"))
    return " · ".join(bits)


def identify(text: str) -> Dict[str, Any]:
    """Characterize a pasted sequence locally. Never claims to have named it."""
    paste = _read_paste(text)

    if paste["format"] == "empty":
        return {"ok": False, "kind": "empty", "error": "Nothing to identify.",
                "next": "Paste a sequence, or an accession to fetch one."}

    if paste["format"] == "accession":
        token = paste["token"]
        family, molecule = _acc.classify(token)
        return {
            "ok": True, "kind": "accession", "identified": False,
            "input_format": "accession", "accession": token,
            "family": family, "molecule": molecule,
            "summary": (f"'{token}' is a {family} identifier"
                        + (f" for {molecule}" if molecule else "")
                        + ", not a sequence."),
            "next": (f"Call fetch_sequence with '{token}' to retrieve it. "
                     f"An accession names the sequence already — there is "
                     f"nothing to identify."),
        }

    residues = paste["residues"]
    if len(residues) < MIN_LEN:
        return {"ok": False, "kind": "too_short",
                "error": f"Only {len(residues)} residues; {MIN_LEN} is the "
                         f"minimum worth analysing.",
                "next": "Short oligos are better handled by design_primers."}

    molecule = _seq.classify_sequence(residues)
    if molecule == "dna" and "U" in residues and "T" not in residues:
        molecule = "rna"

    if molecule == "ambiguous":
        return {
            "ok": True, "kind": "ambiguous", "identified": False,
            "input_format": paste["format"], "molecule": "ambiguous",
            "length": len(residues),
            "summary": (f"{len(residues)} residues of A/C/G/T/U/N only — at "
                        f"this length that reads equally well as a short "
                        f"peptide or a DNA fragment."),
            "next": ("Ask the user which it is before analysing further. "
                     "Guessing here silently misroutes the whole job."),
        }

    detail = (_analyse_protein(residues) if molecule == "protein"
              else _analyse_dna(residues))

    claim = _header_accession(paste["header"]) if paste.get("header") else None
    result: Dict[str, Any] = {
        "ok": True,
        "kind": "characterized",
        "input_format": paste["format"],
        "molecule": molecule,
        "length": len(residues),
        "summary": _summary(molecule, len(residues), detail),
        "detail": detail,
        # The whole point. Local analysis describes; it does not name.
        "identified": False,
        "to_identify_it": (
            "Nothing above names this sequence — it is computed from the "
            "residues against local catalogues. To find out WHAT it is, call "
            "blast_sequence, which searches NCBI. That sends the sequence "
            "outside this Space and the user is asked to approve it first."),
        "caveat": (
            "Motif and junction matches are exact string matches against a "
            "curated list, so a hit is real but an absence means only 'not in "
            "the list' — the catalogue is small and deliberately conservative."),
    }

    if paste.get("header"):
        result["header_claim"] = {
            "text": paste["header"],
            "accession": claim,
            "note": ("This came from the FASTA header, which is whatever the "
                     "person who wrote the file typed. Treat it as a claim to "
                     "check, not as evidence — fetch the accession and compare "
                     "if it matters."),
        }
    return result