Source code for bionetgen.atomizer.atomizeTool

import bionetgen.atomizer.libsbml2bngl as ls2b
from bionetgen.core.defaults import BNGDefaults
import yaml, os

from bionetgen.core.utils.logging import BNGLogger, log_level

d = BNGDefaults()

[docs]class AtomizeTool: def __init__( self, input_file=None, options_dict=None, parser_namespace=None, app=None ): = app self.logger = BNGLogger( self.logger.debug( "Setting up AtomizeTool object", loc=f"{__file__} : AtomizeTool.__init__()" ) # we generate our defaults first and override it with # the dictionary first and then the namespace config = { "input": None, # we need this, check at the end and fail if we don't have it "annotation": False, "output": None, "convention_file": None, "naming_conventions": None, "user_structures": None, "molecule_id": False, "convert_units": False, # currently not supported "atomize": False, # default is flat translation "pathwaycommons": True, # requires connection so default is false "bionetgen_analysis": os.path.join( d.bng_path, "" ), # TODO: get it from app config "isomorphism_check": False, # wtf do we do here? "ignore": False, # wtf do we do here? "memoized_resolver": False, "keep_local_parameters": False, "quiet_mode": False, "obs_map_file": None, "log_level": "DEBUG", # options are "CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG" } # input file if input_file is not None: config["input"] = input_file # dictionary override if options_dict is not None: for key in config: if key in options_dict: config[key] = options_dict[key] # namespace override if parser_namespace is not None: for key in config: if hasattr(parser_namespace, key): config[key] = getattr(parser_namespace, key) # special handling of log level if log_level is not None: config["log_level"] = log_level elif is not None: # we called this from the CLI if config["log_level"] = "DEBUG" elif is not None: config["log_level"] = else: # we called the CLI but didn't give any log_level info config["log_level"] = "INFO" else: # we called from library but no log_level info exists config["log_level"] = "INFO" # check config options self.config = self.checkConfig(config)
[docs] def checkConfig(self, config): self.logger.debug( "Validating config options", loc=f"{__file__} : AtomizeTool.checkConfig()" ) options = {} options["inputFile"] = config["input"] # TODO: ensure this is not None conv, useID, naming = ls2b.selectReactionDefinitions(options["inputFile"]) options["outputFile"] = ( config["output"] if config["output"] is not None else options["inputFile"] + ".bngl" ) options["conventionFile"] = ( config["convention_file"] if config["convention_file"] is not None else conv ) options["userStructure"] = config["user_structures"] options["namingConventions"] = ( config["naming_conventions"] if config["naming_conventions"] is not None else naming ) options["useId"] = config["molecule_id"] options["annotation"] = config["annotation"] options["atomize"] = config["atomize"] options["pathwaycommons"] = config["pathwaycommons"] options["bionetgenAnalysis"] = config["bionetgen_analysis"] options["isomorphismCheck"] = config["isomorphism_check"] options["ignore"] = config["ignore"] options["noConversion"] = not config["convert_units"] options["memoizedResolver"] = config["memoized_resolver"] options["replaceLocParams"] = not config["keep_local_parameters"] options["quietMode"] = config["quiet_mode"] options["obs_map_file"] = config["obs_map_file"] assert config["log_level"] in [ "CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", ], "Logging level {} is not an allowed level".format(config["log_level"]) options["logLevel"] = config["log_level"] return options
[docs] def run(self): # TODO: Make atomizer also use cement app logging # this involves changing a lot of code in atomizer! self.logger.debug("Analyzing SBML file", loc=f"{__file__} :") self.returnArray = ls2b.analyzeFile( self.config["inputFile"], self.config["conventionFile"], self.config["useId"], self.config["namingConventions"], self.config["outputFile"], speciesEquivalence=self.config["userStructure"], atomize=self.config["atomize"], bioGrid=False, pathwaycommons=self.config["pathwaycommons"], ignore=self.config["ignore"], noConversion=self.config["noConversion"], memoizedResolver=self.config["memoizedResolver"], replaceLocParams=self.config["replaceLocParams"], quietMode=self.config["quietMode"], logLevel=self.config["logLevel"], obs_map_file=self.config["obs_map_file"],, ) self.logger.debug("Post-analysis", loc=f"{__file__} :") try: if self.config["bionetgenAnalysis"] and self.returnArray: ls2b.postAnalyzeFile( self.config["outputFile"], self.config["bionetgenAnalysis"], self.returnArray.database, replaceLocParams=self.config["replaceLocParams"], obs_map_file=self.config["obs_map_file"], ) except Exception as e: self.logger.warning( "Post-analysis failed", loc=f"{__file__} :" ) print("Post analysis failed") print(e) self.logger.debug( "Writing annotation file", loc=f"{__file__} :" ) try: if self.config["annotation"] and self.returnArray: with open(self.config["outputFile"] + ".yml", "w") as f: f.write( yaml.dump(self.returnArray.annotation, default_flow_style=False) ) except Exception as e: self.logger.warning( "Failed to write annotation file", loc=f"{__file__} :" ) print(e) return self.returnArray