Skip to content

Documentation for Samoduleagent Module

SAModuleAgent

Bases: ABC

Abstract base class representing a Situational Awareness (SA) module agent.

This interface defines the essential methods that any SA agent must implement to participate in the suggestion and arbitration pipeline. Agents are responsible for registering themselves, suggesting actions in the form of commands, and notifying when all suggestions related to an event are complete.

Methods: - get_agent(): Return a unique identifier or name of the agent. - register_sa_agent(): Perform initialization or registration steps for the agent. - suggest_action(sac): Submit a suggested command (SACommand) for arbitration. - notify_all_suggestions_done(event_type): Indicate that all suggestions for a given event are complete.

Source code in nebula/core/situationalawareness/awareness/sautils/samoduleagent.py 
 6
 7
 8
 9
10
11
12
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
class SAModuleAgent(ABC):
    """
    Abstract base class representing a Situational Awareness (SA) module agent.

    This interface defines the essential methods that any SA agent must implement
    to participate in the suggestion and arbitration pipeline. Agents are responsible
    for registering themselves, suggesting actions in the form of commands, and
    notifying when all suggestions related to an event are complete.

    Methods:
    - get_agent(): Return a unique identifier or name of the agent.
    - register_sa_agent(): Perform initialization or registration steps for the agent.
    - suggest_action(sac): Submit a suggested command (SACommand) for arbitration.
    - notify_all_suggestions_done(event_type): Indicate that all suggestions for a given event are complete.
    """

    @abstractmethod
    async def get_agent(self) -> str:
        """
        Return the unique identifier or name of the agent.

        Returns:
            str: The identifier or label representing this SA agent.
        """
        raise NotImplementedError

    @abstractmethod
    async def register_sa_agent(self):
        """
        Perform initialization logic required to register this SA agent
        within the system (e.g., announcing its presence or preparing state).
        """
        raise NotImplementedError

    @abstractmethod
    async def suggest_action(self, sac: SACommand):
        """
        Submit a suggested action in the form of a SACommand for a given context.

        Parameters:
            sac (SACommand): The command proposed by the agent for execution.
        """
        raise NotImplementedError

    @abstractmethod
    async def notify_all_suggestions_done(self, event_type):
        """
        Notify that this agent has completed all its suggestions for a particular event.

        Parameters:
            event_type (Type[NodeEvent]): The type of the event for which suggestions are now complete.
        """
        raise NotImplementedError

get_agent() abstractmethod async

Return the unique identifier or name of the agent.

Returns:

Name Type Description
str str

The identifier or label representing this SA agent.
Source code in nebula/core/situationalawareness/awareness/sautils/samoduleagent.py 
22
23
24
25
26
27
28
29
30
@abstractmethod
async def get_agent(self) -> str:
    """
    Return the unique identifier or name of the agent.

    Returns:
        str: The identifier or label representing this SA agent.
    """
    raise NotImplementedError

notify_all_suggestions_done(event_type) abstractmethod async

Notify that this agent has completed all its suggestions for a particular event.

Parameters:

Name Type Description Default
event_type Type[NodeEvent]

The type of the event for which suggestions are now complete.

 required
Source code in nebula/core/situationalawareness/awareness/sautils/samoduleagent.py 
50
51
52
53
54
55
56
57
58
@abstractmethod
async def notify_all_suggestions_done(self, event_type):
    """
    Notify that this agent has completed all its suggestions for a particular event.

    Parameters:
        event_type (Type[NodeEvent]): The type of the event for which suggestions are now complete.
    """
    raise NotImplementedError

register_sa_agent() abstractmethod async

Perform initialization logic required to register this SA agent within the system (e.g., announcing its presence or preparing state).

Source code in nebula/core/situationalawareness/awareness/sautils/samoduleagent.py 
32
33
34
35
36
37
38
@abstractmethod
async def register_sa_agent(self):
    """
    Perform initialization logic required to register this SA agent
    within the system (e.g., announcing its presence or preparing state).
    """
    raise NotImplementedError

suggest_action(sac) abstractmethod async

Submit a suggested action in the form of a SACommand for a given context.

Parameters:

Name Type Description Default
sac SACommand

The command proposed by the agent for execution.

 required
Source code in nebula/core/situationalawareness/awareness/sautils/samoduleagent.py 
40
41
42
43
44
45
46
47
48
@abstractmethod
async def suggest_action(self, sac: SACommand):
    """
    Submit a suggested action in the form of a SACommand for a given context.

    Parameters:
        sac (SACommand): The command proposed by the agent for execution.
    """
    raise NotImplementedError