Optional decorator that wraps the task function in extra utilities. Makes multirun more resistant to failure. Utilities: - Calling the `utils.extras()` before the task is started - Calling the `utils.close_loggers()` after the task is finished - Logging the exception if occurs
(task_func: Callable)
| 22 | |
| 23 | |
| 24 | def task_wrapper(task_func: Callable) -> Callable: |
| 25 | """Optional decorator that wraps the task function in extra utilities. |
| 26 | |
| 27 | Makes multirun more resistant to failure. |
| 28 | |
| 29 | Utilities: |
| 30 | - Calling the `utils.extras()` before the task is started |
| 31 | - Calling the `utils.close_loggers()` after the task is finished |
| 32 | - Logging the exception if occurs |
| 33 | - Logging the task total execution time |
| 34 | - Logging the output dir |
| 35 | """ |
| 36 | |
| 37 | def wrap(cfg: DictConfig): |
| 38 | |
| 39 | # apply extra utilities |
| 40 | extras(cfg) |
| 41 | |
| 42 | # execute the task |
| 43 | try: |
| 44 | start_time = time.time() |
| 45 | metric_dict, object_dict = task_func(cfg=cfg) |
| 46 | except Exception as ex: |
| 47 | log.exception("") # save exception to `.log` file |
| 48 | raise ex |
| 49 | finally: |
| 50 | path = Path(cfg.paths.output_dir, "exec_time.log") |
| 51 | content = f"'{cfg.task_name}' execution time: {time.time() - start_time} (s)" |
| 52 | save_file(path, content) # save task execution time (even if exception occurs) |
| 53 | close_loggers() # close loggers (even if exception occurs so multirun won't fail) |
| 54 | |
| 55 | log.info(f"Output dir: {cfg.paths.output_dir}") |
| 56 | |
| 57 | return metric_dict, object_dict |
| 58 | |
| 59 | return wrap |
| 60 | |
| 61 | |
| 62 | def extras(cfg: DictConfig) -> None: |
nothing calls this directly
no outgoing calls
no test coverage detected