diff --git a/spatio_flux/processes/dfba.py b/spatio_flux/processes/dfba.py index da90d33..705e3c7 100644 --- a/spatio_flux/processes/dfba.py +++ b/spatio_flux/processes/dfba.py @@ -32,54 +32,54 @@ class DynamicFBA(Process): """ config_schema = { - 'model_file': 'string', # TODO -- register a 'path' type - 'kinetic_params': 'map[tuple[float,float]]', - 'substrate_update_reactions': 'map[string]', - 'biomass_identifier': 'string', - 'bounds': 'map[bounds]', + "model_file": "string", # TODO -- register a "path" type + "kinetic_params": "map[tuple[float,float]]", + "substrate_update_reactions": "map[string]", + "biomass_identifier": "string", + "bounds": "map[bounds]", } def __init__(self, config, core): super().__init__(config, core) - if not 'xml' in self.config['model_file']: + if not "xml" in self.config["model_file"]: # use the textbook model if no model file is provided - self.model = load_model(self.config['model_file']) - elif isinstance(self.config['model_file'], str): - self.model = cobra.io.read_sbml_model(self.config['model_file']) + self.model = load_model(self.config["model_file"]) + elif isinstance(self.config["model_file"], str): + self.model = cobra.io.read_sbml_model(self.config["model_file"]) else: # error handling - raise ValueError('Invalid model file') + raise ValueError("Invalid model file") - for reaction_id, bounds in self.config['bounds'].items(): - if bounds['lower'] is not None: - self.model.reactions.get_by_id(reaction_id).lower_bound = bounds['lower'] - if bounds['upper'] is not None: - self.model.reactions.get_by_id(reaction_id).upper_bound = bounds['upper'] + for reaction_id, bounds in self.config["bounds"].items(): + if bounds["lower"] is not None: + self.model.reactions.get_by_id(reaction_id).lower_bound = bounds["lower"] + if bounds["upper"] is not None: + self.model.reactions.get_by_id(reaction_id).upper_bound = bounds["upper"] def inputs(self): return { - 'substrates': 'map[positive_float]' # TODO this should be map[concentration] - # 'enzymes': 'map[positive_float]' # TODO this should be map[concentration] + "substrates": "map[positive_float]" # TODO this should be map[concentration] + # "enzymes": "map[positive_float]" # TODO this should be map[concentration] } def outputs(self): return { - 'substrates': 'map[positive_float]' + "substrates": "map[positive_float]" } # def interface(self): # return { - # 'inputs': {'substrates': 'map[positive_float]'}, - # 'outputs': {'substrates': 'map[positive_float]'}, + # "inputs": {"substrates": "map[positive_float]"}, + # "outputs": {"substrates": "map[positive_float]"}, # } # TODO -- can we just put the inputs/outputs directly in the function? def update(self, inputs, interval): - substrates_input = inputs['substrates'] + substrates_input = inputs["substrates"] - for substrate, reaction_id in self.config['substrate_update_reactions'].items(): - Km, Vmax = self.config['kinetic_params'][substrate] + for substrate, reaction_id in self.config["substrate_update_reactions"].items(): + Km, Vmax = self.config["kinetic_params"][substrate] substrate_concentration = substrates_input[substrate] uptake_rate = Vmax * substrate_concentration / (Km + substrate_concentration) self.model.reactions.get_by_id(reaction_id).lower_bound = -uptake_rate @@ -87,12 +87,12 @@ def update(self, inputs, interval): substrate_update = {} solution = self.model.optimize() - if solution.status == 'optimal': - current_biomass = substrates_input[self.config['biomass_identifier']] + if solution.status == "optimal": + current_biomass = substrates_input[self.config["biomass_identifier"]] biomass_growth_rate = solution.objective_value - substrate_update[self.config['biomass_identifier']] = biomass_growth_rate * current_biomass * interval + substrate_update[self.config["biomass_identifier"]] = biomass_growth_rate * current_biomass * interval - for substrate, reaction_id in self.config['substrate_update_reactions'].items(): + for substrate, reaction_id in self.config["substrate_update_reactions"].items(): flux = solution.fluxes[reaction_id] * current_biomass * interval old_concentration = substrates_input[substrate] new_concentration = max(old_concentration + flux, 0) # keep above 0 -- TODO this should not happen @@ -100,46 +100,46 @@ def update(self, inputs, interval): # TODO -- assert not negative? else: # Handle non-optimal solutions if necessary - # print('Non-optimal solution, skipping update') - for substrate, reaction_id in self.config['substrate_update_reactions'].items(): + # print("Non-optimal solution, skipping update") + for substrate, reaction_id in self.config["substrate_update_reactions"].items(): substrate_update[substrate] = 0 return { - 'substrates': substrate_update, + "substrates": substrate_update, } # Helper functions to get specs and states def dfba_config( - model_file='textbook', + model_file="textbook", kinetic_params=None, substrate_update_reactions=None, - biomass_identifier='biomass', + biomass_identifier="biomass", bounds=None ): if substrate_update_reactions is None: substrate_update_reactions = { - 'glucose': 'EX_glc__D_e', - 'acetate': 'EX_ac_e'} + "glucose": "EX_glc__D_e", + "acetate": "EX_ac_e"} if bounds is None: bounds = { - 'EX_o2_e': {'lower': -2, 'upper': None}, - 'ATPM': {'lower': 1, 'upper': 1}} + "EX_o2_e": {"lower": -2, "upper": None}, + "ATPM": {"lower": 1, "upper": 1}} if kinetic_params is None: kinetic_params = { - 'glucose': (0.5, 1), - 'acetate': (0.5, 2)} + "glucose": (0.5, 1), + "acetate": (0.5, 2)} return { - 'model_file': model_file, - 'kinetic_params': kinetic_params, - 'substrate_update_reactions': substrate_update_reactions, - 'biomass_identifier': biomass_identifier, - 'bounds': bounds + "model_file": model_file, + "kinetic_params": kinetic_params, + "substrate_update_reactions": substrate_update_reactions, + "biomass_identifier": biomass_identifier, + "bounds": bounds } def get_single_dfba_spec( - model_file='textbook', + model_file="textbook", mol_ids=None, path=None, i=None, @@ -153,8 +153,8 @@ def get_single_dfba_spec( Parameters: mol_ids (list of str, optional): List of molecule IDs to include in the process. Defaults to - ['glucose', 'acetate', 'biomass']. - path (list of str, optional): The base path to prepend to each molecule ID. Defaults to ['..', 'fields']. + ["glucose", "acetate", "biomass"]. + path (list of str, optional): The base path to prepend to each molecule ID. Defaults to ["..", "fields"]. i (int, optional): The first index to append to the path for each molecule, if not None. j (int, optional): The second index to append to the path for each molecule, if not None. @@ -163,9 +163,9 @@ def get_single_dfba_spec( and outputs based on the specified molecule IDs and indices. """ if path is None: - path = ['..', 'fields'] + path = ["..", "fields"] if mol_ids is None: - mol_ids = ['glucose', 'acetate', 'biomass'] + mol_ids = ["glucose", "acetate", "biomass"] # Function to build the path with optional indices def build_path(mol_id): @@ -177,25 +177,25 @@ def build_path(mol_id): return base_path return { - '_type': 'process', - 'address': 'local:DynamicFBA', - 'config': dfba_config(model_file=model_file), - 'inputs': { - 'substrates': {mol_id: build_path(mol_id) for mol_id in mol_ids} + "_type": "process", + "address": "local:DynamicFBA", + "config": dfba_config(model_file=model_file), + "inputs": { + "substrates": {mol_id: build_path(mol_id) for mol_id in mol_ids} }, - 'outputs': { - 'substrates': {mol_id: build_path(mol_id) for mol_id in mol_ids} + "outputs": { + "substrates": {mol_id: build_path(mol_id) for mol_id in mol_ids} } } def get_spatial_dfba_spec(n_bins=(5, 5), mol_ids=None): if mol_ids is None: - mol_ids = ['glucose', 'acetate', 'biomass'] + mol_ids = ["glucose", "acetate", "biomass"] dfba_processes_dict = {} for i in range(n_bins[0]): for j in range(n_bins[1]): - dfba_processes_dict[f'[{i},{j}]'] = get_single_dfba_spec(mol_ids=mol_ids, path=['..', 'fields'], i=i, j=j) + dfba_processes_dict[f"[{i},{j}]"] = get_single_dfba_spec(mol_ids=mol_ids, path=["..", "fields"], i=i, j=j) return dfba_processes_dict @@ -205,9 +205,9 @@ def get_spatial_dfba_state( initial_min_max=None, # {mol_id: (min, max)} ): if mol_ids is None: - mol_ids = ['glucose', 'acetate', 'biomass'] + mol_ids = ["glucose", "acetate", "biomass"] if initial_min_max is None: - initial_min_max = {'glucose': (0, 20), 'acetate': (0,0 ), 'biomass': (0, 0.1)} + initial_min_max = {"glucose": (0, 20), "acetate": (0,0 ), "biomass": (0, 0.1)} initial_fields = { mol_id: np.random.uniform(low=initial_min_max[mol_id][0], @@ -216,16 +216,16 @@ def get_spatial_dfba_state( for mol_id in mol_ids} return { - 'fields': { - '_type': 'map', - '_value': { - '_type': 'array', - '_shape': n_bins, - '_data': 'positive_float' + "fields": { + "_type": "map", + "_value": { + "_type": "array", + "_shape": n_bins, + "_data": "positive_float" }, **initial_fields, }, - 'spatial_dfba': get_spatial_dfba_spec(n_bins=n_bins, mol_ids=mol_ids) + "spatial_dfba": get_spatial_dfba_spec(n_bins=n_bins, mol_ids=mol_ids) }