Mercurial > repos > bimib > cobraxy

comparison COBRAxy/ras_to_bounds.py @ 4:41f35c2f0c7b draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author luca_milaz
date Wed, 18 Sep 2024 10:59:10 +0000
parents
children fac6930e6385
comparison
equal deleted inserted replaced
3:1f3ac6fd9867 4:41f35c2f0c7b
1 import argparse
2 import utils.general_utils as utils
3 from typing import Optional, List
4 import os
5 import numpy as np
6 import pandas as pd
7 import cobra
8 import sys
9 import csv
10 from joblib import Parallel, delayed, cpu_count
11
12 ################################# process args ###############################
13 def process_args(args :List[str]) -> argparse.Namespace:
14 """
15 Processes command-line arguments.
16
17 Args:
18 args (list): List of command-line arguments.
19
20 Returns:
21 Namespace: An object containing parsed arguments.
22 """
23 parser = argparse.ArgumentParser(usage = '%(prog)s [options]',
24 description = 'process some value\'s')
25
26 parser.add_argument(
27 '-ms', '--model_selector',
28 type = utils.Model, default = utils.Model.ENGRO2, choices = [utils.Model.ENGRO2, utils.Model.Custom],
29 help = 'chose which type of model you want use')
30
31 parser.add_argument("-mo", "--model", type = str,
32 help = "path to input file with custom rules, if provided")
33
34 parser.add_argument("-mn", "--model_name", type = str, help = "custom mode name")
35
36 parser.add_argument(
37 '-mes', '--medium_selector',
38 default = "allOpen",
39 help = 'chose which type of medium you want use')
40
41 parser.add_argument("-meo", "--medium", type = str,
42 help = "path to input file with custom medium, if provided")
43
44 parser.add_argument('-ol', '--out_log',
45 help = "Output log")
46
47 parser.add_argument('-td', '--tool_dir',
48 type = str,
49 required = True,
50 help = 'your tool directory')
51
52 parser.add_argument('-ir', '--input_ras',
53 type=str,
54 required = False,
55 help = 'input ras')
56
57 parser.add_argument('-rs', '--ras_selector',
58 required = True,
59 type=utils.Bool("using_RAS"),
60 help = 'ras selector')
61
62 ARGS = parser.parse_args()
63 return ARGS
64
65 ########################### warning ###########################################
66 def warning(s :str) -> None:
67 """
68 Log a warning message to an output log file and print it to the console.
69
70 Args:
71 s (str): The warning message to be logged and printed.
72
73 Returns:
74 None
75 """
76 with open(ARGS.out_log, 'a') as log:
77 log.write(s + "\n\n")
78 print(s)
79
80 ############################ dataset input ####################################
81 def read_dataset(data :str, name :str) -> pd.DataFrame:
82 """
83 Read a dataset from a CSV file and return it as a pandas DataFrame.
84
85 Args:
86 data (str): Path to the CSV file containing the dataset.
87 name (str): Name of the dataset, used in error messages.
88
89 Returns:
90 pandas.DataFrame: DataFrame containing the dataset.
91
92 Raises:
93 pd.errors.EmptyDataError: If the CSV file is empty.
94 sys.exit: If the CSV file has the wrong format, the execution is aborted.
95 """
96 try:
97 dataset = pd.read_csv(data, sep = '\t', header = 0, engine='python')
98 except pd.errors.EmptyDataError:
99 sys.exit('Execution aborted: wrong format of ' + name + '\n')
100 if len(dataset.columns) < 2:
101 sys.exit('Execution aborted: wrong format of ' + name + '\n')
102 return dataset
103
104
105 def apply_ras_bounds(model, ras_row, rxns_ids):
106 """
107 Adjust the bounds of reactions in the model based on RAS values.
108
109 Args:
110 model (cobra.Model): The metabolic model to be modified.
111 ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
112 rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
113
114 Returns:
115 None
116 """
117 for reaction in rxns_ids:
118 if reaction in ras_row.index and pd.notna(ras_row[reaction]):
119 rxn = model.reactions.get_by_id(reaction)
120 scaling_factor = ras_row[reaction]
121 rxn.lower_bound *= scaling_factor
122 rxn.upper_bound *= scaling_factor
123
124 def process_ras_cell(cellName, ras_row, model, rxns_ids, output_folder):
125 """
126 Process a single RAS cell, apply bounds, and save the bounds to a CSV file.
127
128 Args:
129 cellName (str): The name of the RAS cell (used for naming the output file).
130 ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
131 model (cobra.Model): The metabolic model to be modified.
132 rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
133 output_folder (str): Folder path where the output CSV file will be saved.
134
135 Returns:
136 None
137 """
138 model_new = model.copy()
139 apply_ras_bounds(model_new, ras_row, rxns_ids)
140 bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
141 bounds.to_csv(output_folder + cellName + ".csv", sep='\t', index=True)
142
143 def generate_bounds(model: cobra.Model, medium: dict, ras=None, output_folder='output/') -> pd.DataFrame:
144 """
145 Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.
146
147 Args:
148 model (cobra.Model): The metabolic model for which bounds will be generated.
149 medium (dict): A dictionary where keys are reaction IDs and values are the medium conditions.
150 ras (pd.DataFrame, optional): A DataFrame with RAS scaling factors for different cell types. Defaults to None.
151 output_folder (str, optional): Folder path where output CSV files will be saved. Defaults to 'output/'.
152
153 Returns:
154 pd.DataFrame: DataFrame containing the bounds of reactions in the model.
155 """
156 rxns_ids = [rxn.id for rxn in model.reactions]
157
158 # Set medium conditions
159 for reaction, value in medium.items():
160 if value is not None:
161 model.reactions.get_by_id(reaction).lower_bound = -float(value)
162
163 # Perform Flux Variability Analysis (FVA)
164 df_FVA = cobra.flux_analysis.flux_variability_analysis(model, fraction_of_optimum=0, processes=1).round(8)
165
166 # Set FVA bounds
167 for reaction in rxns_ids:
168 rxn = model.reactions.get_by_id(reaction)
169 rxn.lower_bound = float(df_FVA.loc[reaction, "minimum"])
170 rxn.upper_bound = float(df_FVA.loc[reaction, "maximum"])
171
172 if ras is not None:
173 Parallel(n_jobs=cpu_count())(delayed(process_ras_cell)(cellName, ras_row, model, rxns_ids, output_folder) for cellName, ras_row in ras.iterrows())
174 else:
175 model_new = model.copy()
176 apply_ras_bounds(model_new, pd.Series([1]*len(rxns_ids), index=rxns_ids), rxns_ids)
177 bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model_new.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
178 bounds.to_csv(output_folder + "bounds.csv", sep='\t', index=True)
179
180
181 ############################# main ###########################################
182 def main() -> None:
183 """
184 Initializes everything and sets the program in motion based on the fronted input arguments.
185
186 Returns:
187 None
188 """
189 if not os.path.exists('ras_to_bounds'):
190 os.makedirs('ras_to_bounds')
191
192
193 global ARGS
194 ARGS = process_args(sys.argv)
195
196 ARGS.output_folder = 'ras_to_bounds/'
197
198 if(ARGS.ras_selector == True):
199 ras = read_dataset(ARGS.input_ras, "ras dataset")
200 ras.replace("None", None, inplace=True)
201 ras.set_index("Reactions", drop=True, inplace=True)
202 ras = ras.T
203 ras = ras.astype(float)
204
205 model_type :utils.Model = ARGS.model_selector
206 if model_type is utils.Model.Custom:
207 model = model_type.getCOBRAmodel(customPath = utils.FilePath.fromStrPath(ARGS.model), customExtension = utils.FilePath.fromStrPath(ARGS.model_name).ext)
208 else:
209 model = model_type.getCOBRAmodel(toolDir=ARGS.tool_dir)
210
211 if(ARGS.medium_selector == "Custom"):
212 medium = read_dataset(ARGS.medium, "medium dataset")
213 medium.set_index(medium.columns[0], inplace=True)
214 medium = medium.astype(float)
215 medium = medium[medium.columns[0]].to_dict()
216 else:
217 df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
218 ARGS.medium_selector = ARGS.medium_selector.replace("_", " ")
219 medium = df_mediums[[ARGS.medium_selector]]
220 medium = medium[ARGS.medium_selector].to_dict()
221
222 if(ARGS.ras_selector == True):
223 generate_bounds(model, medium, ras = ras, output_folder=ARGS.output_folder)
224 else:
225 generate_bounds(model, medium, output_folder=ARGS.output_folder)
226
227 pass
228
229 ##############################################################################
230 if __name__ == "__main__":
231 main()