1#!/usr/bin/env python 

2# coding=utf-8 

3 

4"""Tests for our CLI applications""" 

5 

6import fnmatch 

7import logging 

8import os 

9import re 

10import tempfile 

11 

12from click.testing import CliRunner 

13 

14from bob.extension.scripts.click_helper import assert_click_runner_result 

15 

16from . import mock_dataset 

17 

18stare_datadir, stare_dataset = mock_dataset() 

19 

20 

21def _assert_exit_0(result): 

22 

23 assert_click_runner_result(result) 

24 

25 

26def _check_help(entry_point): 

27 

28 runner = CliRunner() 

29 result = runner.invoke(entry_point, ["--help"]) 

30 _assert_exit_0(result) 

31 assert result.output.startswith("Usage:") 

32 

33 

34def test_main_help(): 

35 from ..script.binseg import binseg 

36 

37 _check_help(binseg) 

38 

39 

40def test_experiment_help(): 

41 from ..script.experiment import experiment 

42 

43 _check_help(experiment) 

44 

45 

46def _str_counter(substr, s): 

47 return sum(1 for _ in re.finditer(substr, s, re.MULTILINE)) 

48 

49 

50def _check_experiment_stare(caplog, overlay, multiprocess=False): 

51 

52 from ..script.experiment import experiment 

53 

54 # ensures we capture only ERROR messages and above by default 

55 caplog.set_level(logging.ERROR) 

56 

57 runner = CliRunner() 

58 with runner.isolated_filesystem(), caplog.at_level( 

59 logging.INFO, logger="bob.ip.binseg" 

60 ), tempfile.NamedTemporaryFile(mode="wt") as config: 

61 

62 # re-write STARE dataset configuration for test 

63 config.write("from bob.ip.binseg.data.stare import _make_dataset\n") 

64 config.write(f"_raw = _make_dataset('{stare_datadir}')\n") 

65 config.write( 

66 "from bob.ip.binseg.configs.datasets.stare import _maker\n" 

67 ) 

68 config.write("dataset = _maker('ah', _raw)\n") 

69 config.write("second_annotator = _maker('vk', _raw)\n") 

70 config.flush() 

71 

72 output_folder = "results" 

73 options = [ 

74 "lwnet", 

75 config.name, 

76 "-vv", 

77 "--epochs=1", 

78 "--batch-size=1", 

79 "--steps=10", 

80 f"--output-folder={output_folder}", 

81 ] 

82 if overlay: 

83 options += ["--overlayed"] 

84 if multiprocess: 

85 options += ["--multiproc-data-loading=1"] 

86 

87 result = runner.invoke(experiment, options) 

88 _assert_exit_0(result) 

89 

90 # check command-line 

91 assert os.path.exists(os.path.join(output_folder, "command.sh")) 

92 

93 # check model was saved 

94 train_folder = os.path.join(output_folder, "model") 

95 assert os.path.exists(os.path.join(train_folder, "model_final.pth")) 

96 assert os.path.exists( 

97 os.path.join(train_folder, "model_lowest_valid_loss.pth") 

98 ) 

99 assert os.path.exists(os.path.join(train_folder, "last_checkpoint")) 

100 assert os.path.exists(os.path.join(train_folder, "constants.csv")) 

101 assert os.path.exists(os.path.join(train_folder, "trainlog.csv")) 

102 assert os.path.exists(os.path.join(train_folder, "model_summary.txt")) 

103 

104 # check predictions are there 

105 predict_folder = os.path.join(output_folder, "predictions") 

106 traindir = os.path.join(predict_folder, "train", "stare-images") 

107 assert os.path.exists(traindir) 

108 assert len(fnmatch.filter(os.listdir(traindir), "*.hdf5")) == 10 

109 testdir = os.path.join(predict_folder, "test", "stare-images") 

110 assert os.path.exists(testdir) 

111 assert len(fnmatch.filter(os.listdir(testdir), "*.hdf5")) == 10 

112 

113 overlay_folder = os.path.join(output_folder, "overlayed", "predictions") 

114 traindir = os.path.join(overlay_folder, "train", "stare-images") 

115 testdir = os.path.join(overlay_folder, "test", "stare-images") 

116 if overlay: 

117 # check overlayed images are there (since we requested them) 

118 assert os.path.exists(traindir) 

119 assert len(fnmatch.filter(os.listdir(traindir), "*.png")) == 10 

120 # check overlayed images are there (since we requested them) 

121 assert os.path.exists(testdir) 

122 assert len(fnmatch.filter(os.listdir(testdir), "*.png")) == 10 

123 else: 

124 assert not os.path.exists(traindir) 

125 assert not os.path.exists(testdir) 

126 

127 # check evaluation outputs 

128 eval_folder = os.path.join(output_folder, "analysis") 

129 assert os.path.exists(os.path.join(eval_folder, "train.csv")) 

130 # checks individual performance figures are there 

131 traindir = os.path.join(eval_folder, "train", "stare-images") 

132 assert os.path.exists(traindir) 

133 assert len(fnmatch.filter(os.listdir(traindir), "*.csv")) == 10 

134 

135 assert os.path.exists(os.path.join(eval_folder, "test.csv")) 

136 # checks individual performance figures are there 

137 testdir = os.path.join(eval_folder, "test", "stare-images") 

138 assert os.path.exists(testdir) 

139 assert len(fnmatch.filter(os.listdir(testdir), "*.csv")) == 10 

140 

141 assert os.path.exists( 

142 os.path.join(eval_folder, "second-annotator", "train.csv") 

143 ) 

144 # checks individual performance figures are there 

145 traindir_sa = os.path.join( 

146 eval_folder, "second-annotator", "train", "stare-images" 

147 ) 

148 assert os.path.exists(traindir_sa) 

149 assert len(fnmatch.filter(os.listdir(traindir_sa), "*.csv")) == 10 

150 

151 assert os.path.exists( 

152 os.path.join(eval_folder, "second-annotator", "test.csv") 

153 ) 

154 testdir_sa = os.path.join( 

155 eval_folder, "second-annotator", "test", "stare-images" 

156 ) 

157 assert os.path.exists(testdir_sa) 

158 assert len(fnmatch.filter(os.listdir(testdir_sa), "*.csv")) == 10 

159 

160 overlay_folder = os.path.join(output_folder, "overlayed", "analysis") 

161 traindir = os.path.join(overlay_folder, "train", "stare-images") 

162 testdir = os.path.join(overlay_folder, "test", "stare-images") 

163 if overlay: 

164 # check overlayed images are there (since we requested them) 

165 assert os.path.exists(traindir) 

166 assert len(fnmatch.filter(os.listdir(traindir), "*.png")) == 10 

167 assert os.path.exists(testdir) 

168 assert len(fnmatch.filter(os.listdir(testdir), "*.png")) == 10 

169 else: 

170 assert not os.path.exists(traindir) 

171 assert not os.path.exists(testdir) 

172 

173 # check overlayed images from first-to-second annotator comparisons 

174 # are there (since we requested them) 

175 overlay_folder = os.path.join( 

176 output_folder, "overlayed", "analysis", "second-annotator" 

177 ) 

178 traindir = os.path.join(overlay_folder, "train", "stare-images") 

179 testdir = os.path.join(overlay_folder, "test", "stare-images") 

180 if overlay: 

181 assert os.path.exists(traindir) 

182 assert len(fnmatch.filter(os.listdir(traindir), "*.png")) == 10 

183 assert os.path.exists(testdir) 

184 assert len(fnmatch.filter(os.listdir(testdir), "*.png")) == 10 

185 else: 

186 assert not os.path.exists(traindir) 

187 assert not os.path.exists(testdir) 

188 

189 # check outcomes of the comparison phase 

190 assert os.path.exists(os.path.join(output_folder, "comparison.pdf")) 

191 assert os.path.exists(os.path.join(output_folder, "comparison.rst")) 

192 

193 keywords = { 

194 r"^Started training$": 1, 

195 r"^Found \(dedicated\) '__train__' set for training$": 1, 

196 r"^Found \(dedicated\) '__valid__' set for validation$": 1, 

197 r"^Will checkpoint lowest loss model on validation set$": 1, 

198 r"^Continuing from epoch 0$": 1, 

199 r"^Saving model summary at.*$": 1, 

200 r"^Model has.*$": 1, 

201 r"^Found new low on validation set.*$": 1, 

202 r"^Saving checkpoint": 2, 

203 r"^Ended training$": 1, 

204 r"^Started prediction$": 1, 

205 r"^Loading checkpoint from": 1, 

206 r"^Ended prediction$": 1, 

207 r"^Started evaluation$": 1, 

208 r"^Maximum F1-score of.*\(chosen \*a posteriori\*\)$": 3, 

209 r"^F1-score of.*\(chosen \*a priori\*\)$": 2, 

210 r"^F1-score of.*\(second annotator; threshold=0.5\)$": 2, 

211 r"^Ended evaluation$": 1, 

212 r"^Started comparison$": 1, 

213 r"^Loading measures from": 4, 

214 r"^Creating and saving plot at": 1, 

215 r"^Tabulating performance summary...": 1, 

216 r"^Saving table at": 1, 

217 r"^Ended comparison.*$": 1, 

218 } 

219 messages = "\n".join([k.getMessage() for k in caplog.records]) 

220 for k, v in keywords.items(): 

221 total = _str_counter(k, messages) 

222 assert total == v, ( 

223 f"message '{k}' appears {total} times, but I expected " 

224 f"it to appear {v} times" 

225 ) 

226 

227 

228def test_experiment_stare_with_overlay(caplog): 

229 _check_experiment_stare(caplog, overlay=True) 

230 

231 

232def test_experiment_stare_without_overlay(caplog): 

233 _check_experiment_stare(caplog, overlay=False) 

234 

235 

236def test_experiment_stare_with_multiprocessing(caplog): 

237 _check_experiment_stare(caplog, overlay=False, multiprocess=True) 

238 

239 

240def _check_train(caplog, runner): 

241 

242 from ..script.train import train 

243 

244 with tempfile.NamedTemporaryFile(mode="wt") as config, caplog.at_level( 

245 logging.INFO, logger="bob.ip.binseg" 

246 ): 

247 

248 # single training set configuration 

249 config.write("from bob.ip.binseg.data.stare import _make_dataset\n") 

250 config.write(f"_raw = _make_dataset('{stare_datadir}')\n") 

251 config.write( 

252 "from bob.ip.binseg.configs.datasets.stare import _maker\n" 

253 ) 

254 config.write("dataset = _maker('ah', _raw)\n") 

255 config.flush() 

256 

257 output_folder = "results" 

258 result = runner.invoke( 

259 train, 

260 [ 

261 "lwnet", 

262 config.name, 

263 "-vv", 

264 "--epochs=1", 

265 "--batch-size=1", 

266 f"--output-folder={output_folder}", 

267 ], 

268 ) 

269 _assert_exit_0(result) 

270 

271 assert os.path.exists(os.path.join(output_folder, "model_final.pth")) 

272 assert os.path.exists( 

273 os.path.join(output_folder, "model_lowest_valid_loss.pth") 

274 ) 

275 assert os.path.exists(os.path.join(output_folder, "last_checkpoint")) 

276 assert os.path.exists(os.path.join(output_folder, "constants.csv")) 

277 assert os.path.exists(os.path.join(output_folder, "trainlog.csv")) 

278 assert os.path.exists(os.path.join(output_folder, "model_summary.txt")) 

279 

280 keywords = { 

281 r"^Found \(dedicated\) '__train__' set for training$": 1, 

282 r"^Found \(dedicated\) '__valid__' set for validation$": 1, 

283 r"^Continuing from epoch 0$": 1, 

284 r"^Saving model summary at.*$": 1, 

285 r"^Model has.*$": 1, 

286 r"^Saving checkpoint to .*/model_lowest_valid_loss.pth$": 1, 

287 r"^Saving checkpoint to .*/model_final.pth$": 1, 

288 r"^Total training time:": 1, 

289 } 

290 

291 messages = "\n".join([k.getMessage() for k in caplog.records]) 

292 for k, v in keywords.items(): 

293 total = _str_counter(k, messages) 

294 assert total == v, ( 

295 f"message '{k}' appears {total} times, but I expected " 

296 f"it to appear {v} times" 

297 ) 

298 

299 

300def _check_predict(caplog, runner): 

301 

302 from ..script.predict import predict 

303 

304 with tempfile.NamedTemporaryFile(mode="wt") as config, caplog.at_level( 

305 logging.INFO, logger="bob.ip.binseg" 

306 ): 

307 

308 # single training set configuration 

309 config.write("from bob.ip.binseg.data.stare import _make_dataset\n") 

310 config.write(f"_raw = _make_dataset('{stare_datadir}')\n") 

311 config.write( 

312 "from bob.ip.binseg.configs.datasets.stare import _maker\n" 

313 ) 

314 config.write("dataset = _maker('ah', _raw)['test']\n") 

315 config.flush() 

316 

317 output_folder = "predictions" 

318 overlay_folder = os.path.join("overlayed", "predictions") 

319 result = runner.invoke( 

320 predict, 

321 [ 

322 "lwnet", 

323 config.name, 

324 "-vv", 

325 "--batch-size=1", 

326 "--weight=results/model_final.pth", 

327 f"--output-folder={output_folder}", 

328 f"--overlayed={overlay_folder}", 

329 ], 

330 ) 

331 _assert_exit_0(result) 

332 

333 # check predictions are there 

334 basedir = os.path.join(output_folder, "test", "stare-images") 

335 assert os.path.exists(basedir) 

336 assert len(fnmatch.filter(os.listdir(basedir), "*.hdf5")) == 10 

337 

338 # check overlayed images are there (since we requested them) 

339 basedir = os.path.join(overlay_folder, "test", "stare-images") 

340 assert os.path.exists(basedir) 

341 assert len(fnmatch.filter(os.listdir(basedir), "*.png")) == 10 

342 

343 keywords = { 

344 r"^Loading checkpoint from.*$": 1, 

345 r"^Total time:.*$": 1, 

346 } 

347 

348 messages = "\n".join([k.getMessage() for k in caplog.records]) 

349 for k, v in keywords.items(): 

350 total = _str_counter(k, messages) 

351 assert total == v, ( 

352 f"message '{k}' appears {total} times, but I expected " 

353 f"it to appear {v} times" 

354 ) 

355 

356 

357def _check_evaluate(caplog, runner): 

358 

359 from ..script.evaluate import evaluate 

360 

361 with tempfile.NamedTemporaryFile(mode="wt") as config, caplog.at_level( 

362 logging.INFO, logger="bob.ip.binseg" 

363 ): 

364 

365 # single training set configuration 

366 config.write("from bob.ip.binseg.data.stare import _make_dataset\n") 

367 config.write(f"_raw = _make_dataset('{stare_datadir}')\n") 

368 config.write( 

369 "from bob.ip.binseg.configs.datasets.stare import _maker\n" 

370 ) 

371 config.write("dataset = _maker('ah', _raw)['test']\n") 

372 config.write("second_annotator = _maker('vk', _raw)['test']\n") 

373 config.flush() 

374 

375 output_folder = "evaluations" 

376 overlay_folder = os.path.join("overlayed", "analysis") 

377 result = runner.invoke( 

378 evaluate, 

379 [ 

380 config.name, 

381 "-vv", 

382 "--steps=10", 

383 f"--output-folder={output_folder}", 

384 "--predictions-folder=predictions", 

385 f"--overlayed={overlay_folder}", 

386 ], 

387 ) 

388 _assert_exit_0(result) 

389 

390 assert os.path.exists(os.path.join(output_folder, "test.csv")) 

391 # checks individual performance figures are there 

392 testdir = os.path.join(output_folder, "test", "stare-images") 

393 assert os.path.exists(testdir) 

394 assert len(fnmatch.filter(os.listdir(testdir), "*.csv")) == 10 

395 

396 assert os.path.exists( 

397 os.path.join(output_folder, "second-annotator", "test.csv") 

398 ) 

399 # checks individual performance figures are there 

400 testdir_sa = os.path.join( 

401 output_folder, "second-annotator", "test", "stare-images" 

402 ) 

403 assert os.path.exists(testdir_sa) 

404 assert len(fnmatch.filter(os.listdir(testdir_sa), "*.csv")) == 10 

405 

406 # check overlayed images are there (since we requested them) 

407 basedir = os.path.join(overlay_folder, "test", "stare-images") 

408 assert os.path.exists(basedir) 

409 assert len(fnmatch.filter(os.listdir(basedir), "*.png")) == 10 

410 

411 keywords = { 

412 r"^Maximum F1-score of.*\(chosen \*a posteriori\*\)$": 1, 

413 r"^F1-score of.*\(chosen \*a priori\*\)$": 1, 

414 r"^F1-score of.*\(second annotator; threshold=0.5\)$": 1, 

415 } 

416 

417 messages = "\n".join([k.getMessage() for k in caplog.records]) 

418 for k, v in keywords.items(): 

419 total = _str_counter(k, messages) 

420 assert total == v, ( 

421 f"message '{k}' appears {total} times, but I expected " 

422 f"it to appear {v} times" 

423 ) 

424 

425 

426def _check_compare(caplog, runner): 

427 

428 from ..script.compare import compare 

429 

430 with caplog.at_level(logging.INFO, logger="bob.ip.binseg"): 

431 

432 output_folder = "evaluations" 

433 result = runner.invoke( 

434 compare, 

435 [ 

436 "-vv", 

437 # label - path to measures 

438 "test", 

439 os.path.join(output_folder, "test.csv"), 

440 "test (2nd. human)", 

441 os.path.join(output_folder, "second-annotator", "test.csv"), 

442 "--output-figure=comparison.pdf", 

443 "--output-table=comparison.rst", 

444 ], 

445 ) 

446 _assert_exit_0(result) 

447 

448 assert os.path.exists("comparison.pdf") 

449 assert os.path.exists("comparison.rst") 

450 

451 keywords = { 

452 r"^Loading measures from": 2, 

453 r"^Creating and saving plot at": 1, 

454 r"^Tabulating performance summary...": 1, 

455 r"^Saving table at": 1, 

456 } 

457 messages = "\n".join([k.getMessage() for k in caplog.records]) 

458 for k, v in keywords.items(): 

459 total = _str_counter(k, messages) 

460 assert total == v, ( 

461 f"message '{k}' appears {total} times, but I expected " 

462 f"it to appear {v} times" 

463 ) 

464 

465 

466def _check_significance(caplog, runner): 

467 

468 from ..script.significance import significance 

469 

470 with tempfile.NamedTemporaryFile(mode="wt") as config, caplog.at_level( 

471 logging.INFO, logger="bob.ip.binseg" 

472 ): 

473 

474 config.write("from bob.ip.binseg.data.stare import _make_dataset\n") 

475 config.write(f"_raw = _make_dataset('{stare_datadir}')\n") 

476 config.write( 

477 "from bob.ip.binseg.configs.datasets.stare import _maker\n" 

478 ) 

479 config.write("dataset = _maker('ah', _raw)\n") 

480 config.flush() 

481 

482 ofolder = "significance" 

483 cfolder = os.path.join(ofolder, "caches") 

484 

485 result = runner.invoke( 

486 significance, 

487 [ 

488 "-vv", 

489 config.name, 

490 "--names=v1", 

491 "v2", 

492 "--predictions=predictions", 

493 "predictions", 

494 "--threshold=0.5", 

495 "--size=64", 

496 "64", 

497 "--stride=32", 

498 "32", 

499 "--figure=accuracy", 

500 f"--output-folder={ofolder}", 

501 f"--checkpoint-folder={cfolder}", 

502 ], 

503 ) 

504 _assert_exit_0(result) 

505 

506 assert os.path.exists(ofolder) 

507 assert os.path.exists(cfolder) 

508 assert os.path.exists(os.path.join(ofolder, "analysis.pdf")) 

509 assert os.path.exists(os.path.join(ofolder, "analysis.txt")) 

510 

511 keywords = { 

512 r"^Evaluating sliding window 'accuracy' on": 2, 

513 r"^Evaluating sliding window 'accuracy' differences on": 1, 

514 # r"^Basic statistics from distributions:$": 1, 

515 r"^Writing analysis figures": 1, 

516 r"^Writing analysis summary": 1, 

517 r"^Differences are exactly zero": 2, 

518 } 

519 messages = "\n".join([k.getMessage() for k in caplog.records]) 

520 for k, v in keywords.items(): 

521 total = _str_counter(k, messages) 

522 assert total == v, ( 

523 f"message '{k}' appears {total} times, but I expected " 

524 f"it to appear {v} times" 

525 ) 

526 

527 

528def test_discrete_experiment_stare(caplog): 

529 

530 # ensures we capture only ERROR messages and above by default 

531 caplog.set_level(logging.ERROR) 

532 

533 runner = CliRunner() 

534 with runner.isolated_filesystem(): 

535 _check_train(caplog, runner) 

536 _check_predict(caplog, runner) 

537 _check_evaluate(caplog, runner) 

538 _check_compare(caplog, runner) 

539 # _check_significance(caplog, runner) 

540 

541 

542def test_train_help(): 

543 from ..script.train import train 

544 

545 _check_help(train) 

546 

547 

548def test_predict_help(): 

549 from ..script.predict import predict 

550 

551 _check_help(predict) 

552 

553 

554def test_evaluate_help(): 

555 from ..script.evaluate import evaluate 

556 

557 _check_help(evaluate) 

558 

559 

560def test_compare_help(): 

561 from ..script.compare import compare 

562 

563 _check_help(compare) 

564 

565 

566def test_mkmask_help(): 

567 from ..script.mkmask import mkmask 

568 

569 _check_help(mkmask) 

570 

571 

572def test_significance_help(): 

573 from ..script.significance import significance 

574 

575 _check_help(significance) 

576 

577 

578def test_config_help(): 

579 from ..script.config import config 

580 

581 _check_help(config) 

582 

583 

584def test_config_list_help(): 

585 from ..script.config import list 

586 

587 _check_help(list) 

588 

589 

590def test_config_list(): 

591 from ..script.config import list 

592 

593 runner = CliRunner() 

594 result = runner.invoke(list) 

595 _assert_exit_0(result) 

596 assert "module: bob.ip.binseg.configs.datasets" in result.output 

597 assert "module: bob.ip.binseg.configs.models" in result.output 

598 

599 

600def test_config_list_v(): 

601 from ..script.config import list 

602 

603 runner = CliRunner() 

604 result = runner.invoke(list, ["--verbose"]) 

605 _assert_exit_0(result) 

606 assert "module: bob.ip.binseg.configs.datasets" in result.output 

607 assert "module: bob.ip.binseg.configs.models" in result.output 

608 

609 

610def test_config_describe_help(): 

611 from ..script.config import describe 

612 

613 _check_help(describe) 

614 

615 

616def test_config_describe_drive(): 

617 from ..script.config import describe 

618 

619 runner = CliRunner() 

620 result = runner.invoke(describe, ["drive"]) 

621 _assert_exit_0(result) 

622 assert "[DRIVE-2004]" in result.output 

623 

624 

625def test_config_copy_help(): 

626 from ..script.config import copy 

627 

628 _check_help(copy) 

629 

630 

631def test_config_copy(): 

632 from ..script.config import copy 

633 

634 runner = CliRunner() 

635 with runner.isolated_filesystem(): 

636 result = runner.invoke(copy, ["drive", "test.py"]) 

637 _assert_exit_0(result) 

638 with open("test.py") as f: 

639 data = f.read() 

640 assert "[DRIVE-2004]" in data 

641 

642 

643def test_dataset_help(): 

644 from ..script.dataset import dataset 

645 

646 _check_help(dataset) 

647 

648 

649def test_dataset_list_help(): 

650 from ..script.dataset import list 

651 

652 _check_help(list) 

653 

654 

655def test_dataset_list(): 

656 from ..script.dataset import list 

657 

658 runner = CliRunner() 

659 result = runner.invoke(list) 

660 _assert_exit_0(result) 

661 assert result.output.startswith("Supported datasets:") 

662 

663 

664def test_dataset_check_help(): 

665 from ..script.dataset import check 

666 

667 _check_help(check) 

668 

669 

670def test_dataset_check(): 

671 from ..script.dataset import check 

672 

673 runner = CliRunner() 

674 result = runner.invoke(check, ["--verbose", "--verbose", "--limit=2"]) 

675 _assert_exit_0(result)