Mercurial > repos > bimib > cobraxy

comparison COBRAxy/custom_data_generator.py @ 403:05092b0cfca0 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Uploaded
author francesco_lapi
date Mon, 08 Sep 2025 13:38:59 +0000
parents de4a373e338b
children 08f1ff359397
comparison
equal deleted inserted replaced
402:ccccb731c953 403:05092b0cfca0
145 for reaction in model.reactions: 145 for reaction in model.reactions:
146 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound] 146 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound]
147 return bounds 147 return bounds
148 148
149 149
150
151 def generate_compartments(model: cobra.Model) -> pd.DataFrame:
152 """
153 Generates a DataFrame containing compartment information for each reaction.
154 Creates columns for each compartment position (Compartment_1, Compartment_2, etc.)
155
156 Args:
157 model: the COBRA model to extract compartment data from.
158
159 Returns:
160 pd.DataFrame: DataFrame with ReactionID and compartment columns
161 """
162 compartment_data = []
163
164 # First pass: determine the maximum number of compartments any reaction has
165 max_compartments = 0
166 reaction_compartments = {}
167
168 for reaction in model.reactions:
169 # Get unique compartments from all metabolites in the reaction
170 if type(reaction.annotation['pathways']) == list:
171 reaction_compartments[reaction.id] = reaction.annotation['pathways']
172 max_compartments = max(max_compartments, len(reaction.annotation['pathways']))
173 else:
174 reaction_compartments[reaction.id] = [reaction.annotation['pathways']]
175
176 # Create column names for compartments
177 compartment_columns = [f"Compartment_{i+1}" for i in range(max_compartments)]
178
179 # Second pass: create the data
180 for reaction_id, compartments in reaction_compartments.items():
181 row = {"ReactionID": reaction_id}
182
183 # Fill compartment columns
184 for i in range(max_compartments):
185 col_name = compartment_columns[i]
186 if i < len(compartments):
187 row[col_name] = compartments[i]
188
189 else:
190 row[col_name] = None # or "" if you prefer empty strings
191
192 compartment_data.append(row)
193
194 return pd.DataFrame(compartment_data)
195
196
150 ###############################- FILE SAVING -################################ 197 ###############################- FILE SAVING -################################
151 def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None: 198 def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None:
152 """ 199 """
153 Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath. 200 Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath.
154 201
227 raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}") 274 raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}")
228 275
229 # Determine final model name: explicit --name overrides, otherwise use the model id 276 # Determine final model name: explicit --name overrides, otherwise use the model id
230 277
231 model_name = ARGS.name if ARGS.name else ARGS.model 278 model_name = ARGS.name if ARGS.name else ARGS.model
232 print(ARGS.name)
233 print(model_name)
234 print(ARGS.medium_selector)
235 279
236 if ARGS.name == "ENGRO2" and ARGS.medium_selector != "Default": 280 if ARGS.name == "ENGRO2" and ARGS.medium_selector != "Default":
237 df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0) 281 df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0)
238 ARGS.medium_selector = ARGS.medium_selector.replace("_", " ") 282 ARGS.medium_selector = ARGS.medium_selector.replace("_", " ")
239 medium = df_mediums[[ARGS.medium_selector]] 283 medium = df_mediums[[ARGS.medium_selector]]
255 # generate data 299 # generate data
256 rules = generate_rules(model, asParsed = False) 300 rules = generate_rules(model, asParsed = False)
257 reactions = generate_reactions(model, asParsed = False) 301 reactions = generate_reactions(model, asParsed = False)
258 bounds = generate_bounds(model) 302 bounds = generate_bounds(model)
259 medium = get_medium(model) 303 medium = get_medium(model)
304 compartments = generate_compartments(model)
260 305
261 df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"]) 306 df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"])
262 df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"]) 307 df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"])
263 308
264 df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"}) 309 df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"})
265 df_medium = medium.rename(columns = {"reaction": "ReactionID"}) 310 df_medium = medium.rename(columns = {"reaction": "ReactionID"})
266 df_medium["InMedium"] = True # flag per indicare la presenza nel medium 311 df_medium["InMedium"] = True # flag per indicare la presenza nel medium
267 312
268 merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer") 313 merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer")
269 merged = merged.merge(df_bounds, on = "ReactionID", how = "outer") 314 merged = merged.merge(df_bounds, on = "ReactionID", how = "outer")
270 315 merged = merged.merge(compartments, on = "ReactionID", how = "outer")
271 merged = merged.merge(df_medium, on = "ReactionID", how = "left") 316 merged = merged.merge(df_medium, on = "ReactionID", how = "left")
272 317
273 merged["InMedium"] = merged["InMedium"].fillna(False) 318 merged["InMedium"] = merged["InMedium"].fillna(False)
274 319
275 merged = merged.sort_values(by = "InMedium", ascending = False) 320 merged = merged.sort_values(by = "InMedium", ascending = False)