|
from typing import List |
|
|
|
import hydra |
|
from omegaconf import DictConfig |
|
from pytorch_lightning import Callback |
|
from pytorch_lightning.loggers import Logger |
|
|
|
from .logger import RankedLogger |
|
|
|
log = RankedLogger(__name__, rank_zero_only=True) |
|
|
|
|
|
def instantiate_callbacks(callbacks_cfg: DictConfig) -> List[Callback]: |
|
"""Instantiates callbacks from config.""" |
|
|
|
callbacks: List[Callback] = [] |
|
|
|
if not callbacks_cfg: |
|
log.warning("No callback configs found! Skipping..") |
|
return callbacks |
|
|
|
if not isinstance(callbacks_cfg, DictConfig): |
|
raise TypeError("Callbacks config must be a DictConfig!") |
|
|
|
for _, cb_conf in callbacks_cfg.items(): |
|
if isinstance(cb_conf, DictConfig) and "_target_" in cb_conf: |
|
log.info(f"Instantiating callback <{cb_conf._target_}>") |
|
callbacks.append(hydra.utils.instantiate(cb_conf)) |
|
|
|
return callbacks |
|
|
|
|
|
def instantiate_loggers(logger_cfg: DictConfig) -> List[Logger]: |
|
"""Instantiates loggers from config.""" |
|
|
|
logger: List[Logger] = [] |
|
|
|
if not logger_cfg: |
|
log.warning("No logger configs found! Skipping...") |
|
return logger |
|
|
|
if not isinstance(logger_cfg, DictConfig): |
|
raise TypeError("Logger config must be a DictConfig!") |
|
|
|
for _, lg_conf in logger_cfg.items(): |
|
if isinstance(lg_conf, DictConfig) and "_target_" in lg_conf: |
|
log.info(f"Instantiating logger <{lg_conf._target_}>") |
|
logger.append(hydra.utils.instantiate(lg_conf)) |
|
|
|
return logger |
|
|