from typing import Dict, Union
from episimmer.read_file import ReadAgents
from episimmer.world import World
[docs]class AgentVD():
"""
Base class for implementing Agent Vulnerability Detection modules.
Args:
world_obj: World object of simulation
"""
def __init__(self, world_obj: World):
self.world_obj: World = world_obj
self.agent_scores: Dict[Union[str, int], float] = {}
[docs] def init_scores(self) -> None:
"""
Initialises scores for all agents with value 0.0.
"""
agents_obj = ReadAgents(self.world_obj.agents_filename,
self.world_obj.config_obj)
for agent_index in agents_obj.agents.keys():
self.agent_scores[agent_index] = 0.0
[docs] def one_run(self, *args, **kwargs) -> object:
"""
Executes a single run of the agent vulnerability detection module.
"""
raise NotImplementedError
[docs] def update_agent_scores(self, *args, **kwargs) -> None:
"""
Function to update the agent scores.
"""
raise NotImplementedError
[docs] def run_detection(self) -> None:
"""
Function to run the complete agent vulnerability detection module. This function will ideally call the one_run
function multiple times.
"""
raise NotImplementedError
[docs] def get_max_score_agents(self, n: int) -> Dict[str, float]:
"""
Function to return the n maximum agent scores as a dictionary.
Args:
n: Number of elements to be returned
Returns:
A dictionary mapping agent keys to agent scores. Scores are ordered in descending order
"""
res_max = {
key: round(value, 4)
for key, value in sorted(self.agent_scores.items(),
key=lambda x: (x[1], x[0]),
reverse=True)[:n]
}
return res_max
[docs] def get_min_score_agents(self, n: int) -> Dict[str, float]:
"""
Function to return the n minimum agent scores as a dictionary.
Args:
n: Number of elements to be returned
Returns:
A dictionary mapping agent keys to agent scores. Scores are ordered in ascending order
"""
res_min = {
key: round(value, 4)
for key, value in sorted(self.agent_scores.items(),
key=lambda x: (x[1], x[0]))[:n]
}
return res_min
[docs] def print_default_output(self, n: int) -> None:
"""
Prints the n maximum and n minimum agent scores.
Args:
n: Number of elements to be printed
"""
print(self.get_max_score_agents(n))
print(self.get_min_score_agents(n))
[docs]class VulnerableAgent(AgentVD):
"""
Class for implementing Vulnerable Agent modules.
Args:
world_obj: World object of simulation
"""
def __init__(self, world_obj: World):
super().__init__(world_obj)
self.type: str = 'Vulnerable Agents'
[docs]class AgentVulnerability(AgentVD):
"""
Class for implementing Agent Vulnerability modules.
Args:
world_obj: World object of simulation
"""
def __init__(self, world_obj: World):
super().__init__(world_obj)
self.type: str = 'Agent Vulnerability'
[docs] def remove_agents(self, *args, **kwargs) -> None:
"""
Function used to remove agents from the list of valid agents.
"""
raise NotImplementedError
[docs]class EventVD:
"""
Base class for implementing Event Vulnerability Detection modules.
Args:
world_obj: World object of simulation
"""
def __init__(self, world_obj: World):
self.world_obj: World = world_obj
self.event_scores: Dict[str, float] = {}
[docs] def init_scores(self) -> None:
"""
Initialises scores for all events..
"""
raise NotImplementedError
[docs] def one_run(self, *args, **kwargs) -> None:
"""
Executes a single run of the event vulnerability detection module.
"""
raise NotImplementedError
[docs] def update_event_scores(self, *args, **kwargs) -> None:
"""
Function to update the event scores.
"""
raise NotImplementedError
[docs] def run_detection(self) -> None:
"""
Function to run the complete event vulnerability detection module. This function will ideally call the one_run
function multiple times.
"""
raise NotImplementedError
[docs] def get_max_score_events(self, n: int) -> Dict[str, float]:
"""
Function to return the n maximum event scores as a dictionary.
Args:
n: Number of elements to be returned
Returns:
A dictionary mapping event keys to event scores. Scores are ordered in descending order
"""
res_max = {
key: round(value, 4)
for key, value in sorted(self.event_scores.items(),
key=lambda x: (x[1], x[0]),
reverse=True)[:n]
}
return res_max
[docs] def get_min_score_events(self, n: int) -> Dict[str, float]:
"""
Function to return the n minimum event scores as a dictionary.
Args:
n: Number of elements to be returned
Returns:
A dictionary mapping event keys to event scores. Scores are ordered in ascending order
"""
res_min = {
key: round(value, 4)
for key, value in sorted(self.event_scores.items(),
key=lambda x: (x[1], x[0]))[:n]
}
return res_min
[docs] def print_default_output(self, n: int) -> None:
"""
Prints the n maximum and n minimum event scores.
Args:
n: Number of elements to be printed
"""
print(self.get_max_score_events(n))
print(self.get_min_score_events(n))
class EventVulnerability(EventVD):
"""
Class for implementing Event Vulnerability modules.
Args:
world_obj: World object of simulation
"""
def __init__(self, world_obj: World):
super().__init__(world_obj)
self.type: str = 'Event Vulnerability'
def select_event(self, *args, **kwargs) -> None:
"""
Selects event to run vulnerability detection.
"""
raise NotImplementedError