|
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
|
|
147
|
13 def process_args(args:List[str] = None) -> argparse.Namespace:
|
|
93
|
14 """
|
|
|
15 Interfaces the script of a module with its frontend, making the user's choices for
|
|
|
16 various parameters available as values in code.
|
|
|
17
|
|
|
18 Args:
|
|
|
19 args : Always obtained (in file) from sys.argv
|
|
|
20
|
|
|
21 Returns:
|
|
|
22 Namespace : An object containing the parsed arguments
|
|
|
23 """
|
|
|
24 parser = argparse.ArgumentParser(
|
|
|
25 usage = "%(prog)s [options]",
|
|
|
26 description = "generate custom data from a given model")
|
|
|
27
|
|
|
28 parser.add_argument("-ol", "--out_log", type = str, required = True, help = "Output log")
|
|
|
29
|
|
|
30 parser.add_argument("-orules", "--out_rules", type = str, required = True, help = "Output rules")
|
|
|
31 parser.add_argument("-orxns", "--out_reactions", type = str, required = True, help = "Output reactions")
|
|
|
32 parser.add_argument("-omedium", "--out_medium", type = str, required = True, help = "Output medium")
|
|
|
33 parser.add_argument("-obnds", "--out_bounds", type = str, required = True, help = "Output bounds")
|
|
|
34
|
|
|
35 parser.add_argument("-id", "--input", type = str, required = True, help = "Input model")
|
|
|
36 parser.add_argument("-mn", "--name", type = str, required = True, help = "Input model name")
|
|
|
37 # ^ I need this because galaxy converts my files into .dat but I need to know what extension they were in
|
|
147
|
38 parser.add_argument('-idop', '--output_path', type = str, default='result', help = 'output path for maps')
|
|
|
39 argsNamespace = parser.parse_args(args)
|
|
93
|
40 # ^ can't get this one to work from xml, there doesn't seem to be a way to get the directory attribute from the collection
|
|
|
41
|
|
|
42 return argsNamespace
|
|
|
43
|
|
|
44 ################################- INPUT DATA LOADING -################################
|
|
|
45 def load_custom_model(file_path :utils.FilePath, ext :Optional[utils.FileFormat] = None) -> cobra.Model:
|
|
|
46 """
|
|
|
47 Loads a custom model from a file, either in JSON or XML format.
|
|
|
48
|
|
|
49 Args:
|
|
|
50 file_path : The path to the file containing the custom model.
|
|
|
51 ext : explicit file extension. Necessary for standard use in galaxy because of its weird behaviour.
|
|
|
52
|
|
|
53 Raises:
|
|
|
54 DataErr : if the file is in an invalid format or cannot be opened for whatever reason.
|
|
|
55
|
|
|
56 Returns:
|
|
|
57 cobra.Model : the model, if successfully opened.
|
|
|
58 """
|
|
|
59 ext = ext if ext else file_path.ext
|
|
|
60 try:
|
|
|
61 if ext is utils.FileFormat.XML:
|
|
|
62 return cobra.io.read_sbml_model(file_path.show())
|
|
|
63
|
|
|
64 if ext is utils.FileFormat.JSON:
|
|
|
65 return cobra.io.load_json_model(file_path.show())
|
|
|
66
|
|
|
67 except Exception as e: raise utils.DataErr(file_path, e.__str__())
|
|
|
68 raise utils.DataErr(file_path,
|
|
|
69 f"Formato \"{file_path.ext}\" non riconosciuto, sono supportati solo file JSON e XML")
|
|
|
70
|
|
|
71 ################################- DATA GENERATION -################################
|
|
|
72 ReactionId = str
|
|
|
73 def generate_rules(model: cobra.Model, *, asParsed = True) -> Union[Dict[ReactionId, rulesUtils.OpList], Dict[ReactionId, str]]:
|
|
|
74 """
|
|
|
75 Generates a dictionary mapping reaction ids to rules from the model.
|
|
|
76
|
|
|
77 Args:
|
|
|
78 model : the model to derive data from.
|
|
|
79 asParsed : if True parses the rules to an optimized runtime format, otherwise leaves them as strings.
|
|
|
80
|
|
|
81 Returns:
|
|
|
82 Dict[ReactionId, rulesUtils.OpList] : the generated dictionary of parsed rules.
|
|
|
83 Dict[ReactionId, str] : the generated dictionary of raw rules.
|
|
|
84 """
|
|
|
85 # Is the below approach convoluted? yes
|
|
|
86 # Ok but is it inefficient? probably
|
|
|
87 # Ok but at least I don't have to repeat the check at every rule (I'm clinically insane)
|
|
|
88 _ruleGetter = lambda reaction : reaction.gene_reaction_rule
|
|
|
89 ruleExtractor = (lambda reaction :
|
|
|
90 rulesUtils.parseRuleToNestedList(_ruleGetter(reaction))) if asParsed else _ruleGetter
|
|
|
91
|
|
|
92 return {
|
|
|
93 reaction.id : ruleExtractor(reaction)
|
|
|
94 for reaction in model.reactions
|
|
|
95 if reaction.gene_reaction_rule }
|
|
|
96
|
|
|
97 def generate_reactions(model :cobra.Model, *, asParsed = True) -> Dict[ReactionId, str]:
|
|
|
98 """
|
|
|
99 Generates a dictionary mapping reaction ids to reaction formulas from the model.
|
|
|
100
|
|
|
101 Args:
|
|
|
102 model : the model to derive data from.
|
|
|
103 asParsed : if True parses the reactions to an optimized runtime format, otherwise leaves them as they are.
|
|
|
104
|
|
|
105 Returns:
|
|
|
106 Dict[ReactionId, str] : the generated dictionary.
|
|
|
107 """
|
|
|
108
|
|
|
109 unparsedReactions = {
|
|
|
110 reaction.id : reaction.reaction
|
|
|
111 for reaction in model.reactions
|
|
|
112 if reaction.reaction
|
|
|
113 }
|
|
|
114
|
|
|
115 if not asParsed: return unparsedReactions
|
|
|
116
|
|
|
117 return reactionUtils.create_reaction_dict(unparsedReactions)
|
|
|
118
|
|
|
119 def get_medium(model:cobra.Model) -> pd.DataFrame:
|
|
|
120 trueMedium=[]
|
|
|
121 for r in model.reactions:
|
|
|
122 positiveCoeff=0
|
|
|
123 for m in r.metabolites:
|
|
|
124 if r.get_coefficient(m.id)>0:
|
|
|
125 positiveCoeff=1;
|
|
|
126 if (positiveCoeff==0 and r.lower_bound<0):
|
|
|
127 trueMedium.append(r.id)
|
|
|
128
|
|
|
129 df_medium = pd.DataFrame()
|
|
|
130 df_medium["reaction"] = trueMedium
|
|
|
131 return df_medium
|
|
|
132
|
|
|
133 def generate_bounds(model:cobra.Model) -> pd.DataFrame:
|
|
|
134
|
|
|
135 rxns = []
|
|
|
136 for reaction in model.reactions:
|
|
|
137 rxns.append(reaction.id)
|
|
|
138
|
|
|
139 bounds = pd.DataFrame(columns = ["lower_bound", "upper_bound"], index=rxns)
|
|
|
140
|
|
|
141 for reaction in model.reactions:
|
|
|
142 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
|
|
|
143 return bounds
|
|
|
144
|
|
|
145
|
|
|
146 ###############################- FILE SAVING -################################
|
|
|
147 def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
|
|
|
148 """
|
|
|
149 Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.
|
|
|
150
|
|
|
151 Args:
|
|
|
152 data : the data to be written to the file.
|
|
|
153 file_path : the path to the .csv file.
|
|
|
154 fieldNames : the names of the fields (columns) in the .csv file.
|
|
|
155
|
|
|
156 Returns:
|
|
|
157 None
|
|
|
158 """
|
|
|
159 with open(file_path.show(), 'w', newline='') as csvfile:
|
|
|
160 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
|
|
|
161 writer.writeheader()
|
|
|
162
|
|
|
163 for key, value in data.items():
|
|
|
164 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
|
|
|
165
|
|
|
166 def save_as_csv(data :dict, file_path :str, fieldNames :Tuple[str, str]) -> None:
|
|
|
167 """
|
|
|
168 Saves any dictionary-shaped data in a .csv file created at the given file_path as string.
|
|
|
169
|
|
|
170 Args:
|
|
|
171 data : the data to be written to the file.
|
|
|
172 file_path : the path to the .csv file.
|
|
|
173 fieldNames : the names of the fields (columns) in the .csv file.
|
|
|
174
|
|
|
175 Returns:
|
|
|
176 None
|
|
|
177 """
|
|
|
178 with open(file_path, 'w', newline='') as csvfile:
|
|
|
179 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
|
|
|
180 writer.writeheader()
|
|
|
181
|
|
|
182 for key, value in data.items():
|
|
|
183 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
|
|
|
184
|
|
|
185 ###############################- ENTRY POINT -################################
|
|
147
|
186 def main(args:List[str] = None) -> None:
|
|
93
|
187 """
|
|
|
188 Initializes everything and sets the program in motion based on the fronted input arguments.
|
|
|
189
|
|
|
190 Returns:
|
|
|
191 None
|
|
|
192 """
|
|
|
193 # get args from frontend (related xml)
|
|
|
194 global ARGS
|
|
147
|
195 ARGS = process_args(args)
|
|
93
|
196
|
|
|
197 # this is the worst thing I've seen so far, congrats to the former MaREA devs for suggesting this!
|
|
147
|
198 if os.path.isdir(ARGS.output_path) == False: os.makedirs(ARGS.output_path)
|
|
93
|
199
|
|
|
200 # load custom model
|
|
|
201 model = load_custom_model(
|
|
|
202 utils.FilePath.fromStrPath(ARGS.input), utils.FilePath.fromStrPath(ARGS.name).ext)
|
|
|
203
|
|
|
204 # generate data
|
|
|
205 rules = generate_rules(model, asParsed = False)
|
|
|
206 reactions = generate_reactions(model, asParsed = False)
|
|
|
207 bounds = generate_bounds(model)
|
|
|
208 medium = get_medium(model)
|
|
|
209
|
|
|
210 # save files out of collection: path coming from xml
|
|
|
211 save_as_csv(rules, ARGS.out_rules, ("ReactionID", "Rule"))
|
|
|
212 save_as_csv(reactions, ARGS.out_reactions, ("ReactionID", "Reaction"))
|
|
|
213 bounds.to_csv(ARGS.out_bounds, sep = '\t')
|
|
|
214 medium.to_csv(ARGS.out_medium, sep = '\t')
|
|
|
215
|
|
|
216 if __name__ == '__main__':
|
|
|
217 main() |
