diff --git a/tools/gem_tools/gem_check_memote.py b/tools/gem_tools/gem_check_memote.py index 21c328baeee..18431e62db9 100644 --- a/tools/gem_tools/gem_check_memote.py +++ b/tools/gem_tools/gem_check_memote.py @@ -1,29 +1,51 @@ import argparse import subprocess -import sys -import os + def __main__(): parser = argparse.ArgumentParser( - prog = "MemoteReport", - description = "This program takes a memote snapshot report of a GEM", - epilog = "Adding an epilog, but doubt it's needed.", + prog="MemoteReport", + description="This program takes a memote snapshot report of a GEM", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-o", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-o", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) output_location = parser.parse_args().out_file cb_model_location = parser.parse_args().cb_model_location - command = f"memote report snapshot --filename '{output_location}' '{cb_model_location}'" + command = ( + f"memote report snapshot --filename '{output_location}' " + f"'{cb_model_location}'" + ) try: # Execute the command - result = subprocess.run(command, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - + result = subprocess.run( + command, + shell=True, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True + ) + # Output the result print("Command executed successfully.") print("Output:\n", result.stdout) @@ -33,6 +55,7 @@ def __main__(): raise Exception(e.stderr) else: print(e.stderr) - + + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_escher_visualization.py b/tools/gem_tools/gem_escher_visualization.py index 3a3e118d356..061120e06cc 100644 --- a/tools/gem_tools/gem_escher_visualization.py +++ b/tools/gem_tools/gem_escher_visualization.py @@ -1,35 +1,82 @@ import argparse import cobra -import escher from escher import Builder import pandas as pd + def __main__(): parser = argparse.ArgumentParser( - prog = "EscherVisualization", - description = "This program visualizes an Escher map", - epilog = "Adding an epilog, but doubt it's needed.", + prog="EscherVisualization", + description="This program visualizes an Escher map", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=False, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=False, + help="The model to use." ) parser.add_argument( - "-f", "--flux_distribution_location", dest="flux_distribution_location", action="store", type=str, default=None, required=False, help="The flux distribution to visualize." + "-f", + "--flux_distribution_location", + dest="flux_distribution_location", + action="store", + type=str, + default=None, + required=False, + help="The flux distribution to visualize." ) parser.add_argument( - "-l", "--model_to_download", dest="model_to_download", action="store", type=str, default=None, required=False, help="The model to download." + "-l", + "--model_to_download", + dest="model_to_download", + action="store", + type=str, + default=None, + required=False, + help="The model to download." ) parser.add_argument( - "--map_load_name", dest="map_load_name", action="store", type=str, default=None, required=False, help="The name of the map to use." + "--map_load_name", + dest="map_load_name", + action="store", + type=str, + default=None, + required=False, + help="The name of the map to use." ) parser.add_argument( - "--map_upload_name", dest="map_upload_name", action="store", type=str, default=None, required=False, help="The name of the map to use." + "--map_upload_name", + dest="map_upload_name", + action="store", + type=str, + default=None, + required=False, + help="The name of the map to use." ) parser.add_argument( - "-u", "--uptake_constraints_file", dest="uptake_constraints_file", action="store", type=str, default=None, required=False, help="File containing new uptake constraits." + "-u", + "--uptake_constraints_file", + dest="uptake_constraints_file", + action="store", + type=str, + default=None, + required=False, + help="File containing new uptake constraits." ) parser.add_argument( - "-output", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-output", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) args = parser.parse_args() @@ -37,54 +84,111 @@ def __main__(): cb_model = None model_name = None if args.model_to_download is not None and args.model_to_download != "None": - if args.cb_model_location is not None and args.cb_model_location != "None": - raise Exception("You cannot specify both a model to download and a model to use.") + if args.cb_model_location is not None \ + and args.cb_model_location != "None": + raise Exception( + "You cannot specify both a model to " + "download and a model to use." + ) model_name = args.model_to_download - elif args.cb_model_location is not None and args.cb_model_location != "None": + elif args.cb_model_location is not None\ + and args.cb_model_location != "None": try: cb_model = cobra.io.read_sbml_model(args.cb_model_location) - except: - raise Exception("The model could not be read. Ensure it is in correct SBML format.") + except Exception as e: + raise Exception( + "The model could not be read. " + "Ensure it is in correct SBML format." + ) from e map_name = None map_location = None if args.map_upload_name is not None and args.map_upload_name != "None": if args.map_load_name is not None and args.map_load_name != "None": - raise Exception("You cannot specify both a map to upload and a map to load.") + raise Exception( + "You cannot specify both a map to upload and a map to load." + ) map_location = args.map_upload_name elif args.map_load_name is not None and args.map_load_name != "None": map_name = args.map_load_name - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": + if args.uptake_constraints_file is not None and \ + args.uptake_constraints_file != "None": if cb_model is None: - raise Exception("You cannot specify uptake constraints without uploading a model.") + raise Exception( + "You cannot specify uptake constraints " + "without uploading a model." + ) else: - constraints_df = pd.read_csv(args.uptake_constraints_file, sep=";", header=0, index_col=False) + constraints_df = pd.read_csv( + args.uptake_constraints_file, + sep=";", + header=0, + index_col=False + ) for index, row in constraints_df.iterrows(): - cb_model.reactions.get_by_id(row["reaction_id"]).lower_bound = row["lower_bound"] - cb_model.reactions.get_by_id(row["reaction_id"]).upper_bound = row["upper_bound"] + rxn_id = row["reaction_id"] + cb_model.reactions.get_by_id(rxn_id).lower_bound = \ + row["lower_bound"] + cb_model.reactions.get_by_id(rxn_id).upper_bound = \ + row["upper_bound"] flux_dict = None - if args.flux_distribution_location is not None and args.flux_distribution_location != "None": + if args.flux_distribution_location is not None and \ + args.flux_distribution_location != "None": if cb_model is None: - raise Exception("You cannot specify a flux distribution without uploading a model.") - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": - raise Exception("You cannot specify both uptake constraints and a flux distribution.") + raise Exception( + "You cannot specify a flux distribution " + "without uploading a model." + ) + if args.uptake_constraints_file is not None and \ + args.uptake_constraints_file != "None": + raise Exception( + "You cannot specify both uptake constraints and a flux " + "distribution." + ) try: - flux_df = pd.read_csv(args.flux_distribution_location, sep=";", header=0, index_col=False) - flux_dict = {key: value for key, value in zip(flux_df['reaction_name'], flux_df['flux'])} - except: - raise Exception("The flux distribution file could not be read. Ensure the file has semicolon-separated columns and a header row.") + flux_df = pd.read_csv( + args.flux_distribution_location, + sep=";", + header=0, + index_col=False + ) + flux_dict = { + key: value for key, value in zip( + flux_df['reaction_name'], + flux_df['flux'] + ) + } + except Exception as e: + raise Exception( + "The flux distribution file could not be read. " + "Ensure the file has semicolon-separated " + "columns and a header row." + ) from e if cb_model is not None and flux_dict is None: solution = cobra.flux_analysis.pfba(cb_model) - # make a dataframe with the reaction names, reaction ids, and flux distribution - flux_distribution = pd.DataFrame(columns=["reaction_name", "reaction_id", "flux"]) - flux_distribution["reaction_name"] = [reaction.name for reaction in cb_model.reactions] - flux_distribution["reaction_id"] = [reaction.id for reaction in cb_model.reactions] - flux_distribution["flux"] = [solution.fluxes[reaction.id] for reaction in cb_model.reactions] - flux_dict = {key: value for key, value in zip(flux_distribution['reaction_name'], flux_distribution['flux'])} + # make a dataframe with the reaction names, reaction ids, and flux + flux_distribution = pd.DataFrame( + columns=["reaction_name", "reaction_id", "flux"] + ) + flux_distribution["reaction_name"] = [ + reaction.name for reaction in cb_model.reactions + ] + flux_distribution["reaction_id"] = [ + reaction.id for reaction in cb_model.reactions + ] + flux_distribution["flux"] = [ + solution.fluxes[reaction.id] for reaction in cb_model.reactions + ] + flux_dict = { + key: value for key, value in zip( + flux_distribution['reaction_name'], + flux_distribution['flux'] + ) + } builder = Builder() if map_name is not None: @@ -104,6 +208,7 @@ def __main__(): builder.reaction_data = flux_dict builder.save_html(args.out_file) - + + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_extract_exchange.py b/tools/gem_tools/gem_extract_exchange.py index d39a20267b5..0f66720e911 100644 --- a/tools/gem_tools/gem_extract_exchange.py +++ b/tools/gem_tools/gem_extract_exchange.py @@ -1,41 +1,70 @@ import argparse import cobra + def read_model(model_location): model = cobra.io.read_sbml_model(model_location) return model + def get_exchange_reactions_info(model): exchange_reactions = model.exchanges exchange_reactions_info = [] for reaction in exchange_reactions: - exchange_reactions_info.append([reaction.id, reaction.name, reaction.reaction, reaction.lower_bound, reaction.upper_bound]) - txt_object = "reaction_id;reaction_name;reaction_stoichiometry;lower_bound;upper_bound\n" + exchange_reactions_info.append([ + reaction.id, + reaction.name, + reaction.reaction, + reaction.lower_bound, + reaction.upper_bound + ]) + txt_object = ( + "reaction_id;reaction_name;reaction_stoichiometry;" + "lower_bound;upper_bound\n" + ) for reaction in exchange_reactions_info: txt_object += ";".join([str(x) for x in reaction]) + "\n" return txt_object + def __main__(): # Parsing arguments parser = argparse.ArgumentParser( - prog = "GEM ", - description = "This program retrieves the exchange fluxes of a GEM model to be used in Galaxy.", - epilog = "Adding an epilog, but doubt it's needed.", + prog="GEM ", + description="This program retrieves the exchange fluxes " + "of a GEM model to be used in Galaxy.", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-output", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-output", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) args = parser.parse_args() # Reading model from file try: cb_model = read_model(args.cb_model_location) - except: - raise Exception("The model could not be read. Ensure it is in correct SBML format.") + except Exception as e: + raise Exception( + "The model could not be read. Ensure it is in correct SBML format." + ) from e # Getting exchange reactions info answer = get_exchange_reactions_info(cb_model) @@ -46,4 +75,4 @@ def __main__(): if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_flux_distribution.py b/tools/gem_tools/gem_flux_distribution.py index c2ac194acb9..e61106274d3 100644 --- a/tools/gem_tools/gem_flux_distribution.py +++ b/tools/gem_tools/gem_flux_distribution.py @@ -2,46 +2,88 @@ import cobra import pandas as pd + def __main__(): parser = argparse.ArgumentParser( - prog = "FluxDistribution", - description = "This program calculates the flux distribution of a GEM", - epilog = "Adding an epilog, but doubt it's needed.", + prog="FluxDistribution", + description="This program calculates the flux distribution of a GEM", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-output", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-output", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) parser.add_argument( - "-u", "--uptake_constraints_file", dest="uptake_constraints_file", action="store", type=str, default=None, required=False, help="File containing new uptake constraits." + "-u", + "--uptake_constraints_file", + dest="uptake_constraints_file", + action="store", + type=str, + default=None, + required=False, + help="File containing new uptake constraits." ) args = parser.parse_args() try: cb_model = cobra.io.read_sbml_model(args.cb_model_location) - except: - raise Exception("The model could not be read. Ensure it is in correct SBML format.") + except Exception as e: + raise Exception( + "The model could not be read. Ensure " + "it is in correct SBML format." + ) from e - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": - constraints_df = pd.read_csv(args.uptake_constraints_file, sep=";", header=0, index_col=False) + if args.uptake_constraints_file is not None\ + and args.uptake_constraints_file != "None": + constraints_df = pd.read_csv( + args.uptake_constraints_file, + sep=";", + header=0, + index_col=False + ) for index, row in constraints_df.iterrows(): - cb_model.reactions.get_by_id(row["reaction_id"]).lower_bound = row["lower_bound"] - cb_model.reactions.get_by_id(row["reaction_id"]).upper_bound = row["upper_bound"] + cb_model.reactions.get_by_id( + row["reaction_id"] + ).lower_bound = row["lower_bound"] + cb_model.reactions.get_by_id( + row["reaction_id"] + ).upper_bound = row["upper_bound"] # do pFBA solution = cobra.flux_analysis.pfba(cb_model) - # make a dataframe with the reaction names, reaction ids, and flux distribution - flux_distribution = pd.DataFrame(columns=["reaction_name", "reaction_id", "flux"]) - - flux_distribution["reaction_name"] = [reaction.name for reaction in cb_model.reactions] - flux_distribution["reaction_id"] = [reaction.id for reaction in cb_model.reactions] - flux_distribution["flux"] = [solution.fluxes[reaction.id] for reaction in cb_model.reactions] - + # make a dataframe with the reaction names, + # reaction ids, and flux distribution + flux_distribution = pd.DataFrame( + columns=["reaction_name", "reaction_id", "flux"] + ) + + flux_distribution["reaction_name"] = \ + [reaction.name for reaction in cb_model.reactions] + flux_distribution["reaction_id"] = \ + [reaction.id for reaction in cb_model.reactions] + flux_distribution["flux"] = \ + [solution.fluxes[reaction.id] for reaction in cb_model.reactions] + flux_distribution.to_csv(args.out_file, sep=";", index=False) - + + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_flux_variability_analysis.py b/tools/gem_tools/gem_flux_variability_analysis.py index cc15269c582..3752a9d9015 100644 --- a/tools/gem_tools/gem_flux_variability_analysis.py +++ b/tools/gem_tools/gem_flux_variability_analysis.py @@ -2,48 +2,119 @@ import cobra import pandas as pd + def __main__(): parser = argparse.ArgumentParser( - prog = "FluxVariabilityAnalysis", - description = "This program performs flux variability analysis on a GEM", - epilog = "Adding an epilog, but doubt it's needed.", + prog="FluxVariabilityAnalysis", + description="This program performs flux variability " + "analysis on a GEM", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-output", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-output", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) parser.add_argument( - "-f", "--fraction", dest="fraction_of_optimum", action="store", type=float, default=None, required=True, help="The fraction of optimum the FVA solutions should come within." + "-f", + "--fraction", + dest="fraction_of_optimum", + action="store", + type=float, + default=None, + required=True, + help="The fraction of optimum the FVA solutions should come within." ) parser.add_argument( - "-u", "--uptake_constraints_file", dest="uptake_constraints_file", action="store", type=str, default=None, required=False, help="File containing new uptake constraits." + "-u", + "--uptake_constraints_file", + dest="uptake_constraints_file", + action="store", + type=str, + default=None, + required=False, + help="File containing new uptake constraits." ) args = parser.parse_args() + # Validate constraints file first if provided + constraints_df = None + if args.uptake_constraints_file is not None\ + and args.uptake_constraints_file != "None": + try: + constraints_df = pd.read_csv( + args.uptake_constraints_file, + sep=";", + header=0, + index_col=False + ) + + required_columns = ['reaction_id', 'lower_bound', 'upper_bound'] + missing_columns = [col for col in required_columns if + col not in constraints_df.columns] + + if missing_columns: + raise ValueError( + f"Constraints file is missing required columns: " + f"{', '.join(missing_columns)}. " + f"Required columns are: {', '.join(required_columns)}" + ) + except FileNotFoundError: + raise FileNotFoundError( + f"Constraints file not found: {args.uptake_constraints_file}" + ) + except pd.errors.EmptyDataError: + raise ValueError("Constraints file is empty") + except Exception as e: + raise ValueError(f"Error processing constraints file: {str(e)}") + + # Load model cb_model = cobra.io.read_sbml_model(args.cb_model_location) - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": - constraints_df = pd.read_csv(args.uptake_constraints_file, sep=";", header=0, index_col=False) + # Apply constraints if they were loaded successfully + if constraints_df is not None: for index, row in constraints_df.iterrows(): - cb_model.reactions.get_by_id(row["reaction_id"]).lower_bound = row["lower_bound"] - cb_model.reactions.get_by_id(row["reaction_id"]).upper_bound = row["upper_bound"] + cb_model.reactions.get_by_id( + row["reaction_id"]).lower_bound = float(row["lower_bound"]) + cb_model.reactions.get_by_id( + row["reaction_id"]).upper_bound = float(row["upper_bound"]) fraction_of_optimum = args.fraction_of_optimum # perform fva - fva_result = cobra.flux_analysis.flux_variability_analysis(cb_model, fraction_of_optimum=fraction_of_optimum) + fva_result = cobra.flux_analysis.flux_variability_analysis( + cb_model, + fraction_of_optimum=fraction_of_optimum + ) # add reaction names and ids to the dataframe fva_result["reaction_id"] = fva_result.index - fva_result["reaction_name"] = fva_result["reaction_id"].apply(lambda x: cb_model.reactions.get_by_id(x).name) + fva_result["reaction_name"] = fva_result["reaction_id"].apply( + lambda x: cb_model.reactions.get_by_id(x).name + ) # reorder the columns - fva_result = fva_result[["reaction_id", "reaction_name", "minimum", "maximum"]] - + fva_result = fva_result[[ + "reaction_id", "reaction_name", "minimum", "maximum" + ]] + fva_result.to_csv(args.out_file, sep=";", index=False, header=True) - + + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_knockout.py b/tools/gem_tools/gem_knockout.py index ad994e844d7..e82573394f4 100644 --- a/tools/gem_tools/gem_knockout.py +++ b/tools/gem_tools/gem_knockout.py @@ -2,26 +2,61 @@ import cobra import pandas as pd + def __main__(): parser = argparse.ArgumentParser( - prog = "FluxDistribution", - description = "Performs FBA knockout analysis on a GEM.", - epilog = "Adding an epilog, but doubt it's needed.", + prog="FluxDistribution", + description="Performs FBA knockout analysis on a GEM.", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-output", "--output", dest="out_file", action="store", type=str, default=None, required=True, help="The output file." + "-output", + "--output", + dest="out_file", + action="store", + type=str, + default=None, + required=True, + help="The output file." ) parser.add_argument( - "-k", "--knockout_type", dest="knockout_type", action="store", type=str, default="single", required=False, help="Type of knockout to perform, single or double" + "-k", + "--knockout_type", + dest="knockout_type", + action="store", + type=str, + default="single", + required=False, + help="Type of knockout to perform, single or double" ) parser.add_argument( - "-g", "--gene_knockouts", dest="gene_knockouts", action="store", type=str, default=None, required=False, help="List of genes to knock out. Defaults to all." + "-g", + "--gene_knockouts", + dest="gene_knockouts", + action="store", + type=str, default=None, + required=False, + help="List of genes to knock out. Defaults to all." ) parser.add_argument( - "-u", "--uptake_constraints_file", dest="uptake_constraints_file", action="store", type=str, default=None, required=False, help="File containing new uptake constraits." + "-u", + "--uptake_constraints_file", + dest="uptake_constraints_file", + action="store", + type=str, + default=None, + required=False, + help="File containing new uptake constraits." ) args = parser.parse_args() @@ -29,37 +64,62 @@ def __main__(): # Reading the model try: cb_model = cobra.io.read_sbml_model(args.cb_model_location) - except: - raise Exception("The model could not be read. Ensure it is in correct SBML format.") + except Exception as e: + raise Exception( + "The model could not be read. " + "Ensure it is in correct SBML format." + ) from e # Verifying the genes are present in the model gene_ids = [gene.id for gene in cb_model.genes] - genes_to_knockout_1 = args.gene_knockouts.split(',') if args.gene_knockouts is not None else [] - gene_bool = [True if gene in gene_ids else False for gene in genes_to_knockout_1] + genes_to_knockout_1 = args.gene_knockouts.split(',')\ + if args.gene_knockouts is not None else [] + gene_bool = [ + True if gene in gene_ids else False for gene in genes_to_knockout_1 + ] if not all(gene_bool): - print(f'Found {sum(gene_bool)} of {len(genes_to_knockout_1)} genes in the model.') - raise Exception("One or more of the genes to knockout are not present in the model.") - + print( + f'Found {sum(gene_bool)} of {len(genes_to_knockout_1)} genes ' + 'in the model.' + ) + raise Exception( + "One or more of the genes to knockout are not present " + "in the model." + ) + # Adding all genes to knockout if none are specified if genes_to_knockout_1 is None or len(genes_to_knockout_1) == 0: genes_to_knockout_1 = [gene.id for gene in cb_model.genes] # Applying uptake constraints - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": - constraints_df = pd.read_csv(args.uptake_constraints_file, sep=";", header=0, index_col=False) + if (args.uptake_constraints_file is not None and + args.uptake_constraints_file != "None"): + constraints_df = pd.read_csv( + args.uptake_constraints_file, + sep=";", + header=0, + index_col=False + ) for index, row in constraints_df.iterrows(): - cb_model.reactions.get_by_id(row["reaction_id"]).lower_bound = row["lower_bound"] - cb_model.reactions.get_by_id(row["reaction_id"]).upper_bound = row["upper_bound"] + reaction = cb_model.reactions.get_by_id(row["reaction_id"]) + reaction.lower_bound = row["lower_bound"] + reaction.upper_bound = row["upper_bound"] - result = pd.DataFrame(columns=["reaction_id", "ko_gene_id_1", "ko_gene_id_2", "reaction", "wildtype_flux", "knockout_flux"]) + result = pd.DataFrame(columns=[ + "reaction_id", "ko_gene_id_1", "ko_gene_id_2", + "reaction", "wildtype_flux", "knockout_flux" + ]) if args.knockout_type == "single": genes_to_knockout_2 = [0] elif args.knockout_type == "double": genes_to_knockout_2 = genes_to_knockout_1.copy() else: - raise Exception(f"Invalid knockout type {args.knockout_type}. Only single and double are allowed.") + raise Exception( + f"Invalid knockout type {args.knockout_type}. " + "Only single and double are allowed." + ) # Wildtype pFBA with cb_model as model: @@ -77,14 +137,16 @@ def __main__(): result = pd.concat([result, pd.DataFrame([{ "reaction_id": reaction.id, "ko_gene_id_1": gene1, - "ko_gene_id_2": gene2 if args.knockout_type == "double" else None, + "ko_gene_id_2": gene2 + if args.knockout_type == "double" else None, "reaction": reaction.reaction, "wildtype_flux": wildtype_solution.fluxes[reaction.id], "knockout_flux": solution.fluxes[reaction.id], }])], ignore_index=True) - + # Writing the results to file result.to_csv(args.out_file, sep=";", index=False) + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__() diff --git a/tools/gem_tools/gem_phenotype_phase_plane.py b/tools/gem_tools/gem_phenotype_phase_plane.py index c93e30154b2..0b3009abf68 100644 --- a/tools/gem_tools/gem_phenotype_phase_plane.py +++ b/tools/gem_tools/gem_phenotype_phase_plane.py @@ -1,54 +1,114 @@ import argparse import cobra import pandas as pd -import numpy as np + def __main__(): parser = argparse.ArgumentParser( - prog = "ExpectedGrowthRate", - description = "This program calculates the expected growth rate of a GEM", - epilog = "Adding an epilog, but doubt it's needed.", + prog="ExpectedGrowthRate", + description="This program calculates the production envelope of a GEM", + epilog="Adding an epilog, but doubt it's needed.", ) parser.add_argument( - "-m", "--cb_model_location", dest="cb_model_location", action="store", type=str, default=None, required=True, help="The model to use." + "-m", + "--cb_model_location", + dest="cb_model_location", + action="store", + type=str, + default=None, + required=True, + help="The model to use." ) parser.add_argument( - "-output_csv", "--output_csv", dest="out_file_csv", action="store", type=str, default=None, required=True, help="The output csv file name." + "-output_csv", + "--output_csv", + dest="out_file_csv", + action="store", + type=str, + default=None, + required=True, + help="The output csv file name." ) parser.add_argument( - "-r1", "--reaction1", dest="reaction1", action="store", type=str, default=None, required=True, help="The first reaction to scan." + "-r1", + "--reaction1", + dest="reaction1", + action="store", + type=str, + default=None, + required=True, + help="The first reaction to scan." ) parser.add_argument( - "-r2", "--reaction2", dest="reaction2", action="store", type=str, default=None, required=True, help="The second reaction to scan." + "-r2", + "--reaction2", + dest="reaction2", + action="store", + type=str, + default=None, + required=True, + help="The second reaction to scan." ) parser.add_argument( - "-p", "--points", dest="points", action="store", type=int, default=10, required=False, help="The number of points to scan." + "-p", + "--points", + dest="points", + action="store", + type=int, + default=10, + required=False, + help="The number of points to scan." ) parser.add_argument( - "-c", "--objective", dest="objective", action="store", type=str, default=None, required=False, help="The reaction to use as objective." + "-c", + "--objective", + dest="objective", + action="store", + type=str, + default=None, + required=False, + help="The reaction to use as objective." ) parser.add_argument( - "-u", "--uptake_constraints_file", dest="uptake_constraints_file", action="store", type=str, default=None, required=False, help="The file containing the uptake constraints." + "-u", + "--uptake_constraints_file", + dest="uptake_constraints_file", + action="store", + type=str, + default=None, + required=False, + help="The file containing the uptake constraints." ) args = parser.parse_args() try: - assert(len(vars(args)) == 7) - except: - raise Exception("{} arguments were received. 7 were expected.".format(len(vars(args)))) + assert len(vars(args)) == 7 + except AssertionError: + raise Exception( + f"{len(vars(args))} arguments were received. 7 were expected." + ) try: cb_model = cobra.io.read_sbml_model(args.cb_model_location) - except: - raise Exception("The model could not be read. Ensure it is in correct SBML format.") + except Exception as e: + raise Exception( + "The model could not be read. Ensure it is in correct SBML format." + ) from e # set the uptake constraints if provided - if args.uptake_constraints_file is not None and args.uptake_constraints_file != "None": - constraints_df = pd.read_csv(args.uptake_constraints_file, sep=";", header=0, index_col=False) + if (args.uptake_constraints_file is not None and + args.uptake_constraints_file != "None"): + constraints_df = pd.read_csv( + args.uptake_constraints_file, + sep=";", + header=0, + index_col=False + ) for index, row in constraints_df.iterrows(): - cb_model.reactions.get_by_id(row["reaction_id"]).lower_bound = row["lower_bound"] - cb_model.reactions.get_by_id(row["reaction_id"]).upper_bound = row["upper_bound"] - + cb_model.reactions.get_by_id(row["reaction_id"])\ + .lower_bound = row["lower_bound"] + cb_model.reactions.get_by_id(row["reaction_id"])\ + .upper_bound = row["upper_bound"] # get the reactions reactions = [args.reaction1, args.reaction2] @@ -56,7 +116,10 @@ def __main__(): # checking if reactions are in model for reaction in reactions: if reaction not in cb_model.reactions: - raise Exception(f"Reaction {reaction} not found in model {args.cb_model_location.split('/')[-1]}") + raise Exception( + f"Reaction {reaction} not found in model " + f"{args.cb_model_location.split('/')[-1]}" + ) # get the points points = args.points @@ -66,19 +129,22 @@ def __main__(): raise Exception("Must have at least one point in the phase plane") # perform phenotype phase plane analysis - objective = None - if args.objective is not None and args.objective != "None": - objective = args.objective + if args.objective is not None and args.objective != "None" \ + and args.objective != '': + obj = args.objective + else: + obj = None results = cobra.flux_analysis.phenotype_phase_plane.production_envelope( - model = cb_model, - reactions = reactions, - points = points, - objective=None, + model=cb_model, + reactions=reactions, + points=points, + objective=obj, ) # save the results results.to_csv(args.out_file_csv, sep=";", header=True, index=False) - + + if __name__ == "__main__": - __main__() \ No newline at end of file + __main__()