Mercurial > repos > bimib > cobraxy
comparison COBRAxy/ras_to_bounds_beta.py @ 418:919b5b71a61c draft
Uploaded
| author | francesco_lapi |
|---|---|
| date | Tue, 09 Sep 2025 07:36:30 +0000 |
| parents | e8dd8dca9618 |
| children | 0877682fff48 |
comparison
equal
deleted
inserted
replaced
| 417:e8dd8dca9618 | 418:919b5b71a61c |
|---|---|
| 10 import sys | 10 import sys |
| 11 import csv | 11 import csv |
| 12 from joblib import Parallel, delayed, cpu_count | 12 from joblib import Parallel, delayed, cpu_count |
| 13 import utils.rule_parsing as rulesUtils | 13 import utils.rule_parsing as rulesUtils |
| 14 import utils.reaction_parsing as reactionUtils | 14 import utils.reaction_parsing as reactionUtils |
| 15 import utils.model_utils as modelUtils | |
| 15 | 16 |
| 16 # , medium | 17 # , medium |
| 17 | 18 |
| 18 ################################# process args ############################### | 19 ################################# process args ############################### |
| 19 def process_args(args :List[str] = None) -> argparse.Namespace: | 20 def process_args(args :List[str] = None) -> argparse.Namespace: |
| 149 if upper_bound!=0 and lower_bound!=0: | 150 if upper_bound!=0 and lower_bound!=0: |
| 150 new_bounds.loc[reaction, "lower_bound"] = valMin | 151 new_bounds.loc[reaction, "lower_bound"] = valMin |
| 151 new_bounds.loc[reaction, "upper_bound"] = valMax | 152 new_bounds.loc[reaction, "upper_bound"] = valMax |
| 152 return new_bounds | 153 return new_bounds |
| 153 | 154 |
| 154 ################################- DATA GENERATION -################################ | |
| 155 ReactionId = str | |
| 156 def generate_rules(model: cobra.Model, *, asParsed = True) -> Union[Dict[ReactionId, rulesUtils.OpList], Dict[ReactionId, str]]: | |
| 157 """ | |
| 158 Generates a dictionary mapping reaction ids to rules from the model. | |
| 159 | |
| 160 Args: | |
| 161 model : the model to derive data from. | |
| 162 asParsed : if True parses the rules to an optimized runtime format, otherwise leaves them as strings. | |
| 163 | |
| 164 Returns: | |
| 165 Dict[ReactionId, rulesUtils.OpList] : the generated dictionary of parsed rules. | |
| 166 Dict[ReactionId, str] : the generated dictionary of raw rules. | |
| 167 """ | |
| 168 # Is the below approach convoluted? yes | |
| 169 # Ok but is it inefficient? probably | |
| 170 # Ok but at least I don't have to repeat the check at every rule (I'm clinically insane) | |
| 171 _ruleGetter = lambda reaction : reaction.gene_reaction_rule | |
| 172 ruleExtractor = (lambda reaction : | |
| 173 rulesUtils.parseRuleToNestedList(_ruleGetter(reaction))) if asParsed else _ruleGetter | |
| 174 | |
| 175 return { | |
| 176 reaction.id : ruleExtractor(reaction) | |
| 177 for reaction in model.reactions | |
| 178 if reaction.gene_reaction_rule } | |
| 179 | |
| 180 def generate_reactions(model :cobra.Model, *, asParsed = True) -> Dict[ReactionId, str]: | |
| 181 """ | |
| 182 Generates a dictionary mapping reaction ids to reaction formulas from the model. | |
| 183 | |
| 184 Args: | |
| 185 model : the model to derive data from. | |
| 186 asParsed : if True parses the reactions to an optimized runtime format, otherwise leaves them as they are. | |
| 187 | |
| 188 Returns: | |
| 189 Dict[ReactionId, str] : the generated dictionary. | |
| 190 """ | |
| 191 | |
| 192 unparsedReactions = { | |
| 193 reaction.id : reaction.reaction | |
| 194 for reaction in model.reactions | |
| 195 if reaction.reaction | |
| 196 } | |
| 197 | |
| 198 if not asParsed: return unparsedReactions | |
| 199 | |
| 200 return reactionUtils.create_reaction_dict(unparsedReactions) | |
| 201 | |
| 202 def get_medium(model:cobra.Model) -> pd.DataFrame: | |
| 203 trueMedium=[] | |
| 204 for r in model.reactions: | |
| 205 positiveCoeff=0 | |
| 206 for m in r.metabolites: | |
| 207 if r.get_coefficient(m.id)>0: | |
| 208 positiveCoeff=1; | |
| 209 if (positiveCoeff==0 and r.lower_bound<0): | |
| 210 trueMedium.append(r.id) | |
| 211 | |
| 212 df_medium = pd.DataFrame() | |
| 213 df_medium["reaction"] = trueMedium | |
| 214 return df_medium | |
| 215 | |
| 216 def generate_bounds(model:cobra.Model) -> pd.DataFrame: | |
| 217 | |
| 218 rxns = [] | |
| 219 for reaction in model.reactions: | |
| 220 rxns.append(reaction.id) | |
| 221 | |
| 222 bounds = pd.DataFrame(columns = ["lower_bound", "upper_bound"], index=rxns) | |
| 223 | |
| 224 for reaction in model.reactions: | |
| 225 bounds.loc[reaction.id] = [reaction.lower_bound, reaction.upper_bound] | |
| 226 return bounds | |
| 227 | |
| 228 | |
| 229 | |
| 230 def generate_compartments(model: cobra.Model) -> pd.DataFrame: | |
| 231 """ | |
| 232 Generates a DataFrame containing compartment information for each reaction. | |
| 233 Creates columns for each compartment position (Compartment_1, Compartment_2, etc.) | |
| 234 | |
| 235 Args: | |
| 236 model: the COBRA model to extract compartment data from. | |
| 237 | |
| 238 Returns: | |
| 239 pd.DataFrame: DataFrame with ReactionID and compartment columns | |
| 240 """ | |
| 241 pathway_data = [] | |
| 242 | |
| 243 # First pass: determine the maximum number of pathways any reaction has | |
| 244 max_pathways = 0 | |
| 245 reaction_pathways = {} | |
| 246 | |
| 247 for reaction in model.reactions: | |
| 248 # Get unique pathways from all metabolites in the reaction | |
| 249 if type(reaction.annotation['pathways']) == list: | |
| 250 reaction_pathways[reaction.id] = reaction.annotation['pathways'] | |
| 251 max_pathways = max(max_pathways, len(reaction.annotation['pathways'])) | |
| 252 else: | |
| 253 reaction_pathways[reaction.id] = [reaction.annotation['pathways']] | |
| 254 | |
| 255 # Create column names for pathways | |
| 256 pathway_columns = [f"Pathway_{i+1}" for i in range(max_pathways)] | |
| 257 | |
| 258 # Second pass: create the data | |
| 259 for reaction_id, pathways in reaction_pathways.items(): | |
| 260 row = {"ReactionID": reaction_id} | |
| 261 | |
| 262 # Fill pathway columns | |
| 263 for i in range(max_pathways): | |
| 264 col_name = pathway_columns[i] | |
| 265 if i < len(pathways): | |
| 266 row[col_name] = pathways[i] | |
| 267 else: | |
| 268 row[col_name] = None # or "" if you prefer empty strings | |
| 269 | |
| 270 pathway_data.append(row) | |
| 271 | |
| 272 return pd.DataFrame(pathway_data) | |
| 273 | 155 |
| 274 def save_model(model, filename, output_folder, file_format='csv'): | 156 def save_model(model, filename, output_folder, file_format='csv'): |
| 275 """ | 157 """ |
| 276 Save a COBRA model to file in the specified format. | 158 Save a COBRA model to file in the specified format. |
| 277 | 159 |
| 290 try: | 172 try: |
| 291 if file_format == 'tabular' or file_format == 'csv': | 173 if file_format == 'tabular' or file_format == 'csv': |
| 292 # Special handling for tabular format using utils functions | 174 # Special handling for tabular format using utils functions |
| 293 filepath = os.path.join(output_folder, f"{filename}.csv") | 175 filepath = os.path.join(output_folder, f"{filename}.csv") |
| 294 | 176 |
| 295 rules = generate_rules(model, asParsed = False) | 177 rules = modelUtils.generate_rules(model, asParsed = False) |
| 296 reactions = generate_reactions(model, asParsed = False) | 178 reactions = modelUtils.generate_reactions(model, asParsed = False) |
| 297 bounds = generate_bounds(model) | 179 bounds = modelUtils.generate_bounds(model) |
| 298 medium = get_medium(model) | 180 medium = modelUtils.get_medium(model) |
| 299 | 181 |
| 300 try: | 182 try: |
| 301 compartments = utils.generate_compartments(model) | 183 compartments = modelUtils.generate_compartments(model) |
| 302 except: | 184 except: |
| 303 compartments = None | 185 compartments = None |
| 304 | 186 |
| 305 df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"]) | 187 df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "Rule"]) |
| 306 df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"]) | 188 df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Reaction"]) |
