Domain

Implement your domain model with pytorch

Aggregate

Bases: Module, ABC

An AGGREGATE is a cluster of associated objects that we treat as a unit for the purpose of data changes. Each AGGREGATE has a root and a boundary. The boundary defines what is inside the AGGREGATE. The root is a single, specific ENTITY contained in the AGGREGATE and provides the IDENTITY of the AGGREGATE. The root is the only member of the AGGREGATE that outside objects are allowed to hold references to.

In deep learning, an AGGREGATE consist not only of a neural network, but also several other components such as optimizers, schedulers, tokenizers, etc. For example, a transformer model is just a neural network, and in order to perform tasks such as text completion or translation, it needs to be part of an AGGREGATE that includes other components like a tokenizer. The AGGREGATE is responsible for coordinating the interactions between these components.

Attributes:

Name	Type	Description
id	Any	The id of the AGGREGATE ROOT. It should be unique within the AGGREGATE boundary.
phase	str	The phase of the AGGREGATE.
events	Events	The domain events of the AGGREGATE.

Methods:

Name	Description
onphase	A hook that is called when the phase changes. Implement this method to add custom behavior.
onepoch	A hook that is called when the epoch changes. Implement this method to add custom behavior.

Example

from torch import Tensor
from torch.nn import Module
from torch.optim import Optimizer
from torchsystem import Aggregate
from torchsystem.registry import gethash

class Classifier(Aggregate):
    def __init__(self, model: Module, criterion: Module, optimizer: Optimizer):
        super().__init__()
        self.epoch = 0
        self.model = model
        self.criterion = criterion
        self.optimizer = optimizer

    @property
    def id(self) -> str:
        return gethash(self.model) # See the registry module for more information.

    def onepoch(self):
        print(f'Epoch: {self.epoch}')

    def onphase(self):
        print(f'Phase: {self.phase}')

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

    def loss(self, output: Tensor, target: Tensor) -> Tensor:
        return self.criterion(output, target)

    def fit(self, input: Tensor, target: Tensor) -> tuple[Tensor, Tensor]:
        self.optimizer.zero_grad()
        output = self(input)
        loss = self.loss(output, target)
        loss.backward()
        self.optimizer.step()
        return output, loss

    def evaluate(self, input: Tensor, target: Tensor) -> tuple[Tensor, Tensor]: 
        output = self(input)
        loss = self.loss(output, target)
        return output, loss

id property

The id of the AGGREGATE ROOT. It should be unique within the AGGREGATE boundary. It's up to the user to define the id of the AGGREGATE ROOT and how it should be generated.

The gethash function from the torchsystem.registry module can usefull for generating unique ids from registered pytorch objects.

phase property writable

The phase of the AGGREGATE. The phase is a property of neural networks that not only describes the current state of the network, but also determines how the network should behave.

During the training phase, the network stores the gradients of the weights and biases, and uses them to update the weights and biases. During the evaluation phase, the network does not store the gradients of the weights and biases, and does not update the weights and biases.

Returns:

Type	Description
Literal['train', 'evaluation']	Literal['train', 'evaluation']: The current phase of the AGGREGATE.

onepoch()

A hook that is called when the epoch changes. Implement this method to add custom behavior.

onphase()

A hook that is called when the phase changes. Implement this method to add custom behavior.

Event

A DOMAIN EVENT is a representation of something that has happened in the DOMAIN.

This class is a base class for creating custom DOMAIN EVENTS. It is a simple class that can be optionally subclassed to write self-documented code when creating custom events.

Events

A collection of DOMAIN EVENTS that have occurred within a Bounded context. The EVENTS class is responsible for managing the events that have occurred within the Bounded context and dispatching them to the appropriate handlers.

When an event is enqueued, it is added to the queue of events to be processed. When the commit method is called, the events are dequeued and dispatched to the appropriate handlers. If no handler is found for an event, the event is ignored, except if the event is an exception.

Exceptions are treated as domain events but they are raised when the commit method is called by default if no handler is found for it's type.

Attributes:

Name	Type	Description
queue	deque[Event]	A queue of DOMAIN EVENTS that have occurred within the Bounded context.
handlers	dict[type[Event], Sequence[Callable]]	A dictionary of handlers that are responsible for handling DOMAIN EVENTS. The key is the type of the event and the value is the handler function.

Example

from torchsystem.domain import Events, Event

class ClsEvent(Event):...

class ObjEvent(Event):
    def __init__(self, value):
        self.value = value

class OtherObjEvent(Event):
    def __init__(self, willbeignored):
        self.value = willbeignored

events = Events()
events.enqueue(ClsEvent)
events.enqueue(KeyError) # Enqueues a KeyError exception event
events.enqueue(ObjEvent('somevalue'))
events.enqueue(OtherObjEvent('willbeignored'))
events.enqueue(StopIteration) # Enqueues a StopIteration exception event

events.handlers[ClsEvent] = lambda: print('ClsEvent was handled.')
events.handlers[KeyError] = lambda: print('KeyError was handled.')
events.handlers[ObjEvent] = lambda event: print(f'ObjEvent was handled with value: {event.value}')
events.handlers[OtherObjEvent] = lambda: print('OtherObjEvent was handled.')

try:
    events.commit()
except StopIteration:
    print('StopIteration exception was raised.')

# Output:
#ClsEvent was handled.
#KeyError was handled.
#ObjEvent was handled with value: somevalue
#OtherObjEvent was handled.
#StopIteration exception was raised. Usefull for early stopping in training loops.

dequeue()

Dequeue a DOMAIN EVENT from the EVENTS queue to be processed by the commit method.

Returns:

Type	Description
Optional[EVENT]	Optional[Event]: The DOMAIN EVENT or exception to be processed.

enqueue(event)

enqueue(event: Event) -> None

enqueue(event: type[Event]) -> None

enqueue(event: Exception) -> None

enqueue(event: type[Exception]) -> None

Enqueue a DOMAIN EVENT into the EVENTS queue to be processed when the commit method is called. Exceptions can also be enqueued as domain events.

Parameters:

Name	Type	Description	Default
event	Event	The DOMAIN EVENT or exception to be enqueued.	required

handle(event)

handle(event: Event) -> None

handle(event: type[Event]) -> None

handle(event: Exception) -> None

handle(event: type[Exception]) -> None

Handles a DOMAIN EVENT by dispatching it to the appropriate handler or group of handlers. If no handler is found for the event, the event is ignored, except if the event is an exception. If the event is an exception, it is raised by default if no handler is found for it's type.

Both classes and instances of DOMAIN EVENTS are supported. The method also will look at the signature of the handler to determine if the event should be passed as an argument to the handler or if the handler should be called without arguments.

Parameters:

Name	Type	Description	Default
event	Event	The DOMAIN EVENT or exception to be handled.	required

Raises:

Type	Description
event	If no handler is found for the event and the event is an exception.