fl_server_ai.notification.training ¶

Bases: TrainingNotification['TrainingFinishedNotification.Body']

              flowchart TD
              fl_server_ai.notification.training.TrainingFinishedNotification[TrainingFinishedNotification]
              fl_server_ai.notification.training.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.training.training.TrainingNotification --> fl_server_ai.notification.training.TrainingFinishedNotification
                                fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                




              click fl_server_ai.notification.training.TrainingFinishedNotification href "" "fl_server_ai.notification.training.TrainingFinishedNotification"
              click fl_server_ai.notification.training.training.TrainingNotification href "" "fl_server_ai.notification.training.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Notification that a training has finished.

Classes:

Methods:

Attributes:

14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

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
        )

type: NotificationType = TRAINING_FINISHED

              flowchart TD
              fl_server_ai.notification.training.TrainingFinishedNotification.Body[Body]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.training.TrainingFinishedNotification.Body
                


              click fl_server_ai.notification.training.TrainingFinishedNotification.Body href "" "fl_server_ai.notification.training.TrainingFinishedNotification.Body"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

22
23
24
25
26
27
28

@dataclass
class Body(Serializable):
    """
    Inner class representing the body of the notification.
    """
    global_model_uuid: UUID
    """The UUID of the global model."""

global_model_uuid: UUID

__init__(global_model_uuid: UUID) -> None

from_training(training: Training)

30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

@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
    )

Bases: TrainingRoundStartNotification

              flowchart TD
              fl_server_ai.notification.training.TrainingModelTestNotification[TrainingModelTestNotification]
              fl_server_ai.notification.training.round_start.TrainingRoundStartNotification[TrainingRoundStartNotification]
              fl_server_ai.notification.training.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.training.round_start.TrainingRoundStartNotification --> fl_server_ai.notification.training.TrainingModelTestNotification
                                fl_server_ai.notification.training.training.TrainingNotification --> fl_server_ai.notification.training.round_start.TrainingRoundStartNotification
                                fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                





              click fl_server_ai.notification.training.TrainingModelTestNotification href "" "fl_server_ai.notification.training.TrainingModelTestNotification"
              click fl_server_ai.notification.training.round_start.TrainingRoundStartNotification href "" "fl_server_ai.notification.training.round_start.TrainingRoundStartNotification"
              click fl_server_ai.notification.training.training.TrainingNotification href "" "fl_server_ai.notification.training.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Notification for the start of a model test round.

Attributes:

 8
 9
10
11
12
13
14

class TrainingModelTestNotification(TrainingRoundStartNotification):
    """
    Notification for the start of a model test round.
    """

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

type: NotificationType = MODEL_TEST_ROUND

Bases: Generic[TBody], Notification[TBody]

              flowchart TD
              fl_server_ai.notification.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                



              click fl_server_ai.notification.training.TrainingNotification href "" "fl_server_ai.notification.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Abstract base class for training notifications.

Methods:

Attributes:

11
12
13
14
15
16
17
18
19
20
21
22
23

@dataclass
class TrainingNotification(Generic[TBody], Notification[TBody]):
    """
    Abstract base class for training notifications.
    """

    training_uuid: UUID
    """The UUID of the training."""

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

training_uuid: UUID

__init__(receivers: list[NotificationReceiver], body: TBody, training_uuid: UUID) -> None

serialize() -> dict[str, Any]

20
21
22
23

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

Bases: TrainingNotification['TrainingRoundStartNotification.Body']

              flowchart TD
              fl_server_ai.notification.training.TrainingRoundStartNotification[TrainingRoundStartNotification]
              fl_server_ai.notification.training.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.training.training.TrainingNotification --> fl_server_ai.notification.training.TrainingRoundStartNotification
                                fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                




              click fl_server_ai.notification.training.TrainingRoundStartNotification href "" "fl_server_ai.notification.training.TrainingRoundStartNotification"
              click fl_server_ai.notification.training.training.TrainingNotification href "" "fl_server_ai.notification.training.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Notification for the start of a training round.

Classes:

Methods:

Attributes:

 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

class TrainingRoundStartNotification(TrainingNotification["TrainingRoundStartNotification.Body"]):
    """
    Notification for the start of a training round.
    """

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

    @property
    def callback_success(self) -> Optional[Signature]:
        return training_notification_callback_success.s(training_uuid=self.training_uuid)

    @property
    def callback_error(self) -> Optional[Signature]:
        return training_notification_callback_failure.s(training_uuid=self.training_uuid)

    @dataclass
    class Body(Serializable):
        """
        Inner class representing the body of the notification.
        """
        round: int
        """The round number."""
        global_model_uuid: UUID
        """The UUID of the global model."""

    @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
        )

callback_error: Signature | None

callback_success: Signature | None

type: NotificationType = UPDATE_ROUND_START

              flowchart TD
              fl_server_ai.notification.training.TrainingRoundStartNotification.Body[Body]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.training.TrainingRoundStartNotification.Body
                


              click fl_server_ai.notification.training.TrainingRoundStartNotification.Body href "" "fl_server_ai.notification.training.TrainingRoundStartNotification.Body"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

81
82
83
84
85
86
87
88
89

@dataclass
class Body(Serializable):
    """
    Inner class representing the body of the notification.
    """
    round: int
    """The round number."""
    global_model_uuid: UUID
    """The UUID of the global model."""

global_model_uuid: UUID

round: int

__init__(round: int, global_model_uuid: UUID) -> None

from_training(training: Training)

 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

@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
    )

Bases: TrainingRoundStartNotification

              flowchart TD
              fl_server_ai.notification.training.TrainingSWAGRoundStartNotification[TrainingSWAGRoundStartNotification]
              fl_server_ai.notification.training.round_start.TrainingRoundStartNotification[TrainingRoundStartNotification]
              fl_server_ai.notification.training.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.training.round_start.TrainingRoundStartNotification --> fl_server_ai.notification.training.TrainingSWAGRoundStartNotification
                                fl_server_ai.notification.training.training.TrainingNotification --> fl_server_ai.notification.training.round_start.TrainingRoundStartNotification
                                fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                





              click fl_server_ai.notification.training.TrainingSWAGRoundStartNotification href "" "fl_server_ai.notification.training.TrainingSWAGRoundStartNotification"
              click fl_server_ai.notification.training.round_start.TrainingRoundStartNotification href "" "fl_server_ai.notification.training.round_start.TrainingRoundStartNotification"
              click fl_server_ai.notification.training.training.TrainingNotification href "" "fl_server_ai.notification.training.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Notification for the start of a SWAG training round.

Attributes:

 8
 9
10
11
12
13
14

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

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

type: NotificationType = SWAG_ROUND_START

Bases: TrainingNotification['TrainingStartNotification.Body']

              flowchart TD
              fl_server_ai.notification.training.TrainingStartNotification[TrainingStartNotification]
              fl_server_ai.notification.training.training.TrainingNotification[TrainingNotification]
              fl_server_ai.notification.notification.Notification[Notification]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.training.training.TrainingNotification --> fl_server_ai.notification.training.TrainingStartNotification
                                fl_server_ai.notification.notification.Notification --> fl_server_ai.notification.training.training.TrainingNotification
                                fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.notification.Notification
                




              click fl_server_ai.notification.training.TrainingStartNotification href "" "fl_server_ai.notification.training.TrainingStartNotification"
              click fl_server_ai.notification.training.training.TrainingNotification href "" "fl_server_ai.notification.training.training.TrainingNotification"
              click fl_server_ai.notification.notification.Notification href "" "fl_server_ai.notification.notification.Notification"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

Class representing a notification for the start of a training.

Classes:

Methods:

Attributes:

14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

class TrainingStartNotification(TrainingNotification["TrainingStartNotification.Body"]):
    """
    Class representing a notification for the start of a training.
    """

    type: NotificationType = NotificationType.TRAINING_START
    """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 `TrainingStartNotification` instance from a training object.

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

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

type: NotificationType = TRAINING_START

              flowchart TD
              fl_server_ai.notification.training.TrainingStartNotification.Body[Body]
              fl_server_ai.notification.serializable.Serializable[Serializable]

                              fl_server_ai.notification.serializable.Serializable --> fl_server_ai.notification.training.TrainingStartNotification.Body
                


              click fl_server_ai.notification.training.TrainingStartNotification.Body href "" "fl_server_ai.notification.training.TrainingStartNotification.Body"
              click fl_server_ai.notification.serializable.Serializable href "" "fl_server_ai.notification.serializable.Serializable"
            

22
23
24
25
26
27
28

@dataclass
class Body(Serializable):
    """
    Inner class representing the body of the notification.
    """
    global_model_uuid: UUID
    """The UUID of the global model."""

global_model_uuid: UUID

__init__(global_model_uuid: UUID) -> None

from_training(training: Training)

30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

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

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

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