Source code for ecoli.processes.listeners.replication_data

"""
=========================
Replication Data Listener
=========================
"""

import numpy as np
from ecoli.library.schema import numpy_schema, listener_schema, attrs
from vivarium.core.process import Step

from ecoli.processes.registries import topology_registry


NAME = "replication_data_listener"
TOPOLOGY = {
    "listeners": ("listeners",),
    "oriCs": ("unique", "oriC"),
    "DnaA_boxes": ("unique", "DnaA_box"),
    "active_replisomes": ("unique", "active_replisome"),
    "global_time": ("global_time",),
    "timestep": ("timestep",),
}
topology_registry.register(NAME, TOPOLOGY)


[docs] class ReplicationData(Step): """ Listener for replication data. """ name = NAME topology = TOPOLOGY defaults = {"time_step": 1, "emit_unique": False}
[docs] def ports_schema(self): return { "listeners": { "replication_data": listener_schema( { "fork_coordinates": [], "fork_domains": [], "fork_unique_index": [], "number_of_oric": [], "free_DnaA_boxes": [], "total_DnaA_boxes": [], } ) }, "oriCs": numpy_schema("oriCs", emit=self.parameters["emit_unique"]), "active_replisomes": numpy_schema( "active_replisomes", emit=self.parameters["emit_unique"] ), "DnaA_boxes": numpy_schema( "DnaA_boxes", 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): fork_coordinates, fork_domains, fork_unique_index = attrs( states["active_replisomes"], ["coordinates", "domain_index", "unique_index"] ) (DnaA_box_bound,) = attrs(states["DnaA_boxes"], ["DnaA_bound"]) update = { "listeners": { "replication_data": { "fork_coordinates": fork_coordinates, "fork_domains": fork_domains, "fork_unique_index": fork_unique_index, "number_of_oric": states["oriCs"]["_entryState"].sum(), "total_DnaA_boxes": len(DnaA_box_bound), "free_DnaA_boxes": np.count_nonzero(np.logical_not(DnaA_box_bound)), } } } return update