classUser(ComponentManager,Agent):"""Class that represents an user."""# Class attributes that allow this class to use helper methods from the ComponentManager_instances=[]_object_count=0def__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 classself.__class__._instances.append(self)# Object's class instance IDself.__class__._object_count+=1ifobj_idisNone:obj_id=self.__class__._object_countself.id=obj_id# User coordinatesself.coordinates_trace=[]self.coordinates=None# List of applications accessed by the userself.applications=[]# Reference to the base station the user is connected toself.base_station=None# User access metadataself.making_requests={}self.access_patterns={}# User mobility modelself.mobility_model=Noneself.mobility_model_parameters={}# List of metadata from applications accessed by the userself.communication_paths={}self.delays={}self.delay_slas={}# Model-specific attributes (defined inside the model's "initialize()" method)self.model=Noneself.unique_id=Nonedef_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={}forapp_id,access_patterninself.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)ifself.mobility_model_parameterselse{},},"relationships":{"access_patterns":access_patterns,"mobility_model":self.mobility_model.__name__,"applications":[{"class":type(app).__name__,"id":app.id}forappinself.applications],"base_station":{"class":type(self.base_station).__name__,"id":self.base_station.id},},}returndictionarydefcollect(self)->dict:"""Method that collects a set of metrics for the object. Returns: metrics (dict): Object metrics. """access_history={}forappinself.applications:access_history[str(app.id)]=self.access_patterns[str(app.id)].historymetrics={"Instance ID":self.id,"Coordinates":self.coordinates,"Base Station":f"{self.base_station} ({self.base_station.coordinates})"ifself.base_stationelseNone,"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),}returnmetricsdefstep(self):"""Method that executes the events involving the object at each time step."""# Updating user accesscurrent_step=self.model.schedule.steps+1forappinself.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.ifself.making_requests[str(app.id)][str(current_step)]==True:iflen([sforsinapp.servicesifs._available])==len(app.services):last_access["access_time"]+=1else:last_access["waiting_time"]+=1# Updating user's making requests attribute for the next time stepifcurrent_step+1>=last_access["start"]andcurrent_step+1<=last_access["end"]:self.making_requests[str(app.id)][str(current_step+1)]=Trueelse:self.making_requests[str(app.id)][str(current_step+1)]=False# Creating new access request if neededifcurrent_step+1==last_access["next_access"]:self.making_requests[str(app.id)][str(current_step+1)]=Trueself.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 simulatoriflen(self.coordinates_trace)<=self.model.schedule.steps:self.mobility_model(self)# Updating user's locationifself.coordinates!=self.coordinates_trace[self.model.schedule.steps]:self.coordinates=self.coordinates_trace[self.model.schedule.steps]# Connecting the user to the closest base stationself.base_station=BaseStation.find_by(attribute_name="coordinates",attribute_value=self.coordinates)forapplicationinself.applications:# Only updates the routing path of apps available (i.e., whose services are available)services_available=len([sforsinapplication.servicesifs._available])ifservices_available==len(application.services):# Recomputing user communication pathsself.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([sforsinapp.servicesifs._available])ifservices_available<len(app.services):# Defining the delay as infinity if any of the application services is not availabledelay=float("inf")else:# Initializes the application's delay with the time it takes to communicate its client and his base stationdelay=self.base_station.wireless_delay# Adding the communication path delay to the application's delayforpathinself.communication_paths[str(app.id)]:delay+=topology.calculate_path_delay(path=[NetworkSwitch.find_by_id(i)foriinpath])ifmetric.lower()=="response time":# We assume that Response Time = Latency * 2delay=delay*2# Updating application delay inside user's 'applications' attributeself.delays[str(app.id)]=delayreturndelaydefset_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 applicationifappinself.communication_paths:path=[[NetworkSwitch.find_by_id(i)foriinp]forpinself.communication_paths[str(app.id)]]topology._release_communication_path(communication_path=path,app=app)# Defining communication pathiflen(communication_path)>0:self.communication_paths[str(app.id)]=communication_pathelse:self.communication_paths[str(app.id)]=[]service_hosts_base_stations=[service.server.base_stationforserviceinapp.servicesifservice.server]communication_chain=[self.base_station]+service_hosts_base_stations# Defining a set of links to connect the items in the application's service chainforiinrange(len(communication_chain)-1):# Defining origin and target nodesorigin=communication_chain[i]target=communication_chain[i+1]# Finding and storing the best communication path between the origin and target nodesiforigin==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 pathself.communication_paths[str(app.id)].append([network_switch.idfornetwork_switchinpath])# Computing the new demand of chosen linkspath=[[NetworkSwitch.find_by_id(i)foriinp]forpinself.communication_paths[str(app.id)]]topology._allocate_communication_path(communication_path=path,app=app)# Computing application's delayself._compute_delay(app=app,metric="latency")returnself.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 applicationself.applications.append(app)app.users.append(self)# Assigning delay and delay SLA attributes. Delay is initially None, and must be overwritten by the service placementself.delay_slas[str(app.id)]=delay_slaself.delays[str(app.id)]=Nonedef_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" attributesself.coordinates=coordinatesself.coordinates_trace=[coordinatesfor_inrange(number_of_replicates-1)]# Connecting the user to the base station that shares his initial positionbase_station=BaseStation.find_by(attribute_name="coordinates",attribute_value=self.coordinates)ifbase_stationisNone:raiseException(f"No base station was found at coordinates {coordinates} to connect to user {self}.")self.base_station=base_stationbase_station.users.append(self)
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 classself.__class__._instances.append(self)# Object's class instance IDself.__class__._object_count+=1ifobj_idisNone:obj_id=self.__class__._object_countself.id=obj_id# User coordinatesself.coordinates_trace=[]self.coordinates=None# List of applications accessed by the userself.applications=[]# Reference to the base station the user is connected toself.base_station=None# User access metadataself.making_requests={}self.access_patterns={}# User mobility modelself.mobility_model=Noneself.mobility_model_parameters={}# List of metadata from applications accessed by the userself.communication_paths={}self.delays={}self.delay_slas={}# Model-specific attributes (defined inside the model's "initialize()" method)self.model=Noneself.unique_id=None
collect()
Method that collects a set of metrics for the object.
defcollect(self)->dict:"""Method that collects a set of metrics for the object. Returns: metrics (dict): Object metrics. """access_history={}forappinself.applications:access_history[str(app.id)]=self.access_patterns[str(app.id)].historymetrics={"Instance ID":self.id,"Coordinates":self.coordinates,"Base Station":f"{self.base_station} ({self.base_station.coordinates})"ifself.base_stationelseNone,"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),}returnmetrics
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 [].
defset_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 applicationifappinself.communication_paths:path=[[NetworkSwitch.find_by_id(i)foriinp]forpinself.communication_paths[str(app.id)]]topology._release_communication_path(communication_path=path,app=app)# Defining communication pathiflen(communication_path)>0:self.communication_paths[str(app.id)]=communication_pathelse:self.communication_paths[str(app.id)]=[]service_hosts_base_stations=[service.server.base_stationforserviceinapp.servicesifservice.server]communication_chain=[self.base_station]+service_hosts_base_stations# Defining a set of links to connect the items in the application's service chainforiinrange(len(communication_chain)-1):# Defining origin and target nodesorigin=communication_chain[i]target=communication_chain[i+1]# Finding and storing the best communication path between the origin and target nodesiforigin==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 pathself.communication_paths[str(app.id)].append([network_switch.idfornetwork_switchinpath])# Computing the new demand of chosen linkspath=[[NetworkSwitch.find_by_id(i)foriinp]forpinself.communication_paths[str(app.id)]]topology._allocate_communication_path(communication_path=path,app=app)# Computing application's delayself._compute_delay(app=app,metric="latency")returnself.communication_paths[str(app.id)]
step()
Method that executes the events involving the object at each time step.
defstep(self):"""Method that executes the events involving the object at each time step."""# Updating user accesscurrent_step=self.model.schedule.steps+1forappinself.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.ifself.making_requests[str(app.id)][str(current_step)]==True:iflen([sforsinapp.servicesifs._available])==len(app.services):last_access["access_time"]+=1else:last_access["waiting_time"]+=1# Updating user's making requests attribute for the next time stepifcurrent_step+1>=last_access["start"]andcurrent_step+1<=last_access["end"]:self.making_requests[str(app.id)][str(current_step+1)]=Trueelse:self.making_requests[str(app.id)][str(current_step+1)]=False# Creating new access request if neededifcurrent_step+1==last_access["next_access"]:self.making_requests[str(app.id)][str(current_step+1)]=Trueself.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 simulatoriflen(self.coordinates_trace)<=self.model.schedule.steps:self.mobility_model(self)# Updating user's locationifself.coordinates!=self.coordinates_trace[self.model.schedule.steps]:self.coordinates=self.coordinates_trace[self.model.schedule.steps]# Connecting the user to the closest base stationself.base_station=BaseStation.find_by(attribute_name="coordinates",attribute_value=self.coordinates)forapplicationinself.applications:# Only updates the routing path of apps available (i.e., whose services are available)services_available=len([sforsinapplication.servicesifs._available])ifservices_available==len(application.services):# Recomputing user communication pathsself.set_communication_path(app=application)else:self.communication_paths[str(application.id)]=[]self._compute_delay(app=application)