Mercurial > repos > bimib > cobraxy
diff COBRAxy/metabolicModel2Tabular.py @ 491:7a413a5ec566 draft
Uploaded
author | francesco_lapi |
---|---|
date | Mon, 29 Sep 2025 15:34:59 +0000 |
parents | |
children | 4ed95023af20 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/COBRAxy/metabolicModel2Tabular.py Mon Sep 29 15:34:59 2025 +0000 @@ -0,0 +1,368 @@ +""" +Scripts to generate a tabular file of a metabolic model (built-in or custom). + +This script loads a COBRA model (built-in or custom), optionally applies +medium and gene nomenclature settings, derives reaction-related metadata +(GPR rules, formulas, bounds, objective coefficients, medium membership, +and compartments for ENGRO2), and writes a tabular summary. +""" + +import os +import csv +import cobra +import argparse +import pandas as pd +import utils.general_utils as utils +from typing import Optional, Tuple, List +import utils.model_utils as modelUtils +import logging +from pathlib import Path + + +ARGS : argparse.Namespace +def process_args(args: List[str] = None) -> argparse.Namespace: + """ + Parse command-line arguments for metabolic_model_setting. + """ + + parser = argparse.ArgumentParser( + usage="%(prog)s [options]", + description="Generate custom data from a given model" + ) + + parser.add_argument("--out_log", type=str, required=True, + help="Output log file") + + parser.add_argument("--model", type=str, + help="Built-in model identifier (e.g., ENGRO2, Recon, HMRcore)") + parser.add_argument("--input", type=str, + help="Custom model file (JSON or XML)") + parser.add_argument("--name", type=str, required=True, + help="Model name (default or custom)") + + parser.add_argument("--medium_selector", type=str, required=True, + help="Medium selection option") + + parser.add_argument("--gene_format", type=str, default="Default", + help="Gene nomenclature format: Default (original), ENSNG, HGNC_SYMBOL, HGNC_ID, ENTREZ") + + parser.add_argument("--out_tabular", type=str, + help="Output file for the merged dataset (CSV or XLSX)") + + parser.add_argument("--tool_dir", type=str, default=os.path.dirname(__file__), + help="Tool directory (passed from Galaxy as $__tool_directory__)") + + + return parser.parse_args(args) + +################################- INPUT DATA LOADING -################################ +def load_custom_model(file_path :utils.FilePath, ext :Optional[utils.FileFormat] = None) -> cobra.Model: + """ + Loads a custom model from a file, either in JSON, XML, MAT, or YML format. + + Args: + file_path : The path to the file containing the custom model. + ext : explicit file extension. Necessary for standard use in galaxy because of its weird behaviour. + + Raises: + DataErr : if the file is in an invalid format or cannot be opened for whatever reason. + + Returns: + cobra.Model : the model, if successfully opened. + """ + ext = ext if ext else file_path.ext + try: + if ext is utils.FileFormat.XML: + return cobra.io.read_sbml_model(file_path.show()) + + if ext is utils.FileFormat.JSON: + return cobra.io.load_json_model(file_path.show()) + + if ext is utils.FileFormat.MAT: + return cobra.io.load_matlab_model(file_path.show()) + + if ext is utils.FileFormat.YML: + return cobra.io.load_yaml_model(file_path.show()) + + except Exception as e: raise utils.DataErr(file_path, e.__str__()) + raise utils.DataErr( + file_path, + f"Unrecognized format '{file_path.ext}'. Only JSON, XML, MAT, YML are supported." + ) + + +###############################- FILE SAVING -################################ +def save_as_csv_filePath(data :dict, file_path :utils.FilePath, fieldNames :Tuple[str, str]) -> None: + """ + Saves any dictionary-shaped data in a .csv file created at the given file_path as FilePath. + + Args: + data : the data to be written to the file. + file_path : the path to the .csv file. + fieldNames : the names of the fields (columns) in the .csv file. + + Returns: + None + """ + with open(file_path.show(), 'w', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab") + writer.writeheader() + + for key, value in data.items(): + writer.writerow({ fieldNames[0] : key, fieldNames[1] : value }) + +def save_as_csv(data :dict, file_path :str, fieldNames :Tuple[str, str]) -> None: + """ + Saves any dictionary-shaped data in a .csv file created at the given file_path as string. + + Args: + data : the data to be written to the file. + file_path : the path to the .csv file. + fieldNames : the names of the fields (columns) in the .csv file. + + Returns: + None + """ + with open(file_path, 'w', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames = fieldNames, dialect="excel-tab") + writer.writeheader() + + for key, value in data.items(): + writer.writerow({ fieldNames[0] : key, fieldNames[1] : value }) + +def save_as_tabular_df(df: pd.DataFrame, path: str) -> None: + """ + Save a pandas DataFrame as a tab-separated file, creating directories as needed. + + Args: + df: The DataFrame to write. + path: Destination file path (will be written as TSV). + + Raises: + DataErr: If writing the output fails for any reason. + + Returns: + None + """ + try: + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + df.to_csv(path, sep="\t", index=False) + except Exception as e: + raise utils.DataErr(path, f"failed writing tabular output: {e}") + +def is_placeholder(gid) -> bool: + """Return True if the gene id looks like a placeholder (e.g., 0/NA/NAN/empty).""" + if gid is None: + return True + s = str(gid).strip().lower() + return s in {"0", "", "na", "nan"} # lowercase for simple matching + +def sample_valid_gene_ids(genes, limit=10): + """Yield up to `limit` valid gene IDs, skipping placeholders (e.g., the first 0 in RECON).""" + out = [] + for g in genes: + gid = getattr(g, "id", getattr(g, "gene_id", g)) + if not is_placeholder(gid): + out.append(str(gid)) + if len(out) >= limit: + break + return out + + +###############################- ENTRY POINT -################################ +def main(args:List[str] = None) -> None: + """ + Initialize and generate custom data based on the frontend input arguments. + + Returns: + None + """ + # Parse args from frontend (Galaxy XML) + global ARGS + ARGS = process_args(args) + + + if ARGS.input: + # Load a custom model from file + model = load_custom_model( + utils.FilePath.fromStrPath(ARGS.input), utils.FilePath.fromStrPath(ARGS.name).ext) + else: + # Load a built-in model + + try: + model_enum = utils.Model[ARGS.model] # e.g., Model['ENGRO2'] + except KeyError: + raise utils.ArgsErr("model", "one of Recon/ENGRO2/HMRcore/Custom_model", ARGS.model) + + # Load built-in model (Model.getCOBRAmodel uses tool_dir to locate local models) + try: + model = model_enum.getCOBRAmodel(toolDir=ARGS.tool_dir) + except Exception as e: + # Wrap/normalize load errors as DataErr for consistency + raise utils.DataErr(ARGS.model, f"failed loading built-in model: {e}") + + # Determine final model name: explicit --name overrides, otherwise use the model id + + model_name = ARGS.name if ARGS.name else ARGS.model + + if ARGS.name == "ENGRO2" and ARGS.medium_selector != "Default": + df_mediums = pd.read_csv(ARGS.tool_dir + "/local/medium/medium.csv", index_col = 0) + ARGS.medium_selector = ARGS.medium_selector.replace("_", " ") + medium = df_mediums[[ARGS.medium_selector]] + medium = medium[ARGS.medium_selector].to_dict() + + # Reset all medium reactions lower bound to zero + for rxn_id, _ in model.medium.items(): + model.reactions.get_by_id(rxn_id).lower_bound = float(0.0) + + # Apply selected medium uptake bounds (negative for uptake) + for reaction, value in medium.items(): + if value is not None: + model.reactions.get_by_id(reaction).lower_bound = -float(value) + + # Initialize translation_issues dictionary + translation_issues = {} + + if (ARGS.name == "Recon" or ARGS.name == "ENGRO2") and ARGS.gene_format != "Default": + logging.basicConfig(level=logging.INFO) + logger = logging.getLogger(__name__) + + model, translation_issues = modelUtils.translate_model_genes( + model=model, + mapping_df= pd.read_csv(ARGS.tool_dir + "/local/mappings/genes_human.csv", dtype={'entrez_id': str}), + target_nomenclature=ARGS.gene_format, + source_nomenclature='HGNC_symbol', + logger=logger + ) + + if ARGS.name == "Custom_model" and ARGS.gene_format != "Default": + logging.basicConfig(level=logging.INFO) + logger = logging.getLogger(__name__) + + tmp_check = [] + for g in model.genes[1:5]: # check first 3 genes only + tmp_check.append(modelUtils.gene_type(g.id, "Custom_model")) + + if len(set(tmp_check)) > 1: + raise utils.DataErr("Custom_model", "The custom model contains genes with mixed or unrecognized nomenclature. Please ensure all genes use the same recognized nomenclature before applying gene_format conversion.") + else: + source_nomenclature = tmp_check[0] + + if source_nomenclature != ARGS.gene_format: + model, translation_issues = modelUtils.translate_model_genes( + model=model, + mapping_df= pd.read_csv(ARGS.tool_dir + "/local/mappings/genes_human.csv", dtype={'entrez_id': str}), + target_nomenclature=ARGS.gene_format, + source_nomenclature=source_nomenclature, + logger=logger + ) + + + + + if ARGS.name == "Custom_model" and ARGS.gene_format != "Default": + logger = logging.getLogger(__name__) + + # Take a small, clean sample of gene IDs (skipping placeholders like 0) + ids_sample = sample_valid_gene_ids(model.genes, limit=10) + if not ids_sample: + raise utils.DataErr( + "Custom_model", + "No valid gene IDs found (many may be placeholders like 0)." + ) + + # Detect source nomenclature on the sample + types = [] + for gid in ids_sample: + try: + t = modelUtils.gene_type(gid, "Custom_model") + except Exception as e: + # Keep it simple: skip problematic IDs + logger.debug(f"gene_type failed for {gid}: {e}") + t = None + if t: + types.append(t) + + if not types: + raise utils.DataErr( + "Custom_model", + "Could not detect a known gene nomenclature from the sample." + ) + + unique_types = set(types) + if len(unique_types) > 1: + raise utils.DataErr( + "Custom_model", + "Mixed or inconsistent gene nomenclatures detected. " + "Please unify them before converting." + ) + + source_nomenclature = types[0] + + # Convert only if needed + if source_nomenclature != ARGS.gene_format: + model, translation_issues = modelUtils.translate_model_genes( + model=model, + mapping_df= pd.read_csv(ARGS.tool_dir + "/local/mappings/genes_human.csv", dtype={'entrez_id': str}), + target_nomenclature=ARGS.gene_format, + source_nomenclature=source_nomenclature, + 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) + + if ARGS.name == "ENGRO2": + 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 ARGS.name == "ENGRO2": + 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) + + 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) + expected = ARGS.out_tabular + + # verify output exists and non-empty + if not expected or not os.path.exists(expected) or os.path.getsize(expected) == 0: + raise utils.DataErr(expected, "Output not created or empty") + + print("Metabolic_model_setting: completed successfully") + +if __name__ == '__main__': + + main()