Documentation for Nebulaevents Module

AddonEvent

Bases: ABC

Abstract base class for all addon-related events in the system.

Source code in nebula/core/nebulaevents.py

class AddonEvent(ABC):
    """
    Abstract base class for all addon-related events in the system.
    """

    @abstractmethod
    async def get_event_data(self):
        """
        Retrieve the data associated with the event.

        Returns:
            Any: The event-specific data payload.
        """
        pass

get_event_data() abstractmethod async

Retrieve the data associated with the event.

Returns:

Name	Type	Description
Any		The event-specific data payload.

Source code in nebula/core/nebulaevents.py

@abstractmethod
async def get_event_data(self):
    """
    Retrieve the data associated with the event.

    Returns:
        Any: The event-specific data payload.
    """
    pass

AggregationEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class AggregationEvent(NodeEvent):
    def __init__(self, updates: dict, expected_nodes: set, missing_nodes: set):
        """Event triggered when model aggregation is ready.

        Args:
            updates (dict): Dictionary containing model updates.
            expected_nodes (set): Set of nodes expected to participate in aggregation.
            missing_nodes (set): Set of nodes that did not send their update.
        """
        self._updates = updates
        self._expected_nodes = expected_nodes
        self._missing_nodes = missing_nodes

    def __str__(self):
        return "Aggregation Ready"

    def update_updates(self, new_updates: dict):
        """Allows an external module to update the updates dictionary."""
        self._updates = new_updates

    async def get_event_data(self) -> tuple[dict, set, set]:
        """Retrieves the aggregation event data.

        Returns:
            tuple[dict, set, set]:
                - updates (dict): Model updates.
                - expected_nodes (set): Expected nodes.
                - missing_nodes (set): Missing nodes.
        """
        return (self._updates, self._expected_nodes, self._missing_nodes)

    async def is_concurrent(self) -> bool:
        return False

__init__(updates, expected_nodes, missing_nodes)

Event triggered when model aggregation is ready.

Parameters:

Name	Type	Description	Default
updates	dict	Dictionary containing model updates.	required
expected_nodes	set	Set of nodes expected to participate in aggregation.	required
missing_nodes	set	Set of nodes that did not send their update.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, updates: dict, expected_nodes: set, missing_nodes: set):
    """Event triggered when model aggregation is ready.

    Args:
        updates (dict): Dictionary containing model updates.
        expected_nodes (set): Set of nodes expected to participate in aggregation.
        missing_nodes (set): Set of nodes that did not send their update.
    """
    self._updates = updates
    self._expected_nodes = expected_nodes
    self._missing_nodes = missing_nodes

get_event_data() async

Retrieves the aggregation event data.

Returns:

Type	Description
tuple[dict, set, set]	tuple[dict, set, set]: - updates (dict): Model updates. - expected_nodes (set): Expected nodes. - missing_nodes (set): Missing nodes.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[dict, set, set]:
    """Retrieves the aggregation event data.

    Returns:
        tuple[dict, set, set]:
            - updates (dict): Model updates.
            - expected_nodes (set): Expected nodes.
            - missing_nodes (set): Missing nodes.
    """
    return (self._updates, self._expected_nodes, self._missing_nodes)

update_updates(new_updates)

Allows an external module to update the updates dictionary.

Source code in nebula/core/nebulaevents.py

def update_updates(self, new_updates: dict):
    """Allows an external module to update the updates dictionary."""
    self._updates = new_updates

BeaconRecievedEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class BeaconRecievedEvent(NodeEvent):
    def __init__(self, source, geoloc):
        """
        Initializes an BeaconRecievedEvent.

        Args:
            source (str): The received beacon source.
            geoloc (tuple): The geolocalzition associated with the received beacon source.
        """
        self._source = source
        self._geoloc = geoloc

    def __str__(self):
        return "Beacon recieved"

    async def get_event_data(self) -> tuple[str, tuple[float, float]]:
        """
        Retrieves the event data.

        Returns:
            tuple[str, tuple[float, float]]: A tuple containing:
                - The beacon's source.
                - the device geolocalization (latitude, longitude).
        """
        return (self._source, self._geoloc)

    async def is_concurrent(self) -> bool:
        return True

__init__(source, geoloc)

Initializes an BeaconRecievedEvent.

Parameters:

Name	Type	Description	Default
source	str	The received beacon source.	required
geoloc	tuple	The geolocalzition associated with the received beacon source.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, source, geoloc):
    """
    Initializes an BeaconRecievedEvent.

    Args:
        source (str): The received beacon source.
        geoloc (tuple): The geolocalzition associated with the received beacon source.
    """
    self._source = source
    self._geoloc = geoloc

get_event_data() async

Retrieves the event data.

Returns:

Type	Description
tuple[str, tuple[float, float]]	tuple[str, tuple[float, float]]: A tuple containing: - The beacon's source. - the device geolocalization (latitude, longitude).

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[str, tuple[float, float]]:
    """
    Retrieves the event data.

    Returns:
        tuple[str, tuple[float, float]]: A tuple containing:
            - The beacon's source.
            - the device geolocalization (latitude, longitude).
    """
    return (self._source, self._geoloc)

ChangeLocationEvent

Bases: AddonEvent

Event used to signal a change in the node's geographical location.

Attributes:

Name	Type	Description
latitude	float	New latitude of the node.
longitude	float	New longitude of the node.

Source code in nebula/core/nebulaevents.py

class ChangeLocationEvent(AddonEvent):
    """
    Event used to signal a change in the node's geographical location.

    Attributes:
        latitude (float): New latitude of the node.
        longitude (float): New longitude of the node.
    """

    def __init__(self, latitude, longitude):
        """
        Initializes a ChangeLocationEvent.

        Args:
            latitude (float): The new latitude value.
            longitude (float): The new longitude value.
        """
        self.latitude = latitude
        self.longitude = longitude

    def __str__(self):
        return "ChangeLocationEvent"

    async def get_event_data(self):
        """
        Returns the new location coordinates associated with this event.

        Returns:
            tuple: A tuple containing latitude and longitude.
        """
        return (self.latitude, self.longitude)

__init__(latitude, longitude)

Initializes a ChangeLocationEvent.

Parameters:

Name	Type	Description	Default
latitude	float	The new latitude value.	required
longitude	float	The new longitude value.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, latitude, longitude):
    """
    Initializes a ChangeLocationEvent.

    Args:
        latitude (float): The new latitude value.
        longitude (float): The new longitude value.
    """
    self.latitude = latitude
    self.longitude = longitude

get_event_data() async

Returns the new location coordinates associated with this event.

Returns:

Name	Type	Description
tuple		A tuple containing latitude and longitude.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self):
    """
    Returns the new location coordinates associated with this event.

    Returns:
        tuple: A tuple containing latitude and longitude.
    """
    return (self.latitude, self.longitude)

DuplicatedMessageEvent

Bases: NodeEvent

Event triggered when a message is received that has already been processed.

Attributes:

Name	Type	Description
source	str	The address of the node that sent the duplicated message.

Source code in nebula/core/nebulaevents.py

class DuplicatedMessageEvent(NodeEvent):
    """
    Event triggered when a message is received that has already been processed.

    Attributes:
        source (str): The address of the node that sent the duplicated message.
    """

    def __init__(self, source: str, message_type: str):
        self.source = source

    def __str__(self):
        return f"DuplicatedMessageEvent from {self.source}"

    async def get_event_data(self) -> tuple[str]:
        return (self.source)

    async def is_concurrent(self) -> bool:
        return True

ExperimentFinishEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class ExperimentFinishEvent(NodeEvent):
    def __init__(self):
        """Event triggered when experiment is going to finish."""

    def __str__(self):
        return "Experiment finished"

    async def get_event_data(self):
        pass

    async def is_concurrent(self):
        return False

__init__()

Event triggered when experiment is going to finish.

Source code in nebula/core/nebulaevents.py

def __init__(self):
    """Event triggered when experiment is going to finish."""

GPSEvent

Bases: AddonEvent

Event triggered by a GPS module providing distance data between nodes.

Attributes:

Name	Type	Description
distances	dict	A dictionary mapping node addresses to their respective distances.

Source code in nebula/core/nebulaevents.py

class GPSEvent(AddonEvent):
    """
    Event triggered by a GPS module providing distance data between nodes.

    Attributes:
        distances (dict): A dictionary mapping node addresses to their respective distances.
    """

    def __init__(self, distances: dict):
        """
        Initializes a GPSEvent.

        Args:
            distances (dict): Dictionary of distances from the current node to others.
        """
        self.distances = distances

    def __str__(self):
        return "GPSEvent"

    async def get_event_data(self) -> dict:
        """
        Returns the distance data associated with this event.

        Returns:
            dict: A copy of the distances dictionary.
        """
        return self.distances.copy()

__init__(distances)

Initializes a GPSEvent.

Parameters:

Name	Type	Description	Default
distances	dict	Dictionary of distances from the current node to others.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, distances: dict):
    """
    Initializes a GPSEvent.

    Args:
        distances (dict): Dictionary of distances from the current node to others.
    """
    self.distances = distances

get_event_data() async

Returns the distance data associated with this event.

Returns:

Name	Type	Description
dict	dict	A copy of the distances dictionary.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> dict:
    """
    Returns the distance data associated with this event.

    Returns:
        dict: A copy of the distances dictionary.
    """
    return self.distances.copy()

MessageEvent

Event class for wrapping received messages in the system.

Attributes:

Name	Type	Description
message_type	str	Type/category of the message.
source	str	Address or identifier of the message sender.
message	Any	The actual message payload.

Source code in nebula/core/nebulaevents.py

class MessageEvent:
    """
    Event class for wrapping received messages in the system.

    Attributes:
        message_type (str): Type/category of the message.
        source (str): Address or identifier of the message sender.
        message (Any): The actual message payload.
    """

    def __init__(self, message_type, source, message):
        """
        Initializes a MessageEvent instance.

        Args:
            message_type (str): Type/category of the message.
            source (str): Address or identifier of the message sender.
            message (Any): The actual message payload.
        """
        self.source = source
        self.message_type = message_type
        self.message = message

__init__(message_type, source, message)

Initializes a MessageEvent instance.

Parameters:

Name	Type	Description	Default
message_type	str	Type/category of the message.	required
source	str	Address or identifier of the message sender.	required
message	Any	The actual message payload.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, message_type, source, message):
    """
    Initializes a MessageEvent instance.

    Args:
        message_type (str): Type/category of the message.
        source (str): Address or identifier of the message sender.
        message (Any): The actual message payload.
    """
    self.source = source
    self.message_type = message_type
    self.message = message

ModelPropagationEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class ModelPropagationEvent(NodeEvent):
    def __init__(self, eligible_neighbors, strategy):
        """Event triggered when model propagation is ready.

        Args:
            eligible_neighbors (set): The elegible neighbors to propagate model.
            strategy (str): Strategy to propagete the model
        """
        self.eligible_neighbors = eligible_neighbors
        self._strategy = strategy

    def __str__(self):
        return f"Model propagation event, strategy: {self._strategy}"

    async def get_event_data(self) -> tuple[set, str]:
        """
        Retrieves the event data.

        Returns:
            tuple[set, str]: A tuple containing:
                - The elegible neighbors to propagate model.
                - The propagation strategy.
        """
        return (self.eligible_neighbors, self._strategy)

    async def is_concurrent(self) -> bool:
        return False    

__init__(eligible_neighbors, strategy)

Event triggered when model propagation is ready.

Parameters:

Name	Type	Description	Default
eligible_neighbors	set	The elegible neighbors to propagate model.	required
strategy	str	Strategy to propagete the model	required

Source code in nebula/core/nebulaevents.py

def __init__(self, eligible_neighbors, strategy):
    """Event triggered when model propagation is ready.

    Args:
        eligible_neighbors (set): The elegible neighbors to propagate model.
        strategy (str): Strategy to propagete the model
    """
    self.eligible_neighbors = eligible_neighbors
    self._strategy = strategy

get_event_data() async

Retrieves the event data.

Returns:

Type	Description
tuple[set, str]	tuple[set, str]: A tuple containing: - The elegible neighbors to propagate model. - The propagation strategy.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[set, str]:
    """
    Retrieves the event data.

    Returns:
        tuple[set, str]: A tuple containing:
            - The elegible neighbors to propagate model.
            - The propagation strategy.
    """
    return (self.eligible_neighbors, self._strategy)

NodeBlacklistedEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class NodeBlacklistedEvent(NodeEvent):
    def __init__(self, node_addr, blacklisted: bool = False):
        """
        Initializes a NodeBlacklistedEvent.

        Args:
            node_addr (str): The address of the node.
            blacklisted (bool, optional): True if the node is blacklisted,
                                          False if it's just marked as recently disconnected.
        """
        self._node_addr = node_addr
        self._blacklisted = blacklisted

    def __str__(self):
        return f"Node addr: {self._node_addr} | Blacklisted: {self._blacklisted} | Recently disconnected: {not self._blacklisted}"

    async def get_event_data(self) -> tuple[str, bool]:
        """
        Retrieves the address of the node and its blacklist status.

        Returns:
            tuple[str, bool]: A tuple containing the node address and blacklist flag.
        """
        return (self._node_addr, self._blacklisted)

    async def is_concurrent(self):
        return True

__init__(node_addr, blacklisted=False)

Initializes a NodeBlacklistedEvent.

Parameters:

Name	Type	Description	Default
node_addr	str	The address of the node.	required
blacklisted	bool	True if the node is blacklisted, False if it's just marked as recently disconnected.	False

Source code in nebula/core/nebulaevents.py

def __init__(self, node_addr, blacklisted: bool = False):
    """
    Initializes a NodeBlacklistedEvent.

    Args:
        node_addr (str): The address of the node.
        blacklisted (bool, optional): True if the node is blacklisted,
                                      False if it's just marked as recently disconnected.
    """
    self._node_addr = node_addr
    self._blacklisted = blacklisted

get_event_data() async

Retrieves the address of the node and its blacklist status.

Returns:

Type	Description
tuple[str, bool]	tuple[str, bool]: A tuple containing the node address and blacklist flag.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[str, bool]:
    """
    Retrieves the address of the node and its blacklist status.

    Returns:
        tuple[str, bool]: A tuple containing the node address and blacklist flag.
    """
    return (self._node_addr, self._blacklisted)

NodeEvent

Bases: ABC

Abstract base class for all node-related events in the system.

Source code in nebula/core/nebulaevents.py

class NodeEvent(ABC):
    """
    Abstract base class for all node-related events in the system.
    """

    @abstractmethod
    async def get_event_data(self):
        """
        Retrieve the data associated with the event.

        Returns:
            Any: The event-specific data payload.
        """
        pass

    @abstractmethod
    async def is_concurrent(self):
        """
        Indicates whether the event can be handled concurrently.

        Returns:
            bool: True if concurrent handling is allowed, False otherwise.
        """
        pass

get_event_data() abstractmethod async

Retrieve the data associated with the event.

Returns:

Name	Type	Description
Any		The event-specific data payload.

Source code in nebula/core/nebulaevents.py

@abstractmethod
async def get_event_data(self):
    """
    Retrieve the data associated with the event.

    Returns:
        Any: The event-specific data payload.
    """
    pass

is_concurrent() abstractmethod async

Indicates whether the event can be handled concurrently.

Returns:

Name	Type	Description
bool		True if concurrent handling is allowed, False otherwise.

Source code in nebula/core/nebulaevents.py

@abstractmethod
async def is_concurrent(self):
    """
    Indicates whether the event can be handled concurrently.

    Returns:
        bool: True if concurrent handling is allowed, False otherwise.
    """
    pass

NodeFoundEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class NodeFoundEvent(NodeEvent):
    def __init__(self, node_addr):
        """Event triggered when a new node is found.

        Args:
            node_addr (str): Address of the neighboring node.
        """
        self._node_addr = node_addr

    def __str__(self):
        return f"Node addr: {self._node_addr} found"

    async def get_event_data(self) -> tuple[str, bool]:
        """Retrieves the node found event data.

        Returns:
            tuple[str, bool]:
                - node_addr (str): Address of the node found.
        """
        return self._node_addr

    async def is_concurrent(self) -> bool:
        return True

__init__(node_addr)

Event triggered when a new node is found.

Parameters:

Name	Type	Description	Default
node_addr	str	Address of the neighboring node.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, node_addr):
    """Event triggered when a new node is found.

    Args:
        node_addr (str): Address of the neighboring node.
    """
    self._node_addr = node_addr

get_event_data() async

Retrieves the node found event data.

Returns:

Type	Description
tuple[str, bool]	tuple[str, bool]: - node_addr (str): Address of the node found.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[str, bool]:
    """Retrieves the node found event data.

    Returns:
        tuple[str, bool]:
            - node_addr (str): Address of the node found.
    """
    return self._node_addr

RoundEndEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class RoundEndEvent(NodeEvent):
    def __init__(self, round, end_time):
        """Event triggered when round is going to start.

        Args:
            round (int): Round number.
            end_time (time): Current time when round has ended.
        """
        self._round_end_time = end_time
        self._round = round

    def __str__(self):
        return "Round ending"

    async def get_event_data(self):
        """Retrieves the round start event data.

        Returns:
            tuple[int, float]:
                -round (int): Round number.
                -end_time (time): Current time when round has ended.
        """
        return (self._round, self._round_end_time)

    async def is_concurrent(self):
        return False

__init__(round, end_time)

Event triggered when round is going to start.

Parameters:

Name	Type	Description	Default
round	int	Round number.	required
end_time	time	Current time when round has ended.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, round, end_time):
    """Event triggered when round is going to start.

    Args:
        round (int): Round number.
        end_time (time): Current time when round has ended.
    """
    self._round_end_time = end_time
    self._round = round

get_event_data() async

Retrieves the round start event data.

Returns:

Type	Description
	tuple[int, float]: -round (int): Round number. -end_time (time): Current time when round has ended.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self):
    """Retrieves the round start event data.

    Returns:
        tuple[int, float]:
            -round (int): Round number.
            -end_time (time): Current time when round has ended.
    """
    return (self._round, self._round_end_time)

RoundStartEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class RoundStartEvent(NodeEvent):
    def __init__(self, round, start_time, expected_nodes):
        """Event triggered when round is going to start.

        Args:
            round (int): Round number.
            start_time (time): Current time when round is going to start.
        """
        self._round_start_time = start_time
        self._round = round
        self._expected_nodes = expected_nodes

    def __str__(self):
        return "Round starting"

    async def get_event_data(self):
        """Retrieves the round start event data.

        Returns:
            tuple[int, float]:
                -round (int): Round number.
                -start_time (time): Current time when round is going to start.
        """
        return (self._round, self._round_start_time, self._expected_nodes)

    async def is_concurrent(self):
        return False

__init__(round, start_time, expected_nodes)

Event triggered when round is going to start.

Parameters:

Name	Type	Description	Default
round	int	Round number.	required
start_time	time	Current time when round is going to start.	required

Source code in nebula/core/nebulaevents.py

def __init__(self, round, start_time, expected_nodes):
    """Event triggered when round is going to start.

    Args:
        round (int): Round number.
        start_time (time): Current time when round is going to start.
    """
    self._round_start_time = start_time
    self._round = round
    self._expected_nodes = expected_nodes

get_event_data() async

Retrieves the round start event data.

Returns:

Type	Description
	tuple[int, float]: -round (int): Round number. -start_time (time): Current time when round is going to start.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self):
    """Retrieves the round start event data.

    Returns:
        tuple[int, float]:
            -round (int): Round number.
            -start_time (time): Current time when round is going to start.
    """
    return (self._round, self._round_start_time, self._expected_nodes)

UpdateNeighborEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class UpdateNeighborEvent(NodeEvent):
    def __init__(self, node_addr, removed=False, joining=False):
        """Event triggered when a neighboring node is updated.

        Args:
            node_addr (str): Address of the neighboring node.
            removed (bool, optional): Indicates whether the node was removed.
                                      Defaults to False.
        """
        self._node_addr = node_addr
        self._removed = removed
        self._joining_federation = joining

    def __str__(self):
        return f"Node addr: {self._node_addr}, removed: {self._removed}"

    async def get_event_data(self) -> tuple[str, bool]:
        """Retrieves the neighbor update event data.

        Returns:
            tuple[str, bool]:
                - node_addr (str): Address of the neighboring node.
                - removed (bool): Whether the node was removed.
        """
        return (self._node_addr, self._removed)

    async def is_concurrent(self) -> bool:
        return False

    def is_joining_federation(self):
        return self._joining_federation

__init__(node_addr, removed=False, joining=False)

Event triggered when a neighboring node is updated.

Parameters:

Name	Type	Description	Default
node_addr	str	Address of the neighboring node.	required
removed	bool	Indicates whether the node was removed. Defaults to False.	False

Source code in nebula/core/nebulaevents.py

def __init__(self, node_addr, removed=False, joining=False):
    """Event triggered when a neighboring node is updated.

    Args:
        node_addr (str): Address of the neighboring node.
        removed (bool, optional): Indicates whether the node was removed.
                                  Defaults to False.
    """
    self._node_addr = node_addr
    self._removed = removed
    self._joining_federation = joining

get_event_data() async

Retrieves the neighbor update event data.

Returns:

Type	Description
tuple[str, bool]	tuple[str, bool]: - node_addr (str): Address of the neighboring node. - removed (bool): Whether the node was removed.

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[str, bool]:
    """Retrieves the neighbor update event data.

    Returns:
        tuple[str, bool]:
            - node_addr (str): Address of the neighboring node.
            - removed (bool): Whether the node was removed.
    """
    return (self._node_addr, self._removed)

UpdateReceivedEvent

Bases: NodeEvent

Source code in nebula/core/nebulaevents.py

class UpdateReceivedEvent(NodeEvent):
    def __init__(self, decoded_model, weight, source, round, local=False):
        """
        Initializes an UpdateReceivedEvent.

        Args:
            decoded_model (Any): The received model update.
            weight (float): The weight associated with the received update.
            source (str): The identifier or address of the node that sent the update.
            round (int): The round number in which the update was received.
            local (bool): Local update
        """
        self._source = source
        self._round = round
        self._model = decoded_model
        self._weight = weight
        self._local = local

    def __str__(self):
        return f"Update received from source: {self._source}, round: {self._round}"

    async def get_event_data(self) -> tuple[object, int, str, int, bool]:
        """
        Retrieves the event data.

        Returns:
            tuple[Any, float, str, int, bool]: A tuple containing:
                - The received model update.
                - The weight associated with the update.
                - The source node identifier.
                - The round number of the update.
                - If the update is local
        """
        return (self._model, self._weight, self._source, self._round, self._local)

    async def is_concurrent(self) -> bool:
        return False

__init__(decoded_model, weight, source, round, local=False)

Initializes an UpdateReceivedEvent.

Parameters:

Name	Type	Description	Default
decoded_model	Any	The received model update.	required
weight	float	The weight associated with the received update.	required
source	str	The identifier or address of the node that sent the update.	required
round	int	The round number in which the update was received.	required
local	bool	Local update	False

Source code in nebula/core/nebulaevents.py

def __init__(self, decoded_model, weight, source, round, local=False):
    """
    Initializes an UpdateReceivedEvent.

    Args:
        decoded_model (Any): The received model update.
        weight (float): The weight associated with the received update.
        source (str): The identifier or address of the node that sent the update.
        round (int): The round number in which the update was received.
        local (bool): Local update
    """
    self._source = source
    self._round = round
    self._model = decoded_model
    self._weight = weight
    self._local = local

get_event_data() async

Retrieves the event data.

Returns:

Type	Description
tuple[object, int, str, int, bool]	tuple[Any, float, str, int, bool]: A tuple containing: - The received model update. - The weight associated with the update. - The source node identifier. - The round number of the update. - If the update is local

Source code in nebula/core/nebulaevents.py

async def get_event_data(self) -> tuple[object, int, str, int, bool]:
    """
    Retrieves the event data.

    Returns:
        tuple[Any, float, str, int, bool]: A tuple containing:
            - The received model update.
            - The weight associated with the update.
            - The source node identifier.
            - The round number of the update.
            - If the update is local
    """
    return (self._model, self._weight, self._source, self._round, self._local)