Coverage for pesummary/cli/summaryrecreate.py: 83.1%
142 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-12-09 22:34 +0000
1#! /usr/bin/env python
3# Licensed under an MIT style license -- see LICENSE.md
5from pesummary.core.cli.parser import ArgumentParser as _ArgumentParser
6from pesummary.core.cli.actions import DelimiterSplitAction
7from pesummary.gw.file.read import read
8from pesummary.utils.utils import logger, make_dir
9from pesummary.utils.dict import edit_dictionary, paths_to_key
10from pesummary.utils.exceptions import InputError
11from pesummary.io import write
13import subprocess
14import os
15import copy
17__author__ = ["Charlie Hoy <charlie.hoy@ligo.org>"]
18__doc__ = """This executable is used to recreate the analysis which was used
19to generate the posterior samples stored in the metafile"""
22def launch_subprocess(command_line):
23 """Run a command line
25 Parameters
26 ----------
27 command_line: str
28 the command line you wish to run
29 """
30 logger.info("Running '{}'".format(command_line))
31 process = subprocess.Popen(command_line, shell=True)
32 process.wait()
35class ArgumentParser(_ArgumentParser):
36 def _pesummary_options(self):
37 options = super(ArgumentParser, self)._pesummary_options()
38 options.update(
39 {
40 "--samples": {
41 "short": "-s",
42 "required": True,
43 "help": (
44 "Path to the PESummary file which stores the analysis "
45 "you wish to recreate."
46 )
47 },
48 "--rundir": {
49 "default": "./",
50 "help": "The run directory of the analysis",
51 "short": "-r",
52 },
53 "--code": {
54 "default": "lalinference",
55 "short": "-c",
56 "help": "The sampling software you wish to run"
57 },
58 "--delimiter": {
59 "default": ":",
60 "help": (
61 "Delimiter used to seperate the existing and new "
62 "quantity"
63 )
64 },
65 "--config_override": {
66 "action": DelimiterSplitAction,
67 "nargs": "+",
68 "default": {},
69 "help": (
70 "Changes you wish to make to the configuration file. "
71 "Must be in the form `key:item` where key is the entry "
72 "in the config file you wish to modify, ':' is the "
73 "default delimiter, and item is the string you wish to "
74 "replace with."
75 )
76 },
77 }
78 )
79 return options
82class LALInference(object):
83 """Class to create a LALInference analysis
84 """
85 @staticmethod
86 def pipe(rundir, config, **kwargs):
87 """Launch the `lalinference_pipe` executable
89 Parameters
90 ----------
91 rundir: str
92 path to the run directory of the analysis
93 config: str
94 path to the configuration file
95 kwargs: dict
96 dictionary of command line arguments that are passed directly to
97 `lalinference_pipe`
98 """
99 command_line = "lalinference_pipe {} -r {}".format(
100 config, os.path.join(rundir, "outdir")
101 )
102 for key, item in kwargs.items():
103 command_line += " --{} {}".format(key, item)
104 launch_subprocess(command_line)
106 @staticmethod
107 def map(parameter, delimiter=":"):
108 """Map a parameter name to LALInference format
110 Parameters
111 ----------
112 parameter: str
113 The parameter you wish to map
114 delimiter: str, optional
115 The delimiter to split the parameter
116 """
117 not_equal_to_length_two_error = (
118 "The {} must be in the form '{}:{}' where '%s' is the default "
119 "delimiter" % (delimiter)
120 )
121 p = parameter.split(delimiter)
122 if "psd" in p:
123 if len(p) != 2:
124 raise ValueError(
125 not_equal_to_length_two_error.format("psd", "ifo", "psd")
126 )
127 return "{}-{}".format(p[0].lower(), p[1])
128 elif "calibration" in p:
129 if len(p) != 2:
130 raise ValueError(
131 not_equal_to_length_two_error.format(
132 "calibration", "ifo", "calibration"
133 )
134 )
135 return "{}-spcal-envelope".format(p[0].lower())
136 else:
137 return delimiter.join(p)
140class Code(object):
141 lalinference = LALInference
144class _Input(object):
145 """Super class to handle the command line arguments
146 """
147 @property
148 def samples(self):
149 return self._samples
151 @samples.setter
152 def samples(self, samples):
153 if not os.path.isfile(samples):
154 raise FileNotFoundError(
155 "File '{}' does not exist".format(samples)
156 )
157 self._samples = read(samples)
159 @property
160 def labels(self):
161 return self._labels
163 @labels.setter
164 def labels(self, labels):
165 self._labels = self.samples.labels
166 if labels is not None:
167 self._labels = labels
168 for label in labels:
169 if label not in self.samples.labels:
170 raise InputError(
171 "The label '{}' is not stored in the result file. "
172 "The list of available labels is {}".format(
173 label, ", ".join(self.samples.labels)
174 )
175 )
176 logger.info(
177 "Recreating the following analyses: '{}'".format(
178 ", ".join(labels)
179 )
180 )
181 else:
182 logger.info(
183 "No labels provided. Recreating all analyses stored in the "
184 "result file: {}".format(", ".join(self.samples.labels))
185 )
187 @property
188 def rundir(self):
189 return self._rundir
191 @rundir.setter
192 def rundir(self, rundir):
193 self._rundir = os.path.abspath(rundir)
194 make_dir(rundir)
195 logger.info(
196 "Setting '{}' to be the run directory of the analysis".format(
197 rundir
198 )
199 )
200 for label in self.labels:
201 make_dir(os.path.join(rundir, label))
203 @property
204 def code(self):
205 return self._code
207 @code.setter
208 def code(self, code):
209 allowed_codes = ["lalinference"]
210 if code.lower() not in allowed_codes:
211 raise InputError(
212 "The 'summaryrecreate' executable is currently only configured "
213 "to run with the following codes: {}".format(
214 ", ".join(allowed_codes)
215 )
216 )
217 self._code = getattr(Code, code.lower())
219 def _write_to_file(self, attribute):
220 """Write an attribute stored in the result file to file
222 Parameters
223 ----------
224 attribute: str
225 name of the attribute that you wish to write to file
226 """
227 f = self.samples
228 data = getattr(f, attribute)
229 if len(data) and all(data[label] != {} for label in self.labels):
230 for label in self.labels:
231 for ifo in data[label].keys():
232 filename = os.path.join(
233 self.rundir, label, "{}_{}.dat".format(ifo, attribute)
234 )
235 data[label][ifo].save_to_file(filename)
236 name = self.code.map(
237 ":".join([ifo, attribute]), delimiter=":"
238 )
239 self.settings_to_change[label][name] = filename
240 else:
241 logger.warning(
242 "No {} data found in the file. Using the {}s stored in the "
243 "configuration file".format(attribute, attribute)
244 )
246 def write_psd_to_file(self):
247 """Write the psd to file
248 """
249 self._write_to_file("psd")
251 def write_calibration_to_file(self):
252 """Write the calibration data to file
253 """
254 self._write_to_file("calibration")
256 def write_config_file(self):
257 """Save the config file to the run directory and return the data
258 stored within
259 """
260 if not len(self.samples.config):
261 raise ValueError(
262 "No configuration file stored in the result file. Unable to "
263 "recreate the analysis"
264 )
265 config_files = {}
266 for label in self.labels:
267 config_data = copy.deepcopy(self.samples.config[label])
268 if self.settings_to_change is not None:
269 for key, item in self.settings_to_change[label].items():
270 try:
271 path, = paths_to_key(key, config_data)
272 config_data = edit_dictionary(config_data, path, item)
273 except ValueError:
274 logger.warning(
275 "Unable to change '{}' to '{}' in the config "
276 "file".format(key, item)
277 )
278 outdir = os.path.join(self.rundir, label)
279 write(
280 config_data, outdir=outdir, filename="config.ini",
281 file_format="ini"
282 )
283 logger.info("Writing the configuration file to: {}".format(outdir))
284 config_files[label] = os.path.join(outdir, "config.ini")
285 return config_files
288class Input(_Input):
289 """Class to handle the command line arguments
291 Parameters
292 ----------
293 opts: argparse.Namespace
294 Namespace object containing the command line options
295 """
296 def __init__(self, opts):
297 logger.info(opts)
298 self.opts = opts
299 self.samples = self.opts.samples
300 self.labels = self.opts.labels
301 self.settings_to_change = {label: {} for label in self.labels}
302 self.rundir = self.opts.rundir
303 self.code = self.opts.code
304 self.write_psd_to_file()
305 self.write_calibration_to_file()
306 for key, item in self.opts.config_override.items():
307 for label in self.labels:
308 name = self.code.map(key)
309 self.settings_to_change[label][name] = item
310 self.config = self.write_config_file()
313def main(args=None):
314 """The main function for the `summaryrecreate` executable
315 """
316 parser = ArgumentParser(description=__doc__)
317 parser.add_known_options_to_parser(
318 [
319 "--samples", "--labels", "--rundir", "--code", "--delimiter",
320 "--config_override"
321 ]
322 )
323 opts = parser.parse_args(args=args)
324 inputs = Input(opts)
325 for label in inputs.labels:
326 inputs.code.pipe(os.path.join(inputs.rundir, label), inputs.config[label])
329if __name__ == "__main__":
330 main()