Module fl_server_ai.notification.training.finished¶
View Source
# SPDX-FileCopyrightText: 2024 Benedikt Franke <benedikt.franke@dlr.de>
# SPDX-FileCopyrightText: 2024 Florian Heinrich <florian.heinrich@dlr.de>
#
# SPDX-License-Identifier: Apache-2.0
from dataclasses import dataclass
from uuid import UUID
from fl_server_core.models import Training as TrainingDB
from ..serializable import Serializable
from ..notification_type import NotificationType
from .training import TrainingNotification
class TrainingFinishedNotification(TrainingNotification["TrainingFinishedNotification.Body"]):
"""
Notification that a training has finished.
"""
type: NotificationType = NotificationType.TRAINING_FINISHED
"""The type of the notification."""
@dataclass
class Body(Serializable):
"""
Inner class representing the body of the notification.
"""
global_model_uuid: UUID
"""The UUID of the global model."""
@classmethod
def from_training(cls, training: TrainingDB):
"""
Create a `TrainingFinishedNotification` instance from a training object.
Args:
training (TrainingDB): The training object to create the notification from.
Returns:
TrainingFinishedNotification: The created notification.
"""
receivers = list(training.participants.all())
if not receivers.__contains__(training.actor):
receivers.append(training.actor)
return cls(
receivers=receivers,
body=cls.Body(
global_model_uuid=training.model.id
),
training_uuid=training.id
)
Classes¶
TrainingFinishedNotification¶
class TrainingFinishedNotification(
receivers: List[fl_server_core.models.user.NotificationReceiver],
body: ~TBody,
training_uuid: uuid.UUID
)
Notification that a training has finished.
View Source
class TrainingFinishedNotification(TrainingNotification["TrainingFinishedNotification.Body"]):
"""
Notification that a training has finished.
"""
type: NotificationType = NotificationType.TRAINING_FINISHED
"""The type of the notification."""
@dataclass
class Body(Serializable):
"""
Inner class representing the body of the notification.
"""
global_model_uuid: UUID
"""The UUID of the global model."""
@classmethod
def from_training(cls, training: TrainingDB):
"""
Create a `TrainingFinishedNotification` instance from a training object.
Args:
training (TrainingDB): The training object to create the notification from.
Returns:
TrainingFinishedNotification: The created notification.
"""
receivers = list(training.participants.all())
if not receivers.__contains__(training.actor):
receivers.append(training.actor)
return cls(
receivers=receivers,
body=cls.Body(
global_model_uuid=training.model.id
),
training_uuid=training.id
)
Ancestors (in MRO)¶
- fl_server_ai.notification.training.training.TrainingNotification
- fl_server_ai.notification.notification.Notification
- typing.Generic
- fl_server_ai.notification.serializable.Serializable
Class variables¶
Static methods¶
from_training¶
Create a TrainingFinishedNotification instance from a training object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| training | TrainingDB | The training object to create the notification from. | None |
Returns:
| Type | Description |
|---|---|
| TrainingFinishedNotification | The created notification. |
View Source
@classmethod
def from_training(cls, training: TrainingDB):
"""
Create a `TrainingFinishedNotification` instance from a training object.
Args:
training (TrainingDB): The training object to create the notification from.
Returns:
TrainingFinishedNotification: The created notification.
"""
receivers = list(training.participants.all())
if not receivers.__contains__(training.actor):
receivers.append(training.actor)
return cls(
receivers=receivers,
body=cls.Body(
global_model_uuid=training.model.id
),
training_uuid=training.id
)
Instance variables¶
The callback to be called on error. By default, this is None.
The callback to be called on success. By default, this is None.
Methods¶
send¶
def send(
self,
callback_success: Optional[celery.canvas.Signature] = None,
callback_error: Optional[celery.canvas.Signature] = None
) -> celery.result.AsyncResult
Send notification to the receivers asynchronously.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| callback_success | Optional[Signature] | The callback to be called on success. Defaults to None. | None |
| callback_error | Optional[Signature] | The callback to be called on error. Defaults to None. | None |
Returns:
| Type | Description |
|---|---|
| AsyncResult | The result of the asynchronous operation. |
View Source
def send(
self,
callback_success: Optional[Signature] = None,
callback_error: Optional[Signature] = None
) -> AsyncResult:
"""
Send notification to the receivers asynchronously.
Args:
callback_success (Optional[Signature], optional): The callback to be called on success. Defaults to None.
callback_error (Optional[Signature], optional): The callback to be called on error. Defaults to None.
Returns:
AsyncResult: The result of the asynchronous operation.
"""
callback_success = callback_success or self.callback_success
callback_error = callback_error or self.callback_error
return send_notifications.s(
notification=self, callback_success=callback_success, callback_error=callback_error
).apply_async(retry=False)
serialize¶
Serialize the notification into a dictionary.
Returns:
| Type | Description |
|---|---|
| dict[str, Any] | The serialized notification. |