|
88
|
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
|
|
|
10 ################################# process args ###############################
|
|
|
11 def process_args(args :List[str]) -> argparse.Namespace:
|
|
|
12 """
|
|
|
13 Processes command-line arguments.
|
|
|
14
|
|
|
15 Args:
|
|
|
16 args (list): List of command-line arguments.
|
|
|
17
|
|
|
18 Returns:
|
|
|
19 Namespace: An object containing parsed arguments.
|
|
|
20 """
|
|
|
21 parser = argparse.ArgumentParser(usage = '%(prog)s [options]',
|
|
|
22 description = 'process some value\'s')
|
|
|
23
|
|
|
24 parser.add_argument(
|
|
|
25 '-ms', '--model_selector',
|
|
|
26 type = utils.Model, default = utils.Model.ENGRO2, choices = [utils.Model.ENGRO2, utils.Model.Custom],
|
|
|
27 help = 'chose which type of model you want use')
|
|
|
28
|
|
|
29 parser.add_argument("-mo", "--model", type = str,
|
|
|
30 help = "path to input file with custom rules, if provided")
|
|
|
31
|
|
|
32 parser.add_argument("-mn", "--model_name", type = str, help = "custom mode name")
|
|
|
33
|
|
|
34 parser.add_argument(
|
|
|
35 '-mes', '--medium_selector',
|
|
98
|
36 default = "allOpen",
|
|
88
|
37 help = 'chose which type of medium you want use')
|
|
|
38
|
|
|
39 parser.add_argument("-meo", "--medium", type = str,
|
|
|
40 help = "path to input file with custom medium, if provided")
|
|
|
41
|
|
|
42 parser.add_argument('-ol', '--out_log',
|
|
|
43 help = "Output log")
|
|
|
44
|
|
|
45 parser.add_argument('-td', '--tool_dir',
|
|
|
46 type = str,
|
|
|
47 required = True,
|
|
|
48 help = 'your tool directory')
|
|
|
49
|
|
|
50 parser.add_argument('-ir', '--input_ras',
|
|
120
|
51 type=str,
|
|
98
|
52 required = False,
|
|
|
53 help = 'input ras')
|
|
|
54
|
|
|
55 parser.add_argument('-rs', '--ras_selector',
|
|
88
|
56 required = True,
|
|
107
|
57 type=bool,
|
|
98
|
58 help = 'ras selector')
|
|
88
|
59
|
|
|
60 ARGS = parser.parse_args()
|
|
|
61 return ARGS
|
|
|
62
|
|
|
63 ########################### warning ###########################################
|
|
|
64 def warning(s :str) -> None:
|
|
|
65 """
|
|
|
66 Log a warning message to an output log file and print it to the console.
|
|
|
67
|
|
|
68 Args:
|
|
|
69 s (str): The warning message to be logged and printed.
|
|
|
70
|
|
|
71 Returns:
|
|
|
72 None
|
|
|
73 """
|
|
|
74 with open(ARGS.out_log, 'a') as log:
|
|
|
75 log.write(s + "\n\n")
|
|
|
76 print(s)
|
|
|
77
|
|
115
|
78 ############################ dataset input ####################################
|
|
|
79 def read_dataset(data :str, name :str) -> pd.DataFrame:
|
|
|
80 """
|
|
|
81 Read a dataset from a CSV file and return it as a pandas DataFrame.
|
|
|
82
|
|
|
83 Args:
|
|
|
84 data (str): Path to the CSV file containing the dataset.
|
|
|
85 name (str): Name of the dataset, used in error messages.
|
|
|
86
|
|
|
87 Returns:
|
|
|
88 pandas.DataFrame: DataFrame containing the dataset.
|
|
|
89
|
|
|
90 Raises:
|
|
|
91 pd.errors.EmptyDataError: If the CSV file is empty.
|
|
|
92 sys.exit: If the CSV file has the wrong format, the execution is aborted.
|
|
|
93 """
|
|
|
94 try:
|
|
|
95 dataset = pd.read_csv(data, sep = '\t', header = 0, engine='python')
|
|
|
96 except pd.errors.EmptyDataError:
|
|
|
97 sys.exit('Execution aborted: wrong format of ' + name + '\n')
|
|
|
98 if len(dataset.columns) < 2:
|
|
|
99 sys.exit('Execution aborted: wrong format of ' + name + '\n')
|
|
|
100 return dataset
|
|
|
101
|
|
107
|
102 def generate_bounds(model:cobra.Model, medium:dict, ras=None) -> pd.DataFrame:
|
|
|
103 model_new = model.copy()
|
|
|
104 rxns_ids = []
|
|
|
105 for rxn in model.reactions:
|
|
|
106 rxns_ids.append(rxn.id)
|
|
|
107 for reaction in medium.keys():
|
|
|
108 if(medium[reaction] != None):
|
|
|
109 model_new.reactions.get_by_id(reaction).lower_bound=-float(medium[reaction])
|
|
|
110 df_FVA = cobra.flux_analysis.flux_variability_analysis(model_new,fraction_of_optimum=0,processes=1).round(8)
|
|
|
111 for reaction in rxns_ids:
|
|
|
112 model_new.reactions.get_by_id(reaction).lower_bound=float(df_FVA.loc[reaction,"minimum"])
|
|
|
113 model_new.reactions.get_by_id(reaction).upper_bound=float(df_FVA.loc[reaction,"maximum"])
|
|
|
114
|
|
116
|
115 if(ras is not None):#iterate over cells
|
|
107
|
116 for reaction in rxns_ids:
|
|
|
117 if reaction in ras.keys():
|
|
|
118 lower_bound=model_new.reactions.get_by_id(reaction).lower_bound
|
|
|
119 upper_bound=model_new.reactions.get_by_id(reaction).upper_bound
|
|
|
120 valMax=float((upper_bound)*ras[reaction])
|
|
|
121 valMin=float((lower_bound)*ras[reaction])
|
|
|
122 if upper_bound!=0 and lower_bound==0:
|
|
|
123 model_new.reactions.get_by_id(reaction).upper_bound=valMax
|
|
|
124 if upper_bound==0 and lower_bound!=0:
|
|
|
125 model_new.reactions.get_by_id(reaction).lower_bound=valMin
|
|
|
126 if upper_bound!=0 and lower_bound!=0:
|
|
|
127 model_new.reactions.get_by_id(reaction).lower_bound=valMin
|
|
|
128 model_new.reactions.get_by_id(reaction).upper_bound=valMax
|
|
|
129 rxns = []
|
|
|
130 for reaction in model_new.reactions:
|
|
|
131 rxns.append(reaction.id)
|
|
88
|
132
|
|
107
|
133 bounds = pd.DataFrame(columns = ["lower_bound", "upper_bound"], index=rxns)
|
|
88
|
134
|
|
107
|
135 for reaction in model.reactions:
|
|
|
136 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
|
|
|
137 return bounds
|
|
88
|
138
|
|
|
139
|
|
|
140 ############################# main ###########################################
|
|
|
141 def main() -> None:
|
|
|
142 """
|
|
|
143 Initializes everything and sets the program in motion based on the fronted input arguments.
|
|
|
144
|
|
|
145 Returns:
|
|
|
146 None
|
|
|
147 """
|
|
107
|
148 if not os.path.exists('ras_to_bounds'):
|
|
|
149 os.makedirs('ras_to_bounds')
|
|
88
|
150
|
|
111
|
151 global ARGS
|
|
113
|
152 ARGS = process_args(sys.argv)
|
|
88
|
153
|
|
111
|
154 ARGS.output_folder = 'ras_to_bounds/'
|
|
88
|
155
|
|
117
|
156 utils.logWarning(
|
|
|
157 ARGS.input_ras,
|
|
|
158 ARGS.out_log)
|
|
116
|
159
|
|
107
|
160 boundsPath = utils.FilePath("bounds", ".csv", prefix = ARGS.output_folder)
|
|
|
161 mediumPath = utils.FilePath("medium", ".csv", prefix = ARGS.output_folder)
|
|
88
|
162
|
|
107
|
163 if(ARGS.ras_selector == True):
|
|
115
|
164 ras = read_dataset(ARGS.input_ras, "ras dataset")
|
|
107
|
165 ras.replace("None", None, inplace=True)
|
|
|
166 ras = ras.astype(float)
|
|
|
167
|
|
88
|
168 model_type :utils.Model = ARGS.model_selector
|
|
|
169 if model_type is utils.Model.Custom:
|
|
|
170 model = model_type.getCOBRAmodel(customPath = utils.FilePath.fromStrPath(ARGS.model), customExtension = utils.FilePath.fromStrPath(ARGS.model_name).ext)
|
|
|
171 else:
|
|
|
172 model = model_type.getCOBRAmodel(toolDir=ARGS.tool_dir)
|
|
|
173
|
|
107
|
174 if(ARGS.medium_selector == "Custom"):
|
|
115
|
175 medium = read_dataset(ARGS.input_medium, "medium dataset")
|
|
107
|
176 medium = medium.astype(float)
|
|
|
177 medium = medium['medium'].to_dict()
|
|
|
178 else:
|
|
|
179 df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
|
|
|
180 medium = df_mediums[[ARGS.medium_selector]]
|
|
|
181 medium = medium[ARGS.medium_selector].to_dict()
|
|
88
|
182
|
|
107
|
183 if(ARGS.ras_selector == True):
|
|
|
184 bounds = generate_bounds(model, medium, ras = ras)
|
|
|
185 else:
|
|
|
186 bounds = generate_bounds(model, medium)
|
|
|
187
|
|
|
188 bounds.to_csv(boundsPath.show())
|
|
|
189 medium.to_csv(mediumPath.show())
|
|
88
|
190 pass
|
|
|
191
|
|
|
192 ##############################################################################
|
|
|
193 if __name__ == "__main__":
|
|
|
194 main() |
