Mercurial > repos > bimib > cobraxy

comparison COBRAxy/custom_data_generator.py @ 4:41f35c2f0c7b draft

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