system/simplecls.py

from dataclasses import dataclass, field
from utils import parse_structure
from typing import Any, Dict, Mapping
from .base import BaseSystemConfig, BaseSystem
from torch import nn, Tensor

import os
import torch
import numpy as np
import models


@dataclass
class SimpleClassificationConfig(BaseSystemConfig):
    pass


class SimpleClassificationSystem(BaseSystem):
    def __init__(self, cfg: Dict, *args: Any, **kwargs: Any) -> BaseSystem:
        super().__init__(cfg, *args, **kwargs)
        self.cfg:SimpleClassificationConfig = parse_structure(SimpleClassificationConfig, cfg)
        self.model: nn.Module = getattr(models, self.cfg.model_type)(self.cfg.model)

    def forward(self, x: Tensor) -> Tensor:
        return self.model(x)

    def training_step(self, batch: Mapping[str, Tensor], batch_idx: int) -> Tensor: 
        x: Tensor = batch[0]
        y: Tensor = batch[1].float()
        
        y_hat: Tensor = self.model(x).squeeze(-1)
        loss = self.criterion(y_hat, y)

        self.log(
            "train/loss", 
            loss, 
            on_step=self.cfg.log_on_step, 
            on_epoch=self.cfg.log_on_epoch, 
            prog_bar=self.cfg.log_prog_bar, 
            logger=self.cfg.log_logger
        )
        self.log_metrics(self.metrics_func(y_hat, y, 'train'))

        return loss

    def validation_step(self, batch: Mapping[str, Tensor], batch_idx: int) -> Tensor:
        x: Tensor = batch[0]
        y: Tensor = batch[1].float()
        
        y_hat: Tensor = self.model(x).squeeze(-1)
        loss = self.criterion(y_hat, y)

        self.log(
            "valid/loss", 
            loss, 
            on_step=self.cfg.log_on_step, 
            on_epoch=self.cfg.log_on_epoch, 
            prog_bar=self.cfg.log_prog_bar, 
            logger=self.cfg.log_logger
        )
        self.log_metrics(self.metrics_func(y_hat, y, 'valid'))

        return loss

    def test_step(self, batch: Mapping[str, Tensor], batch_idx: int) -> Tensor:
        x: Tensor = batch[0]
        y: Tensor = batch[1].float()
        
        y_hat: Tensor = self.model(x).squeeze(-1)
        loss = self.criterion(y_hat, y)

        self.log(
            "test/loss", 
            loss, 
            on_step=self.cfg.log_on_step, 
            on_epoch=self.cfg.log_on_epoch, 
            prog_bar=self.cfg.log_prog_bar, 
            logger=self.cfg.log_logger
        )
        metrics_dict = self.metrics_func(y_hat, y, 'test')
        self.log_metrics(metrics_dict)