Source code for ecoli.library.json_state

import ast
import json
import numpy as np
import concurrent.futures

from vivarium.core.serialize import deserialize_value
from wholecell.utils import units


[docs] def load_states(path): with open(path, "r") as states_file: states = json.load(states_file) return states
[docs] def numpy_molecules(states): """ Loads unique and bulk molecule data as Numpy structured arrays """ if "bulk_dtypes" in states: bulk_dtypes = ast.literal_eval(states.pop("bulk_dtypes")) bulk_tuples = [tuple(mol) for mol in states["bulk"]] states["bulk"] = np.array(bulk_tuples, dtype=bulk_dtypes) # Numpy arrays are read-only outside of updater states["bulk"].flags.writeable = False if "unique_dtypes" in states: for key, dtypes in states.pop("unique_dtypes").items(): dtypes = ast.literal_eval(dtypes) unique_tuples = [tuple(mol) for mol in states["unique"][key]] states["unique"][key] = np.array(unique_tuples, dtype=dtypes) states["unique"][key].flags.writeable = False if "environment" in states: if "exchange_data" in states["environment"]: states["environment"]["exchange_data"]["constrained"] = { mol: units.mmol / (units.g * units.h) * rate for mol, rate in states["environment"]["exchange_data"][ "constrained" ].items() } else: # Load aerobic minimal media exchange data by default states["environment"]["exchange_data"] = { "unconstrained": sorted( [ "CL-[p]", "FE+2[p]", "FE+3[p]", "CO+2[p]", "MG+2[p]", "NA+[p]", "CARBON-DIOXIDE[p]", "OXYGEN-MOLECULE[p]", "MN+2[p]", "L-SELENOCYSTEINE[c]", "K+[p]", "SULFATE[p]", "ZN+2[p]", "CA+2[p]", "Pi[p]", "NI+2[p]", "WATER[p]", "AMMONIUM[c]", ] ), "constrained": {"GLC[p]": 20.0 * units.mmol / (units.g * units.h)}, } return states
[docs] def get_state_from_file( path="data/wcecoli_t0.json", ): serialized_state = load_states(path) # Parallelize deserialization of colony states if "agents" in serialized_state: agents = serialized_state.pop("agents") n_agents = len(agents) with concurrent.futures.ProcessPoolExecutor(n_agents) as executor: deserialized_agents = executor.map(deserialize_value, agents.values()) numpy_agents = [] for agent in deserialized_agents: numpy_agents.append(numpy_molecules(agent)) agents = dict(zip(agents.keys(), numpy_agents)) states = deserialize_value(serialized_state) states["agents"] = agents return states deserialized_states = deserialize_value(serialized_state) states = numpy_molecules(deserialized_states) # TODO: Add timeline process to set up media ID states.setdefault("environment", {})["media_id"] = "minimal" return states