Skip to content

Topology

Bases: ComponentManager, nx.Graph, Agent

Class that represents a network topology.

Source code in edge_sim_py/components/topology.py 
 13
 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
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class Topology(ComponentManager, nx.Graph, Agent):
    """Class that represents a network topology."""

    # Class attributes that allow this class to use helper methods from ComponentManager
    _instances = []
    _object_count = 0

    def __init__(self, obj_id: int = None, existing_graph: nx.Graph = None) -> object:
        """Creates a Topology object backed by NetworkX functionality.

        Args:
            obj_id (int, optional): Object identifier.
            existing_graph (nx.Graph, optional): NetworkX graph representing the network topology.

        Returns:
            object: Created Topology object.
        """
        # Adding the new object to the list of instances of its class
        self.__class__._instances.append(self)

        # Object's class instance ID
        self.__class__._object_count += 1
        if obj_id is None:
            obj_id = self.__class__._object_count
        self.id = obj_id

        # Initializing the NetworkX topology
        if existing_graph is None:
            nx.Graph.__init__(self)
        else:
            nx.Graph.__init__(self, incoming_graph_data=existing_graph)

        # Model-specific attributes (defined inside the model's "initialize()" method)
        self.model = None
        self.unique_id = None

    def _to_dict(self) -> dict:
        """Method that overrides the way the object is formatted to JSON."

        Returns:
            dict: JSON-friendly representation of the object as a dictionary.
        """
        dictionary = {"attributes": {"id": self.id}}
        return dictionary

    def collect(self) -> dict:
        """Method that collects a set of metrics for the object.

        Returns:
            metrics (dict): Object metrics.
        """
        metrics = {}
        return metrics

    def step(self):
        """Method that executes the events involving the object at each time step."""
        self.model.network_flow_scheduling_algorithm(topology=self, flows=NetworkFlow.all())

    def _remove_path_duplicates(self, path: list) -> list:
        """Removes side-by-side duplicated nodes on network paths to avoid NetworkX crashes.

        Args:
            path (list): Original network path.

        Returns:
            modified_path (list): Modified network path without duplicates.
        """
        modified_path = []

        for i in range(len(path)):
            if len(modified_path) == 0 or len(modified_path) >= 1 and modified_path[-1] != path[i]:
                modified_path.append(path[i])

        return modified_path

    def calculate_path_delay(self, path: list) -> int:
        """Calculates the communication delay of a network path.

        Args:
            path (list): Network path whose delay will be calculated.

        Returns:
            path_delay (int): Network path delay.
        """
        path_delay = 0

        # Calculates the communication delay based on the delay property of each network link in the path
        path_delay = nx.classes.function.path_weight(G=self, path=path, weight="delay")

        return path_delay

    def _allocate_communication_path(self, communication_path: list, app: object):
        """Adds the demand of a given application to a set of links that comprehend a communication path.

        Args:
            communication_path (list): Communication path.
            app (object): Application object.
        """
        for path in communication_path:
            if len(path) > 1:
                for i in range(len(path) - 1):
                    node1 = path[i]
                    node2 = path[i + 1]

                    link = self[node1][node2]

                    if app not in link["applications"]:
                        link["applications"].append(app)

    def _release_communication_path(self, communication_path: list, app: object):
        """Releases the demand of a given application from a set of links that comprehend a communication path.

        Args:
            communication_path (list): Communication path.
            app (object): Application object.
        """
        for path in communication_path:
            if len(path) > 1:
                for i in range(len(path) - 1):
                    node1 = path[i]
                    node2 = path[i + 1]
                    link = self[node1][node2]

                    if app in link["applications"]:
                        link["applications"].remove(app)

__init__(obj_id=None, existing_graph=None)

Creates a Topology object backed by NetworkX functionality.

Parameters:

Name Type Description Default
obj_id int

Object identifier.

 None
existing_graph nx.Graph

NetworkX graph representing the network topology.

 None

Returns:

Name Type Description
object object

Created Topology object.
Source code in edge_sim_py/components/topology.py 
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
def __init__(self, obj_id: int = None, existing_graph: nx.Graph = None) -> object:
    """Creates a Topology object backed by NetworkX functionality.

    Args:
        obj_id (int, optional): Object identifier.
        existing_graph (nx.Graph, optional): NetworkX graph representing the network topology.

    Returns:
        object: Created Topology object.
    """
    # Adding the new object to the list of instances of its class
    self.__class__._instances.append(self)

    # Object's class instance ID
    self.__class__._object_count += 1
    if obj_id is None:
        obj_id = self.__class__._object_count
    self.id = obj_id

    # Initializing the NetworkX topology
    if existing_graph is None:
        nx.Graph.__init__(self)
    else:
        nx.Graph.__init__(self, incoming_graph_data=existing_graph)

    # Model-specific attributes (defined inside the model's "initialize()" method)
    self.model = None
    self.unique_id = None

calculate_path_delay(path)

Calculates the communication delay of a network path.

Parameters:

Name Type Description Default
path list

Network path whose delay will be calculated.

 required

Returns:

Name Type Description
path_delay int

Network path delay.
Source code in edge_sim_py/components/topology.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def calculate_path_delay(self, path: list) -> int:
    """Calculates the communication delay of a network path.

    Args:
        path (list): Network path whose delay will be calculated.

    Returns:
        path_delay (int): Network path delay.
    """
    path_delay = 0

    # Calculates the communication delay based on the delay property of each network link in the path
    path_delay = nx.classes.function.path_weight(G=self, path=path, weight="delay")

    return path_delay

collect()

Method that collects a set of metrics for the object.

Returns:

Name Type Description
metrics dict

Object metrics.
Source code in edge_sim_py/components/topology.py 
58
59
60
61
62
63
64
65
def collect(self) -> dict:
    """Method that collects a set of metrics for the object.

    Returns:
        metrics (dict): Object metrics.
    """
    metrics = {}
    return metrics

step()

Method that executes the events involving the object at each time step.

Source code in edge_sim_py/components/topology.py 
67
68
69
def step(self):
    """Method that executes the events involving the object at each time step."""
    self.model.network_flow_scheduling_algorithm(topology=self, flows=NetworkFlow.all())