base

Evaluation(config, model_path, report_folder=None)

Bases: Task

Base Evaluation Task with deployment models.

Parameters:

• config (DictConfig) –

  The experiment configuration

• model_path (str) –

  The model path.

• report_folder (Optional[str]) –

  The report folder. Defaults to None.

Raises:

• ValueError –

  If the experiment path is not provided

Source code in quadra/tasks/base.py

def __init__(
    self,
    config: DictConfig,
    model_path: str,
    report_folder: Optional[str] = None,
):
    super().__init__(config=config)
    self.config = config
    self.metadata = {"report_files": []}
    self.model_path = model_path
    self.device = utils.get_device()
    self.report_folder = report_folder
    self._deployment_model: RecursiveScriptModule
    self.deployment_model_type: str
    if self.report_folder is None:
        log.warning("Report folder is not provided, using default report folder")
        self.report_folder = "report"

deployment_model: RecursiveScriptModule property writable

prepare()

Prepare the evaluation.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the evaluation."""
    self.deployment_model, self.deployment_model_type = import_deployment_model(self.model_path, self.device)

LightningTask(config, checkpoint_path=None, run_test=False, report=False, export_type=None)

Bases: Generic[DataModuleT], Task[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• checkpoint_path (Optional[str]) –

  The path to the checkpoint to load the model from. Defaults to None.

• run_test (bool) –

  Whether to run the test after training. Defaults to False.

• report (bool) –

  Whether to generate a report. Defaults to False.

• export_type (Optional[List[str]]) –

  List of export method for the model, e.g. [torchscript]. Defaults to None.

Source code in quadra/tasks/base.py

def __init__(
    self,
    config: DictConfig,
    checkpoint_path: Optional[str] = None,
    run_test: bool = False,
    report: bool = False,
    export_type: Optional[List[str]] = None,
):
    super().__init__(config, export_type=export_type)
    self.config = config
    self.checkpoint_path = checkpoint_path
    self.run_test = run_test
    self.report = report
    self._module: LightningModule
    self._devices: Union[int, List[int]]
    self._callbacks: List[Callback]
    self._logger: List[Logger]
    self._trainer: Trainer

callbacks: List[Callback] property writable

List[Callback]: The callbacks.

devices: Union[int, List[int]] property writable

List[int]: The devices ids.

logger: List[Logger] property writable

List[Logger]: The loggers.

module: LightningModule property writable

trainer: Trainer property writable

add_callback(callback)

Add a callback to the trainer.

Parameters:

• callback (Callback) –

  The callback to add

Source code in quadra/tasks/base.py

def add_callback(self, callback: Callback):
    """Add a callback to the trainer.

    Args:
        callback: The callback to add
    """
    if hasattr(self.trainer, "callbacks") and isinstance(self.trainer.callbacks, list):
        self.trainer.callbacks.append(callback)

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    if self.run_test:
        self.test()
    if self.export_type is not None and len(self.export_type) > 0:
        self.export()
    if self.report:
        self.generate_report()
    self.finalize()

finalize()

Finalize the experiment.

Source code in quadra/tasks/base.py

def finalize(self) -> None:
    """Finalize the experiment."""
    super().finalize()
    utils.finish(
        config=self.config,
        model=self.module,
        datamodule=self.datamodule,
        trainer=self.trainer,
        callbacks=self.callbacks,
        logger=self.logger,
    )

    if not self.config.trainer.get("fast_dev_run"):
        if self.trainer.checkpoint_callback is not None and hasattr(
            self.trainer.checkpoint_callback, "best_model_path"
        ):
            log.info("Best model ckpt: %s", self.trainer.checkpoint_callback.best_model_path)

prepare()

Prepare the experiment.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the experiment."""
    super().prepare()

    if "callbacks" in self.config:
        self.callbacks = self.config.callbacks

    if "logger" in self.config:
        self.logger = self.config.logger
    self.devices = self.config.trainer.devices
    self.trainer = self.config.trainer

test()

Test the model.

Source code in quadra/tasks/base.py

def test(self) -> Any:
    """Test the model."""
    log.info("Starting testing!")
    return self.trainer.test(model=self.module, datamodule=self.datamodule)

train()

Train the model.

Source code in quadra/tasks/base.py

def train(self) -> None:
    """Train the model."""
    log.info("Starting training!")
    utils.log_hyperparameters(
        config=self.config,
        model=self.module,
        trainer=self.trainer,
    )
    self.trainer.fit(model=self.module, datamodule=self.datamodule)

PlaceholderTask

Bases: Task

Placeholder task.

execute()

Execute the task and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the task and all the steps."""
    log.info("Running Placeholder Task.")
    log.info("Quadra Version: %s", str(get_version()))
    log.info("If you are reading this, it means that library is installed correctly!")

Task(config, export_type=None)

Bases: Generic[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration.

• export_type (Optional[List[str]]) –

  List of export method for the model, e.g. [torchscript]. Defaults to None.

Source code in quadra/tasks/base.py

def __init__(self, config: DictConfig, export_type: Optional[List[str]] = None):
    self.config = config
    self.export_type = export_type
    self._datamodule: DataModuleT
    self.metadata: Dict[str, Any]
    self.save_config()

datamodule: DataModuleT property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    self.test()
    if self.export_type is not None and len(self.export_type) > 0:
        self.export()
    self.generate_report()
    self.finalize()

export()

Export model for production.

Source code in quadra/tasks/base.py

def export(self) -> None:
    """Export model for production."""
    log.info("Export model for production not implemented for this task!")

finalize()

Finalize the experiment.

Source code in quadra/tasks/base.py

def finalize(self) -> None:
    """Finalize the experiment."""
    log.info("Results are saved in %s", os.getcwd())

generate_report()

Generate a report.

Source code in quadra/tasks/base.py

def generate_report(self) -> None:
    """Generate a report."""
    log.info("Report generation not implemented for this task!")

prepare()

Prepare the experiment.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the experiment."""
    self.datamodule = self.config.datamodule

save_config()

Save the experiment configuration when running an Hydra experiment.

Source code in quadra/tasks/base.py

def save_config(self) -> None:
    """Save the experiment configuration when running an Hydra experiment."""
    if HydraConfig.initialized():
        with open("config_resolved.yaml", "w") as fp:
            OmegaConf.save(config=OmegaConf.to_container(self.config, resolve=True), f=fp.name)

test()

Test the model.

Source code in quadra/tasks/base.py

def test(self) -> Any:
    """Test the model."""
    log.info("Testing not implemented for this task!")

train()

Train the model.

Source code in quadra/tasks/base.py

def train(self) -> Any:
    """Train the model."""
    log.info("Training not implemented for this task!")