__author__ = "Johannes Köster"
__copyright__ = "Copyright 2016-2019, Johannes Köster"
__email__ = "johannes.koester@protonmail.com"
__license__ = "MIT"
from functools import update_wrapper
import platform
import hashlib
import inspect
import uuid
import os
from ._version import get_versions
__version__ = get_versions()["version"]
del get_versions
MIN_PY_VERSION = (3, 5)
DYNAMIC_FILL = "__snakemake_dynamic__"
SNAKEMAKE_SEARCHPATH = os.path.dirname(os.path.dirname(__file__))
UUID_NAMESPACE = uuid.uuid5(uuid.NAMESPACE_URL, "https://snakemake.readthedocs.io")
ON_WINDOWS = platform.system() == "Windows"
[docs]class TBDInt(int):
"""An integer that prints into <TBD>"""
def __str__(self):
return "<TBD>"
# A string that prints as TBD
TBDString = "<TBD>"
[docs]def num_if_possible(s):
"""Convert string to number if possible, otherwise return string."""
try:
return int(s)
except ValueError:
try:
return float(s)
except ValueError:
return s
[docs]def get_last_stable_version():
return __version__.split("+")[0]
[docs]def get_container_image():
return "snakemake/snakemake:v{}".format(get_last_stable_version())
[docs]def get_uuid(name):
return uuid.uuid5(UUID_NAMESPACE, name)
[docs]def get_file_hash(filename, algorithm="sha256"):
"""find the SHA256 hash string of a file. We use this so that the
user can choose to cache working directories in storage.
"""
from snakemake.logging import logger
# The algorithm must be available
try:
hasher = hashlib.new(algorithm)
except ValueError as ex:
logger.error("%s is not an available algorithm." % algorithm)
raise ex
with open(filename, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
return hasher.hexdigest()
[docs]def bytesto(bytes, to, bsize=1024):
"""convert bytes to megabytes.
bytes to mb: bytesto(bytes, 'm')
bytes to gb: bytesto(bytes, 'g' etc.
From https://gist.github.com/shawnbutts/3906915
"""
levels = {"k": 1, "m": 2, "g": 3, "t": 4, "p": 5, "e": 6}
answer = float(bytes)
for _ in range(levels[to]):
answer = answer / bsize
return answer
[docs]class Mode:
"""
Enum for execution mode of Snakemake.
This handles the behavior of e.g. the logger.
"""
default = 0
subprocess = 1
cluster = 2
[docs]class lazy_property(property):
__slots__ = ["method", "cached", "__doc__"]
[docs] @staticmethod
def clean(instance, method):
delattr(instance, method)
def __init__(self, method):
self.method = method
self.cached = "_{}".format(method.__name__)
super().__init__(method, doc=method.__doc__)
def __get__(self, instance, owner):
cached = (
getattr(instance, self.cached) if hasattr(instance, self.cached) else None
)
if cached is not None:
return cached
value = self.method(instance)
setattr(instance, self.cached, value)
return value
[docs]def strip_prefix(text, prefix):
if text.startswith(prefix):
return text[len(prefix) :]
return text
[docs]def log_location(msg):
from snakemake.logging import logger
callerframerecord = inspect.stack()[1]
frame = callerframerecord[0]
info = inspect.getframeinfo(frame)
logger.debug(
"{}: {info.filename}, {info.function}, {info.lineno}".format(msg, info=info)
)