Mercurial > repos > bimib > cobraxy

comparison COBRAxy/custom_data_generator.py @ 93:7e703e546998 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author luca_milaz
date Sun, 13 Oct 2024 11:41:34 +0000
parents
children 3fca9b568faf
comparison
equal deleted inserted replaced
92:fdf713bb5772 93:7e703e546998
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
9 from typing import Optional, Tuple, Union, Dict
10 import utils.reaction_parsing as reactionUtils
11
12 ARGS : argparse.Namespace
13 def process_args() -> argparse.Namespace:
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
38
39 argsNamespace = parser.parse_args()
40 argsNamespace.out_dir = "result"
41 # ^ 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
42
43 return argsNamespace
44
45 ################################- INPUT DATA LOADING -################################
46 def load_custom_model(file_path :utils.FilePath, ext :Optional[utils.FileFormat] = None) -> cobra.Model:
47 """
48 Loads a custom model from a file, either in JSON or XML format.
49
50 Args:
51 file_path : The path to the file containing the custom model.
52 ext : explicit file extension. Necessary for standard use in galaxy because of its weird behaviour.
53
54 Raises:
55 DataErr : if the file is in an invalid format or cannot be opened for whatever reason.
56
57 Returns:
58 cobra.Model : the model, if successfully opened.
59 """
60 ext = ext if ext else file_path.ext
61 try:
62 if ext is utils.FileFormat.XML:
63 return cobra.io.read_sbml_model(file_path.show())
64
65 if ext is utils.FileFormat.JSON:
66 return cobra.io.load_json_model(file_path.show())
67
68 except Exception as e: raise utils.DataErr(file_path, e.__str__())
69 raise utils.DataErr(file_path,
70 f"Formato \"{file_path.ext}\" non riconosciuto, sono supportati solo file JSON e XML")
71
72 ################################- DATA GENERATION -################################
73 ReactionId = str
74 def generate_rules(model: cobra.Model, *, asParsed = True) -> Union[Dict[ReactionId, rulesUtils.OpList], Dict[ReactionId, str]]:
75 """
76 Generates a dictionary mapping reaction ids to rules from the model.
77
78 Args:
79 model : the model to derive data from.
80 asParsed : if True parses the rules to an optimized runtime format, otherwise leaves them as strings.
81
82 Returns:
83 Dict[ReactionId, rulesUtils.OpList] : the generated dictionary of parsed rules.
84 Dict[ReactionId, str] : the generated dictionary of raw rules.
85 """
86 # Is the below approach convoluted? yes
87 # Ok but is it inefficient? probably
88 # Ok but at least I don't have to repeat the check at every rule (I'm clinically insane)
89 _ruleGetter = lambda reaction : reaction.gene_reaction_rule
90 ruleExtractor = (lambda reaction :
91 rulesUtils.parseRuleToNestedList(_ruleGetter(reaction))) if asParsed else _ruleGetter
92
93 return {
94 reaction.id : ruleExtractor(reaction)
95 for reaction in model.reactions
96 if reaction.gene_reaction_rule }
97
98 def generate_reactions(model :cobra.Model, *, asParsed = True) -> Dict[ReactionId, str]:
99 """
100 Generates a dictionary mapping reaction ids to reaction formulas from the model.
101
102 Args:
103 model : the model to derive data from.
104 asParsed : if True parses the reactions to an optimized runtime format, otherwise leaves them as they are.
105
106 Returns:
107 Dict[ReactionId, str] : the generated dictionary.
108 """
109
110 unparsedReactions = {
111 reaction.id : reaction.reaction
112 for reaction in model.reactions
113 if reaction.reaction
114 }
115
116 if not asParsed: return unparsedReactions
117
118 return reactionUtils.create_reaction_dict(unparsedReactions)
119
120 def get_medium(model:cobra.Model) -> pd.DataFrame:
121 trueMedium=[]
122 for r in model.reactions:
123 positiveCoeff=0
124 for m in r.metabolites:
125 if r.get_coefficient(m.id)>0:
126 positiveCoeff=1;
127 if (positiveCoeff==0 and r.lower_bound<0):
128 trueMedium.append(r.id)
129
130 df_medium = pd.DataFrame()
131 df_medium["reaction"] = trueMedium
132 return df_medium
133
134 def generate_bounds(model:cobra.Model) -> pd.DataFrame:
135
136 rxns = []
137 for reaction in model.reactions:
138 rxns.append(reaction.id)
139
140 bounds = pd.DataFrame(columns = ["lower_bound", "upper_bound"], index=rxns)
141
142 for reaction in model.reactions:
143 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
144 return bounds
145
146
147 ###############################- FILE SAVING -################################
148 def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
149 """
150 Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.
151
152 Args:
153 data : the data to be written to the file.
154 file_path : the path to the .csv file.
155 fieldNames : the names of the fields (columns) in the .csv file.
156
157 Returns:
158 None
159 """
160 with open(file_path.show(), 'w', newline='') as csvfile:
161 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
162 writer.writeheader()
163
164 for key, value in data.items():
165 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
166
167 def save_as_csv(data :dict, file_path :str, fieldNames :Tuple[str, str]) -> None:
168 """
169 Saves any dictionary-shaped data in a .csv file created at the given file_path as string.
170
171 Args:
172 data : the data to be written to the file.
173 file_path : the path to the .csv file.
174 fieldNames : the names of the fields (columns) in the .csv file.
175
176 Returns:
177 None
178 """
179 with open(file_path, 'w', newline='') as csvfile:
180 writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab")
181 writer.writeheader()
182
183 for key, value in data.items():
184 writer.writerow({ fieldNames[0] : key, fieldNames[1] : value })
185
186 ###############################- ENTRY POINT -################################
187 def main() -> None:
188 """
189 Initializes everything and sets the program in motion based on the fronted input arguments.
190
191 Returns:
192 None
193 """
194 # get args from frontend (related xml)
195 global ARGS
196 ARGS = process_args()
197
198 # this is the worst thing I've seen so far, congrats to the former MaREA devs for suggesting this!
199 if os.path.isdir(ARGS.out_dir) == False: os.makedirs(ARGS.out_dir)
200
201 # load custom model
202 model = load_custom_model(
203 utils.FilePath.fromStrPath(ARGS.input), utils.FilePath.fromStrPath(ARGS.name).ext)
204
205 # generate data
206 rules = generate_rules(model, asParsed = False)
207 reactions = generate_reactions(model, asParsed = False)
208 bounds = generate_bounds(model)
209 medium = get_medium(model)
210
211 # save files out of collection: path coming from xml
212 save_as_csv(rules, ARGS.out_rules, ("ReactionID", "Rule"))
213 save_as_csv(reactions, ARGS.out_reactions, ("ReactionID", "Reaction"))
214 bounds.to_csv(ARGS.out_bounds, sep = '\t')
215 medium.to_csv(ARGS.out_medium, sep = '\t')
216
217 if __name__ == '__main__':
218 main()