forked from yarden/MISO
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsettings.py
More file actions
152 lines (123 loc) · 4.72 KB
/
Copy pathsettings.py
File metadata and controls
152 lines (123 loc) · 4.72 KB
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
##
## Settings with relevant directories
##
from parse_csv import *
import ConfigParser
import os
miso_path = os.path.dirname(os.path.abspath(os.path.expanduser(__file__)))
class Settings(object):
@classmethod
def load(cls, path):
"""
Reads in settings from a ConfigParser formatted file
ignores section headers, so make sure each option is unique in the file
returns a dictionary with all the options mapped to their values.
"""
config = ConfigParser.ConfigParser()
if path != None:
cls.settings_path = path
else:
# Use default settings file if none was given
cls.settings_path = os.path.join(miso_path,
"settings",
"miso_settings.txt")
print "Loading settings from: %s" %(cls.settings_path)
cls.parsed_settings = config.read(cls.settings_path)
cls.global_settings = {}
print "Settings: "
for section in config.sections():
for option in config.options(section):
# Load cluster options as strings, without attempting to evaluate them
# Avoids misinterpretation of words like "long" as a data type
if section == "cluster":
cls.global_settings[option] = str(config.get(section, option))
else:
cls.global_settings[option] = tryEval(config.get(section, option))
print " ", option, cls.global_settings[option]
# Set directory paths specific to pipeline
if 'pipeline_results_dir' in cls.global_settings:
cls.global_settings['analysis_dir'] = os.path.join(cls.global_settings['pipeline_results_dir'],
'analysis')
cls.global_settings['rna_events_dir'] = os.path.join(cls.global_settings['analysis_dir'],
'rna_events')
@classmethod
def get_sampler_params(cls):
"""
Return sampler parameters.
"""
param_names = ['burn_in', 'lag', 'num_iters']
sampler_params = {}
for name in param_names:
if name not in cls.global_settings:
print "settings: ", cls.global_settings
raise Exception, "Error: need %s parameter to be set in settings file." \
%(name)
sampler_params[name] = cls.global_settings[name]
return sampler_params
@classmethod
def get_cluster_command(cls):
"""
Return the name of the command to use for cluster submission
(e.g. 'qsub')
"""
if 'cluster_command' in cls.global_settings:
return cls.global_settings['cluster_command']
else:
return None
@classmethod
def get_long_queue_name(cls):
"""
Return the name of the long queue (for long jobs.)
"""
if 'long_queue_name' in cls.global_settings:
return cls.global_settings['long_queue_name']
else:
return None
@classmethod
def get_short_queue_name(cls):
"""
Return the name of the short queue (for short jobs.)
"""
if 'long_queue_name' in cls.global_settings:
return cls.global_settings['short_queue_name']
else:
return None
@classmethod
def get_min_event_reads(cls):
"""
Return minimum number of reads an event should have.
"""
min_event_reads = cls.global_settings["min_event_reads"]
return min_event_reads
@classmethod
def get_counts_dir(cls, event_type):
"""
Return counts directory for given event type.
"""
if 'rna_events_dir' in cls.global_settings:
return os.path.join(cls.global_settings['rna_events_dir'],
event_type)
return None
@classmethod
def get_counts_filename(cls, sample_label, event_type):
"""
Return counts filename for a given sample and its type.
"""
return os.path.join(cls.get_counts_dir(event_type),
'%s.counts' %(sample_label))
@classmethod
def get_filters(cls, event_type):
pass
@classmethod
def get(cls):
return cls.global_settings
def load_settings(settings_filename):
Settings.load(settings_filename)
return Settings.get()
def main():
settings_filename = 'settings/miso_settings.txt'
Settings.load(settings_filename)
print Settings.get()
print Settings.get_counts_dir('TandemUTR')
if __name__ == '__main__':
main()