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 identity of the AGGREGATE ROOT.

In Domain-Driven Design, the AGGREGATE ROOT is the single entry point to a cluster of domain objects, and its identity uniquely distinguishes the AGGREGATE within its boundary. This property returns that identity.

Key points

Immutable once assigned to preserve AGGREGATE consistency and invariants.
Accessing the ID before initialization raises an error to prevent unsafe operations on an uninitialized AGGREGATE ROOT.
Ensures that all domain operations relying on identity are deterministic and safe.

Raises:

Type	Description
`ValueError`	If the AGGREGATE ROOT's ID has not been initialized.

Returns:

Name	Type	Description
`Any`	`Any`	The unique identifier of the AGGREGATE ROOT within its boundary.

`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.

`initialize(id)`

Initializes the identity of the AGGREGATE ROOT.

This method allows deferred assignment of identity when the creation of the AGGREGATE depends on external factors (e.g., database-generated IDs, hashes of components). Once set, the ID is immutable to uphold the AGGREGATE's invariants and consistency rules.

Parameters:

Name	Type	Description	Default
`id`	`Any`	A unique identifier for the AGGREGATE ROOT.	required

Raises:

Type	Description
`ValueError`	If the AGGREGATE ROOT already has an initialized identity.

`is_initialized()`

Checks whether the AGGREGATE ROOT has an assigned identity.

Some domain operations require that the AGGREGATE ROOT has a valid identity before executing (e.g., persisting the aggregate, emitting domain events). This method allows the domain logic to verify readiness safely.

Returns:

Name	Type	Description
`bool`	`bool`	True if the AGGREGATE ROOT has an initialized identity, False otherwise.

Warning

If the id property is overridden in a subclass (e.g., computed dynamically instead of stored in self.__id), this method may return False even if the aggregate has a valid identity.

`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: type[Exception]) -> None

handle(event: 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.

Domain

Implement your domain model with pytorch

Aggregate

id property

phase property writable

initialize(id)

is_initialized()

onepoch()

onphase()

Event

Events

dequeue()

enqueue(event)

handle(event)

`Aggregate`

`id` `property`

`phase` `property` `writable`

`initialize(id)`

`is_initialized()`

`onepoch()`

`onphase()`

`Event`

`Events`

`dequeue()`

`enqueue(event)`

`handle(event)`