Mercurial > repos > bimib > cobraxy
changeset 508:ca98c149ec61 draft default tip
Uploaded
author | francesco_lapi |
---|---|
date | Wed, 01 Oct 2025 14:21:26 +0000 |
parents | 20e135a73cad |
children | |
files | COBRAxy/metabolicModel2Tabular.py COBRAxy/ras_to_bounds.py COBRAxy/test_class COBRAxy/utils/model_utils.py |
diffstat | 4 files changed, 89 insertions(+), 67 deletions(-) [+] |
line wrap: on
line diff
--- a/COBRAxy/metabolicModel2Tabular.py Wed Oct 01 13:51:50 2025 +0000 +++ b/COBRAxy/metabolicModel2Tabular.py Wed Oct 01 14:21:26 2025 +0000 @@ -325,51 +325,17 @@ logger=logger ) - # generate data - rules = modelUtils.generate_rules(model, asParsed = False) - reactions = modelUtils.generate_reactions(model, asParsed = False) - bounds = modelUtils.generate_bounds(model) - medium = modelUtils.get_medium(model) - objective_function = modelUtils.extract_objective_coefficients(model) - - compartments = modelUtils.generate_compartments(model) - - df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "GPR"]) - df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Formula"]) - - # Create DataFrame for translation issues - df_translation_issues = pd.DataFrame([ - {"ReactionID": rxn_id, "TranslationIssues": issues} - for rxn_id, issues in translation_issues.items() - ]) - - df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"}) - df_medium = medium.rename(columns = {"reaction": "ReactionID"}) - df_medium["InMedium"] = True - - merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer") - merged = merged.merge(df_bounds, on = "ReactionID", how = "outer") - merged = merged.merge(objective_function, on = "ReactionID", how = "outer") - if compartments is not None: - merged = merged.merge(compartments, on = "ReactionID", how = "outer") - merged = merged.merge(df_medium, on = "ReactionID", how = "left") - - # Add translation issues column - if not df_translation_issues.empty: - merged = merged.merge(df_translation_issues, on = "ReactionID", how = "left") - merged["TranslationIssues"] = merged["TranslationIssues"].fillna("") - else: - # Add empty TranslationIssues column if no issues found - #merged["TranslationIssues"] = "" - pass - - merged["InMedium"] = merged["InMedium"].fillna(False) - - merged = merged.sort_values(by = "InMedium", ascending = False) - + # generate data using unified function if not ARGS.out_tabular: raise utils.ArgsErr("out_tabular", "output path (--out_tabular) is required when output_format == tabular", ARGS.out_tabular) - save_as_tabular_df(merged, ARGS.out_tabular) + + merged = modelUtils.export_model_to_tabular( + model=model, + output_path=ARGS.out_tabular, + translation_issues=translation_issues, + include_objective=True, + save_function=save_as_tabular_df + ) expected = ARGS.out_tabular # verify output exists and non-empty
--- a/COBRAxy/ras_to_bounds.py Wed Oct 01 13:51:50 2025 +0000 +++ b/COBRAxy/ras_to_bounds.py Wed Oct 01 14:21:26 2025 +0000 @@ -173,30 +173,12 @@ # Special handling for tabular format using utils functions filepath = os.path.join(output_folder, f"{filename}.csv") - rules = modelUtils.generate_rules(model, asParsed = False) - reactions = modelUtils.generate_reactions(model, asParsed = False) - bounds = modelUtils.generate_bounds(model) - medium = modelUtils.get_medium(model) - - compartments = modelUtils.generate_compartments(model) - - df_rules = pd.DataFrame(list(rules.items()), columns = ["ReactionID", "GPR"]) - df_reactions = pd.DataFrame(list(reactions.items()), columns = ["ReactionID", "Formula"]) - df_bounds = bounds.reset_index().rename(columns = {"index": "ReactionID"}) - df_medium = medium.rename(columns = {"reaction": "ReactionID"}) - df_medium["InMedium"] = True - - merged = df_reactions.merge(df_rules, on = "ReactionID", how = "outer") - merged = merged.merge(df_bounds, on = "ReactionID", how = "outer") - # Add compartments only if they exist - if compartments is not None: - merged = merged.merge(compartments, on = "ReactionID", how = "outer") - - merged = merged.merge(df_medium, on = "ReactionID", how = "left") - merged["InMedium"] = merged["InMedium"].fillna(False) - merged = merged.sort_values(by = "InMedium", ascending = False) - - merged.to_csv(filepath, sep="\t", index=False) + # Use unified function for tabular export + merged = modelUtils.export_model_to_tabular( + model=model, + output_path=filepath, + include_objective=True + ) else: # Standard COBRA formats
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/COBRAxy/test_class Wed Oct 01 14:21:26 2025 +0000 @@ -0,0 +1,2 @@ +Patient_ID Class +test_class test
--- a/COBRAxy/utils/model_utils.py Wed Oct 01 13:51:50 2025 +0000 +++ b/COBRAxy/utils/model_utils.py Wed Oct 01 14:21:26 2025 +0000 @@ -1155,6 +1155,78 @@ logger.info(f"Model genes updated: removed {removed}, added {added}") +def export_model_to_tabular(model: cobraModel, + output_path: str, + translation_issues: Dict = None, + include_objective: bool = True, + save_function = None) -> pd.DataFrame: + """ + Export a COBRA model to tabular format with optional components. + + Args: + model: COBRA model to export + output_path: Path where to save the tabular file + translation_issues: Optional dict of {reaction_id: issues} from gene translation + include_objective: Whether to include objective coefficient column + save_function: Optional custom save function, if None uses pd.DataFrame.to_csv + + Returns: + pd.DataFrame: The merged tabular data + """ + # Generate model data + rules = generate_rules(model, asParsed=False) + + reactions = generate_reactions(model, asParsed=False) + bounds = generate_bounds(model) + medium = get_medium(model) + compartments = generate_compartments(model) + + # Create base DataFrames + df_rules = pd.DataFrame(list(rules.items()), columns=["ReactionID", "GPR"]) + df_reactions = pd.DataFrame(list(reactions.items()), columns=["ReactionID", "Formula"]) + df_bounds = bounds.reset_index().rename(columns={"index": "ReactionID"}) + df_medium = medium.rename(columns={"reaction": "ReactionID"}) + df_medium["InMedium"] = True + + # Start merging + merged = df_reactions.merge(df_rules, on="ReactionID", how="outer") + merged = merged.merge(df_bounds, on="ReactionID", how="outer") + + # Add objective coefficients if requested + if include_objective: + objective_function = extract_objective_coefficients(model) + merged = merged.merge(objective_function, on="ReactionID", how="outer") + + # Add compartments/pathways if they exist + if compartments is not None: + merged = merged.merge(compartments, on="ReactionID", how="outer") + + # Add medium information + merged = merged.merge(df_medium, on="ReactionID", how="left") + + # Add translation issues if provided + if translation_issues: + df_translation_issues = pd.DataFrame([ + {"ReactionID": rxn_id, "TranslationIssues": issues} + for rxn_id, issues in translation_issues.items() + ]) + if not df_translation_issues.empty: + merged = merged.merge(df_translation_issues, on="ReactionID", how="left") + merged["TranslationIssues"] = merged["TranslationIssues"].fillna("") + + # Final processing + merged["InMedium"] = merged["InMedium"].fillna(False) + merged = merged.sort_values(by="InMedium", ascending=False) + + # Save the file + if save_function: + save_function(merged, output_path) + else: + merged.to_csv(output_path, sep="\t", index=False) + + return merged + + def _log_translation_statistics(stats: Dict[str, int], unmapped_genes: List[str], multi_mapping_genes: List[Tuple[str, List[str]]],