Skip to content

User

Bases: ComponentManager, Agent

Class that represents an user.

Source code in edge_sim_py/components/user.py 
 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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
class User(ComponentManager, Agent):
    """Class that represents an user."""

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

    def __init__(self, obj_id: int = None) -> object:
        """Creates an User object.

        Args:
            obj_id (int, optional): Object identifier. Defaults to None.

        Returns:
            object: Created User 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

        # User coordinates
        self.coordinates_trace = []
        self.coordinates = None

        # List of applications accessed by the user
        self.applications = []

        # Reference to the base station the user is connected to
        self.base_station = None

        # User access metadata
        self.making_requests = {}
        self.access_patterns = {}

        # User mobility model
        self.mobility_model = None
        self.mobility_model_parameters = {}

        # List of metadata from applications accessed by the user
        self.communication_paths = {}
        self.delays = {}
        self.delay_slas = {}

        # 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.
        """
        access_patterns = {}
        for app_id, access_pattern in self.access_patterns.items():
            access_patterns[app_id] = {"class": access_pattern.__class__.__name__, "id": access_pattern.id}

        dictionary = {
            "attributes": {
                "id": self.id,
                "coordinates": self.coordinates,
                "coordinates_trace": self.coordinates_trace,
                "delays": copy.deepcopy(self.delays),
                "delay_slas": copy.deepcopy(self.delay_slas),
                "communication_paths": copy.deepcopy(self.communication_paths),
                "making_requests": copy.deepcopy(self.making_requests),
                "mobility_model_parameters": copy.deepcopy(self.mobility_model_parameters)
                if self.mobility_model_parameters
                else {},
            },
            "relationships": {
                "access_patterns": access_patterns,
                "mobility_model": self.mobility_model.__name__,
                "applications": [{"class": type(app).__name__, "id": app.id} for app in self.applications],
                "base_station": {"class": type(self.base_station).__name__, "id": self.base_station.id},
            },
        }
        return dictionary

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

        Returns:
            metrics (dict): Object metrics.
        """
        access_history = {}
        for app in self.applications:
            access_history[str(app.id)] = self.access_patterns[str(app.id)].history

        metrics = {
            "Instance ID": self.id,
            "Coordinates": self.coordinates,
            "Base Station": f"{self.base_station} ({self.base_station.coordinates})" if self.base_station else None,
            "Delays": copy.deepcopy(self.delays),
            "Communication Paths": copy.deepcopy(self.communication_paths),
            "Making Requests": copy.deepcopy(self.making_requests),
            "Access History": copy.deepcopy(access_history),
        }
        return metrics

    def step(self):
        """Method that executes the events involving the object at each time step."""
        # Updating user access
        current_step = self.model.schedule.steps + 1
        for app in self.applications:
            last_access = self.access_patterns[str(app.id)].history[-1]

            # Updating user access waiting and access times. Waiting time represents the period in which the user is waiting for
            # his application to be provisioned. Access time represents the period in which the user is successfully accessing
            # his application, meaning his application is available. We assume that an application is only available when all its
            # services are available.
            if self.making_requests[str(app.id)][str(current_step)] == True:
                if len([s for s in app.services if s._available]) == len(app.services):
                    last_access["access_time"] += 1
                else:
                    last_access["waiting_time"] += 1

            # Updating user's making requests attribute for the next time step
            if current_step + 1 >= last_access["start"] and current_step + 1 <= last_access["end"]:
                self.making_requests[str(app.id)][str(current_step + 1)] = True
            else:
                self.making_requests[str(app.id)][str(current_step + 1)] = False

            # Creating new access request if needed
            if current_step + 1 == last_access["next_access"]:
                self.making_requests[str(app.id)][str(current_step + 1)] = True
                self.access_patterns[str(app.id)].get_next_access(start=current_step + 1)

        # Re-executing user's mobility model in case no future mobility track is known by the simulator
        if len(self.coordinates_trace) <= self.model.schedule.steps:
            self.mobility_model(self)

        # Updating user's location
        if self.coordinates != self.coordinates_trace[self.model.schedule.steps]:
            self.coordinates = self.coordinates_trace[self.model.schedule.steps]

            # Connecting the user to the closest base station
            self.base_station = BaseStation.find_by(attribute_name="coordinates", attribute_value=self.coordinates)

            for application in self.applications:
                # Only updates the routing path of apps available (i.e., whose services are available)
                services_available = len([s for s in application.services if s._available])
                if services_available == len(application.services):
                    # Recomputing user communication paths
                    self.set_communication_path(app=application)
                else:
                    self.communication_paths[str(application.id)] = []
                    self._compute_delay(app=application)

    def _compute_delay(self, app: object, metric: str = "latency") -> int:
        """Computes the delay of an application accessed by the user.

        Args:
            metric (str, optional): Delay measure (valid options: 'latency' and 'response time'). Defaults to 'latency'.
            app (object): Application accessed by the user.

        Returns:
            delay (int): User-perceived delay when accessing application "app".
        """
        topology = Topology.first()

        services_available = len([s for s in app.services if s._available])
        if services_available < len(app.services):
            # Defining the delay as infinity if any of the application services is not available
            delay = float("inf")
        else:
            # Initializes the application's delay with the time it takes to communicate its client and his base station
            delay = self.base_station.wireless_delay

            # Adding the communication path delay to the application's delay
            for path in self.communication_paths[str(app.id)]:
                delay += topology.calculate_path_delay(path=[NetworkSwitch.find_by_id(i) for i in path])

            if metric.lower() == "response time":
                # We assume that Response Time = Latency * 2
                delay = delay * 2

        # Updating application delay inside user's 'applications' attribute
        self.delays[str(app.id)] = delay

        return delay

    def set_communication_path(self, app: object, communication_path: list = []) -> list:
        """Updates the set of links used during the communication of user and its application.

        Args:
            app (object): User application.
            communication_path (list, optional): User-specified communication path. Defaults to [].

        Returns:
            list: Updated communication path.
        """
        topology = Topology.first()

        # Releasing links used in the past to connect the user with its application
        if app in self.communication_paths:
            path = [[NetworkSwitch.find_by_id(i) for i in p] for p in self.communication_paths[str(app.id)]]
            topology._release_communication_path(communication_path=path, app=app)

        # Defining communication path
        if len(communication_path) > 0:
            self.communication_paths[str(app.id)] = communication_path
        else:
            self.communication_paths[str(app.id)] = []

            service_hosts_base_stations = [service.server.base_station for service in app.services if service.server]
            communication_chain = [self.base_station] + service_hosts_base_stations

            # Defining a set of links to connect the items in the application's service chain
            for i in range(len(communication_chain) - 1):

                # Defining origin and target nodes
                origin = communication_chain[i]
                target = communication_chain[i + 1]

                # Finding and storing the best communication path between the origin and target nodes
                if origin == target:
                    path = []
                else:
                    path = nx.shortest_path(
                        G=topology,
                        source=origin.network_switch,
                        target=target.network_switch,
                        weight="delay",
                        method="dijkstra",
                    )

                # Adding the best path found to the communication path
                self.communication_paths[str(app.id)].append([network_switch.id for network_switch in path])

                # Computing the new demand of chosen links
                path = [[NetworkSwitch.find_by_id(i) for i in p] for p in self.communication_paths[str(app.id)]]
                topology._allocate_communication_path(communication_path=path, app=app)

        # Computing application's delay
        self._compute_delay(app=app, metric="latency")

        return self.communication_paths[str(app.id)]

    def _connect_to_application(self, app: object, delay_sla: float) -> object:
        """Connects the user to a given application, establishing all the relationship attributes in both objects.

        Args:
            app (object): Application that will be connected to the user.
            delay_sla (float): Delay threshold for the user regarding the specified application.

        Returns:
            self (object): Updated user object.
        """
        # Defining the relationship attributes between the user and its new application
        self.applications.append(app)
        app.users.append(self)

        # Assigning delay and delay SLA attributes. Delay is initially None, and must be overwritten by the service placement
        self.delay_slas[str(app.id)] = delay_sla
        self.delays[str(app.id)] = None

    def _set_initial_position(self, coordinates: list, number_of_replicates: int = 0) -> object:
        """Defines the initial coordinates for the user, automatically connecting to a base station in that position.

        Args:
            coordinates (list): Initial user coordinates.
            number_of_replicates (int, optional): Number of times the initial coordinates will replicated in the coordinates trace. Defaults to 0.

        Returns:
            self (object): Updated user object.
        """
        # Defining the "coordinates" and "coordinates_trace" attributes
        self.coordinates = coordinates
        self.coordinates_trace = [coordinates for _ in range(number_of_replicates - 1)]

        # Connecting the user to the base station that shares his initial position
        base_station = BaseStation.find_by(attribute_name="coordinates", attribute_value=self.coordinates)

        if base_station is None:
            raise Exception(f"No base station was found at coordinates {coordinates} to connect to user {self}.")

        self.base_station = base_station
        base_station.users.append(self)

__init__(obj_id=None)

Creates an User object.

Parameters:

Name Type Description Default
obj_id int

Object identifier. Defaults to None.

 None

Returns:

Name Type Description
object object

Created User object.
Source code in edge_sim_py/components/user.py 
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
def __init__(self, obj_id: int = None) -> object:
    """Creates an User object.

    Args:
        obj_id (int, optional): Object identifier. Defaults to None.

    Returns:
        object: Created User 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

    # User coordinates
    self.coordinates_trace = []
    self.coordinates = None

    # List of applications accessed by the user
    self.applications = []

    # Reference to the base station the user is connected to
    self.base_station = None

    # User access metadata
    self.making_requests = {}
    self.access_patterns = {}

    # User mobility model
    self.mobility_model = None
    self.mobility_model_parameters = {}

    # List of metadata from applications accessed by the user
    self.communication_paths = {}
    self.delays = {}
    self.delay_slas = {}

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

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/user.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def collect(self) -> dict:
    """Method that collects a set of metrics for the object.

    Returns:
        metrics (dict): Object metrics.
    """
    access_history = {}
    for app in self.applications:
        access_history[str(app.id)] = self.access_patterns[str(app.id)].history

    metrics = {
        "Instance ID": self.id,
        "Coordinates": self.coordinates,
        "Base Station": f"{self.base_station} ({self.base_station.coordinates})" if self.base_station else None,
        "Delays": copy.deepcopy(self.delays),
        "Communication Paths": copy.deepcopy(self.communication_paths),
        "Making Requests": copy.deepcopy(self.making_requests),
        "Access History": copy.deepcopy(access_history),
    }
    return metrics

set_communication_path(app, communication_path=[])

Updates the set of links used during the communication of user and its application.

Parameters:

Name Type Description Default
app object

User application.

 required
communication_path list

User-specified communication path. Defaults to [].

 []

Returns:

Name Type Description
list list

Updated communication path.
Source code in edge_sim_py/components/user.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def set_communication_path(self, app: object, communication_path: list = []) -> list:
    """Updates the set of links used during the communication of user and its application.

    Args:
        app (object): User application.
        communication_path (list, optional): User-specified communication path. Defaults to [].

    Returns:
        list: Updated communication path.
    """
    topology = Topology.first()

    # Releasing links used in the past to connect the user with its application
    if app in self.communication_paths:
        path = [[NetworkSwitch.find_by_id(i) for i in p] for p in self.communication_paths[str(app.id)]]
        topology._release_communication_path(communication_path=path, app=app)

    # Defining communication path
    if len(communication_path) > 0:
        self.communication_paths[str(app.id)] = communication_path
    else:
        self.communication_paths[str(app.id)] = []

        service_hosts_base_stations = [service.server.base_station for service in app.services if service.server]
        communication_chain = [self.base_station] + service_hosts_base_stations

        # Defining a set of links to connect the items in the application's service chain
        for i in range(len(communication_chain) - 1):

            # Defining origin and target nodes
            origin = communication_chain[i]
            target = communication_chain[i + 1]

            # Finding and storing the best communication path between the origin and target nodes
            if origin == target:
                path = []
            else:
                path = nx.shortest_path(
                    G=topology,
                    source=origin.network_switch,
                    target=target.network_switch,
                    weight="delay",
                    method="dijkstra",
                )

            # Adding the best path found to the communication path
            self.communication_paths[str(app.id)].append([network_switch.id for network_switch in path])

            # Computing the new demand of chosen links
            path = [[NetworkSwitch.find_by_id(i) for i in p] for p in self.communication_paths[str(app.id)]]
            topology._allocate_communication_path(communication_path=path, app=app)

    # Computing application's delay
    self._compute_delay(app=app, metric="latency")

    return self.communication_paths[str(app.id)]

step()

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

Source code in edge_sim_py/components/user.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def step(self):
    """Method that executes the events involving the object at each time step."""
    # Updating user access
    current_step = self.model.schedule.steps + 1
    for app in self.applications:
        last_access = self.access_patterns[str(app.id)].history[-1]

        # Updating user access waiting and access times. Waiting time represents the period in which the user is waiting for
        # his application to be provisioned. Access time represents the period in which the user is successfully accessing
        # his application, meaning his application is available. We assume that an application is only available when all its
        # services are available.
        if self.making_requests[str(app.id)][str(current_step)] == True:
            if len([s for s in app.services if s._available]) == len(app.services):
                last_access["access_time"] += 1
            else:
                last_access["waiting_time"] += 1

        # Updating user's making requests attribute for the next time step
        if current_step + 1 >= last_access["start"] and current_step + 1 <= last_access["end"]:
            self.making_requests[str(app.id)][str(current_step + 1)] = True
        else:
            self.making_requests[str(app.id)][str(current_step + 1)] = False

        # Creating new access request if needed
        if current_step + 1 == last_access["next_access"]:
            self.making_requests[str(app.id)][str(current_step + 1)] = True
            self.access_patterns[str(app.id)].get_next_access(start=current_step + 1)

    # Re-executing user's mobility model in case no future mobility track is known by the simulator
    if len(self.coordinates_trace) <= self.model.schedule.steps:
        self.mobility_model(self)

    # Updating user's location
    if self.coordinates != self.coordinates_trace[self.model.schedule.steps]:
        self.coordinates = self.coordinates_trace[self.model.schedule.steps]

        # Connecting the user to the closest base station
        self.base_station = BaseStation.find_by(attribute_name="coordinates", attribute_value=self.coordinates)

        for application in self.applications:
            # Only updates the routing path of apps available (i.e., whose services are available)
            services_available = len([s for s in application.services if s._available])
            if services_available == len(application.services):
                # Recomputing user communication paths
                self.set_communication_path(app=application)
            else:
                self.communication_paths[str(application.id)] = []
                self._compute_delay(app=application)