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

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 

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 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) 

96 

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 

102 

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) 

113 

114 

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) 

124 

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) 

130 

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) 

137 

138 

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) 

148 

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) 

154 

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) 

161 

162 

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) 

172 

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) 

178 

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) 

185 

186 

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) 

196 

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)) 

204 

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) 

211 

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) 

223 

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) 

234 

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) 

246 

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) 

254 

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

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

257 

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) 

266 

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) 

278 

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

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

281 

282 from pesummary.core.cli.parser import ArgumentParser 

283 

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) 

296 

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

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

299 

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)) 

304 

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) 

314 

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) 

320 

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) 

327 

328 

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) 

338 

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) 

344 

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) 

351 

352 

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) 

362 

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) 

368 

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) 

375 

376 

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) 

386 

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) 

392 

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) 

399 

400 

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) 

410 

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) 

416 

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) 

423 

424 

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) 

434 

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)) 

442 

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) 

449 

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) 

461 

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) 

475 

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) 

487 

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) 

495 

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) 

506 

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) 

518 

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

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

521 

522 from pesummary.gw.cli.parser import ArgumentParser 

523 

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) 

537 

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

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

540 

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))