Coverage for pesummary/tests/workflow_test.py: 25.5%
322 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-12-09 22:34 +0000
« prev ^ index » next coverage.py v7.4.4, created at 2024-12-09 22:34 +0000
1# Licensed under an MIT style license -- see LICENSE.md
3import os
4import shutil
5import glob
6import pytest
7import numpy as np
9from .base import make_argparse, get_list_of_plots, get_list_of_files
10from .base import read_result_file
11from pesummary.utils.utils import functions
12from pesummary.cli.summarypages import WebpageGeneration
13from pesummary.cli.summaryplots import PlotGeneration
14import tempfile
16__author__ = ["Charlie Hoy <charlie.hoy@ligo.org>"]
19class Base(object):
20 """Base class to test the full workflow
21 """
22 @pytest.mark.workflowtest
23 def test_single_run(self, extension, bilby=False):
24 """Test the full workflow for a single result file case
26 Parameters
27 ----------
28 result_file: str
29 path to result file you wish to run with
30 """
31 opts, inputs = make_argparse(outdir=self.tmpdir,
32 gw=False, extension=extension, bilby=bilby)
33 func = functions(opts)
34 PlotGeneration(inputs)
35 WebpageGeneration(inputs)
36 func["MetaFile"](inputs)
37 func["FinishingTouches"](inputs)
39 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
40 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
41 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, gw=False)))
42 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=False))
43 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=False) for i in plots)
44 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, gw=False)))
45 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=False))
46 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=False) for i in files)
47 self.check_samples(extension, bilby=bilby)
49 def check_samples(self, extension, bilby=False):
50 """Check that the samples in the result file are consistent with the
51 inputs
52 """
53 from pesummary.core.file.read import read
55 initial_samples = read_result_file(extension=extension, bilby=bilby, outdir=self.tmpdir)
56 data = read("{}/samples/posterior_samples.h5".format(self.tmpdir))
57 samples = data.samples_dict
58 label = data.labels[0]
59 for param in initial_samples.keys():
60 for i, j in zip(initial_samples[param], samples[label][param]):
61 assert np.round(i, 8) == np.round(j, 8)
64class GWBase(Base):
65 """Base class to test the full workflow including gw specific options
66 """
67 @pytest.mark.workflowtest
68 def test_single_run(self, extension, bilby=False, lalinference=False):
69 """Test the full workflow for a single result file case
71 Parameters
72 ----------
73 result_file: str
74 path to result file you wish to run with
75 """
76 opts, inputs = make_argparse(outdir=self.tmpdir,
77 gw=True, extension=extension, bilby=bilby, lalinference=lalinference)
78 print(opts)
79 func = functions(opts)
80 PlotGeneration(inputs, gw=True)
81 WebpageGeneration(inputs, gw=True)
82 func["MetaFile"](inputs)
83 func["FinishingTouches"](inputs)
85 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
86 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
87 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, gw=True)))
88 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True))
89 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True) for i in plots)
90 for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, gw=True)):
91 print(i, j)
92 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, gw=True)))
93 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True))
94 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True) for i in files)
95 self.check_samples(extension, bilby=bilby, lalinference=lalinference)
97 def check_samples(self, extension, bilby=False, lalinference=False):
98 """Check that the samples in the result file are consistent with the
99 inputs
100 """
101 from pesummary.core.file.read import read
103 initial_samples = read_result_file(
104 extension=extension, bilby=bilby, lalinference=lalinference,
105 outdir=self.tmpdir
106 )
107 data = read("{}/samples/posterior_samples.h5".format(self.tmpdir))
108 samples = data.samples_dict
109 label = data.labels[0]
110 for param in initial_samples.keys():
111 for i, j in zip(initial_samples[param], samples[label][param]):
112 assert np.round(i, 8) == np.round(j, 8)
115class TestCoreDat(Base):
116 """Test the full workflow with a core dat file
117 """
118 def setup_method(self):
119 """Setup the TestCoreDat class
120 """
121 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
122 if not os.path.isdir(self.tmpdir):
123 os.mkdir(self.tmpdir)
125 def teardown_method(self):
126 """Remove the files and directories created from this class
127 """
128 if os.path.isdir(self.tmpdir):
129 shutil.rmtree(self.tmpdir)
131 @pytest.mark.workflowtest
132 def test_single_run(self):
133 """Test the full workflow with a core dat result file
134 """
135 extension = "dat"
136 super(TestCoreDat, self).test_single_run(extension)
139class TestCoreJson(Base):
140 """Test the full workflow with a core json file
141 """
142 def setup_method(self):
143 """Setup the TestCoreJson class
144 """
145 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
146 if not os.path.isdir(self.tmpdir):
147 os.mkdir(self.tmpdir)
149 def teardown_method(self):
150 """Remove the files and directories created from this class
151 """
152 if os.path.isdir(self.tmpdir):
153 shutil.rmtree(self.tmpdir)
155 @pytest.mark.workflowtest
156 def test_single_run(self):
157 """Test the full workflow with a core json result file
158 """
159 extension = "json"
160 super(TestCoreJson, self).test_single_run(extension)
163class TestCoreHDF5(Base):
164 """Test the full workflow with a core hdf5 file
165 """
166 def setup_method(self):
167 """Setup the TestCoreHDF5 class
168 """
169 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
170 if not os.path.isdir(self.tmpdir):
171 os.mkdir(self.tmpdir)
173 def teardown_method(self):
174 """Remove the files and directories created from this class
175 """
176 if os.path.isdir(self.tmpdir):
177 shutil.rmtree(self.tmpdir)
179 @pytest.mark.workflowtest
180 def test_single_run(self):
181 """Test the full workflow with a core hdf5 result file
182 """
183 extension = "h5"
184 super(TestCoreHDF5, self).test_single_run(extension)
187class TestCoreBilbyJson(Base):
188 """Test the full workflow with a core json bilby file
189 """
190 def setup_method(self):
191 """Setup the TestCoreBilby class
192 """
193 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
194 if not os.path.isdir(self.tmpdir):
195 os.mkdir(self.tmpdir)
197 def teardown_method(self):
198 """Remove the files and directories created from this class
199 """
200 if os.path.isdir(self.tmpdir):
201 shutil.rmtree(self.tmpdir)
202 if os.path.isdir("{}_pesummary".format(self.tmpdir)):
203 shutil.rmtree("{}_pesummary".format(self.tmpdir))
205 @pytest.mark.workflowtest
206 def test_single_run(self):
207 """Test the full workflow with a core bilby result file
208 """
209 extension = "json"
210 super(TestCoreBilbyJson, self).test_single_run(extension, bilby=True)
212 @pytest.mark.workflowtest
213 def test_double_run(self):
214 """Test the full workflow for 2 lalinference result files
215 """
216 opts, inputs = make_argparse(outdir=self.tmpdir,
217 gw=False, extension="json", bilby=True, number=2)
218 func = functions(opts)
219 PlotGeneration(inputs)
220 WebpageGeneration(inputs)
221 func["MetaFile"](inputs)
222 func["FinishingTouches"](inputs)
224 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
225 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
226 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir,
227 gw=False, number=2)))
228 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2))
229 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2) for i in plots)
230 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir,
231 gw=False, number=2)))
232 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2))
233 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2) for i in files)
235 @pytest.mark.workflowtest
236 def test_existing_run(self):
237 """Test the fill workflow for when you add to an existing webpage
238 """
239 opts, inputs = make_argparse(outdir=self.tmpdir,
240 gw=False, extension="json", bilby=True)
241 func = functions(opts)
242 PlotGeneration(inputs)
243 WebpageGeneration(inputs)
244 func["MetaFile"](inputs)
245 func["FinishingTouches"](inputs)
247 opts, inputs = make_argparse(outdir=self.tmpdir,
248 gw=False, extension="json", bilby=True, existing=True)
249 func = functions(opts)
250 PlotGeneration(inputs)
251 WebpageGeneration(inputs)
252 func["MetaFile"](inputs)
253 func["FinishingTouches"](inputs)
255 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
256 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
258 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir,
259 gw=False, number=2)))
260 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2))
261 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2) for i in plots)
262 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir,
263 gw=False, number=2)))
264 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2))
265 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2) for i in files)
267 @pytest.mark.workflowtest
268 def test_pesummary_input(self):
269 """Test the full workflow for a pesummary input file
270 """
271 opts, inputs = make_argparse(outdir=self.tmpdir,
272 gw=False, extension="json", bilby=True, number=2)
273 func = functions(opts)
274 PlotGeneration(inputs)
275 WebpageGeneration(inputs)
276 func["MetaFile"](inputs)
277 func["FinishingTouches"](inputs)
279 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
280 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
282 from pesummary.core.cli.parser import ArgumentParser
284 parser = ArgumentParser()
285 parser.add_all_known_options_to_parser()
286 default_args = ["--webdir", "{}_pesummary".format(self.tmpdir),
287 "--samples", "{}/samples/posterior_samples.h5".format(self.tmpdir),
288 "--disable_expert"]
289 opts = parser.parse_args(default_args)
290 func = functions(opts)
291 inputs = func["input"](opts)
292 PlotGeneration(inputs)
293 WebpageGeneration(inputs)
294 func["MetaFile"](inputs)
295 func["FinishingTouches"](inputs)
297 plots_pesummary = sorted(glob.glob("{}_pesummary/plots/*.png".format(self.tmpdir)))
298 files_pesummary = sorted(glob.glob("{}_pesummary/html/*.html".format(self.tmpdir)))
300 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip(
301 plots, plots_pesummary))
302 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip(
303 files, files_pesummary))
305class TestCoreBilbyHDF5(Base):
306 """Test the full workflow with a core hdf5 bilby file
307 """
308 def setup_method(self):
309 """Setup the TestCoreBilby class
310 """
311 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
312 if not os.path.isdir(self.tmpdir):
313 os.mkdir(self.tmpdir)
315 def teardown_method(self):
316 """Remove the files and directories created from this class
317 """
318 if os.path.isdir(self.tmpdir):
319 shutil.rmtree(self.tmpdir)
321 @pytest.mark.workflowtest
322 def test_single_run(self):
323 """Test the full workflow with a core bilby result file
324 """
325 extension = "hdf5"
326 super(TestCoreBilbyHDF5, self).test_single_run(extension, bilby=True)
329class TestGWDat(GWBase):
330 """Test the full workflow with a gw dat file
331 """
332 def setup_method(self):
333 """Setup the TestCoreDat class
334 """
335 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
336 if not os.path.isdir(self.tmpdir):
337 os.mkdir(self.tmpdir)
339 def teardown_method(self):
340 """Remove the files and directories created from this class
341 """
342 if os.path.isdir(self.tmpdir):
343 shutil.rmtree(self.tmpdir)
345 @pytest.mark.workflowtest
346 def test_single_run(self):
347 """Test the full workflow with a gw dat result file
348 """
349 extension = "dat"
350 super(TestGWDat, self).test_single_run(extension)
353class TestGWJson(GWBase):
354 """Test the full workflow with a json dat file
355 """
356 def setup_method(self):
357 """Setup the TestGWJson class
358 """
359 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
360 if not os.path.isdir(self.tmpdir):
361 os.mkdir(self.tmpdir)
363 def teardown_method(self):
364 """Remove the files and directories created from this class
365 """
366 if os.path.isdir(self.tmpdir):
367 shutil.rmtree(self.tmpdir)
369 @pytest.mark.workflowtest
370 def test_single_run(self):
371 """Test the full workflow with a gw json result file
372 """
373 extension = "json"
374 super(TestGWJson, self).test_single_run(extension)
377class TestGWBilbyJson(GWBase):
378 """Test the full workflow with a gw bilby json file
379 """
380 def setup_method(self):
381 """Setup the TestGWJson class
382 """
383 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
384 if not os.path.isdir(self.tmpdir):
385 os.mkdir(self.tmpdir)
387 def teardown_method(self):
388 """Remove the files and directories created from this class
389 """
390 if os.path.isdir(self.tmpdir):
391 shutil.rmtree(self.tmpdir)
393 @pytest.mark.workflowtest
394 def test_single_run(self):
395 """Test the full workflow with a gw bilby json result file
396 """
397 extension = "json"
398 super(TestGWBilbyJson, self).test_single_run(extension, bilby=True)
401class TestGWBilbyHDF5(GWBase):
402 """Test the full workflow with a gw bilby HDF5 file
403 """
404 def setup_method(self):
405 """Setup the TestGWJson class
406 """
407 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
408 if not os.path.isdir(self.tmpdir):
409 os.mkdir(self.tmpdir)
411 def teardown_method(self):
412 """Remove the files and directories created from this class
413 """
414 if os.path.isdir(self.tmpdir):
415 shutil.rmtree(self.tmpdir)
417 @pytest.mark.workflowtest
418 def test_single_run(self):
419 """Test the full workflow with a gw bilby HDF5 result file
420 """
421 extension = "hdf5"
422 super(TestGWBilbyHDF5, self).test_single_run(extension, bilby=True)
425class TestGWLALInference(GWBase):
426 """Test the full workflow with a lalinference file
427 """
428 def setup_method(self):
429 """Setup the TestGWJson class
430 """
431 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name
432 if not os.path.isdir(self.tmpdir):
433 os.mkdir(self.tmpdir)
435 def teardown_method(self):
436 """Remove the files and directories created from this class
437 """
438 if os.path.isdir(self.tmpdir):
439 shutil.rmtree(self.tmpdir)
440 if os.path.isdir("{}_pesummary".format(self.tmpdir)):
441 shutil.rmtree("{}_pesummary".format(self.tmpdir))
443 @pytest.mark.workflowtest
444 def test_single_run(self):
445 """Test the full workflow with a lalinference result file
446 """
447 extension = "hdf5"
448 super(TestGWLALInference, self).test_single_run(extension, lalinference=True)
450 @pytest.mark.workflowtest
451 def test_double_run(self):
452 """Test the full workflow for 2 lalinference result files
453 """
454 opts, inputs = make_argparse(outdir=self.tmpdir,
455 gw=True, extension="hdf5", lalinference=True, number=2)
456 func = functions(opts)
457 PlotGeneration(inputs, gw=True)
458 WebpageGeneration(inputs, gw=True)
459 func["MetaFile"](inputs)
460 func["FinishingTouches"](inputs)
462 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
463 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
464 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir,
465 gw=True, number=2)))
466 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2))
467 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2) for i in plots)
468 for i, j in zip(files, get_list_of_files(outdir=self.tmpdir,
469 gw=True, number=2)):
470 print(i, j)
471 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir,
472 gw=True, number=2)))
473 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2))
474 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2) for i in files)
476 @pytest.mark.workflowtest
477 def test_existing_run(self):
478 """Test the fill workflow for when you add to an existing webpage
479 """
480 opts, inputs = make_argparse(outdir=self.tmpdir,
481 gw=True, extension="hdf5", lalinference=True)
482 func = functions(opts)
483 PlotGeneration(inputs, gw=True)
484 WebpageGeneration(inputs, gw=True)
485 func["MetaFile"](inputs)
486 func["FinishingTouches"](inputs)
488 opts, inputs = make_argparse(outdir=self.tmpdir,
489 gw=True, extension="hdf5", lalinference=True, existing=True)
490 func = functions(opts)
491 PlotGeneration(inputs, gw=True)
492 WebpageGeneration(inputs, gw=True)
493 func["MetaFile"](inputs)
494 func["FinishingTouches"](inputs)
496 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
497 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
498 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir,
499 gw=True, number=2)))
500 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2))
501 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2) for i in plots)
502 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir,
503 gw=True, number=2)))
504 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2))
505 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2) for i in files)
507 @pytest.mark.workflowtest
508 def test_pesummary_input(self):
509 """Test the full workflow for a pesummary input file
510 """
511 opts, inputs = make_argparse(outdir=self.tmpdir,
512 gw=True, extension="hdf5", lalinference=True, number=2)
513 func = functions(opts)
514 PlotGeneration(inputs, gw=True)
515 WebpageGeneration(inputs, gw=True)
516 func["MetaFile"](inputs)
517 func["FinishingTouches"](inputs)
519 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir)))
520 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir)))
522 from pesummary.gw.cli.parser import ArgumentParser
524 parser = ArgumentParser()
525 parser.add_all_known_options_to_parser()
526 default_args = ["--webdir", "{}_pesummary".format(self.tmpdir),
527 "--samples", "{}/samples/posterior_samples.h5".format(self.tmpdir),
528 "--gw", "--disable_expert"]
529 from pesummary.gw.file.read import read
530 f = read("{}/samples/posterior_samples.h5".format(self.tmpdir))
531 opts = parser.parse_args(default_args)
532 inputs = func["input"](opts)
533 PlotGeneration(inputs, gw=True)
534 WebpageGeneration(inputs, gw=True)
535 func["MetaFile"](inputs)
536 func["FinishingTouches"](inputs)
538 plots_pesummary = sorted(glob.glob("{}_pesummary/plots/*.png".format(self.tmpdir)))
539 files_pesummary = sorted(glob.glob("{}_pesummary/html/*.html".format(self.tmpdir)))
541 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip(
542 plots, plots_pesummary))
543 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip(
544 files, files_pesummary))