dee/core/expression.py

"""Why a construct expresses badly — the causes that are visible in the DNA.

WHAT IS HONEST HERE, AND WHAT IS NOT
------------------------------------
#24 is rated High because a design that does not express is a wasted month,
and nothing in the product currently looks at it. But an expression LEVEL —
"expect 40 mg/L" — is not something this engine can honestly produce. Yield
depends on the host strain, the promoter, plasmid copy number, induction
temperature, media, growth phase, and whether the protein is toxic to the
cell. None of that is in a sequence, and no model trained on it is available
here. So there is no predicted yield in this module and there is no score.

What IS in the sequence is a specific, enumerable set of causes that are known
to break translation, and those are computed exactly:

* **Rare codons**, from the engine's real usage tables in `codon.py` — the same
  tables the library encoder uses, so a codon this module calls rare and a
  codon the DE pipeline installs are the same opinion, not two.
* **Rare-codon RUNS**, which matter far more than the overall count. A
  scattered rare codon is absorbed; three in a row is where a ribosome stalls.
* **The 5' ramp**, because the first ~50 codons carry the initiation region and
  a rare codon there costs more than the same codon at residue 400.
* **Reading-frame integrity**, because a construct with an internal stop
  expresses a truncated product and every other number would describe a
  protein that is never made.

THE RARE-CODON THRESHOLD IS DERIVED, NOT CHOSEN
------------------------------------------------
Relative adaptiveness w = f(codon) / f(the most-used synonym), which is the
Sharp & Li definition. At w < 0.15 the E. coli table selects AGA, AGG, ATA and
CTA — precisely the set that Rosetta / CodonPlus strains exist to supply
(argU, ileY, leuW). The cutoff was checked against the tables rather than
picked, and the codons it selects are reported so the call can be audited.

The same cutoff selects three arginine codons in yeast and NOTHING in human —
which is a real result, not a failure. Mammalian codon usage is much flatter
and the rare-codon story is genuinely weaker there. Reporting that is more
useful than manufacturing a warning to look thorough.

HOSTS THIS HAS NO TABLE FOR
----------------------------
`codon.py` carries E. coli, S. cerevisiae and human. For anything else this
refuses and says so. That is not a capability gate on the organism — it is a
missing data file, and silently scoring a plant construct against the E. coli
table would be far worse than declining. Audit #11 (Kazusa tables) is the fix.
"""
from __future__ import annotations

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

from dee.core import codon as _codon
from dee.core import edits as _edits

# Relative adaptiveness below which a codon is called rare. Derived: see module
# docstring. The selected codons travel with every result so this is auditable.
RARE_BELOW = 0.15
# Consecutive rare codons at or above this length is a stall risk rather than
# noise. A lone rare codon is absorbed by the tRNA pool; a run depletes it.
RUN_LENGTH = 3
# The initiation region, where a rare codon costs disproportionately.
RAMP_CODONS = 50
# Below this there is not enough coding sequence for the statistics to mean
# anything.
MIN_CODONS = 30
STOPS = {"TAA", "TAG", "TGA"}


def _clean_dna(dna: str) -> str:
    return re.sub(r"[^ACGTUacgtu]", "", dna or "").upper().replace("U", "T")


def _adaptiveness(host: str) -> Dict[str, Any]:
    """codon -> (amino acid, relative adaptiveness), from the engine's tables."""
    table = _codon._resolve_table(host)
    w: Dict[str, float] = {}
    aa_of: Dict[str, str] = {}
    for aa, usage in table.items():
        top = max(usage.values()) or 1.0
        for cod, freq in usage.items():
            w[cod] = freq / top
            aa_of[cod] = aa
    return {"w": w, "aa": aa_of}


def _runs(flags: List[bool], codons: List[str]) -> List[Dict[str, Any]]:
    out: List[Dict[str, Any]] = []
    start = None
    for i, bad in enumerate(flags + [False]):
        if bad and start is None:
            start = i
        elif not bad and start is not None:
            if i - start >= RUN_LENGTH:
                out.append({"start_codon": start + 1, "end_codon": i,
                            "length": i - start,
                            "codons": codons[start:i]})
            start = None
    return out


def assess(dna: str, host: str) -> Dict[str, Any]:
    """Sequence-visible reasons a CDS may express poorly in `host`."""
    seq = _clean_dna(dna)
    if not host or not str(host).strip():
        return {"ok": False, "error": "Which host? Expression is host-specific "
                                      "— the same CDS behaves differently in "
                                      "E. coli and in human cells."}
    try:
        adapt = _adaptiveness(str(host))
    except ValueError as exc:
        return {
            "ok": False, "kind": "no_codon_table",
            "error": str(exc),
            "next": ("This is a missing data file, not an unsupported "
                     "organism. Scoring the construct against a different "
                     "host's table would produce confident nonsense, so it "
                     "declines instead. Everything else about the sequence "
                     "(check_synthesis, GC, reading frame) still works."),
        }

    if len(seq) < MIN_CODONS * 3:
        return {"ok": False, "kind": "too_short",
                "error": f"Need at least {MIN_CODONS} codons; got "
                         f"{len(seq) // 3}.",
                "next": "Paste the full coding sequence."}

    frame_ok = len(seq) % 3 == 0
    codons = [seq[i:i + 3] for i in range(0, len(seq) - len(seq) % 3, 3)]
    # The terminal stop is expected; internal ones are the finding.
    terminal_stop = bool(codons) and codons[-1] in STOPS
    body = codons[:-1] if terminal_stop else codons
    internal_stops = [i + 1 for i, c in enumerate(body) if c in STOPS]

    w_map, aa_map = adapt["w"], adapt["aa"]
    sense = [c for c in body if c not in STOPS]
    unknown = [c for c in sense if c not in w_map]
    scored = [c for c in sense if c in w_map]
    # CAI excludes the single-codon families (Met, Trp): with one option there
    # is no choice to adapt, and including them only pulls every score toward 1.
    cai_pool = [w_map[c] for c in scored if aa_map[c] not in ("M", "W")]
    cai = None
    if cai_pool:
        # Geometric mean, in log space — the direct product underflows on a
        # long CDS and silently returns 0.0.
        import math
        cai = round(math.exp(sum(math.log(max(x, 1e-6)) for x in cai_pool)
                             / len(cai_pool)), 3)

    rare_flags = [c in w_map and w_map[c] < RARE_BELOW for c in body]
    rare = [{"codon_number": i + 1, "codon": c, "amino_acid": aa_map.get(c),
             "adaptiveness": round(w_map[c], 3)}
            for i, c in enumerate(body)
            if c in w_map and w_map[c] < RARE_BELOW]
    runs = _runs(rare_flags, body)
    ramp = [r for r in rare if r["codon_number"] <= RAMP_CODONS]

    gc = round(100.0 * sum(c in "GC" for c in seq) / len(seq), 1)
    ramp_seq = seq[:RAMP_CODONS * 3]
    gc_ramp = round(100.0 * sum(c in "GC" for c in ramp_seq) / len(ramp_seq), 1)

    rare_set = sorted({c for c, x in w_map.items() if x < RARE_BELOW
                       and c not in STOPS})

    flags: List[Dict[str, str]] = []
    if not frame_ok:
        flags.append({"factor": "not a whole number of codons",
                      "observed": f"{len(seq)} nt",
                      "why": ("The sequence is not in frame as given. Every "
                              "codon-level number below is computed on the "
                              "frame starting at base 1, which may not be the "
                              "intended one.")})
    if internal_stops:
        flags.append({"factor": "internal stop codon",
                      "observed": f"codon {internal_stops[0]}"
                                  + (f" (+{len(internal_stops) - 1} more)"
                                     if len(internal_stops) > 1 else ""),
                      "why": ("Translation terminates here. The product is a "
                              "truncated fragment, so nothing downstream "
                              "describes the intended protein.")})
    if not terminal_stop:
        flags.append({"factor": "no terminal stop codon",
                      "observed": f"ends {codons[-1] if codons else '?'}",
                      "why": ("Without a stop the ribosome reads into the "
                              "vector, adding an unintended C-terminal tail.")})
    for run in runs:
        flags.append({"factor": f"run of {run['length']} consecutive rare codons",
                      "observed": (f"codons {run['start_codon']}-"
                                   f"{run['end_codon']}: "
                                   + " ".join(run["codons"])),
                      "why": ("Consecutive rare codons deplete the local "
                              "charged-tRNA pool and stall the ribosome, which "
                              "causes truncation and frameshifting. This is "
                              "the single most actionable finding here — "
                              "recoding just this run is usually enough.")})
    if ramp:
        flags.append({"factor": f"{len(ramp)} rare codon(s) in the first "
                                f"{RAMP_CODONS} codons",
                      "observed": ", ".join(f"{r['codon']}@{r['codon_number']}"
                                            for r in ramp[:6]),
                      "why": ("The 5' region sets the rate of translation "
                              "initiation, so a rare codon here costs more "
                              "than the same codon further in.")})

    return {
        "ok": True,
        "host": host,
        "codons": len(body),
        "codon_adaptation_index": cai,
        "gc_percent": gc,
        "gc_percent_5prime": gc_ramp,
        "rare_codons": rare,
        "rare_codon_count": len(rare),
        "rare_codon_runs": runs,
        "rare_in_ramp": ramp,
        "internal_stops": internal_stops,
        "has_terminal_stop": terminal_stop,
        "in_frame": frame_ok,
        "is_clean_cds": _edits.is_coding(seq),
        "unrecognised_codons": len(unknown),
        "flags": flags,
        "summary": _summary(host, len(body), cai, rare, runs, flags),
        "rare_codons_in_this_host": rare_set,
        "method": (f"Relative adaptiveness w = f(codon) / f(most-used synonym) "
                   f"(Sharp & Li), from the engine's own {host} usage table — "
                   f"the same table the library encoder writes with. Rare means "
                   f"w < {RARE_BELOW}, which in this host selects: "
                   f"{' '.join(rare_set) if rare_set else 'nothing'}. CAI is "
                   f"the geometric mean of w over sense codons, excluding Met "
                   f"and Trp, which have no synonym to choose between."),
        # The line this module exists to be able to say.
        "no_yield": ("This does NOT predict an expression level. Yield depends "
                     "on strain, promoter, copy number, induction temperature, "
                     "media and protein toxicity — none of which is in a "
                     "sequence, and no model trained on them is available "
                     "here. These are the sequence-visible causes of failure, "
                     "not a forecast."),
        "not_covered": ("Also not checked here: internal Shine-Dalgarno-like "
                        "pausing sites, mRNA secondary structure over the RBS "
                        "(audit #38/#39, which need an RNA folding engine), "
                        "and protein-level causes — for those run "
                        "assess_solubility on the translated product."),
    }


def _summary(host: str, n: int, cai: Optional[float], rare: List[Dict[str, Any]],
             runs: List[Dict[str, Any]], flags: List[Dict[str, str]]) -> str:
    bits = [f"{n} codons in {host}"]
    if cai is not None:
        bits.append(f"CAI {cai}")
    bits.append(f"{len(rare)} rare codon(s)")
    if runs:
        bits.append(f"{len(runs)} rare run(s) — codons "
                    + ", ".join(f"{r['start_codon']}-{r['end_codon']}"
                                for r in runs[:3]))
    if not flags:
        bits.append("nothing flagged")
    return " · ".join(bits)