Source code for flowcraft.templates.flowcraft_utils.flowcraft_base

"""

"""

import os
import sys
import json
import logging
import traceback

from time import gmtime, strftime


[docs]def get_logger(filepath, level=logging.DEBUG): # create logger logger = logging.getLogger(os.path.basename(filepath)) logger.setLevel(level) # create console handler and set level to debug ch = logging.StreamHandler() ch.setLevel(level) # create formatter formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') # add formatter to ch ch.setFormatter(formatter) # add ch to logger logger.addHandler(ch) return logger
[docs]def log_error(): """Nextflow specific function that logs an error upon unexpected failing """ with open(".status", "w") as status_fh: status_fh.write("error")
[docs]class MainWrapper: def __init__(self, f): self.f = f self.context = self.f.__globals__ self.logger = self.context.get("logger", None) def __call__(self, *args, **kwargs): self.logger.debug("Starting template at {}".format( strftime("%Y-%m-%d %H:%M:%S", gmtime()))) self.logger.debug("Working directory: {}".format(os.getcwd())) try: self.build_versions() self.f(*args, **kwargs) except SystemExit as e: sys.exit(e) except: if self.logger: self.logger.error("Module exited unexpectedly with error:" "\\n{}".format(traceback.format_exc())) log_error() self.logger.debug("Finished template at {}".format( strftime("%Y-%m-%d %H:%M:%S", gmtime())))
[docs] def build_versions(self): """Writes versions JSON for a template file This method creates the JSON file ``.versions`` based on the metadata and specific functions that are present in a given template script. It starts by fetching the template metadata, which can be specified via the ``__version__``, ``__template__`` and ``__build__`` attributes. If all of these attributes exist, it starts to populate a JSON/dict array (Note that the absence of any one of them will prevent the version from being written). Then, it will search the template scope for functions that start with the substring ``__set_version`` (For example ``def __set_version_fastqc()`). These functions should gather the version of an arbitrary program and return a JSON/dict object with the following information:: { "program": <program_name>, "version": <version> "build": <build> } This JSON/dict object is then written in the ``.versions`` file. """ version_storage = [] template_version = self.context.get("__version__", None) template_program = self.context.get("__template__", None) template_build = self.context.get("__build__", None) if template_version and template_program and template_build: if self.logger: self.logger.debug("Adding template version: {}; {}; " "{}".format(template_program, template_version, template_build)) version_storage.append({ "program": template_program, "version": template_version, "build": template_build }) for var, obj in self.context.items(): if var.startswith("__get_version"): ver = obj() version_storage.append(ver) if self.logger: self.logger.debug("Found additional software version" "{}".format(ver)) with open(".versions", "w") as fh: fh.write(json.dumps(version_storage, separators=(",", ":")))