diff --git a/src/raven_python/io/__init__.py b/src/raven_python/io/__init__.py index bc70511..1fc937a 100644 --- a/src/raven_python/io/__init__.py +++ b/src/raven_python/io/__init__.py @@ -1,12 +1,15 @@ -"""RAVEN-specific I/O: YAML (cobra + Metabolic Atlas / Human-GEM extensions), SIF, -Excel export, and the Standard-GEM ``model//…`` git layout. +"""RAVEN-specific I/O: YAML (cobra + Metabolic Atlas / Human-GEM extensions, plus +the GECKO ec-model substructure), SIF, Excel export, and the Standard-GEM +``model//…`` git layout. """ +from raven_python.io.ec_data import EcData from raven_python.io.excel import export_to_excel from raven_python.io.git import export_for_git from raven_python.io.sif import export_model_to_sif from raven_python.io.yaml import read_yaml_model, write_yaml_model __all__ = [ + "EcData", "export_for_git", "export_model_to_sif", "export_to_excel", diff --git a/src/raven_python/io/ec_data.py b/src/raven_python/io/ec_data.py new file mode 100644 index 0000000..850c8e8 --- /dev/null +++ b/src/raven_python/io/ec_data.py @@ -0,0 +1,383 @@ +"""Typed enzyme-constrained ecModel substructure (``model.ec``). + +Aligned with MATLAB GECKO's ``model.ec`` struct. Holds the per-reaction and +per-enzyme arrays that the GECKO toolbox attaches to a metabolic model when +making it enzyme-constrained: kcat values, enzyme molecular weights, the +sparse reaction-to-enzyme coupling matrix, and assorted bookkeeping +(provenance source tags, free-text notes, EC numbers, sequences). + +This module owns: + +- the ``EcData`` dataclass (in-memory shape), +- the YAML schema for the ``ec-rxns`` / ``ec-enzymes`` / ``gecko_light`` + top-level sections, +- the (de)serialisation helpers `ec_data_from_yaml_sections` and + `ec_data_to_yaml_sections`. + +It does NOT touch the cobra-shaped portion of the document — that stays +with ``raven_python.io.yaml``, which calls into here when ec sections are +present. The split mirrors RAVEN MATLAB: ``readYAMLmodel.m`` populates +``model.ec`` whenever the YAML defines it; downstream consumers +(geckopy / GECKO) operate on the populated struct. + +YAML schema (one entry per row): + +:: + + ec-rxns: + - id: R1_EXP_1 + kcat: 12.5 # turnover number, 1/s (0 == "no kcat assigned") + source: brenda # optional, omitted if empty + notes: free-text # optional, omitted if empty + eccodes: "1.1.1.1" # optional; scalar OR a list when multiple + enzymes: # column -> stoichiometric subunit count + P12345: 1.0 + P67890: 2.0 # heteromeric complex with 2 copies of P67890 + ec-enzymes: + - genes: G1 # gene name as it appears in cobra + enzymes: P12345 # uniprot accession (or KEGG id) + mw: 50000.0 # Da; omitted when unknown + sequence: MAGIC # protein sequence; omitted when empty + concs: 0.005 # proteomics-measured concentration mg/gDCW; + # omitted when not measured + gecko_light: false # top-level bool; defaults to false on load +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Any + +import numpy as np +from scipy import sparse + + +@dataclass +class EcData: + """Typed enzyme-constrained ecModel substructure attached as ``model.ec``. + + Field semantics match MATLAB GECKO's ``model.ec`` struct one-to-one. + Two parallel index spaces: + + - per-reaction arrays (``rxns``, ``kcat``, ``source``, ``notes``, + ``eccodes``) of length ``n_rxns``; + - per-enzyme arrays (``genes``, ``enzymes``, ``mw``, ``sequence``, + ``concs``) of length ``n_enzymes``. + + Connected by the sparse ``rxn_enz_mat`` of shape ``(n_rxns, n_enzymes)`` + whose ``[i, j]`` entry is the subunit count of enzyme j in reaction i + (typically 0 or 1; >1 for heteromeric complexes). + + Sentinels (mirror MATLAB GECKO): + + - ``kcat == 0`` means "no kcat assigned" (zero is the unset state; + real turnover numbers are always positive). + - ``mw == nan`` means "MW unknown" (the writer omits NaN mw entries). + - ``concs == nan`` means "not measured" (the writer omits NaN concs). + - empty strings in ``source`` / ``notes`` / ``eccodes`` / ``sequence`` + are omitted on write and restored as ``""`` on load. + + ``gecko_light`` marks the gecko-light layout: cobra reactions stay + singular, ec.rxns carries one entry per isozyme distinguished by a + ``###_`` counter prefix, and per-enzyme ``prot_`` / usage reactions + are skipped in favour of the shared protein pool. ``False`` is the + default (full layout, where ``ec.rxns`` matches cobra reactions + one-to-one after isozyme expansion). + """ + gecko_light: bool = False + rxns: list[str] = field(default_factory=list) + kcat: np.ndarray = field(default_factory=lambda: np.empty(0, dtype=float)) + source: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + eccodes: list[str] = field(default_factory=list) + genes: list[str] = field(default_factory=list) + enzymes: list[str] = field(default_factory=list) + mw: np.ndarray = field(default_factory=lambda: np.empty(0, dtype=float)) + sequence: list[str] = field(default_factory=list) + concs: np.ndarray = field(default_factory=lambda: np.empty(0, dtype=float)) + rxn_enz_mat: sparse.csr_matrix = field( + default_factory=lambda: sparse.csr_matrix((0, 0), dtype=float), + ) + + @property + def n_rxns(self) -> int: + return len(self.rxns) + + @property + def n_enzymes(self) -> int: + return len(self.enzymes) + + def validate(self) -> None: + """Raise ``ValueError`` if internal field lengths are inconsistent. + + Cheap sanity check: catches accidental drift between the per-rxn + arrays, the per-enzyme arrays, and the coupling matrix shape. + Called by pipeline stages after they mutate the data, and by the + YAML loader after construction. + """ + n_r, n_e = self.n_rxns, self.n_enzymes + + rxn_lengths = { + "kcat": len(self.kcat), + "source": len(self.source), + "notes": len(self.notes), + "eccodes": len(self.eccodes), + } + for name, length in rxn_lengths.items(): + if length != n_r: + raise ValueError( + f"ec.{name} has length {length}, expected {n_r} " + f"(matching ec.rxns)" + ) + + # `ec.enzymes` itself is the reference length; check the remaining + # per-enzyme arrays against it. + enz_lengths = { + "genes": len(self.genes), + "mw": len(self.mw), + "sequence": len(self.sequence), + "concs": len(self.concs), + } + for name, length in enz_lengths.items(): + if length != n_e: + raise ValueError( + f"ec.{name} has length {length}, expected {n_e} " + f"(matching ec.enzymes)" + ) + + if self.rxn_enz_mat.shape != (n_r, n_e): + raise ValueError( + f"ec.rxn_enz_mat has shape {self.rxn_enz_mat.shape}, " + f"expected ({n_r}, {n_e})" + ) + + @staticmethod + def empty(n_rxns: int, n_enzymes: int = 0, *, + gecko_light: bool = False) -> EcData: + """Preallocate an ``EcData`` with the canonical sentinel values. + + Per-rxn fields get empty strings; per-enzyme fields get empty + strings and NaN arrays. ``kcat`` starts at 0 (0 marks "no kcat + assigned"). ``mw`` and ``concs`` start at NaN, since their + physical default is "unknown" rather than zero. + + Used by makeEcModel-style builders that allocate the structure + up-front, then fill it row by row. + """ + return EcData( + gecko_light=gecko_light, + rxns=[""] * n_rxns, + kcat=np.zeros(n_rxns, dtype=float), + source=[""] * n_rxns, + notes=[""] * n_rxns, + eccodes=[""] * n_rxns, + genes=[""] * n_enzymes, + enzymes=[""] * n_enzymes, + mw=np.full(n_enzymes, np.nan, dtype=float), + sequence=[""] * n_enzymes, + concs=np.full(n_enzymes, np.nan, dtype=float), + rxn_enz_mat=sparse.lil_matrix( + (n_rxns, n_enzymes), dtype=float, + ).tocsr(), + ) + + +# --------------------------------------------------------------------------- # +# Load +# --------------------------------------------------------------------------- # + +def ec_data_from_yaml_sections(sections: dict) -> EcData | None: + """Build an ``EcData`` from the ec-* top-level YAML sections. + + Returns ``None`` when ``ec-rxns`` and ``ec-enzymes`` are both absent — + the caller treats that as "this YAML is not an ec-model" and leaves + ``model.ec = None``. If exactly one of the two is present, the YAML + is malformed: raise ValueError. + + ``sections`` is the dict of foreign top-level keys captured by the YAML + loader. ``gecko_light`` defaults to ``False`` when the key is absent. + """ + has_rxns = "ec-rxns" in sections + has_enzymes = "ec-enzymes" in sections + if not has_rxns and not has_enzymes: + return None + if has_rxns != has_enzymes: + missing = "ec-enzymes" if has_rxns else "ec-rxns" + raise ValueError( + f"ecModel YAML is missing the `{missing}` top-level section; " + "both ec-rxns and ec-enzymes are required." + ) + + return _build_ec_data( + sections["ec-rxns"], + sections["ec-enzymes"], + gecko_light=bool(sections.get("gecko_light", False)), + ) + + +def _build_ec_data( + ec_rxns_raw: list, + ec_enzymes_raw: list, + *, + gecko_light: bool, +) -> EcData: + """Construct an ``EcData`` from the parsed YAML lists. + + Missing optional fields are filled with sentinels (NaN for mw/concs, + empty string for source/notes/eccodes/sequence, 0.0 for kcat). + Validates that every enzyme referenced from an ec-rxns row exists in + ec-enzymes; raises ValueError otherwise (catches a common authoring + bug where the two sections drifted out of sync). + """ + n_e = len(ec_enzymes_raw) + genes = [str(e["genes"]) for e in ec_enzymes_raw] + enzymes = [str(e["enzymes"]) for e in ec_enzymes_raw] + mw = np.array( + [float(e.get("mw", np.nan)) for e in ec_enzymes_raw], dtype=float, + ) + sequence = [str(e.get("sequence", "")) for e in ec_enzymes_raw] + concs = np.array( + [float(e.get("concs", np.nan)) for e in ec_enzymes_raw], dtype=float, + ) + + enz_index = {eid: i for i, eid in enumerate(enzymes)} + + n_r = len(ec_rxns_raw) + rxns = [str(r["id"]) for r in ec_rxns_raw] + # 0 == "no kcat assigned"; real turnover numbers are always positive. + kcat = np.array( + [float(r.get("kcat", 0.0)) for r in ec_rxns_raw], dtype=float, + ) + source = [str(r.get("source", "")) for r in ec_rxns_raw] + notes = [str(r.get("notes", "")) for r in ec_rxns_raw] + eccodes = [_canonicalize_eccodes(r.get("eccodes", "")) for r in ec_rxns_raw] + + mat = sparse.lil_matrix((n_r, n_e), dtype=float) + for i, r in enumerate(ec_rxns_raw): + for enz_id, stoich in (r.get("enzymes") or {}).items(): + j = enz_index.get(str(enz_id)) + if j is None: + raise ValueError( + f"ec-rxns[{i}] (id={r.get('id')!r}) references enzyme " + f"{enz_id!r} that is not present in ec-enzymes." + ) + mat[i, j] = float(stoich) + + return EcData( + gecko_light=gecko_light, + rxns=rxns, + kcat=kcat, + source=source, + notes=notes, + eccodes=eccodes, + genes=genes, + enzymes=enzymes, + mw=mw, + sequence=sequence, + concs=concs, + rxn_enz_mat=mat.tocsr(), + ) + + +# --------------------------------------------------------------------------- # +# Save +# --------------------------------------------------------------------------- # + +def ec_data_to_yaml_sections(ec: EcData) -> dict[str, Any]: + """Serialise an ``EcData`` to a dict suitable for YAML emission. + + Returns a fresh dict with three keys: ``gecko_light`` (bool), + ``ec-rxns`` (list of mappings), ``ec-enzymes`` (list of mappings). + Values are native Python primitives — no numpy/ruamel scalars — so + the YAML writer can dump them directly without further coercion. + + Empty optional fields are omitted to keep the file compact; the + loader fills them back in. + """ + return { + "gecko_light": bool(ec.gecko_light), + "ec-rxns": _build_ec_rxns_list(ec), + "ec-enzymes": _build_ec_enzymes_list(ec), + } + + +def _build_ec_rxns_list(ec: EcData) -> list[dict[str, Any]]: + """Translate per-rxn ec fields + ``rxn_enz_mat`` rows to the + list-of-mappings YAML form. + + Empty ``source`` / ``notes`` / ``eccodes`` strings are omitted. + ``kcat`` is always written: a real turnover number when set, + otherwise ``0`` (0 marks "no kcat assigned"). + """ + coo = ec.rxn_enz_mat.tocoo() + per_row_enzymes: list[dict[str, float]] = [{} for _ in range(ec.n_rxns)] + # All three arrays come from the same COO matrix, so they're guaranteed + # equal length; strict=True turns any future drift into a loud TypeError + # instead of silent truncation. + for i, j, v in zip(coo.row, coo.col, coo.data, strict=True): + per_row_enzymes[int(i)][ec.enzymes[int(j)]] = float(v) + + out: list[dict[str, Any]] = [] + for i in range(ec.n_rxns): + entry: dict[str, Any] = { + "id": ec.rxns[i], + "kcat": float(ec.kcat[i]), + } + if ec.source[i]: + entry["source"] = ec.source[i] + if ec.notes[i]: + entry["notes"] = ec.notes[i] + if ec.eccodes[i]: + entry["eccodes"] = _eccodes_to_yaml(ec.eccodes[i]) + entry["enzymes"] = per_row_enzymes[i] + out.append(entry) + return out + + +def _build_ec_enzymes_list(ec: EcData) -> list[dict[str, Any]]: + """Translate per-enzyme ec fields to the list-of-mappings YAML form. + + NaN ``mw`` / ``concs`` and empty ``sequence`` are omitted; the loader + restores them as NaN / empty string. + """ + out: list[dict[str, Any]] = [] + for j in range(ec.n_enzymes): + entry: dict[str, Any] = { + "genes": ec.genes[j], + "enzymes": ec.enzymes[j], + } + if not math.isnan(ec.mw[j]): + entry["mw"] = float(ec.mw[j]) + if ec.sequence[j]: + entry["sequence"] = ec.sequence[j] + if not math.isnan(ec.concs[j]): + entry["concs"] = float(ec.concs[j]) + out.append(entry) + return out + + +# --------------------------------------------------------------------------- # +# eccodes representation helpers +# --------------------------------------------------------------------------- # + +def _canonicalize_eccodes(value) -> str: + """Coerce an EC-codes field to a single `;`-joined string. + + The schema accepts either a scalar string (`"1.1.1.1"`) or a list of + strings (`["1.1.1.1", "1.1.99.40"]`); both round-trip to the same + internal representation. + """ + if value is None: + return "" + if isinstance(value, str): + return value + return ";".join(str(v) for v in value) + + +def _eccodes_to_yaml(eccodes: str): + """Convert the internal `;`-joined eccodes string back to the YAML form: + a scalar string for one EC, a list for multiple.""" + parts = [p for p in eccodes.split(";") if p] + if len(parts) <= 1: + return eccodes + return parts diff --git a/src/raven_python/io/yaml.py b/src/raven_python/io/yaml.py index e3da8ac..f24c642 100644 --- a/src/raven_python/io/yaml.py +++ b/src/raven_python/io/yaml.py @@ -16,15 +16,31 @@ ``confidence_score``, ``references``, ``rxnFrom``, ``deltaG`` and ``notes`` (rxnNotes) on reactions; ``protein`` on genes. These are stashed in the cobra object's ``.notes`` dict on read and lifted back to top-level keys on write. -* **Model-level extras** cobra ignores: ``version``, the ``metaData`` provenance - block, and the GECKO sections (``gecko_light``/``ec-rxns``/``ec-enzymes``), - preserved on ``model.notes`` for round-tripping. - -The reader also accepts the older RAVEN files (id/name nested in ``metaData``). +* **Model-level extras** cobra ignores: ``version`` and the ``metaData`` + provenance block, preserved on ``model.notes`` for round-tripping. +* **The GECKO ec sections** (``ec-rxns``/``ec-enzymes``/``gecko_light``): on + read, parsed into a typed :class:`~raven_python.io.ec_data.EcData` and + attached as ``model.ec``; on write, serialised back to top-level sections + whenever ``model.ec`` is present. This mirrors RAVEN ``readYAMLmodel.m`` / + ``writeYAMLmodel.m``, which populate the ``model.ec`` struct when the YAML + defines it. Any other unknown top-level keys are still preserved opaquely + via ``model.notes['_yaml_sections']``. + +Legacy quirks the reader also accepts (silent normalisation): + +* older RAVEN files with ``id`` / ``name`` nested in ``metaData``; +* per-metabolite top-level ``smiles`` (lifted into ``annotation['smiles']``); +* very old RAVEN files written as a bare ``-`` sequence of single-key mappings + rather than one big mapping; +* MATLAB GECKO ecModels whose ``usage_prot_*`` and ``prot_pool_exchange`` + reactions use the older reverse-direction convention (negative lower bound, + swapped stoichiometry signs); these are flipped on load so consumers always + see the forward convention. """ from __future__ import annotations import gzip +import warnings from collections import OrderedDict from pathlib import Path @@ -32,6 +48,12 @@ from cobra.io.dict import model_from_dict, model_to_dict from cobra.io.yaml import yaml as _cobra_yaml # ruamel round-trip YAML (handles !!omap) +from raven_python.io.ec_data import ( + EcData, + ec_data_from_yaml_sections, + ec_data_to_yaml_sections, +) + def _open_text(path: str | Path, mode: str): """Open ``path`` as a text handle, transparently gzipping when it ends ``.gz``.""" @@ -54,6 +76,10 @@ def _open_text(path: str | Path, mode: str): _COBRA_TOP_KEYS = frozenset({"metabolites", "reactions", "genes", "compartments", "id", "name"}) +# Top-level keys consumed by the typed EcData layer; not re-emitted as opaque +# `_yaml_sections` on round-trip (the source of truth is ``model.ec``). +_EC_TOP_KEYS = frozenset({"ec-rxns", "ec-enzymes", "gecko_light"}) + def _to_plain(obj): if isinstance(obj, dict): @@ -90,33 +116,65 @@ def read_yaml_model(path: str | Path) -> cobra.Model: Convenience wrapper around :func:`model_from_yaml_data` that opens the file (transparently un-gzipping ``.gz``) and parses the YAML. Callers - that need to pre-process the document (e.g. lift legacy fields that - cobra doesn't recognise) can read+normalise themselves and call - :func:`model_from_yaml_data` with the resulting dict. + that need to pre-process the document (e.g. lift further non-standard + fields that cobra doesn't recognise) can read+normalise themselves and + call :func:`model_from_yaml_data` with the resulting dict. + + Accepts both the cobra `!!omap` shape and a very old RAVEN shape where + the document root is a bare ``-`` sequence of single-key mappings; the + latter is merged into one mapping before parsing. """ with _open_text(path, "r") as handle: raw = _to_plain(_cobra_yaml.load(handle)) + if isinstance(raw, list): + # Very old RAVEN files: a sequence of one-key mappings instead of + # one big !!omap. Merge into a single dict before parsing. + merged: dict = {} + for item in raw: + if isinstance(item, dict): + merged.update(item) + raw = merged + if not isinstance(raw, dict): - raise ValueError(f"{path}: top-level YAML is a {type(raw).__name__}, not a mapping.") + raise ValueError( + f"{path}: top-level YAML must be a mapping or a sequence of " + f"single-key mappings, got {type(raw).__name__}." + ) return model_from_yaml_data(raw) def model_from_yaml_data(raw: dict) -> cobra.Model: """Build a ``cobra.Model`` from an already-parsed RAVEN/cobrapy YAML dict. - Strips and restores the RAVEN per-entry side-fields onto each entry's - ``.notes``, lifts ``id``/``name`` out of legacy ``metaData``, and - stashes everything else (``version``, the ``metaData`` block itself, - and any unknown top-level keys such as ``ec-rxns``/``ec-enzymes``) - onto ``model.notes`` so a round-trip via :func:`write_yaml_model` - preserves them. ``raw`` is mutated in place — copy it first if the - caller needs the original. + Performs three jobs in order: + + 1. **cobra-shaped portion:** strips and restores RAVEN-only per-entry + side-fields onto each entry's ``.notes``; lifts ``id`` / ``name`` + out of legacy ``metaData``; preserves ``version`` and ``metaData`` + on ``model.notes`` for round-trip. + 2. **legacy quirks:** lifts per-metabolite top-level ``smiles`` into + ``annotation['smiles']`` (older MATLAB GECKO ecModels emitted it + at the top level); flips the older reverse-direction + ``usage_prot_*`` / ``prot_pool_exchange`` convention to the + forward convention. + 3. **GECKO ec sections:** when ``ec-rxns`` / ``ec-enzymes`` are + present, parses them into a typed :class:`EcData` and attaches + it as ``model.ec``. Other unknown top-level keys land opaquely on + ``model.notes['_yaml_sections']`` for round-trip. + + ``raw`` is mutated in place — copy it first if the caller needs the + original. """ metadata = raw.pop("metaData", None) or {} version = raw.pop("version", None) foreign = {k: raw.pop(k) for k in list(raw) if k not in _COBRA_TOP_KEYS} + # Legacy quirk: per-metabolite top-level `smiles` -> annotation.smiles. + # Done before model_from_dict so cobra sees the annotation in its + # canonical place. No-op on current files. + _lift_smiles_to_annotation(raw.get("metabolites")) + met_notes = _capture_entry_fields(raw.get("metabolites", []), _MET_FIELDS) rxn_notes = _capture_entry_fields(raw.get("reactions", []), _RXN_FIELDS) gene_notes = _capture_entry_fields(raw.get("genes", []), _GENE_FIELDS) @@ -130,6 +188,10 @@ def model_from_yaml_data(raw: dict) -> cobra.Model: for gene, notes in zip(model.genes, gene_notes, strict=False): gene.notes = notes + # Legacy quirk: flip reverse-direction usage_prot_* / prot_pool_exchange. + # No-op when none match. + _flip_legacy_prot_direction(model) + # Legacy files keep id/name inside metaData; restore them if cobra found none. if metadata.get("id") and not model.id: model.id = metadata["id"] @@ -139,12 +201,87 @@ def model_from_yaml_data(raw: dict) -> cobra.Model: model.notes["metaData"] = metadata if version is not None: model.notes["version"] = version + + # Pop the ec sections out of `foreign` and into a typed EcData. + # The remaining unknown keys round-trip opaquely. + ec_sections = {k: foreign.pop(k) for k in list(foreign) if k in _EC_TOP_KEYS} + ec_data = ec_data_from_yaml_sections(ec_sections) + if ec_data is not None: + model.ec = ec_data if foreign: model.notes["_yaml_sections"] = foreign return model +# --------------------------------------------------------------------------- # +# Legacy quirk normalisers +# --------------------------------------------------------------------------- # + +def _lift_smiles_to_annotation(metabolites) -> None: + """Move per-metabolite top-level ``smiles`` into ``annotation['smiles']``. + + Older MATLAB GECKO ecModel writers placed SMILES at the metabolite top + level; the cobra/raven convention is to nest them inside ``annotation``. + Normalises in place; no-op when no metabolite carries a top-level + ``smiles`` key. + """ + if not isinstance(metabolites, list): + return + for met in metabolites: + if not (isinstance(met, dict) and "smiles" in met): + continue + smiles = met.pop("smiles") + annotation = met.get("annotation") + if not isinstance(annotation, dict): + annotation = {} + met["annotation"] = annotation + if "smiles" not in annotation and smiles: + annotation["smiles"] = ( + smiles if isinstance(smiles, list) else [smiles] + ) + + +def _flip_legacy_prot_direction(model: cobra.Model) -> None: + """Flip pre-forward-direction protein reactions in place. + + Older MATLAB GECKO ecModels defined ``usage_prot_*`` and + ``prot_pool_exchange`` as "reverse" reactions: their flux was + negative, and the stoichiometry signs were correspondingly swapped. + The current convention (in both geckopy and recent MATLAB GECKO) + treats them as ordinary forward reactions with positive flux. When a + loaded model still uses the older convention we flip the affected + reactions in place so consumers never have to handle two shapes. + + The signature we look for is any ``usage_prot_*`` or + ``prot_pool_exchange`` reaction whose lower bound is negative. + """ + flipped: list[str] = [] + for rxn in model.reactions: + if not ( + rxn.id.startswith("usage_prot_") + or rxn.id == "prot_pool_exchange" + ): + continue + if rxn.lower_bound >= -1e-9: + continue + rxn.add_metabolites( + {met: -2.0 * coef for met, coef in rxn.metabolites.items()}, + combine=True, + ) + lb, ub = rxn.lower_bound, rxn.upper_bound + rxn.lower_bound = -ub + rxn.upper_bound = -lb + flipped.append(rxn.id) + if flipped: + warnings.warn( + f"ecModel uses the older reverse-direction convention for " + f"{len(flipped)} protein usage/pool reaction(s); flipping to " + "the current forward convention.", + stacklevel=3, + ) + + def _emit_entry_fields(entries, fields): """Lift RAVEN-only keys out of each entry's ``notes`` dict to top level.""" for entry in entries: @@ -169,8 +306,14 @@ def write_yaml_model( ) -> None: """Write a ``cobra.Model`` to RAVEN/cobrapy (``!!omap``) YAML. - With ``sort_ids=True`` metabolites/reactions/genes/compartments are written - in alphabetical order (diff-friendly), without modifying ``model``. + When ``model.ec`` is a populated :class:`EcData`, the ``gecko_light`` + flag and the ``ec-rxns`` / ``ec-enzymes`` top-level sections are + emitted from it (numpy/ruamel scalars are coerced to plain Python + primitives en route, so the dumper never sees them). + + With ``sort_ids=True`` metabolites/reactions/genes/compartments are + written in alphabetical order (diff-friendly), without modifying + ``model``. """ model_notes = dict(model.notes or {}) stored_meta = model_notes.pop("metaData", None) or {} @@ -201,6 +344,18 @@ def write_yaml_model( _emit_entry_fields(doc.get("reactions", []), _RXN_FIELDS) _emit_entry_fields(doc.get("genes", []), _GENE_FIELDS) + # ec sections come from the typed model.ec (when present), not from the + # opaque foreign-keys stash. Drop any stale ec-* entries in `foreign` so + # they can't conflict with the EcData-derived ones. + for ec_key in _EC_TOP_KEYS: + foreign.pop(ec_key, None) + ec = getattr(model, "ec", None) + ec_sections = ( + _to_plain(ec_data_to_yaml_sections(ec)) + if isinstance(ec, EcData) and (ec.n_rxns or ec.n_enzymes) + else None + ) + # cobra dict order is metabolites, reactions, genes, id, name, compartments; # append version / gecko_light / metaData / ec-* like RAVEN's writer. if version is not None: @@ -210,11 +365,13 @@ def write_yaml_model( metadata.setdefault("id", model.id) if model.name: metadata.setdefault("name", model.name) - for key in ("gecko_light",): - if key in foreign: - doc[key] = foreign.pop(key) + if ec_sections is not None: + doc["gecko_light"] = ec_sections["gecko_light"] if metadata: doc["metaData"] = metadata + if ec_sections is not None: + doc["ec-rxns"] = ec_sections["ec-rxns"] + doc["ec-enzymes"] = ec_sections["ec-enzymes"] for key, value in foreign.items(): doc[key] = value diff --git a/tests/test_io_yaml.py b/tests/test_io_yaml.py index 510af5f..c406857 100644 --- a/tests/test_io_yaml.py +++ b/tests/test_io_yaml.py @@ -9,7 +9,8 @@ # A model laid out exactly as RAVEN writeYAMLmodel (fa281a1) emits: cobra-native # structure, RAVEN-only fields as top-level per-entry keys, smiles/ec-code inside -# the annotation block, metaData provenance-only, id/name/version top-level. +# the annotation block, metaData provenance-only, id/name/version top-level, +# plus the GECKO ec-* sections that populate `model.ec`. RAVEN_DOC = { "metabolites": [ { @@ -51,7 +52,13 @@ "compartments": {"c": "cytoplasm"}, "version": "1.0", "metaData": {"date": "2026-05-23", "taxonomy": "taxonomy/559292", "defaultLB": "-1000"}, - "ec-rxns": [{"id": "R1", "kcat": 100.0}], + "gecko_light": False, + "ec-rxns": [ + {"id": "R1", "kcat": 100.0, "source": "brenda", "enzymes": {"P12345": 1.0}}, + ], + "ec-enzymes": [ + {"genes": "G1", "enzymes": "P12345", "mw": 50000.0, "sequence": "MAGIC"}, + ], } @@ -101,10 +108,13 @@ def test_raven_only_fields_captured(yaml_file): def test_model_level_extras(yaml_file): + """metaData / version round-trip via model.notes; the ec-* sections + are consumed into a typed model.ec, not stashed in _yaml_sections.""" model = read_yaml_model(yaml_file) assert model.notes["metaData"]["taxonomy"] == "taxonomy/559292" assert model.notes["version"] == "1.0" - assert model.notes["_yaml_sections"]["ec-rxns"][0]["kcat"] == 100.0 + # ec sections live on model.ec, not in _yaml_sections. + assert "_yaml_sections" not in model.notes def test_round_trip(yaml_file, tmp_path): @@ -123,7 +133,12 @@ def test_round_trip(yaml_file, tmp_path): r = reloaded.reactions.get_by_id("R1") assert r.notes["confidence_score"] == 2 assert reloaded.genes.get_by_id("G1").notes["protein"] == "P12345" - assert reloaded.notes["_yaml_sections"]["ec-rxns"][0]["id"] == "R1" + # ec round-trip: same kcat, same coupling entry. + assert reloaded.ec.rxns == ["R1"] + assert reloaded.ec.kcat[0] == 100.0 + assert reloaded.ec.source[0] == "brenda" + assert reloaded.ec.enzymes == ["P12345"] + assert reloaded.ec.mw[0] == 50000.0 def test_extra_notes_not_dropped_when_free_text_note_present(yaml_file, tmp_path): diff --git a/tests/test_io_yaml_ec_data.py b/tests/test_io_yaml_ec_data.py new file mode 100644 index 0000000..94c2ebb --- /dev/null +++ b/tests/test_io_yaml_ec_data.py @@ -0,0 +1,413 @@ +"""Tests for raven_python.io.yaml's GECKO ec-model support. + +Covers: +- model.ec populated from `ec-rxns` / `ec-enzymes` / `gecko_light` sections; +- model.ec serialised back to those sections (overwriting any stale + `_yaml_sections` entries); +- numpy/sparse coercion happens at the boundary (writer must not see + numpy scalars); +- legacy quirks: top-level smiles, reverse-direction usage_prot_*, bare-`-` + document root; +- malformed inputs: half a pair of ec-* sections, dangling enzyme reference. +""" +from __future__ import annotations + +from pathlib import Path + +import cobra +import numpy as np +import pytest +from cobra.io.yaml import yaml as cobra_yaml +from scipy import sparse + +from raven_python.io import EcData, read_yaml_model, write_yaml_model +from raven_python.io.yaml import model_from_yaml_data + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # + +def _minimal_model_doc() -> dict: + """A cobra-shaped doc with two mets, one rxn, one gene — minimum needed + for cobra to parse a model. Tests add the ec-* sections as needed.""" + return { + "metabolites": [ + {"id": "a", "name": "A", "compartment": "c"}, + {"id": "b", "name": "B", "compartment": "c"}, + ], + "reactions": [ + { + "id": "R1", + "metabolites": {"a": -1, "b": 1}, + "lower_bound": 0.0, + "upper_bound": 1000.0, + "gene_reaction_rule": "G1", + }, + ], + "genes": [{"id": "G1", "name": "g one"}], + "id": "m", + "compartments": {"c": "cytoplasm"}, + } + + +def _write_yaml(doc: dict, path: Path) -> Path: + with open(path, "w", encoding="utf-8") as fh: + cobra_yaml.dump(doc, fh) + return path + + +# --------------------------------------------------------------------------- # +# Load: ec sections populate model.ec +# --------------------------------------------------------------------------- # + +def test_load_populates_ec_data(tmp_path): + doc = _minimal_model_doc() + doc["ec-rxns"] = [ + {"id": "R1", "kcat": 12.5, "source": "manual", "enzymes": {"P1": 1.0}}, + ] + doc["ec-enzymes"] = [ + {"genes": "G1", "enzymes": "P1", "mw": 30000.0, "sequence": "MAGIC"}, + ] + doc["gecko_light"] = False + model = read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + + assert isinstance(model.ec, EcData) + assert model.ec.gecko_light is False + assert model.ec.rxns == ["R1"] + assert model.ec.kcat[0] == 12.5 + assert model.ec.source[0] == "manual" + assert model.ec.enzymes == ["P1"] + assert model.ec.mw[0] == 30000.0 + assert model.ec.rxn_enz_mat.shape == (1, 1) + assert model.ec.rxn_enz_mat[0, 0] == 1.0 + + +def test_load_missing_optional_fields_get_sentinels(tmp_path): + """Optional fields omitted from a row come back as the right sentinels + on the loaded EcData.""" + doc = _minimal_model_doc() + doc["ec-rxns"] = [{"id": "R1", "kcat": 0.0, "enzymes": {"P1": 1.0}}] + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1"}] # no mw, sequence, concs + model = read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + + assert model.ec.source == [""] + assert model.ec.notes == [""] + assert model.ec.eccodes == [""] + assert np.isnan(model.ec.mw[0]) + assert model.ec.sequence == [""] + assert np.isnan(model.ec.concs[0]) + + +def test_load_gecko_light_flag(tmp_path): + doc = _minimal_model_doc() + doc["ec-rxns"] = [ + {"id": "001_R1", "kcat": 5.0, "enzymes": {"P1": 1.0}}, + ] + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1", "mw": 100.0}] + doc["gecko_light"] = True + model = read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + assert model.ec.gecko_light is True + + +def test_load_without_ec_sections_leaves_model_without_ec(tmp_path): + """Non-ecmodel YAML loads as a plain cobra.Model — no model.ec.""" + model = read_yaml_model(_write_yaml(_minimal_model_doc(), tmp_path / "m.yml")) + assert not hasattr(model, "ec") or model.ec is None or model.ec == EcData() + # The attribute is simply not set in the no-ec case. + assert "ec" not in vars(model) + + +def test_load_eccodes_scalar_or_list_both_round_trip(tmp_path): + """The schema accepts a scalar string OR a list of strings for `eccodes`; + both flavours land in the same `;`-joined internal form.""" + doc = _minimal_model_doc() + doc["ec-rxns"] = [ + {"id": "R1", "kcat": 1.0, "eccodes": "1.1.1.1", "enzymes": {"P1": 1.0}}, + ] + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1", "mw": 1.0}] + m1 = read_yaml_model(_write_yaml(doc, tmp_path / "m1.yml")) + assert m1.ec.eccodes == ["1.1.1.1"] + + doc["ec-rxns"][0]["eccodes"] = ["1.1.1.1", "1.1.99.40"] + m2 = read_yaml_model(_write_yaml(doc, tmp_path / "m2.yml")) + assert m2.ec.eccodes == ["1.1.1.1;1.1.99.40"] + + +# --------------------------------------------------------------------------- # +# Save: model.ec serialised back to top-level sections +# --------------------------------------------------------------------------- # + +def _make_ec_model() -> cobra.Model: + """A minimal cobra.Model with a populated model.ec attached by hand.""" + model = cobra.Model("m") + a = cobra.Metabolite("a", compartment="c") + b = cobra.Metabolite("b", compartment="c") + model.add_metabolites([a, b]) + r = cobra.Reaction("R1", lower_bound=0.0, upper_bound=1000.0) + r.add_metabolites({a: -1, b: 1}) + r.gene_reaction_rule = "G1" + model.add_reactions([r]) + + mat = sparse.lil_matrix((1, 1), dtype=float) + mat[0, 0] = 1.0 + model.ec = EcData( + gecko_light=False, + rxns=["R1"], + kcat=np.array([42.0]), + source=["manual"], + notes=[""], + eccodes=["1.1.1.1"], + genes=["G1"], + enzymes=["P1"], + mw=np.array([12345.0]), + sequence=["MAGIC"], + concs=np.array([np.nan]), + rxn_enz_mat=mat.tocsr(), + ) + return model + + +def test_save_emits_ec_sections(tmp_path): + out = tmp_path / "out.yml" + write_yaml_model(_make_ec_model(), out) + text = out.read_text() + assert "ec-rxns:" in text + assert "ec-enzymes:" in text + assert "gecko_light:" in text + assert "42" in text and "12345" in text and "MAGIC" in text + + +def test_save_round_trip_preserves_all_ec_fields(tmp_path): + out = tmp_path / "out.yml" + write_yaml_model(_make_ec_model(), out) + reloaded = read_yaml_model(out) + assert reloaded.ec.rxns == ["R1"] + assert reloaded.ec.kcat[0] == 42.0 + assert reloaded.ec.source == ["manual"] + assert reloaded.ec.eccodes == ["1.1.1.1"] + assert reloaded.ec.enzymes == ["P1"] + assert reloaded.ec.mw[0] == 12345.0 + assert reloaded.ec.sequence == ["MAGIC"] + assert np.isnan(reloaded.ec.concs[0]) # NaN omitted on write, restored on load + assert reloaded.ec.rxn_enz_mat[0, 0] == 1.0 + + +def test_save_skips_nan_and_empty_optional_fields(tmp_path): + """NaN mw/concs and empty source/notes/eccodes/sequence get omitted + from the YAML to keep files compact.""" + model = _make_ec_model() + model.ec.source = [""] + model.ec.notes = [""] + model.ec.eccodes = [""] + model.ec.sequence = [""] + model.ec.mw = np.array([np.nan]) + model.ec.concs = np.array([np.nan]) + + out = tmp_path / "out.yml" + write_yaml_model(model, out) + text = out.read_text() + assert "source:" not in text + assert "notes:" not in text + assert "eccodes:" not in text + assert "sequence:" not in text + # kcat is always written (even 0); mw / concs omitted when NaN. + assert "kcat:" in text + + +def test_save_coerces_numpy_scalars(tmp_path): + """ec arrays hold numpy types; the writer must coerce so the YAML + dumper never sees an np.float64 or np.int64.""" + model = _make_ec_model() + model.ec.kcat = np.array([np.float32(7.5)]) + model.ec.mw = np.array([np.int64(20000)]) + out = tmp_path / "out.yml" + write_yaml_model(model, out) # must not raise + reloaded = read_yaml_model(out) + assert reloaded.ec.kcat[0] == pytest.approx(7.5) + assert reloaded.ec.mw[0] == 20000.0 + + +def test_save_without_ec_omits_sections(tmp_path): + model = cobra.Model("m") + a = cobra.Metabolite("a", compartment="c") + model.add_metabolites([a]) + out = tmp_path / "out.yml" + write_yaml_model(model, out) + text = out.read_text() + assert "ec-rxns:" not in text + assert "ec-enzymes:" not in text + assert "gecko_light:" not in text + + +def test_save_overrides_stale_yaml_sections_for_ec_keys(tmp_path): + """If a loaded model carried stale ec-* in _yaml_sections AND also has + a populated model.ec (shouldn't happen via the normal load path, + but a caller could construct it), the writer uses model.ec and + drops the stale stash so the file isn't ambiguous.""" + model = _make_ec_model() + model.notes["_yaml_sections"] = { + "ec-rxns": [{"id": "STALE", "kcat": 999.0}], + "ec-enzymes": [{"genes": "GHOST", "enzymes": "PGHOST"}], + } + out = tmp_path / "out.yml" + write_yaml_model(model, out) + reloaded = read_yaml_model(out) + assert reloaded.ec.rxns == ["R1"] # not "STALE" + assert reloaded.ec.enzymes == ["P1"] # not "PGHOST" + + +# --------------------------------------------------------------------------- # +# Legacy quirks +# --------------------------------------------------------------------------- # + +def test_legacy_top_level_smiles_lifted_to_annotation(tmp_path): + """Old MATLAB GECKO ecModels put SMILES at the metabolite top level + rather than inside annotation. The loader normalises both flavours.""" + doc = _minimal_model_doc() + doc["metabolites"][0]["smiles"] = "CC(=O)O" + model = read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + a = model.metabolites.get_by_id("a") + assert a.annotation["smiles"] == ["CC(=O)O"] + # Should NOT survive as a stray notes key. + assert "smiles" not in a.notes + + +def test_legacy_reverse_direction_prot_flipped(tmp_path): + """`usage_prot_*` reactions with a negative lower bound and swapped + stoichiometry get flipped to the forward convention on load.""" + doc = _minimal_model_doc() + doc["metabolites"].append({"id": "prot_P1", "name": "prot", "compartment": "c"}) + doc["metabolites"].append({"id": "prot_pool", "name": "pool", "compartment": "c"}) + doc["reactions"].append( + { + "id": "usage_prot_P1", + "metabolites": {"prot_P1": -1, "prot_pool": 1}, # swapped signs + "lower_bound": -1000.0, + "upper_bound": 0.0, + } + ) + with pytest.warns(UserWarning, match="reverse-direction"): + model = read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + r = model.reactions.get_by_id("usage_prot_P1") + assert r.lower_bound == 0.0 + assert r.upper_bound == 1000.0 + # Stoichiometry signs flipped. + coefs = {m.id: c for m, c in r.metabolites.items()} + assert coefs["prot_P1"] == 1.0 + assert coefs["prot_pool"] == -1.0 + + +def test_legacy_bare_sequence_root_merged(tmp_path): + """Very old RAVEN files were written as a bare `-` sequence of + single-key mappings; reader merges to one dict.""" + legacy_text = ( + "- metabolites:\n" + " - id: a\n" + " name: A\n" + " compartment: c\n" + "- reactions: []\n" + "- genes: []\n" + "- id: legacy_m\n" + "- compartments:\n" + " c: cyt\n" + ) + p = tmp_path / "bare.yml" + p.write_text(legacy_text) + model = read_yaml_model(p) + assert model.id == "legacy_m" + assert {m.id for m in model.metabolites} == {"a"} + + +# --------------------------------------------------------------------------- # +# Error paths +# --------------------------------------------------------------------------- # + +def test_load_rxns_only_without_enzymes_raises(tmp_path): + doc = _minimal_model_doc() + doc["ec-rxns"] = [{"id": "R1", "kcat": 1.0}] + with pytest.raises(ValueError, match="ec-enzymes"): + read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + + +def test_load_enzymes_only_without_rxns_raises(tmp_path): + doc = _minimal_model_doc() + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1"}] + with pytest.raises(ValueError, match="ec-rxns"): + read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + + +def test_load_dangling_enzyme_reference_raises(tmp_path): + """An ec-rxns row whose enzymes mapping references an accession not + listed in ec-enzymes is a hard error — catches the common authoring + bug where the two sections drifted apart.""" + doc = _minimal_model_doc() + doc["ec-rxns"] = [ + {"id": "R1", "kcat": 1.0, "enzymes": {"PGHOST": 1.0}}, + ] + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1", "mw": 1.0}] + with pytest.raises(ValueError, match="PGHOST"): + read_yaml_model(_write_yaml(doc, tmp_path / "m.yml")) + + +# --------------------------------------------------------------------------- # +# In-memory model_from_yaml_data (no file I/O) +# --------------------------------------------------------------------------- # + +def test_model_from_yaml_data_mutates_in_place(): + """`model_from_yaml_data` pops sections off its dict input. Documented + behaviour — verify callers passing a fresh dict see it drained.""" + doc = _minimal_model_doc() + doc["ec-rxns"] = [{"id": "R1", "kcat": 1.0, "enzymes": {"P1": 1.0}}] + doc["ec-enzymes"] = [{"genes": "G1", "enzymes": "P1", "mw": 1.0}] + doc["gecko_light"] = False + model_from_yaml_data(doc) + assert "ec-rxns" not in doc + assert "ec-enzymes" not in doc + assert "gecko_light" not in doc + + +# --------------------------------------------------------------------------- # +# EcData.validate / EcData.empty +# --------------------------------------------------------------------------- # + +def test_empty_has_canonical_sentinels(): + """`EcData.empty(n, m)` preallocates with the documented sentinels.""" + ec = EcData.empty(3, 2) + assert ec.n_rxns == 3 + assert ec.n_enzymes == 2 + assert ec.rxns == ["", "", ""] + assert (ec.kcat == 0).all() + assert np.isnan(ec.mw).all() + assert np.isnan(ec.concs).all() + assert ec.rxn_enz_mat.shape == (3, 2) + assert ec.rxn_enz_mat.nnz == 0 + + +def test_empty_round_trips_through_validate(): + EcData.empty(5, 4).validate() # must not raise + + +def test_validate_catches_per_rxn_length_drift(): + ec = EcData.empty(3, 2) + ec.kcat = np.array([1.0, 2.0]) # length 2, should be 3 + with pytest.raises(ValueError, match="ec.kcat has length 2, expected 3"): + ec.validate() + + +def test_validate_catches_per_enzyme_length_drift(): + ec = EcData.empty(3, 2) + ec.mw = np.array([1.0]) # length 1, should be 2 + with pytest.raises(ValueError, match="ec.mw has length 1, expected 2"): + ec.validate() + + +def test_validate_catches_coupling_matrix_shape_drift(): + ec = EcData.empty(3, 2) + ec.rxn_enz_mat = sparse.csr_matrix((3, 5), dtype=float) + with pytest.raises(ValueError, match=r"ec.rxn_enz_mat has shape \(3, 5\)"): + ec.validate() + + +def test_empty_gecko_light_flag_propagates(): + assert EcData.empty(1, 1, gecko_light=True).gecko_light is True + assert EcData.empty(1, 1).gecko_light is False