# Copyright (C) 2017 by Per Unneberg
import os
import re
import csv
import ast
import subprocess as sp
[docs]def to_min(timestr):
h, m, s = timestr.split(':')
return int(h) * 60 + int(m)
[docs]def get_samples(config, logger):
if config['settings'].get('sampleinfo') is None:
logger.info("[snakemake_rules.settings]: no sampleinfo provided; not setting config['_sampleinfo']")
return
try:
## Get the sample information
_include = config.get('samples', [])
if _include is None:
_include = []
assert isinstance(_include, list), print("config['samples'] should be a list")
_exclude = config.get('ignore_samples', [])
if _exclude is None:
_exclude = []
assert isinstance(_exclude, list), print("config['ignore_samples] should be a list")
_ignored = []
if len(_include) == 0:
config['samples'] = []
with open(config['settings']['sampleinfo']) as csvfile:
dialect = csv.Sniffer().sniff(csvfile.read(10240), delimiters=[',',';'])
config['_sampleinfo_delim'] = dialect.delimiter # save for later use by QC rules
csvfile.seek(0)
reader = csv.DictReader(csvfile, dialect=dialect)
rows = [row for row in reader]
# check if sample in config but not in sampleinfo:
if len(_include)>0 and len([s for s in _include if s not in [row['SM'] for row in rows]])>0:
logger.info("[snakemake_rules.settings]: warning: sample listed in config['samples'] not in '{}'".format(config['settings']['sampleinfo']))
if len(_exclude)>0 and len([s for s in _exclude if s not in [row['SM'] for row in rows]])>0:
logger.info("[snakemake_rules.settings]: warning: sample listed in config['ignore_samples'] not in '{}'".format(config['settings']['sampleinfo']))
# check overlap include exclude:
if len(_include)>0 and len(_exclude)>0 and not set(_include).isdisjoint(_exclude):
logger.info("[snakemake_rules.settings]: warning: same sample(s) listed in config['samples'] and config['ignore_samples']")
config['_sampleinfo'] = []
for row in rows:
# No samples specified: include all in sampleinfo except ones listed in ignore_samples
if len(_include) == 0 and row['SM'] not in _exclude:
config['_sampleinfo'].append(row)
config['samples'].append(row['SM'])
# Samples specified: include those, regardless of ignore_samples
elif row['SM'] in _include:
config['_sampleinfo'].append(row)
# ignore remaining samples in sampleinfo
else:
_ignored.append(row['SM'])
logger.info("[snakemake_rules.settings]: successfully loaded {} sample(s) from '{}'".format(len(config['_sampleinfo']), config['settings']['sampleinfo']))
if len(_include) == 0:
config['samples'] = sorted(list(set(config['samples'])))
logger.info("[snakemake_rules.settings]: no samples listed in config; added {} samples to config['samples']".format(len(config['samples'])))
if len(_ignored) > 0:
logger.info("[snakemake_rules.settings]: ignored {} sample(s) listed in '{}'".format(len(_ignored), config['settings']['sampleinfo']))
except Exception as e:
logger.info("[snakemake_rules.settings]: parsing sampleinfo failed; not setting config['_sampleinfo']: {}".format(e))
[docs]def python2_path(config, logger):
"""Add python 2 path if possible"""
try:
output = sp.check_output(['conda', 'env', 'list', '--json'])
envs = ast.literal_eval(output.decode("utf-8"))
py2 = [x for x in envs['envs'] if re.search("{}{}$".format(os.sep, config['settings']['conda']['python2']), x)][0]
py2bin = os.path.join(py2, "bin")
os.environ["PATH"] = ":".join([os.environ["PATH"], py2bin])
logger.info("[snakemake_rules.settings]: Add python2 bin path '{}' to PATH".format(py2bin))
except Exception as e:
logger.info("[snakemake_rules.settings]: failed to add conda python2 environment '{py2}' to PATH".format(py2=config['settings']['conda']['python2']))
print(e)
raise