-
Notifications
You must be signed in to change notification settings - Fork 7
/
Copy pathconfig.py
182 lines (144 loc) · 6.34 KB
/
config.py
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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
import sys
import copy
import os
import errors
import debug # Imported for convenience!!
base_config_dir = os.getenv("COASTGUARD_CFG", None)
if base_config_dir is None:
raise ValueError("COASTGUARD_CFG environment variable must be set. "
"(It should point to the CoastGuard configurations "
"directory to use.)")
execfile(os.path.join(base_config_dir, "global.cfg"), {}, locals())
class ConfigDict(dict):
def __add__(self, other):
newcfg = copy.deepcopy(self)
newcfg.update(other)
return newcfg
def __str__(self):
lines = []
for key in sorted(self.keys()):
lines.append("%s: %r" % (key, self[key]))
return "\n".join(lines)
def read_file(fn, required=False):
cfgdict = ConfigDict()
if os.path.isfile(fn):
if not fn.endswith('.cfg'):
raise ValueError("Coast Guard configuration files must "
"end with the extention '.cfg'.")
key = os.path.split(fn)[-1][:-4]
execfile(fn, {}, cfgdict)
elif required:
raise ValueError("Configuration file (%s) doesn't exist "
"and is required!" % fn)
return cfgdict
class CoastGuardConfigs(object):
def __init__(self, base_config_dir=base_config_dir):
self.base_config_dir = base_config_dir
default_config_fn = os.path.join(self.base_config_dir, "default.cfg")
self.defaults = read_file(default_config_fn, required=True)
self.obsconfigs = ConfigDict()
self.overrides = ConfigDict()
def __getattr__(self, key):
return self.__getitem__(key)
def __getitem__(self, key):
if key in self.overrides:
#utils.print_debug("Config '%s' found in Overrides" % key, 'config', stepsback=3)
val = self.overrides[key]
elif key in self.obsconfigs:
#utils.print_debug("Config '%s' found in Observation configs" % key, 'config', stepsback=3)
val = self.obsconfigs[key]
elif key in self.defaults:
#utils.print_debug("Config '%s' found in Default" % key, 'config', stepsback=3)
val = self.defaults[key]
else:
raise errors.ConfigurationError("The configuration '%s' "
"cannot be found!" % key)
return val
def __str__(self):
allkeys = set.union(set(self.defaults.keys()),
set(self.obsconfigs.keys()),
set(self.overrides.keys()))
lines = ["Current configurations:"]
#for key in allkeys:
# lines.append(" %s: %s" % (key, self[key]))
lines.append(" "+str(self.defaults+self.obsconfigs+self.overrides).replace("\n", "\n "))
lines.append("Overrides:")
lines.append(" "+str(self.overrides).replace("\n", "\n "))
lines.append("Observation configurations:")
lines.append(" "+str(self.obsconfigs).replace("\n", "\n "))
lines.append("Defaults:")
lines.append(" "+str(self.defaults).replace("\n", "\n "))
return "\n".join(lines)
def clear_obsconfigs(self):
self.obsconfigs.clear()
def clear_overrides(self):
self.overrides.clear()
def set_override_config(self, key, val):
self.overrides[key] = val
def load_configs_for_archive(self, arfn):
"""Given a psrchive archive file set current configurations to the values
pre-set for this observation, pulsar, backend, receiver, telescope.
Inputs:
fn: The psrchive archive to get configurations for.
Outputs:
None
"""
self.clear_obsconfigs()
config_files = [] # A list of configuration files to look for
telescope = arfn['telname']
precedence = [arfn['telname'].lower(),
arfn['rcvr'].lower(),
arfn['backend'].lower()]
cfgdir = self.base_config_dir
for dirname in precedence:
cfgdir = os.path.join(cfgdir, dirname)
config_files.append(os.path.join(cfgdir, 'configs.cfg'))
#config_files.append(os.path.join(self.base_config_dir, 'telescopes',
# "%s.cfg" % telescope.lower()))
#config_files.append(os.path.join(self.base_config_dir, 'receivers',
# "%s.cfg" % arfn['rcvr'].lower()))
#config_files.append(os.path.join(self.base_config_dir, 'backends',
# "%s.cfg" % arfn['backend'].lower()))
#config_files.append(os.path.join(self.base_config_dir, 'pulsars',
# "%s.cfg" % arfn['name'].upper()))
#config_files.append(os.path.join(self.base_config_dir, 'observations',
# "%s.cfg" % os.path.split(arfn.fn)[-1]))
#msg = "\n ".join(["Checking for the following configurations:"] + \
# config_files)
#utils.print_debug(msg, 'config')
for fn in config_files:
self.obsconfigs += read_file(fn)
#utils.print_debug("Current configurations:\n%s" % self, 'config')
class ConfigManager(object):
"""An object to hold and manage CoastCuardConfigs objects
from multiple threads.
This is important because each thread may require
different configurations.
"""
def __init__(self):
self.configs = {}
def __contains__(self, name):
return name in self.configs
def get(self):
name = os.getpid()
if name not in self:
self.configs[name] = CoastGuardConfigs()
#utils.print_debug("Getting configs for process %s" % name, 'config')
return self.configs[name]
def load_configs_for_archive(self, arf):
#utils.print_debug("Loading configs for %s" % arf.fn, 'config')
self.get().load_configs_for_archive(arf)
def __getattr__(self, key):
val = self.get()[key]
#utils.print_debug("Getting config %s = %s" % (key, val), 'config')
return val
cfg = ConfigManager()
def main():
import utils
if len(sys.argv) > 1:
arf = utils.ArchiveFile(sys.argv[1])
cfg.set_override_config("something", 'newvalue!')
cfg.load_configs_for_archive(arf)
print cfg.get()
if __name__ == '__main__':
main()