Source code for ecoli.processes.listeners.dna_supercoiling

"""
=========================
DNA Supercoiling 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 = "dna_supercoiling_listener"
TOPOLOGY = {
    "listeners": ("listeners",),
    "chromosomal_segments": ("unique", "chromosomal_segment"),
    "global_time": ("global_time",),
    "timestep": ("timestep",),
}
topology_registry.register(NAME, TOPOLOGY)


[docs] class DnaSupercoiling(Step): """ Listener for DNA supercoiling data. """ name = NAME topology = TOPOLOGY defaults = { "relaxed_DNA_base_pairs_per_turn": 0, "emit_unique": False, "time_step": 1, } def __init__(self, parameters=None): super().__init__(parameters) self.relaxed_DNA_base_pairs_per_turn = self.parameters[ "relaxed_DNA_base_pairs_per_turn" ]
[docs] def ports_schema(self): return { "listeners": { "dna_supercoiling": listener_schema( { "segment_left_boundary_coordinates": [], "segment_right_boundary_coordinates": [], "segment_domain_indexes": [], "segment_superhelical_densities": [], } ) }, "chromosomal_segments": numpy_schema( "chromosomal_segments", 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): boundary_coordinates, domain_indexes, linking_numbers = attrs( states["chromosomal_segments"], ["boundary_coordinates", "domain_index", "linking_number"], ) if len(boundary_coordinates) == 0: return { "listeners": { "dna_supercoiling": { "segment_left_boundary_coordinates": [], "segment_right_boundary_coordinates": [], "segment_domain_indexes": [], "segment_superhelical_densities": [], } } } # Get mask for segments with nonzero lengths segment_lengths = boundary_coordinates[:, 1] - boundary_coordinates[:, 0] assert np.all(segment_lengths >= 0) nonzero_length_mask = segment_lengths > 0 # Calculate superhelical densities linking_numbers_relaxed_DNA = ( segment_lengths[nonzero_length_mask] / self.relaxed_DNA_base_pairs_per_turn ) update = { "listeners": { "dna_supercoiling": { "segment_left_boundary_coordinates": boundary_coordinates[ nonzero_length_mask, 0 ], "segment_right_boundary_coordinates": boundary_coordinates[ nonzero_length_mask, 1 ], "segment_domain_indexes": domain_indexes[nonzero_length_mask], "segment_superhelical_densities": np.divide( linking_numbers[nonzero_length_mask] - linking_numbers_relaxed_DNA, linking_numbers_relaxed_DNA, ), } } } return update