Source code for ecoli.processes.listeners.monomer_counts

"""
=======================
Monomer Counts Listener
=======================
"""

import numpy as np
from ecoli.library.schema import numpy_schema, counts, bulk_name_to_idx
from vivarium.core.process import Step

from ecoli.processes.registries import topology_registry


NAME = "monomer_counts_listener"
TOPOLOGY = {
    "listeners": ("listeners",),
    "bulk": ("bulk",),
    "unique": ("unique",),
    "global_time": ("global_time",),
    "timestep": ("timestep",),
}
topology_registry.register(NAME, TOPOLOGY)


[docs] class MonomerCounts(Step): """ Listener for the counts of each protein monomer species. """ name = NAME topology = TOPOLOGY defaults = { "bulk_molecule_ids": [], "unique_ids": [], "complexation_molecule_ids": [], "complexation_complex_ids": [], "equilibrium_molecule_ids": [], "equilibrium_complex_ids": [], "monomer_ids": [], "two_component_system_molecule_ids": [], "two_component_system_complex_ids": [], "ribosome_50s_subunits": [], "ribosome_30s_subunits": [], "rnap_subunits": [], "replisome_trimer_subunits": [], "replisome_monomer_subunits": [], "complexation_stoich": [], "equilibrium_stoich": [], "two_component_system_stoich": [], "emit_unique": False, "time_step": 1, } def __init__(self, parameters=None): super().__init__(parameters) # Get IDs of all bulk molecules self.bulk_molecule_ids = self.parameters["bulk_molecule_ids"] # Get IDs of molecules involved in complexation and equilibrium self.complexation_molecule_ids = self.parameters["complexation_molecule_ids"] self.complexation_complex_ids = self.parameters["complexation_complex_ids"] self.equilibrium_molecule_ids = self.parameters["equilibrium_molecule_ids"] self.equilibrium_complex_ids = self.parameters["equilibrium_complex_ids"] self.monomer_ids = self.parameters["monomer_ids"] # Get IDs of complexed molecules monomers involved in two # component system self.two_component_system_molecule_ids = self.parameters[ "two_component_system_molecule_ids" ] self.two_component_system_complex_ids = self.parameters[ "two_component_system_complex_ids" ] # Get IDs of ribosome subunits ribosome_50s_subunits = self.parameters["ribosome_50s_subunits"] ribosome_30s_subunits = self.parameters["ribosome_30s_subunits"] self.ribosome_subunit_ids = ( ribosome_50s_subunits["subunitIds"].tolist() + ribosome_30s_subunits["subunitIds"].tolist() ) # Get IDs of RNA polymerase subunits rnap_subunits = self.parameters["rnap_subunits"] self.rnap_subunit_ids = rnap_subunits["subunitIds"].tolist() # Get IDs of replisome subunits replisome_trimer_subunits = self.parameters["replisome_trimer_subunits"] replisome_monomer_subunits = self.parameters["replisome_monomer_subunits"] self.replisome_subunit_ids = ( replisome_trimer_subunits + replisome_monomer_subunits ) # Get stoichiometric matrices for complexation, equilibrium, two # component system and the assembly of unique molecules self.complexation_stoich = self.parameters["complexation_stoich"] self.equilibrium_stoich = self.parameters["equilibrium_stoich"] self.two_component_system_stoich = self.parameters[ "two_component_system_stoich" ] self.ribosome_stoich = np.hstack( ( ribosome_50s_subunits["subunitStoich"], ribosome_30s_subunits["subunitStoich"], ) ) self.rnap_stoich = rnap_subunits["subunitStoich"] self.replisome_stoich = np.hstack( ( 3 * np.ones(len(replisome_trimer_subunits)), np.ones(len(replisome_monomer_subunits)), ) ) # Helper indices for Numpy indexing self.monomer_idx = None
[docs] def ports_schema(self): return { "listeners": { "monomer_counts": { "_default": [], "_updater": "set", "_emit": True, "_properties": {"metadata": self.monomer_ids}, } }, "bulk": numpy_schema("bulk"), "unique": { "active_ribosome": numpy_schema( "active_ribosome", emit=self.parameters["emit_unique"] ), "active_RNAP": numpy_schema( "active_RNAPs", emit=self.parameters["emit_unique"] ), "active_replisome": numpy_schema( "active_replisomes", emit=self.parameters["emit_unique"] ), }, "global_time": {"_default": 0.0}, "timestep": {"_default": self.parameters["time_step"]}, }
[docs] def update_condition(self, timestep, states): return (states["global_time"] % states["timestep"]) == 0
[docs] def next_update(self, timestep, states): if self.monomer_idx is None: bulk_ids = states["bulk"]["id"] self.bulk_molecule_idx = bulk_name_to_idx(self.bulk_molecule_ids, bulk_ids) self.monomer_idx = bulk_name_to_idx(self.monomer_ids, bulk_ids) self.complexation_molecule_idx = bulk_name_to_idx( self.complexation_molecule_ids, bulk_ids ) self.complexation_complex_idx = bulk_name_to_idx( self.complexation_complex_ids, bulk_ids ) self.equilibrium_molecule_idx = bulk_name_to_idx( self.equilibrium_molecule_ids, bulk_ids ) self.equilibrium_complex_idx = bulk_name_to_idx( self.equilibrium_complex_ids, bulk_ids ) self.two_component_system_molecule_idx = bulk_name_to_idx( self.two_component_system_molecule_ids, bulk_ids ) self.two_component_system_complex_idx = bulk_name_to_idx( self.two_component_system_complex_ids, bulk_ids ) self.ribosome_subunit_idx = bulk_name_to_idx( self.ribosome_subunit_ids, bulk_ids ) self.rnap_subunit_idx = bulk_name_to_idx(self.rnap_subunit_ids, bulk_ids) self.replisome_subunit_idx = bulk_name_to_idx( self.replisome_subunit_ids, bulk_ids ) # Get current counts of bulk and unique molecules bulkMoleculeCounts = counts(states["bulk"], self.bulk_molecule_idx) n_active_ribosome = states["unique"]["active_ribosome"]["_entryState"].sum() n_active_rnap = states["unique"]["active_RNAP"]["_entryState"].sum() n_active_replisome = states["unique"]["active_replisome"]["_entryState"].sum() # Account for monomers in bulk molecule complexes complex_monomer_counts = np.dot( self.complexation_stoich, np.negative(bulkMoleculeCounts[self.complexation_complex_idx]), ) equilibrium_monomer_counts = np.dot( self.equilibrium_stoich, np.negative(bulkMoleculeCounts[self.equilibrium_complex_idx]), ) two_component_monomer_counts = np.dot( self.two_component_system_stoich, np.negative(bulkMoleculeCounts[self.two_component_system_complex_idx]), ) bulkMoleculeCounts[self.complexation_molecule_idx] += ( complex_monomer_counts.astype(np.int32) ) bulkMoleculeCounts[self.equilibrium_molecule_idx] += ( equilibrium_monomer_counts.astype(np.int32) ) bulkMoleculeCounts[self.two_component_system_molecule_idx] += ( two_component_monomer_counts.astype(np.int32) ) # Account for monomers in unique molecule complexes n_ribosome_subunit = n_active_ribosome * self.ribosome_stoich n_rnap_subunit = n_active_rnap * self.rnap_stoich n_replisome_subunit = n_active_replisome * self.replisome_stoich bulkMoleculeCounts[self.ribosome_subunit_idx] += n_ribosome_subunit.astype( np.int32 ) bulkMoleculeCounts[self.rnap_subunit_idx] += n_rnap_subunit.astype(np.int32) bulkMoleculeCounts[self.replisome_subunit_idx] += n_replisome_subunit.astype( np.int32 ) # Update monomerCounts monomer_counts = bulkMoleculeCounts[self.monomer_idx] update = {"listeners": {"monomer_counts": monomer_counts}} return update
def test_monomer_counts_listener(): from ecoli.experiments.ecoli_master_sim import EcoliSim sim = EcoliSim.from_file() sim.total_time = 2 sim.raw_output = False sim.build_ecoli() sim.run() listeners = sim.query()["agents"]["0"]["listeners"] assert isinstance(listeners["monomer_counts"][0], list) assert isinstance(listeners["monomer_counts"][1], list) # python ecoli/processes/listeners/monomer_counts.py if __name__ == "__main__": test_monomer_counts_listener()