Ideas for Writer architecture

[NicolasGensollen]

Hi,

I've been doing some thinking after our discussion yesterday.
If you want to group the writing logic in the same module without having a huge class doing everything, you can have an architecture that would look more or less like that (classes and methods are only unrealistic toy examples to show the idea):

import json
from pathlib import Path
from abc import ABC, abstractmethod


# --- in the writers module for example ---
class Writable(ABC):
    """All object that can be written in MAPS implement this interface."""
    @abstractmethod
    def write(self, folder: Path) -> None:
        ...

class Writer(ABC):
    """Base class for writers."""
    @abstractmethod
    def to_dict(self) -> dict:
        ...

    @abstractmethod
    def get_filename(self) -> str:
        ...

    def write(self, folder: Path) -> None:
        folder.mkdir(parents=True, exist_ok=True)
        with open(folder / self.get_filename(), "w") as fp:
            json.dump(self.to_dict(), fp)

class CAPSDatasetWriter(Writer):
    """Writer for CAPSDataset."""
    def __init__(self, dataset_size: int):
        self.dataset_size = dataset_size

    @classmethod
    def from_dataset(cls, dataset: CAPSDataset):
        """Alternative constructor."""
        return cls(dataset.size)

    def to_dict(self) -> dict:
        return {"DatasetSize": self.dataset_size}

    def get_filename(self) -> str:
        return "dataset.json"

class AdamOptimizerWriter(Writer):
    """Writer for Adam optimizer."""
    def __init__(self, max_iteration: int):
        self.max_iteration = max_iteration

    @classmethod
    def from_optimizer(cls, optimizer: AdamOptimizer):
        """Alternative constructor."""
        return cls(optimizer.max_iteration)

    def to_dict(self) -> dict:
        return {"MaxIteration": self.max_iteration}

    def get_filename(self) -> str:
        return "optimizer.json"
    
# other writers...
# --- in the dataset module for example ----
class Dataset(ABC):
    "All dataset implement this interface."""
    @abstractmethod
    def some_common_dataset_method(self) -> None:
        ...

class CAPSDataset(Dataset, Writable):
    def __init__(self, size: int):
        self.size = size

    def some_common_dataset_method(self) -> None:
        print("I'm a CAPS dataset.")

    def write(self, folder: Path) -> None:
        writer = CAPSDatasetWriter.from_dataset(self)
        writer.write(folder)

# other datasets...

# --- in the optimizer module for example ----
class Optimizer(ABC):
    """All optimizer implement this interface."""
    @abstractmethod
    def optimize(self, iteration: int) -> None:
        ...

class AdamOptimizer(Optimizer, Writable):
    def __init__(self, max_iteration: int):
        self.max_iteration = max_iteration

    def optimize(self, iteration: int) -> None:
        for i in range(min(iteration, self.max_iteration)):
            print("Optimizing...")

    def write(self, folder: Path) -> None:
        writer = AdamOptimizerWriter.from_optimizer(self)
        writer.write(folder)

# other optimizers...

This would be used in the following way:

# User decides where to save stuffs
maps = Path.cwd() / "maps"

# User initializes the objects needed
caps = CAPSDataset(6)
optimizer = AdamOptimizer(10)

# User uses the objects
caps.some_common_dataset_method()
optimizer.optimize(20)

# User saves to maps folder
caps.write(maps)
optimizer.write(maps)

[NicolasGensollen]

A slightly modified implementation with more dynamicity using the factory pattern:

import json
from pathlib import Path
from abc import ABC, abstractmethod
from typing import Type, Any
from enum import Enum


class WriterName(str, Enum):
    CAPS_DATASET = "caps_dataset"
    ADAM_OPTIMIER = "adam_optimizer"


class Writable(ABC):
    """All object that can be written in MAPS implement this interface."""
    @abstractmethod
    def get_writer_name(self) -> WriterName:
        ...
    
    def write(self, folder: Path) -> None:
        writer = writer_factory(self.get_writer_name()).from_instance(self)
        writer.write(folder)

class Writer(ABC):
    """Base class for writers."""
    @abstractmethod
    def to_dict(self) -> dict:
        ...

    @abstractmethod
    def get_filename(self) -> str:
        ...

    @classmethod
    @abstractmethod
    def from_instance(cls, instance: Any):
        ...

    def write(self, folder: Path) -> None:
        folder.mkdir(parents=True, exist_ok=True)
        with open(folder / self.get_filename(), "w") as fp:
            json.dump(self.to_dict(), fp)


def writer_factory(name: str | WriterName) -> Type[Writer]:
    name = WriterName(name)
    if name == WriterName.CAPS_DATASET:
        return CAPSDatasetWriter
    if name == WriterName.ADAM_OPTIMIER:
        return AdamOptimizerWriter


class Dataset(ABC):
    "All dataset implement this interface."""
    @abstractmethod
    def some_common_dataset_method(self) -> None:
        ...


class CAPSDataset(Dataset, Writable):
    def __init__(self, size: int):
        self.size = size

    def some_common_dataset_method(self) -> None:
        print("I'm a CAPS dataset.")

    def get_writer_name(self) -> WriterName:
        return WriterName.CAPS_DATASET


class CAPSDatasetWriter(Writer):
    """Writer for CAPSDataset."""
    def __init__(self, dataset_size: int):
        self.dataset_size = dataset_size

    @classmethod
    def from_instance(cls, dataset: CAPSDataset):
        """Alternative constructor."""
        return cls(dataset.size)

    def to_dict(self) -> dict:
        return {"DatasetSize": self.dataset_size}

    def get_filename(self) -> str:
        return "dataset.json"

class Optimizer(ABC):
    """All optimizer implement this interface."""
    @abstractmethod
    def optimize(self, iteration: int) -> None:
        ...


class AdamOptimizer(Optimizer, Writable):
    def __init__(self, max_iteration: int):
        self.max_iteration = max_iteration

    def optimize(self, iteration: int) -> None:
        for i in range(min(iteration, self.max_iteration)):
            print("Optimizing...")

    def get_writer_name(self) -> WriterName:
        return WriterName.ADAM_OPTIMIER


class AdamOptimizerWriter(Writer):
    """Writer for Adam optimizer."""
    def __init__(self, max_iteration: int):
        self.max_iteration = max_iteration

    @classmethod
    def from_instance(cls, optimizer: AdamOptimizer):
        """Alternative constructor."""
        return cls(optimizer.max_iteration)

    def to_dict(self) -> dict:
        return {"MaxIteration": self.max_iteration}

    def get_filename(self) -> str:
        return "optimizer.json"

[NicolasGensollen]

Finally, a more straightforward design without the Writer classes which are strongly coupled to their counterparts and which might not be useful:

import json
from pathlib import Path
from abc import ABC, abstractmethod


class Writable(ABC):
    """All object that can be written in MAPS implement this interface."""
    @abstractmethod
    def to_dict(self) -> dict:
        ...

    @abstractmethod
    def get_filename(self) -> str:
        ...

    def write(self, folder: Path) -> None:
        folder.mkdir(parents=True, exist_ok=True)
        with open(folder / self.get_filename(), "w") as fp:
            json.dump(self.to_dict(), fp)


class Dataset(ABC):
    "All dataset implement this interface."""
    @abstractmethod
    def some_common_dataset_method(self) -> None:
        ...


class CAPSDataset(Dataset, Writable):
    def __init__(self, size: int):
        self.size = size

    def some_common_dataset_method(self) -> None:
        print("I'm a CAPS dataset.")

    def to_dict(self) -> dict:
        return {"DatasetSize": self.size}

    def get_filename(self) -> str:
        return "dataset.json"


class Optimizer(ABC):
    """All optimizer implement this interface."""
    @abstractmethod
    def optimize(self, iteration: int) -> None:
        ...


class AdamOptimizer(Optimizer, Writable):
    def __init__(self, max_iteration: int):
        self.max_iteration = max_iteration

    def optimize(self, iteration: int) -> None:
        for i in range(min(iteration, self.max_iteration)):
            print("Optimizing...")

    def to_dict(self) -> dict:
        return {"MaxIteration": self.max_iteration}

    def get_filename(self) -> str:
        return "optimizer.json"