|
93
|
1 import os
|
|
|
2 import csv
|
|
|
3 import cobra
|
|
|
4 import pickle
|
|
|
5 import argparse
|
|
|
6 import pandas as pd
|
|
|
7 import utils.general_utils as utils
|
|
|
8 import utils.rule_parsing as rulesUtils
|
|
147
|
9 from typing import Optional, Tuple, Union, List, Dict
|
|
93
|
10 import utils.reaction_parsing as reactionUtils
|
|
|
11
|
|
|
12 ARGS : argparse.Namespace
|
|
343
|
13 def process_args(args: List[str] = None) -> argparse.Namespace:
|
|
|
14 """
|
|
|
15 Parse command-line arguments for CustomDataGenerator.
|
|
93
|
16 """
|
|
343
|
17
|
|
|
18 parser = argparse.ArgumentParser(
|
|
|
19 usage="%(prog)s [options]",
|
|
|
20 description="Generate custom data from a given model"
|
|
|
21 )
|
|
93
|
22
|
|
343
|
23 parser.add_argument("--out_log", type=str, required=True,
|
|
|
24 help="Output log file")
|
|
|
25 parser.add_argument("--out_data", type=str, required=True,
|
|
|
26 help="Single output dataset (CSV or Excel)")
|
|
93
|
27
|
|
343
|
28 parser.add_argument("--model", type=str,
|
|
|
29 help="Built-in model identifier (e.g., ENGRO2, Recon, HMRcore)")
|
|
|
30 parser.add_argument("--input", type=str,
|
|
|
31 help="Custom model file (JSON or XML)")
|
|
|
32 parser.add_argument("--name", type=str, required=True,
|
|
|
33 help="Model name (default or custom)")
|
|
93
|
34
|
|
343
|
35 parser.add_argument("--medium_selector", type=str, required=True,
|
|
|
36 help="Medium selection option (default/custom)")
|
|
|
37 parser.add_argument("--medium", type=str,
|
|
|
38 help="Custom medium file if medium_selector=Custom")
|
|
|
39
|
|
|
40 parser.add_argument("--output_format", type=str, choices=["tabular", "xlsx"], required=True,
|
|
|
41 help="Output format: CSV (tabular) or Excel (xlsx)")
|
|
|
42
|
|
|
43 parser.add_argument('-idop', '--output_path', type = str, default='result',
|
|
|
44 help = 'output path for the result files (default: result)')
|
|
|
45
|
|
93
|
46
|
|
343
|
47 return parser.parse_args(args)
|
|
93
|
48
|
|
|
49 ################################- INPUT DATA LOADING -################################
|
|
|
50 def load_custom_model(file_path :utils.FilePath, ext :Optional[utils.FileFormat] = None) -> cobra.Model:
|
|
|
51 """
|
|
|
52 Loads a custom model from a file, either in JSON or XML format.
|
|
|
53
|
|
|
54 Args:
|
|
|
55 file_path : The path to the file containing the custom model.
|
|
|
56 ext : explicit file extension. Necessary for standard use in galaxy because of its weird behaviour.
|
|
|
57
|
|
|
58 Raises:
|
|
|
59 DataErr : if the file is in an invalid format or cannot be opened for whatever reason.
|
|
|
60
|
|
|
61 Returns:
|
|
|
62 cobra.Model : the model, if successfully opened.
|
|
|
63 """
|
|
|
64 ext = ext if ext else file_path.ext
|
|
|
65 try:
|
|
|
66 if ext is utils.FileFormat.XML:
|
|
|
67 return cobra.io.read_sbml_model(file_path.show())
|
|
|
68
|
|
|
69 if ext is utils.FileFormat.JSON:
|
|
|
70 return cobra.io.load_json_model(file_path.show())
|
|
|
71
|
|
|
72 except Exception as e: raise utils.DataErr(file_path, e.__str__())
|
|
|
73 raise utils.DataErr(file_path,
|
|
|
74 f"Formato \"{file_path.ext}\" non riconosciuto, sono supportati solo file JSON e XML")
|
|
|
75
|
|
|
76 ################################- DATA GENERATION -################################
|
|
|
77 ReactionId = str
|
|
|
78 def generate_rules(model: cobra.Model, *, asParsed = True) -> Union[Dict[ReactionId, rulesUtils.OpList], Dict[ReactionId, str]]:
|
|
|
79 """
|
|
|
80 Generates a dictionary mapping reaction ids to rules from the model.
|
|
|
81
|
|
|
82 Args:
|
|
|
83 model : the model to derive data from.
|
|
|
84 asParsed : if True parses the rules to an optimized runtime format, otherwise leaves them as strings.
|
|
|
85
|
|
|
86 Returns:
|
|
|
87 Dict[ReactionId, rulesUtils.OpList] : the generated dictionary of parsed rules.
|
|
|
88 Dict[ReactionId, str] : the generated dictionary of raw rules.
|
|
|
89 """
|
|
|
90 # Is the below approach convoluted? yes
|
|
|
91 # Ok but is it inefficient? probably
|
|
|
92 # Ok but at least I don't have to repeat the check at every rule (I'm clinically insane)
|
|
|
93 _ruleGetter = lambda reaction : reaction.gene_reaction_rule
|
|
|
94 ruleExtractor = (lambda reaction :
|
|
|
95 rulesUtils.parseRuleToNestedList(_ruleGetter(reaction))) if asParsed else _ruleGetter
|
|
|
96
|
|
|
97 return {
|
|
|
98 reaction.id : ruleExtractor(reaction)
|
|
|
99 for reaction in model.reactions
|
|
|
100 if reaction.gene_reaction_rule }
|
|
|
101
|
|
|
102 def generate_reactions(model :cobra.Model, *, asParsed = True) -> Dict[ReactionId, str]:
|
|
|
103 """
|
|
|
104 Generates a dictionary mapping reaction ids to reaction formulas from the model.
|
|
|
105
|
|
|
106 Args:
|
|
|
107 model : the model to derive data from.
|
|
|
108 asParsed : if True parses the reactions to an optimized runtime format, otherwise leaves them as they are.
|
|
|
109
|
|
|
110 Returns:
|
|
|
111 Dict[ReactionId, str] : the generated dictionary.
|
|
|
112 """
|
|
|
113
|
|
|
114 unparsedReactions = {
|
|
|
115 reaction.id : reaction.reaction
|
|
|
116 for reaction in model.reactions
|
|
|
117 if reaction.reaction
|
|
|
118 }
|
|
|
119
|
|
|
120 if not asParsed: return unparsedReactions
|
|
|
121
|
|
|
122 return reactionUtils.create_reaction_dict(unparsedReactions)
|
|
|
123
|
|
|
124 def get_medium(model:cobra.Model) -> pd.DataFrame:
|
|
|
125 trueMedium=[]
|
|
|
126 for r in model.reactions:
|
|
|
127 positiveCoeff=0
|
|
|
128 for m in r.metabolites:
|
|
|
129 if r.get_coefficient(m.id)>0:
|
|
|
130 positiveCoeff=1;
|
|
|
131 if (positiveCoeff==0 and r.lower_bound<0):
|
|
|
132 trueMedium.append(r.id)
|
|
|
133
|
|
|
134 df_medium = pd.DataFrame()
|
|
|
135 df_medium["reaction"] = trueMedium
|
|
|
136 return df_medium
|
|
|
137
|
|
|
138 def generate_bounds(model:cobra.Model) -> pd.DataFrame:
|
|
|
139
|
|
|
140 rxns = []
|
|
|
141 for reaction in model.reactions:
|
|
|
142 rxns.append(reaction.id)
|
|
|
143
|
|
|
144 bounds = pd.DataFrame(columns = ["lower_bound", "upper_bound"], index=rxns)
|
|
|
145
|
|
|
146 for reaction in model.reactions:
|
|
|
147 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
|
|
|
148 return bounds
|
|
|
149
|
|
|
150
|
|
|
151 ###############################- FILE SAVING -################################
|
|
|
152 def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
|
|
|
153 """
|
|
|
154 Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.
|
|
|
155
|
|
|
156 Args:
|
|
|
157 data : the data to be written to the file.
|
|
|
158 file_path : the path to the .csv file.
|
|
|
159 fieldNames : the names of the fields (columns) in the .csv file.
|
|
|
160
|
|
|
161 Returns:
|
|
|
162 None
|
|
|
163 """
|
|
|
164 with open(file_path.show(), 'w', newline='') as csvfile:
|
|
|
165 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
|
|
|
166 writer.writeheader()
|
|
|
167
|
|
|
168 for key, value in data.items():
|
|
|
169 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
|
|
|
170
|
|
|
171 def save_as_csv(data :dict, file_path :str, fieldNames :Tuple[str, str]) -> None:
|
|
|
172 """
|
|
|
173 Saves any dictionary-shaped data in a .csv file created at the given file_path as string.
|
|
|
174
|
|
|
175 Args:
|
|
|
176 data : the data to be written to the file.
|
|
|
177 file_path : the path to the .csv file.
|
|
|
178 fieldNames : the names of the fields (columns) in the .csv file.
|
|
|
179
|
|
|
180 Returns:
|
|
|
181 None
|
|
|
182 """
|
|
|
183 with open(file_path, 'w', newline='') as csvfile:
|
|
|
184 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
|
|
|
185 writer.writeheader()
|
|
|
186
|
|
|
187 for key, value in data.items():
|
|
|
188 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
|
|
|
189
|
|
|
190 ###############################- ENTRY POINT -################################
|
|
147
|
191 def main(args:List[str] = None) -> None:
|
|
93
|
192 """
|
|
|
193 Initializes everything and sets the program in motion based on the fronted input arguments.
|
|
|
194
|
|
|
195 Returns:
|
|
|
196 None
|
|
|
197 """
|
|
|
198 # get args from frontend (related xml)
|
|
|
199 global ARGS
|
|
147
|
200 ARGS = process_args(args)
|
|
93
|
201
|
|
|
202 # this is the worst thing I've seen so far, congrats to the former MaREA devs for suggesting this!
|
|
343
|
203 if os.path.isdir(ARGS.output_path) == False:
|
|
|
204 os.makedirs(ARGS.output_path)
|
|
|
205
|
|
|
206 if ARGS.input != "None":
|
|
|
207 # load custom model
|
|
|
208 model = load_custom_model(
|
|
|
209 utils.FilePath.fromStrPath(ARGS.input), utils.FilePath.fromStrPath(ARGS.name).ext)
|
|
|
210 else:
|
|
|
211 # load built-in model
|
|
93
|
212
|
|
343
|
213 try:
|
|
|
214 model_enum = utils.Model[ARGS.model] # e.g., Model['ENGRO2']
|
|
|
215 except KeyError:
|
|
|
216 raise utils.ArgsErr("model", "one of Recon/ENGRO2/HMRcore/Custom_model", ARGS.model)
|
|
|
217
|
|
|
218 # Load built-in model (Model.getCOBRAmodel uses tool_dir to locate local models)
|
|
|
219 try:
|
|
|
220 model = model_enum.getCOBRAmodel(toolDir='../../COBRAxy')
|
|
|
221 except Exception as e:
|
|
|
222 # Wrap/normalize load errors as DataErr for consistency
|
|
|
223 raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}")
|
|
|
224
|
|
|
225 # Determine final model name: explicit --name overrides, otherwise use the model id
|
|
|
226 model_name = ARGS.name if ARGS.name else ARGS.model
|
|
93
|
227
|
|
|
228 # generate data
|
|
|
229 rules = generate_rules(model, asParsed = False)
|
|
|
230 reactions = generate_reactions(model, asParsed = False)
|
|
|
231 bounds = generate_bounds(model)
|
|
|
232 medium = get_medium(model)
|
|
|
233
|
|
343
|
234 df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"])
|
|
|
235 df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"])
|
|
|
236
|
|
|
237 df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"})
|
|
|
238 df_medium = medium.rename(columns = {"reaction": "ReactionID"})
|
|
|
239 df_medium["InMedium"] = True # flag per indicare la presenza nel medium
|
|
|
240
|
|
|
241 merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer")
|
|
|
242 merged = merged.merge(df_bounds, on = "ReactionID", how = "outer")
|
|
|
243
|
|
|
244 merged = merged.merge(df_medium, on = "ReactionID", how = "left")
|
|
|
245
|
|
|
246 merged["InMedium"] = merged["InMedium"].fillna(False)
|
|
|
247
|
|
|
248 merged = merged.sort_values(by = "InMedium", ascending = False)
|
|
|
249
|
|
|
250 out_file = os.path.join(ARGS.output_path, f"{os.path.basename(ARGS.name).split('.')[0]}_custom_data")
|
|
|
251
|
|
|
252 #merged.to_csv(out_file, sep = '\t', index = False)
|
|
|
253
|
|
|
254
|
|
|
255 ####
|
|
|
256 out_data_path = out_file #ARGS.out_data
|
|
|
257
|
|
|
258 # If Galaxy provided a .dat name, ensure a correct extension according to output_format
|
|
|
259 if ARGS.output_format == "xlsx":
|
|
|
260 if not out_data_path.lower().endswith(".xlsx"):
|
|
|
261 out_data_path = out_data_path + ".xlsx"
|
|
|
262 merged.to_excel(out_data_path, index=False)
|
|
|
263 else:
|
|
|
264 # 'tabular' -> tab-separated, extension .csv is fine and common for Galaxy tabular
|
|
|
265 if not (out_data_path.lower().endswith(".csv") or out_data_path.lower().endswith(".tsv")):
|
|
|
266 out_data_path = out_data_path + ".csv"
|
|
|
267 merged.to_csv(out_data_path, sep="\t", index=False)
|
|
|
268
|
|
93
|
269
|
|
|
270 if __name__ == '__main__':
|
|
|
271 main() |
