Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import os
import shutil
import gzip
import json
import bz2
from copy import deepcopy
import subprocess
def dict_merge(a, b):
"""
Deep merge 2 dicts together
"""
if not isinstance(b, dict):
return b
result = deepcopy(a)
for k, v in b.items():
if k in result and isinstance(result[k], dict):
result[k] = dict_merge(result[k], v)
else:
result[k] = deepcopy(v)
return result
# default configuration file
configfile:
"src/config.imp.json"
# default executable for snakmake
shell.executable("bash")
# custom configuration file
CUSTOM_CONFIG_PATH = os.environ.get("CONFIGFILE", "conf/userconfig.imp.json")
# merge 2 configurations files together
if os.path.exists(CUSTOM_CONFIG_PATH):
with open(CUSTOM_CONFIG_PATH, 'r') as rhandle:
data = json.load(rhandle)
config = dict_merge(config, data)
# some parameters
SRCDIR = os.environ.get("SRCDIR", config['imp_src'])
# get parameters from the command line
OUTPUTDIR = os.environ.get("OUTPUTDIR", config['outputdir'])
MG = os.environ.get("MG", config['raws']['Metagenomics']).split()
MT = os.environ.get("MT", config['raws']['Metatranscriptomics']).split()
SAMPLE = os.environ.get("SAMPLE", config['sample'])
DBPATH = os.environ.get("DBPATH", config['db_path'])
if not os.path.exists(DBPATH):
os.makedirs(DBPATH)
# Get general parameters
THREADS = os.environ.get("THREADS", config['threads'])
MEMTOTAL = os.environ.get("MEMTOTAL", config['memory_total_gb'])
MEMCORE = os.environ.get("MEMCORE", config['memory_per_core_gb'])
# temporary directory will be stored inside the OUTPUTDIR directory
# unless a absolute path is set
TMPDIR = os.environ.get("TMPDIR", config['tmp_dir'])
if not os.path.isabs(TMPDIR):
TMPDIR = os.path.join(OUTPUTDIR, TMPDIR)
if not os.path.exists(TMPDIR):
os.makedirs(TMPDIR)