|
406
|
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] = None) -> 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('-rn', '--name',
|
|
|
58 type=str,
|
|
|
59 help = 'ras class names')
|
|
|
60
|
|
|
61 parser.add_argument('-rs', '--ras_selector',
|
|
|
62 required = True,
|
|
|
63 type=utils.Bool("using_RAS"),
|
|
|
64 help = 'ras selector')
|
|
|
65
|
|
|
66 parser.add_argument('-cc', '--cell_class',
|
|
|
67 type = str,
|
|
|
68 help = 'output of cell class')
|
|
|
69 parser.add_argument(
|
|
|
70 '-idop', '--output_path',
|
|
|
71 type = str,
|
|
|
72 default='ras_to_bounds/',
|
|
|
73 help = 'output path for maps')
|
|
|
74
|
|
|
75
|
|
|
76 ARGS = parser.parse_args(args)
|
|
|
77 return ARGS
|
|
|
78
|
|
|
79 ########################### warning ###########################################
|
|
|
80 def warning(s :str) -> None:
|
|
|
81 """
|
|
|
82 Log a warning message to an output log file and print it to the console.
|
|
|
83
|
|
|
84 Args:
|
|
|
85 s (str): The warning message to be logged and printed.
|
|
|
86
|
|
|
87 Returns:
|
|
|
88 None
|
|
|
89 """
|
|
|
90 with open(ARGS.out_log, 'a') as log:
|
|
|
91 log.write(s + "\n\n")
|
|
|
92 print(s)
|
|
|
93
|
|
|
94 ############################ dataset input ####################################
|
|
|
95 def read_dataset(data :str, name :str) -> pd.DataFrame:
|
|
|
96 """
|
|
|
97 Read a dataset from a CSV file and return it as a pandas DataFrame.
|
|
|
98
|
|
|
99 Args:
|
|
|
100 data (str): Path to the CSV file containing the dataset.
|
|
|
101 name (str): Name of the dataset, used in error messages.
|
|
|
102
|
|
|
103 Returns:
|
|
|
104 pandas.DataFrame: DataFrame containing the dataset.
|
|
|
105
|
|
|
106 Raises:
|
|
|
107 pd.errors.EmptyDataError: If the CSV file is empty.
|
|
|
108 sys.exit: If the CSV file has the wrong format, the execution is aborted.
|
|
|
109 """
|
|
|
110 try:
|
|
|
111 dataset = pd.read_csv(data, sep = '\t', header = 0, engine='python')
|
|
|
112 except pd.errors.EmptyDataError:
|
|
|
113 sys.exit('Execution aborted: wrong format of ' + name + '\n')
|
|
|
114 if len(dataset.columns) < 2:
|
|
|
115 sys.exit('Execution aborted: wrong format of ' + name + '\n')
|
|
|
116 return dataset
|
|
|
117
|
|
|
118
|
|
|
119 def apply_ras_bounds(bounds, ras_row):
|
|
|
120 """
|
|
|
121 Adjust the bounds of reactions in the model based on RAS values.
|
|
|
122
|
|
|
123 Args:
|
|
|
124 bounds (pd.DataFrame): Model bounds.
|
|
|
125 ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
|
|
|
126 Returns:
|
|
|
127 new_bounds (pd.DataFrame): integrated bounds.
|
|
|
128 """
|
|
|
129 new_bounds = bounds.copy()
|
|
|
130 for reaction in ras_row.index:
|
|
|
131 scaling_factor = ras_row[reaction]
|
|
|
132 if not np.isnan(scaling_factor):
|
|
|
133 lower_bound=bounds.loc[reaction, "lower_bound"]
|
|
|
134 upper_bound=bounds.loc[reaction, "upper_bound"]
|
|
|
135 valMax=float((upper_bound)*scaling_factor)
|
|
|
136 valMin=float((lower_bound)*scaling_factor)
|
|
|
137 if upper_bound!=0 and lower_bound==0:
|
|
|
138 new_bounds.loc[reaction, "upper_bound"] = valMax
|
|
|
139 if upper_bound==0 and lower_bound!=0:
|
|
|
140 new_bounds.loc[reaction, "lower_bound"] = valMin
|
|
|
141 if upper_bound!=0 and lower_bound!=0:
|
|
|
142 new_bounds.loc[reaction, "lower_bound"] = valMin
|
|
|
143 new_bounds.loc[reaction, "upper_bound"] = valMax
|
|
|
144 return new_bounds
|
|
|
145
|
|
|
146 def process_ras_cell(cellName, ras_row, model, rxns_ids, output_folder):
|
|
|
147 """
|
|
|
148 Process a single RAS cell, apply bounds, and save the bounds to a CSV file.
|
|
|
149
|
|
|
150 Args:
|
|
|
151 cellName (str): The name of the RAS cell (used for naming the output file).
|
|
|
152 ras_row (pd.Series): A row from a RAS DataFrame containing scaling factors for reaction bounds.
|
|
|
153 model (cobra.Model): The metabolic model to be modified.
|
|
|
154 rxns_ids (list of str): List of reaction IDs to which the scaling factors will be applied.
|
|
|
155 output_folder (str): Folder path where the output CSV file will be saved.
|
|
|
156
|
|
|
157 Returns:
|
|
|
158 None
|
|
|
159 """
|
|
|
160 bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
|
|
|
161 new_bounds = apply_ras_bounds(bounds, ras_row)
|
|
|
162 new_bounds.to_csv(output_folder + cellName + ".csv", sep='\t', index=True)
|
|
|
163 pass
|
|
|
164
|
|
|
165 def generate_bounds(model: cobra.Model, medium: dict, ras=None, output_folder='output/') -> pd.DataFrame:
|
|
|
166 """
|
|
|
167 Generate reaction bounds for a metabolic model based on medium conditions and optional RAS adjustments.
|
|
|
168
|
|
|
169 Args:
|
|
|
170 model (cobra.Model): The metabolic model for which bounds will be generated.
|
|
|
171 medium (dict): A dictionary where keys are reaction IDs and values are the medium conditions.
|
|
|
172 ras (pd.DataFrame, optional): RAS pandas dataframe. Defaults to None.
|
|
|
173 output_folder (str, optional): Folder path where output CSV files will be saved. Defaults to 'output/'.
|
|
|
174
|
|
|
175 Returns:
|
|
|
176 pd.DataFrame: DataFrame containing the bounds of reactions in the model.
|
|
|
177 """
|
|
|
178 rxns_ids = [rxn.id for rxn in model.reactions]
|
|
|
179
|
|
|
180 # Set all reactions to zero in the medium
|
|
|
181 for rxn_id, _ in model.medium.items():
|
|
|
182 model.reactions.get_by_id(rxn_id).lower_bound = float(0.0)
|
|
|
183
|
|
|
184 # Set medium conditions
|
|
|
185 for reaction, value in medium.items():
|
|
|
186 if value is not None:
|
|
|
187 model.reactions.get_by_id(reaction).lower_bound = -float(value)
|
|
|
188
|
|
|
189
|
|
|
190 # Perform Flux Variability Analysis (FVA) on this medium
|
|
|
191 df_FVA = cobra.flux_analysis.flux_variability_analysis(model, fraction_of_optimum=0, processes=1).round(8)
|
|
|
192
|
|
|
193 # Set FVA bounds
|
|
|
194 for reaction in rxns_ids:
|
|
|
195 model.reactions.get_by_id(reaction).lower_bound = float(df_FVA.loc[reaction, "minimum"])
|
|
|
196 model.reactions.get_by_id(reaction).upper_bound = float(df_FVA.loc[reaction, "maximum"])
|
|
|
197
|
|
|
198 if ras is not None:
|
|
|
199 Parallel(n_jobs=cpu_count())(delayed(process_ras_cell)(cellName, ras_row, model, rxns_ids, output_folder) for cellName, ras_row in ras.iterrows())
|
|
|
200 else:
|
|
|
201 bounds = pd.DataFrame([(rxn.lower_bound, rxn.upper_bound) for rxn in model.reactions], index=rxns_ids, columns=["lower_bound", "upper_bound"])
|
|
|
202 newBounds = apply_ras_bounds(bounds, pd.Series([1]*len(rxns_ids), index=rxns_ids))
|
|
|
203 newBounds.to_csv(output_folder + "bounds.csv", sep='\t', index=True)
|
|
|
204 pass
|
|
|
205
|
|
|
206
|
|
|
207
|
|
|
208 ############################# main ###########################################
|
|
|
209 def main(args:List[str] = None) -> None:
|
|
|
210 """
|
|
|
211 Initializes everything and sets the program in motion based on the fronted input arguments.
|
|
|
212
|
|
|
213 Returns:
|
|
|
214 None
|
|
|
215 """
|
|
|
216 if not os.path.exists('ras_to_bounds'):
|
|
|
217 os.makedirs('ras_to_bounds')
|
|
|
218
|
|
|
219
|
|
|
220 global ARGS
|
|
|
221 ARGS = process_args(args)
|
|
|
222
|
|
|
223 if(ARGS.ras_selector == True):
|
|
|
224 ras_file_list = ARGS.input_ras.split(",")
|
|
|
225 ras_file_names = ARGS.name.split(",")
|
|
|
226 if len(ras_file_names) != len(set(ras_file_names)):
|
|
|
227 error_message = "Duplicated file names in the uploaded RAS matrices."
|
|
|
228 warning(error_message)
|
|
|
229 raise ValueError(error_message)
|
|
|
230 pass
|
|
|
231 ras_class_names = []
|
|
|
232 for file in ras_file_names:
|
|
|
233 ras_class_names.append(file.rsplit(".", 1)[0])
|
|
|
234 ras_list = []
|
|
|
235 class_assignments = pd.DataFrame(columns=["Patient_ID", "Class"])
|
|
|
236 for ras_matrix, ras_class_name in zip(ras_file_list, ras_class_names):
|
|
|
237 ras = read_dataset(ras_matrix, "ras dataset")
|
|
|
238 ras.replace("None", None, inplace=True)
|
|
|
239 ras.set_index("Reactions", drop=True, inplace=True)
|
|
|
240 ras = ras.T
|
|
|
241 ras = ras.astype(float)
|
|
|
242 if(len(ras_file_list)>1):
|
|
|
243 #append class name to patient id (dataframe index)
|
|
|
244 ras.index = [f"{idx}_{ras_class_name}" for idx in ras.index]
|
|
|
245 else:
|
|
|
246 ras.index = [f"{idx}" for idx in ras.index]
|
|
|
247 ras_list.append(ras)
|
|
|
248 for patient_id in ras.index:
|
|
|
249 class_assignments.loc[class_assignments.shape[0]] = [patient_id, ras_class_name]
|
|
|
250
|
|
|
251
|
|
|
252 # Concatenate all ras DataFrames into a single DataFrame
|
|
|
253 ras_combined = pd.concat(ras_list, axis=0)
|
|
|
254 # Normalize the RAS values by max RAS
|
|
|
255 ras_combined = ras_combined.div(ras_combined.max(axis=0))
|
|
|
256 ras_combined.dropna(axis=1, how='all', inplace=True)
|
|
|
257
|
|
|
258
|
|
|
259
|
|
|
260 model_type :utils.Model = ARGS.model_selector
|
|
|
261 if model_type is utils.Model.Custom:
|
|
|
262 model = model_type.getCOBRAmodel(customPath = utils.FilePath.fromStrPath(ARGS.model), customExtension = utils.FilePath.fromStrPath(ARGS.model_name).ext)
|
|
|
263 else:
|
|
|
264 model = model_type.getCOBRAmodel(toolDir=ARGS.tool_dir)
|
|
|
265
|
|
|
266 if(ARGS.medium_selector == "Custom"):
|
|
|
267 medium = read_dataset(ARGS.medium, "medium dataset")
|
|
|
268 medium.set_index(medium.columns[0], inplace=True)
|
|
|
269 medium = medium.astype(float)
|
|
|
270 medium = medium[medium.columns[0]].to_dict()
|
|
|
271 else:
|
|
|
272 df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
|
|
|
273 ARGS.medium_selector = ARGS.medium_selector.replace("_", " ")
|
|
|
274 medium = df_mediums[[ARGS.medium_selector]]
|
|
|
275 medium = medium[ARGS.medium_selector].to_dict()
|
|
|
276
|
|
|
277 if(ARGS.ras_selector == True):
|
|
|
278 generate_bounds(model, medium, ras = ras_combined, output_folder=ARGS.output_path)
|
|
|
279 class_assignments.to_csv(ARGS.cell_class, sep = '\t', index = False)
|
|
|
280 else:
|
|
|
281 generate_bounds(model, medium, output_folder=ARGS.output_path)
|
|
|
282
|
|
|
283 pass
|
|
|
284
|
|
|
285 ##############################################################################
|
|
|
286 if __name__ == "__main__":
|
|
|
287 main() |
