runner.py

"""Unified entry point to run any of the three dynamic community-FBA methods.

``run_dynamic`` builds and simulates the requested method from the same set of
inputs and returns the (already simulated) result object. It is used directly as
a convenience API and internally by the essentiality analysis, which re-runs a
simulation many times with different media.
"""
from __future__ import annotations

from .community import build_community_model
from .joint import DynamicJointFBA
from .parallel import DynamicParallelFBA
from .endpoint import EndPointFBA


def _as_biomass_dict(initial_biomasses, organism_ids):
    if isinstance(initial_biomasses, dict):
        return {o: float(initial_biomasses[o]) for o in organism_ids}
    return {o: float(b) for o, b in zip(organism_ids, initial_biomasses)}


def run_dynamic(method, models, biomass_ids, organism_ids=None,
                initial_biomasses=None, initial_concentrations=None,
                dt=0.1, n=1000, solver="glpk", kinetics=None,
                endpoint_n=None, epsilon=1e-6, diet=None, max_uptake=None):
    """Build + simulate a dynamic community FBA.

    Parameters
    ----------
    method : {"joint", "parallel", "endpoint"}
    models : list[cobra.Model]           individual organism GSMMs (not mutated)
    biomass_ids : list[str]              biomass reaction id in each model
    organism_ids : list[str], optional   short unique ids (default: model ids)
    initial_biomasses : dict|list        gDW/L per organism
    initial_concentrations : dict        extracellular metabolite id -> mmol/L
    dt, n : float, int                   step size / max steps (n = #steps for endpoint)
    solver : str                         "glpk" (default) or "gurobi"/...
    kinetics : Kinetics | dict           Monod kinetics (dict per organism for parallel)
    endpoint_n : int, optional           #time points for endpoint (default: n)

    Returns
    -------
    A simulated DynamicJointFBA / DynamicParallelFBA / EndPointFBA object.
    """
    method = method.lower()
    if organism_ids is None:
        organism_ids = [m.id for m in models]
    biomass = _as_biomass_dict(initial_biomasses, organism_ids)
    concs = dict(initial_concentrations or {})

    if method == "joint":
        cm = build_community_model(models, biomass_ids, organism_ids)
        r = DynamicJointFBA(cm, biomass, concs, kinetics=kinetics, solver=solver,
                            diet=diet, max_uptake=max_uptake)
        r.simulate(dt=dt, n=n, epsilon=epsilon)
        return r

    if method == "parallel":
        r = DynamicParallelFBA(models, biomass_ids, organism_ids,
                               initial_biomasses=biomass,
                               initial_concentrations=concs,
                               kinetics=kinetics or {}, solver=solver,
                               diet=diet, max_uptake=max_uptake)
        r.simulate(dt=dt, n=n, epsilon=epsilon)
        return r

    if method == "endpoint":
        cm = build_community_model(models, biomass_ids, organism_ids)
        r = EndPointFBA(cm, n=endpoint_n or n, initial_biomasses=biomass,
                        initial_concentrations=concs, dt=dt, solver=solver)
        r.simulate()
        return r

    raise ValueError(f"unknown method '{method}' (use joint|parallel|endpoint)")