Skip to content

Commit

Permalink
linting
Browse files Browse the repository at this point in the history
  • Loading branch information
@eagmon
eagmon committed Dec 5, 2024
1 parent d5db2e6 commit 61c0650
Showing 1 changed file with 66 additions and 66 deletions.
132 changes: 66 additions & 66 deletions spatio_flux/processes/dfba.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,114 +32,114 @@ 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

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
substrate_update[substrate] = new_concentration - old_concentration
# 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,
Expand All @@ -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.
Expand All @@ -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):
Expand All @@ -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


Expand All @@ -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],
Expand All @@ -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)
}


0 comments on commit 61c0650

Please sign in to comment.