Coverage for pesummary/tests/workflow_test.py: 25.3%

1# Licensed under an MIT style license -- see LICENSE.md 

2 

3import os 

4import shutil 

5import glob 

6import pytest 

7import numpy as np 

8 

9from .base import make_argparse, get_list_of_plots, get_list_of_files 

10from .base import read_result_file, testing_dir 

11from pesummary.utils.utils import functions 

12from pesummary.cli.summarypages import WebpageGeneration 

13from pesummary.cli.summaryplots import PlotGeneration 

14import tempfile 

15 

16__author__ = ["Charlie Hoy <charlie.hoy@ligo.org>"] 

17 

18 

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 

25 

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) 

38 

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) 

48 

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 

54 

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) 

62 

63 

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 

70 

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) 

84 

85 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

86 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

87 for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, gw=True)): 

88 print(i, j) 

89 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, gw=True))) 

90 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True)) 

91 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True) for i in plots) 

92 for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, gw=True)): 

93 print(i, j) 

94 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, gw=True))) 

95 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True)) 

96 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True) for i in files) 

97 self.check_samples(extension, bilby=bilby, lalinference=lalinference) 

98 

99 def check_samples(self, extension, bilby=False, lalinference=False): 

100 """Check that the samples in the result file are consistent with the 

101 inputs 

102 """ 

103 from pesummary.core.file.read import read 

104 

105 initial_samples = read_result_file( 

106 extension=extension, bilby=bilby, lalinference=lalinference, 

107 outdir=self.tmpdir 

108 ) 

109 data = read("{}/samples/posterior_samples.h5".format(self.tmpdir)) 

110 samples = data.samples_dict 

111 label = data.labels[0] 

112 for param in initial_samples.keys(): 

113 for i, j in zip(initial_samples[param], samples[label][param]): 

114 assert np.round(i, 8) == np.round(j, 8) 

115 

116 

117class TestCoreDat(Base): 

118 """Test the full workflow with a core dat file 

119 """ 

120 def setup_method(self): 

121 """Setup the TestCoreDat class 

122 """ 

123 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

124 if not os.path.isdir(self.tmpdir): 

125 os.mkdir(self.tmpdir) 

126 

127 def teardown_method(self): 

128 """Remove the files and directories created from this class 

129 """ 

130 if os.path.isdir(self.tmpdir): 

131 shutil.rmtree(self.tmpdir) 

132 

133 @pytest.mark.workflowtest 

134 def test_single_run(self): 

135 """Test the full workflow with a core dat result file 

136 """ 

137 extension = "dat" 

138 super(TestCoreDat, self).test_single_run(extension) 

139 

140 

141class TestCoreJson(Base): 

142 """Test the full workflow with a core json file 

143 """ 

144 def setup_method(self): 

145 """Setup the TestCoreJson class 

146 """ 

147 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

148 if not os.path.isdir(self.tmpdir): 

149 os.mkdir(self.tmpdir) 

150 

151 def teardown_method(self): 

152 """Remove the files and directories created from this class 

153 """ 

154 if os.path.isdir(self.tmpdir): 

155 shutil.rmtree(self.tmpdir) 

156 

157 @pytest.mark.workflowtest 

158 def test_single_run(self): 

159 """Test the full workflow with a core json result file 

160 """ 

161 extension = "json" 

162 super(TestCoreJson, self).test_single_run(extension) 

163 

164 

165class TestCoreHDF5(Base): 

166 """Test the full workflow with a core hdf5 file 

167 """ 

168 def setup_method(self): 

169 """Setup the TestCoreHDF5 class 

170 """ 

171 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

172 if not os.path.isdir(self.tmpdir): 

173 os.mkdir(self.tmpdir) 

174 

175 def teardown_method(self): 

176 """Remove the files and directories created from this class 

177 """ 

178 if os.path.isdir(self.tmpdir): 

179 shutil.rmtree(self.tmpdir) 

180 

181 @pytest.mark.workflowtest 

182 def test_single_run(self): 

183 """Test the full workflow with a core hdf5 result file 

184 """ 

185 extension = "h5" 

186 super(TestCoreHDF5, self).test_single_run(extension) 

187 

188 

189class TestCoreBilbyJson(Base): 

190 """Test the full workflow with a core json bilby file 

191 """ 

192 def setup_method(self): 

193 """Setup the TestCoreBilby class 

194 """ 

195 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

196 if not os.path.isdir(self.tmpdir): 

197 os.mkdir(self.tmpdir) 

198 

199 def teardown_method(self): 

200 """Remove the files and directories created from this class 

201 """ 

202 if os.path.isdir(self.tmpdir): 

203 shutil.rmtree(self.tmpdir) 

204 if os.path.isdir("{}_pesummary".format(self.tmpdir)): 

205 shutil.rmtree("{}_pesummary".format(self.tmpdir)) 

206 

207 @pytest.mark.workflowtest 

208 def test_single_run(self): 

209 """Test the full workflow with a core bilby result file 

210 """ 

211 extension = "json" 

212 super(TestCoreBilbyJson, self).test_single_run(extension, bilby=True) 

213 

214 @pytest.mark.workflowtest 

215 def test_double_run(self): 

216 """Test the full workflow for 2 lalinference result files 

217 """ 

218 opts, inputs = make_argparse(outdir=self.tmpdir, 

219 gw=False, extension="json", bilby=True, number=2) 

220 func = functions(opts) 

221 PlotGeneration(inputs) 

222 WebpageGeneration(inputs) 

223 func["MetaFile"](inputs) 

224 func["FinishingTouches"](inputs) 

225 

226 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

227 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

228 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, 

229 gw=False, number=2))) 

230 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2)) 

231 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2) for i in plots) 

232 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, 

233 gw=False, number=2))) 

234 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2)) 

235 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2) for i in files) 

236 

237 @pytest.mark.workflowtest 

238 def test_existing_run(self): 

239 """Test the fill workflow for when you add to an existing webpage 

240 """ 

241 opts, inputs = make_argparse(outdir=self.tmpdir, 

242 gw=False, extension="json", bilby=True) 

243 func = functions(opts) 

244 PlotGeneration(inputs) 

245 WebpageGeneration(inputs) 

246 func["MetaFile"](inputs) 

247 func["FinishingTouches"](inputs) 

248 

249 opts, inputs = make_argparse(outdir=self.tmpdir, 

250 gw=False, extension="json", bilby=True, existing=True) 

251 func = functions(opts) 

252 PlotGeneration(inputs) 

253 WebpageGeneration(inputs) 

254 func["MetaFile"](inputs) 

255 func["FinishingTouches"](inputs) 

256 

257 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

258 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

259 

260 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, 

261 gw=False, number=2))) 

262 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2)) 

263 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=False, number=2) for i in plots) 

264 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, 

265 gw=False, number=2))) 

266 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2)) 

267 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=False, number=2) for i in files) 

268 

269 @pytest.mark.workflowtest 

270 def test_pesummary_input(self): 

271 """Test the full workflow for a pesummary input file 

272 """ 

273 opts, inputs = make_argparse(outdir=self.tmpdir, 

274 gw=False, extension="json", bilby=True, number=2) 

275 func = functions(opts) 

276 PlotGeneration(inputs) 

277 WebpageGeneration(inputs) 

278 func["MetaFile"](inputs) 

279 func["FinishingTouches"](inputs) 

280 

281 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

282 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

283 

284 from pesummary.core.cli.parser import ArgumentParser 

285 

286 parser = ArgumentParser() 

287 parser.add_all_known_options_to_parser() 

288 default_args = ["--webdir", "{}_pesummary".format(self.tmpdir), 

289 "--samples", "{}/samples/posterior_samples.h5".format(self.tmpdir), 

290 "--disable_expert"] 

291 opts = parser.parse_args(default_args) 

292 func = functions(opts) 

293 inputs = func["input"](opts) 

294 PlotGeneration(inputs) 

295 WebpageGeneration(inputs) 

296 func["MetaFile"](inputs) 

297 func["FinishingTouches"](inputs) 

298 

299 plots_pesummary = sorted(glob.glob("{}_pesummary/plots/*.png".format(self.tmpdir))) 

300 files_pesummary = sorted(glob.glob("{}_pesummary/html/*.html".format(self.tmpdir))) 

301 

302 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip( 

303 plots, plots_pesummary)) 

304 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip( 

305 files, files_pesummary)) 

306 

307class TestCoreBilbyHDF5(Base): 

308 """Test the full workflow with a core hdf5 bilby file 

309 """ 

310 def setup_method(self): 

311 """Setup the TestCoreBilby class 

312 """ 

313 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

314 if not os.path.isdir(self.tmpdir): 

315 os.mkdir(self.tmpdir) 

316 

317 def teardown_method(self): 

318 """Remove the files and directories created from this class 

319 """ 

320 if os.path.isdir(self.tmpdir): 

321 shutil.rmtree(self.tmpdir) 

322 

323 @pytest.mark.workflowtest 

324 def test_single_run(self): 

325 """Test the full workflow with a core bilby result file 

326 """ 

327 extension = "hdf5" 

328 super(TestCoreBilbyHDF5, self).test_single_run(extension, bilby=True) 

329 

330 

331class TestGWDat(GWBase): 

332 """Test the full workflow with a gw dat file 

333 """ 

334 def setup_method(self): 

335 """Setup the TestCoreDat class 

336 """ 

337 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

338 if not os.path.isdir(self.tmpdir): 

339 os.mkdir(self.tmpdir) 

340 

341 def teardown_method(self): 

342 """Remove the files and directories created from this class 

343 """ 

344 if os.path.isdir(self.tmpdir): 

345 shutil.rmtree(self.tmpdir) 

346 

347 @pytest.mark.workflowtest 

348 def test_single_run(self): 

349 """Test the full workflow with a gw dat result file 

350 """ 

351 extension = "dat" 

352 super(TestGWDat, self).test_single_run(extension) 

353 

354 

355class TestGWJson(GWBase): 

356 """Test the full workflow with a json dat file 

357 """ 

358 def setup_method(self): 

359 """Setup the TestGWJson class 

360 """ 

361 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

362 if not os.path.isdir(self.tmpdir): 

363 os.mkdir(self.tmpdir) 

364 

365 def teardown_method(self): 

366 """Remove the files and directories created from this class 

367 """ 

368 if os.path.isdir(self.tmpdir): 

369 shutil.rmtree(self.tmpdir) 

370 

371 @pytest.mark.workflowtest 

372 def test_single_run(self): 

373 """Test the full workflow with a gw json result file 

374 """ 

375 extension = "json" 

376 super(TestGWJson, self).test_single_run(extension) 

377 

378 

379class TestGWBilbyJson(GWBase): 

380 """Test the full workflow with a gw bilby json file 

381 """ 

382 def setup_method(self): 

383 """Setup the TestGWJson class 

384 """ 

385 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

386 if not os.path.isdir(self.tmpdir): 

387 os.mkdir(self.tmpdir) 

388 

389 def teardown_method(self): 

390 """Remove the files and directories created from this class 

391 """ 

392 if os.path.isdir(self.tmpdir): 

393 shutil.rmtree(self.tmpdir) 

394 

395 @pytest.mark.workflowtest 

396 def test_single_run(self): 

397 """Test the full workflow with a gw bilby json result file 

398 """ 

399 extension = "json" 

400 super(TestGWBilbyJson, self).test_single_run(extension, bilby=True) 

401 

402 

403class TestGWBilbyHDF5(GWBase): 

404 """Test the full workflow with a gw bilby HDF5 file 

405 """ 

406 def setup_method(self): 

407 """Setup the TestGWJson class 

408 """ 

409 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

410 if not os.path.isdir(self.tmpdir): 

411 os.mkdir(self.tmpdir) 

412 

413 def teardown_method(self): 

414 """Remove the files and directories created from this class 

415 """ 

416 if os.path.isdir(self.tmpdir): 

417 shutil.rmtree(self.tmpdir) 

418 

419 @pytest.mark.workflowtest 

420 def test_single_run(self): 

421 """Test the full workflow with a gw bilby HDF5 result file 

422 """ 

423 extension = "hdf5" 

424 super(TestGWBilbyHDF5, self).test_single_run(extension, bilby=True) 

425 

426 

427class TestGWLALInference(GWBase): 

428 """Test the full workflow with a lalinference file 

429 """ 

430 def setup_method(self): 

431 """Setup the TestGWJson class 

432 """ 

433 self.tmpdir = tempfile.TemporaryDirectory(prefix=".", dir=".").name 

434 if not os.path.isdir(self.tmpdir): 

435 os.mkdir(self.tmpdir) 

436 

437 def teardown_method(self): 

438 """Remove the files and directories created from this class 

439 """ 

440 if os.path.isdir(self.tmpdir): 

441 shutil.rmtree(self.tmpdir) 

442 if os.path.isdir("{}_pesummary".format(self.tmpdir)): 

443 shutil.rmtree("{}_pesummary".format(self.tmpdir)) 

444 

445 @pytest.mark.workflowtest 

446 def test_single_run(self): 

447 """Test the full workflow with a lalinference result file 

448 """ 

449 extension = "hdf5" 

450 super(TestGWLALInference, self).test_single_run(extension, lalinference=True) 

451 

452 @pytest.mark.workflowtest 

453 def test_double_run(self): 

454 """Test the full workflow for 2 lalinference result files 

455 """ 

456 opts, inputs = make_argparse(outdir=self.tmpdir, 

457 gw=True, extension="hdf5", lalinference=True, number=2) 

458 func = functions(opts) 

459 PlotGeneration(inputs, gw=True) 

460 WebpageGeneration(inputs, gw=True) 

461 func["MetaFile"](inputs) 

462 func["FinishingTouches"](inputs) 

463 

464 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

465 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

466 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, 

467 gw=True, number=2))) 

468 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2)) 

469 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2) for i in plots) 

470 for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, 

471 gw=True, number=2)): 

472 print(i, j) 

473 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, 

474 gw=True, number=2))) 

475 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2)) 

476 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2) for i in files) 

477 

478 @pytest.mark.workflowtest 

479 def test_existing_run(self): 

480 """Test the fill workflow for when you add to an existing webpage 

481 """ 

482 opts, inputs = make_argparse(outdir=self.tmpdir, 

483 gw=True, extension="hdf5", lalinference=True) 

484 func = functions(opts) 

485 PlotGeneration(inputs, gw=True) 

486 WebpageGeneration(inputs, gw=True) 

487 func["MetaFile"](inputs) 

488 func["FinishingTouches"](inputs) 

489 

490 opts, inputs = make_argparse(outdir=self.tmpdir, 

491 gw=True, extension="hdf5", lalinference=True, existing=True) 

492 func = functions(opts) 

493 PlotGeneration(inputs, gw=True) 

494 WebpageGeneration(inputs, gw=True) 

495 func["MetaFile"](inputs) 

496 func["FinishingTouches"](inputs) 

497 

498 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

499 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

500 assert all(i == j for i, j in zip(plots, get_list_of_plots(outdir=self.tmpdir, 

501 gw=True, number=2))) 

502 assert all(i in plots for i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2)) 

503 assert all(i in get_list_of_plots(outdir=self.tmpdir, gw=True, number=2) for i in plots) 

504 assert all(i == j for i, j in zip(files, get_list_of_files(outdir=self.tmpdir, 

505 gw=True, number=2))) 

506 assert all(i in files for i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2)) 

507 assert all(i in get_list_of_files(outdir=self.tmpdir, gw=True, number=2) for i in files) 

508 

509 @pytest.mark.workflowtest 

510 def test_pesummary_input(self): 

511 """Test the full workflow for a pesummary input file 

512 """ 

513 opts, inputs = make_argparse(outdir=self.tmpdir, 

514 gw=True, extension="hdf5", lalinference=True, number=2) 

515 func = functions(opts) 

516 PlotGeneration(inputs, gw=True) 

517 WebpageGeneration(inputs, gw=True) 

518 func["MetaFile"](inputs) 

519 func["FinishingTouches"](inputs) 

520 

521 plots = sorted(glob.glob("{}/plots/*.png".format(self.tmpdir))) 

522 files = sorted(glob.glob("{}/html/*.html".format(self.tmpdir))) 

523 

524 from pesummary.gw.cli.parser import ArgumentParser 

525 

526 parser = ArgumentParser() 

527 parser.add_all_known_options_to_parser() 

528 default_args = ["--webdir", "{}_pesummary".format(self.tmpdir), 

529 "--samples", "{}/samples/posterior_samples.h5".format(self.tmpdir), 

530 "--gw", "--disable_expert", "--pastro_category_file", 

531 "{}/rates.yml".format(testing_dir), 

532 "--catch_terrestrial_probability_error"] 

533 from pesummary.gw.file.read import read 

534 f = read("{}/samples/posterior_samples.h5".format(self.tmpdir)) 

535 opts = parser.parse_args(default_args) 

536 inputs = func["input"](opts) 

537 PlotGeneration(inputs, gw=True) 

538 WebpageGeneration(inputs, gw=True) 

539 func["MetaFile"](inputs) 

540 func["FinishingTouches"](inputs) 

541 

542 plots_pesummary = sorted(glob.glob("{}_pesummary/plots/*.png".format(self.tmpdir))) 

543 files_pesummary = sorted(glob.glob("{}_pesummary/html/*.html".format(self.tmpdir))) 

544 

545 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip( 

546 plots, plots_pesummary)) 

547 assert all(i.split("/")[-1] == j.split("/")[-1] for i, j in zip( 

548 files, files_pesummary))