Module fl_server_ai.notification.training.swag¶

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 ..notification_type import NotificationType
from .round_start import TrainingRoundStartNotification


class TrainingSWAGRoundStartNotification(TrainingRoundStartNotification):
    """
    Notification for the start of a SWAG training round.
    """

    type: NotificationType = NotificationType.SWAG_ROUND_START
    """The type of the notification."""

Classes¶

TrainingSWAGRoundStartNotification¶

class TrainingSWAGRoundStartNotification(
    receivers: List[fl_server_core.models.user.NotificationReceiver],
    body: ~TBody,
    training_uuid: uuid.UUID
)

Notification for the start of a SWAG training round.

View Source

class TrainingSWAGRoundStartNotification(TrainingRoundStartNotification):
    """
    Notification for the start of a SWAG training round.
    """

    type: NotificationType = NotificationType.SWAG_ROUND_START
    """The type of the notification."""

Ancestors (in MRO)¶

fl_server_ai.notification.training.round_start.TrainingRoundStartNotification
fl_server_ai.notification.training.training.TrainingNotification
fl_server_ai.notification.notification.Notification
typing.Generic
fl_server_ai.notification.serializable.Serializable

Class variables¶

Body

type

Static methods¶

from_training¶

def from_training(
    training: fl_server_core.models.training.Training
)

Create a TrainingRoundStartNotification instance from a training object.

Parameters:

Name	Type	Description	Default
training	TrainingDB	The training object to create the notification from.	None

Returns:

Type	Description
TrainingRoundStartNotification	The created notification.

View Source

    @classmethod
    def from_training(cls, training: TrainingDB):
        """
        Create a `TrainingRoundStartNotification` instance from a training object.

        Args:
            training (TrainingDB): The training object to create the notification from.

        Returns:
            TrainingRoundStartNotification: The created notification.
        """
        return cls(
            receivers=training.participants.all(),
            body=cls.Body(
                round=training.model.round,
                global_model_uuid=training.model.id
            ),
            training_uuid=training.id
        )

Instance variables¶

callback_error

callback_success

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¶

def serialize(
    self
) -> dict[str, typing.Any]

Serialize the notification into a dictionary.

Returns:

Type	Description
dict[str, Any]	The serialized notification.

View Source

    def serialize(self) -> dict[str, Any]:
        data = super().serialize()
        data["training_uuid"] = str(self.training_uuid)
        return data