Source code for ecoli.processes.listeners.rnap_data

"""
====================
RNAP Data Listener
====================
"""

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

from ecoli.processes.registries import topology_registry
from ecoli.processes.transcript_elongation import get_mapping_arrays


NAME = "rnap_data_listener"
TOPOLOGY = {
    "listeners": ("listeners",),
    "active_RNAPs": ("unique", "active_RNAP"),
    "RNAs": ("unique", "RNA"),
    "active_ribosomes": ("unique", "active_ribosome"),
    "global_time": ("global_time",),
    "timestep": ("timestep",),
    "next_update_time": ("next_update_time", NAME),
}
topology_registry.register(NAME, TOPOLOGY)


[docs] class RnapData(Step): """ Listener for RNAP data. """ name = NAME topology = TOPOLOGY defaults = { "stable_RNA_indexes": [], "cistron_ids": [], "cistron_tu_mapping_matrix": [], "time_step": 1, "emit_unique": False, } def __init__(self, parameters=None): super().__init__(parameters) self.stable_RNA_indexes = self.parameters["stable_RNA_indexes"] self.cistron_ids = self.parameters["cistron_ids"] self.cistron_tu_mapping_matrix = self.parameters["cistron_tu_mapping_matrix"]
[docs] def ports_schema(self): n_TUs = self.cistron_tu_mapping_matrix.shape[1] ports = { "listeners": { "rnap_data": listener_schema( { "rna_init_event": np.zeros(n_TUs, dtype=np.int64), "active_rnap_coordinates": [], "active_rnap_domain_indexes": [], "active_rnap_unique_indexes": [], "active_rnap_on_stable_RNA_indexes": [], "active_rnap_n_bound_ribosomes": [], "rna_init_event_per_cistron": ( [0] * len(self.cistron_ids), self.cistron_ids, ), } ) }, "active_RNAPs": numpy_schema( "active_RNAPs", emit=self.parameters["emit_unique"] ), "RNAs": numpy_schema("RNAs", emit=self.parameters["emit_unique"]), "active_ribosomes": numpy_schema( "active_ribosome", emit=self.parameters["emit_unique"] ), "global_time": {"_default": 0.0}, "timestep": {"_default": self.parameters["time_step"]}, "next_update_time": { "_default": self.parameters["time_step"], "_updater": "set", "_divider": "set", }, } return ports
[docs] def update_condition(self, timestep, states): """ See :py:meth:`~ecoli.processes.partition.Requester.update_condition`. """ if states["next_update_time"] <= states["global_time"]: if states["next_update_time"] < states["global_time"]: warnings.warn( f"{self.name} updated at t=" f"{states['global_time']} instead of t=" f"{states['next_update_time']}. Decrease the " "timestep for the global clock process for more " "accurate timekeeping." ) return True return False
[docs] def next_update(self, timestep, states): # Read coordinates of all active RNAPs coordinates, domain_indexes, RNAP_unique_indexes = attrs( states["active_RNAPs"], ["coordinates", "domain_index", "unique_index"] ) (RNA_RNAP_index, is_full_transcript, RNA_unique_indexes, TU_indexes) = attrs( states["RNAs"], ["RNAP_index", "is_full_transcript", "unique_index", "TU_index"], ) is_partial_transcript = np.logical_not(is_full_transcript) is_stable_RNA = np.isin(TU_indexes, self.stable_RNA_indexes) partial_RNA_RNAP_indexes = RNA_RNAP_index[is_partial_transcript] partial_RNA_unique_indexes = RNA_unique_indexes[is_partial_transcript] (ribosome_RNA_index,) = attrs(states["active_ribosomes"], ["mRNA_index"]) RNA_index_counts = dict(zip(*np.unique(ribosome_RNA_index, return_counts=True))) partial_RNA_to_RNAP_mapping, _ = get_mapping_arrays( partial_RNA_RNAP_indexes, RNAP_unique_indexes ) update = { "listeners": { "rnap_data": { "active_rnap_coordinates": coordinates, "active_rnap_domain_indexes": domain_indexes, "active_rnap_unique_indexes": RNAP_unique_indexes, "active_rnap_on_stable_RNA_indexes": RNA_RNAP_index[ np.logical_and(is_stable_RNA, is_partial_transcript) ], "active_rnap_n_bound_ribosomes": np.array( [ RNA_index_counts.get(partial_RNA_unique_indexes[i], 0) for i in partial_RNA_to_RNAP_mapping ] ), # Calculate hypothetical RNA initiation events per cistron "rna_init_event_per_cistron": self.cistron_tu_mapping_matrix.dot( states["listeners"]["rnap_data"]["rna_init_event"] ), } }, "next_update_time": states["global_time"] + states["timestep"], } return update