Source code for ecoli.composites.environment.grow_divide

import os

from vivarium.library.units import units
from vivarium.core.composer import Composer
from vivarium.core.composition import (
    composer_in_experiment,
    COMPOSITE_OUT_DIR,
)
from vivarium.plots.agents_multigen import plot_agents_multigen

# processes
from vivarium.processes.growth_rate import GrowthRate
from vivarium.processes.divide_condition import DivideCondition
from vivarium.processes.meta_division import MetaDivision
from ecoli.processes.environment.derive_globals import DeriveGlobals
from ecoli.processes.environment.exchange import Exchange


NAME = "grow_divide"


GROW_DIVIDE_DEFAULTS = {
    "growth": {
        "variables": ["mass"],
    },
    "divide_condition": {"threshold": 2000.0 * units.fg},
    "boundary_path": ("boundary",),
    "agents_path": (
        "..",
        "..",
        "agents",
    ),
    "_schema": {
        "growth": {
            "variables": {
                "mass": {
                    "_default": 1000.0 * units.fg,
                    "_divider": "split",
                }
            }
        }
    },
}


[docs] class GrowDivide(Composer): defaults = GROW_DIVIDE_DEFAULTS
[docs] def generate_processes(self, config): # division config agent_id = config["agent_id"] division_config = dict( config.get("division", {}), agent_id=agent_id, composer=self, ) return { "growth": GrowthRate(config["growth"]), "globals_deriver": DeriveGlobals(), "divide_condition": DivideCondition(config["divide_condition"]), "division": MetaDivision(division_config), }
[docs] def generate_topology(self, config): boundary_path = config["boundary_path"] agents_path = config["agents_path"] return { "growth": { "variables": boundary_path, "rates": ("rates",), }, "globals_deriver": {"global": boundary_path}, "divide_condition": { "variable": boundary_path + ("mass",), "divide": boundary_path + ("divide",), }, "division": {"global": boundary_path, "agents": agents_path}, }
[docs] class GrowDivideExchange(GrowDivide): name = "grow_divide_exchange" defaults = GROW_DIVIDE_DEFAULTS defaults.update( { "exchange": { "molecules": ["A"], }, "fields_path": ("..", "..", "fields"), "dimensions_path": ( "..", "..", "dimensions", ), } )
[docs] def generate_processes(self, config): processes = super().generate_processes(config) added_processes = { "exchange": Exchange(config["exchange"]), } processes.update(added_processes) return processes
[docs] def generate_topology(self, config): topology = super().generate_topology(config) boundary_path = config["boundary_path"] added_topology = { "exchange": { "exchanges": boundary_path + ("exchanges",), "external": boundary_path + ("external",), "internal": ("internal",), }, } topology.update(added_topology) return topology
def test_grow_divide(total_time=2000, return_data=False): agent_id = "0" composite = GrowDivide( { "agent_id": agent_id, "growth": { "growth_rate": 0.006, # very fast growth "default_growth_noise": 1e-3, }, } ) initial_state = {"agents": {agent_id: {"global": {"mass": 1000 * units.fg}}}} settings = {"experiment_id": "grow_divide"} experiment = composer_in_experiment( composite, initial_state=initial_state, outer_path=("agents", agent_id), settings=settings, ) experiment.update(total_time) output = experiment.emitter.get_data_unitless() # assert division occurred assert list(output[0.0]["agents"].keys()) == [agent_id] assert agent_id not in list(output[total_time]["agents"].keys()) assert len(output[0.0]["agents"]) == 1 assert len(output[total_time]["agents"]) > 1 if return_data: return output def test_grow_divide_exchange(total_time=2000, return_data=False): agent_id = "0" molecule_id = "A" composite = GrowDivideExchange( { "agent_id": agent_id, "growth": { "growth_rate": 0.006, # very fast growth "default_growth_noise": 1e-3, }, "exchange": { "molecules": [molecule_id], }, } ) initial_state = { "agents": { agent_id: { "global": {"mass": 1000 * units.fg}, "external": {molecule_id: 10.0 * units.mmol / units.L}, "internal": {molecule_id: 0.0 * units.mmol / units.L}, } } } settings = {"experiment_id": "grow_divide_exchange"} experiment = composer_in_experiment( composite, initial_state=initial_state, outer_path=("agents", agent_id), settings=settings, ) experiment.update(total_time) output = experiment.emitter.get_data_unitless() # assert # external starts at 1, goes down until death, and then back up # internal does the inverse assert list(output[0.0]["agents"].keys()) == [agent_id] assert agent_id not in list(output[total_time]["agents"].keys()) assert len(output[0.0]["agents"]) == 1 assert len(output[total_time]["agents"]) > 1 if return_data: return output
[docs] def main(): out_dir = os.path.join(COMPOSITE_OUT_DIR, NAME) if not os.path.exists(out_dir): os.makedirs(out_dir) # if grow_divide: output = test_grow_divide(2000, return_data=True) plot_settings = {} plot_agents_multigen(output, plot_settings, out_dir, "grow_divide") # if grow_divide_exchange: output = test_grow_divide_exchange(2000, return_data=True) plot_settings = {} plot_agents_multigen(output, plot_settings, out_dir, "grow_divide_exchange")
if __name__ == "__main__": main()