Source code for ecoli.composites.ecoli_master_tests

"""
======================
Tests for Ecoli Master
======================
"""

import os
import numpy as np
import pytest
import warnings

from vivarium.core.engine import Engine
from vivarium.core.control import run_library_cli

from ecoli.library.schema import bulk_name_to_idx
from ecoli.analysis.colony.snapshots import (
    plot_snapshots,
    format_snapshot_data,
    make_video,
)
from ecoli.composites.ecoli_configs import (
    ECOLI_DEFAULT_PROCESSES,
    ECOLI_DEFAULT_TOPOLOGY,
)
from ecoli.experiments.ecoli_master_sim import EcoliSim, CONFIG_DIR_PATH


@pytest.mark.slow
def test_division(agent_id="0", total_time=4):
    """tests that a cell can be divided and keep running"""

    # get initial mass from Ecoli composer
    sim = EcoliSim.from_file()
    sim.config["initial_state_file"] = "vivecoli_t2527"
    sim.config["divide"] = True
    sim.config["division_variable"] = ("listeners", "mass", "dry_mass")
    sim.config["agent_id"] = agent_id
    sim.config["total_time"] = total_time
    # Ensure unique molecules are emitted
    sim.config["emit_unique"] = True
    sim.build_ecoli()

    # Set threshold so that mother divides after first timestep
    sim.generated_initial_state["agents"]["0"]["division_threshold"] = 724.4
    sim.run()

    # retrieve output
    output = sim.ecoli_experiment.emitter.get_data()
    # metabolism runs and greatly changes molecule counts after
    # previoius emit but before division, so ignore metabolites
    metabolite_idx = sim.ecoli_experiment.state["agents"]["00"][
        "ecoli-metabolism"
    ].value.metabolite_idx
    # ignore trna charging molecules, which also change before division
    polypep_elong = sim.ecoli_experiment.state["agents"]["00"]["process"][
        "ecoli-polypeptide-elongation"
    ].value[0]
    trna_charging_idx = polypep_elong.charging_molecule_idx
    ppgpp_idx = np.array([polypep_elong.ppgpp_idx])
    # ignore inactive ribosomes, which also change before division
    polypep_init = sim.ecoli_experiment.state["agents"]["00"]["process"][
        "ecoli-polypeptide-initiation"
    ].value[0]
    ribosome_idx = np.array(
        [polypep_init.ribosome30S_idx, polypep_init.ribosome50S_idx]
    )
    # ignore fragment bases, which also change before division
    fragment_base_idx = (
        sim.ecoli_experiment.state["agents"]["00"]["process"]["ecoli-rna-maturation"]
        .value[0]
        .fragment_base_idx
    )
    # ignore high-count membrane-related proteins, which also change before division
    membrane_idx = bulk_name_to_idx(
        ["EG10544-MONOMER[m]", "EG10669-MONOMER[o]", "EG50003-MONOMER[c]"],
        sim.ecoli_experiment.state["agents"]["00"]["bulk"].value["id"],
    )
    ignore_idx = np.concatenate(
        [
            metabolite_idx,
            trna_charging_idx,
            ppgpp_idx,
            ribosome_idx,
            fragment_base_idx,
            membrane_idx,
        ]
    )
    mother_state = next(iter(output[0]["agents"].values()))
    mother_bulk = np.delete(mother_state["bulk"], ignore_idx)
    daughter_states = list(output[1]["agents"].values())
    daughter_bulk = [np.delete(ds["bulk"], ignore_idx) for ds in daughter_states]

    # compare the counts of bulk molecules between the mother and daughters
    # this is not exact because the mother grew slightly in the timestep
    # after its last emit but before being split into two daughter cells
    assert np.allclose(
        mother_bulk, np.array(daughter_bulk[0]) + np.array(daughter_bulk[1]), atol=20
    )

    # compare the counts of unique molecules between the mother and daughters
    for name, mols in mother_state["unique"].items():
        d1_state = daughter_states[0]["unique"][name]
        d2_state = daughter_states[1]["unique"][name]
        m_state = np.array(mols)
        entryState_col = np.where(np.array(d1_state.dtype.names) == "_entryState")[0]
        n_mother = m_state[entryState_col].sum()
        n_daughter = d1_state["_entryState"].sum() + d2_state["_entryState"].sum()
        if name == "chromosome_domain":
            # Chromosome domain 0 is lost after division because
            # it has been fully split into child domains 1 and 2
            n_daughter += 1
        assert np.isclose(
            n_mother, n_daughter, rtol=0.01
        ), f"{name}: mother has {n_mother}, daughters have {n_daughter}"
        # Assert that no unique mol is in both daughters
        assert not (set(d1_state["unique_index"]) & set(d2_state["unique_index"]))

    # asserts
    final_agents = output[total_time]["agents"].keys()
    print(f"initial agent id: {agent_id}")
    print(f"final agent ids: {final_agents}")
    assert len(final_agents) == 2


def test_division_topology():
    """test that the topology is correctly dividing"""
    timestep = 2
    agent_id = "0"
    # get initial mass from Ecoli composer
    sim = EcoliSim.from_file()
    sim.config["seed"] = 1
    sim.config["initial_state_file"] = "vivecoli_t2527"
    sim.config["divide"] = True
    sim.config["agent_id"] = agent_id
    sim.build_ecoli()

    # make the experiment
    experiment_config = {
        "processes": sim.ecoli.processes,
        "steps": sim.ecoli.steps,
        "flow": sim.ecoli.flow,
        "topology": sim.ecoli.topology,
        "initial_state": sim.generated_initial_state,
    }

    # Since unique numpy updater is an class method, internal
    # deepcopying in vivarium-core causes this warning to appear
    warnings.filterwarnings(
        "ignore",
        message="Incompatible schema "
        "assignment at .+ Trying to assign the value <bound method "
        "UniqueNumpyUpdater\.updater .+ to key updater, which already "
        "has the value <bound method UniqueNumpyUpdater\.updater",
    )
    sim.ecoli_experiment = Engine(**experiment_config)

    # Only emit designated stores
    sim.ecoli_experiment.state.set_emit_values([tuple()], False)
    sim.ecoli_experiment.state.set_emit_values(
        sim.config["emit_paths"],
        True,
    )
    # Divide immediately
    sim.ecoli_experiment.state["agents"]["0"]["divide"] = True

    # Clean up unnecessary references
    sim.generated_initial_state = None
    sim.ecoli_experiment.initial_state = None

    full_topology = sim.ecoli_experiment.state.get_topology()
    mother_topology = full_topology["agents"][agent_id].copy()

    # update one time step at a time until division
    while len(full_topology["agents"]) <= 1:
        sim.ecoli_experiment.update(timestep)
        full_topology = sim.ecoli_experiment.state.get_topology()
    sim.ecoli_experiment.end()

    # assert that the daughter topologies are the same as the mother topology
    daughter_ids = list(full_topology["agents"].keys())
    for daughter_id in daughter_ids:
        daughter_topology = full_topology["agents"][daughter_id]
        assert daughter_topology == mother_topology


def test_ecoli_generate():
    sim = EcoliSim.from_file()
    sim.divide = False
    sim.build_ecoli()

    # asserts to ecoli_composite['processes'] and ecoli_composite['topology']
    assert all(
        "_requester" in k
        or "_evolver" in k
        or "allocator" in k
        or "unique_update" in k
        or k == "division"
        or k == "mark_d_period"
        or isinstance(v, ECOLI_DEFAULT_PROCESSES[k])
        for k, v in sim.ecoli["steps"].items()
    )
    ignore_keys = ["request", "process", "allocate", "global_time", "next_update_time"]
    for k, v in sim.ecoli["topology"].items():
        proc_name = k.split("_requester")[0].split("_evolver")[0]
        # Clock topology is not registered in registry
        if proc_name == "clock":
            pass
        elif proc_name in ECOLI_DEFAULT_TOPOLOGY:
            if "_requester" in k or "_evolver" in k:
                # Ignore partitioning and timestepping keys
                for ignore_key in ignore_keys:
                    v.pop(ignore_key, None)
            assert ECOLI_DEFAULT_TOPOLOGY[proc_name] == v


def test_lattice_lysis(plot=False):
    """
    Run plots:
    '''
    > python ecoli/composites/ecoli_master_tests.py -n 4 -o plot=True
    '''

    ANTIBIOTIC_KEY = 'nitrocefin'
    PUMP_KEY = 'TRANS-CPLX-201[s]'
    PORIN_KEY = 'porin'
    BETA_LACTAMASE_KEY = 'EG10040-MONOMER[p]'

    TODO: connect glucose! through local_field
    """
    sim = EcoliSim.from_file(CONFIG_DIR_PATH + "lysis.json")
    sim.total_time = 10
    sim.process_configs.update({"global_clock": {"time_step": 2}})
    sim.build_ecoli()
    # Add beta-lactam to bulk store
    initial_state = sim.generated_initial_state
    zero_mass = (0,) * 9
    initial_state["agents"]["0"]["bulk"] = np.append(
        initial_state["agents"]["0"]["bulk"],
        np.array(
            [
                ("beta-lactam[p]", 0) + zero_mass,
                ("hydrolyzed-beta-lactam[p]", 0) + zero_mass,
            ],
            dtype=initial_state["agents"]["0"]["bulk"].dtype,
        ),
    )
    sim.run()
    data = sim.query()

    if plot:
        plot_spatial_snapshots(data, sim, experiment_dir="ecoli_lysis")


[docs] def plot_spatial_snapshots(data, sim, experiment_dir="ecoli_test"): out_dir = os.path.join("out", "experiments", experiment_dir) os.makedirs(out_dir, exist_ok=True) bounds = sim.config["spatial_environment_config"]["multibody"]["bounds"] # format the data for plot_snapshots agents, fields = format_snapshot_data(data) plot_snapshots( bounds, agents=agents, fields=fields, n_snapshots=5, out_dir=out_dir, filename="snapshots", ) # make snapshot video make_video( data, bounds, plot_type="fields", out_dir=out_dir, filename="video", )
test_library = { "1": test_division, "2": test_division_topology, "3": test_ecoli_generate, "4": test_lattice_lysis, } # run experiments in test_library from the command line with: # python ecoli/composites/ecoli_master_tests.py -n [experiment id] if __name__ == "__main__": run_library_cli(test_library)