Skip to content

Agent Engine

AgentType

Bases: IntEnum

Enum representing different types of agents.

Source code in gamms/typing/agent_engine.py
class AgentType(IntEnum):
    """
    Enum representing different types of agents.
    """
    BASIC = 0
    AERIAL = 1

IAerialAgent

Bases: IAgent

Abstract base class representing an aerial agent in the system.

This class extends the basic agent functionality to include aerial-specific behaviors.

Requires a start node ID and speed for initialization.

Source code in gamms/typing/agent_engine.py
class IAerialAgent(IAgent):
    """
    Abstract base class representing an aerial agent in the system.

    This class extends the basic agent functionality to include aerial-specific behaviors.

    Requires a start node ID and speed for initialization.
    """

    @property
    @abstractmethod
    def quat(self) -> Tuple[float, float, float, float]:
        """
        Get the quaternion representation of the agent's orientation.
        Formatted as (w, x, y, z).

        Returns:
            Tuple[float, float, float, float]: The quaternion representing the agent's orientation.
        """
        pass

    @property
    @abstractmethod
    def orientation(self) -> Tuple[float, float]:
        """
        Get the orientation of the agent in the x-y plane.

        Returns:
            Tuple[float, float]: The current orientation of the agent as a tuple (sin, cos).
        """
        pass

    @property
    @abstractmethod
    def position(self) -> Tuple[float, float, float]:
        """
        Get the current position of the agent in 3D space.

        Returns:
            Tuple[float, float, float]: The current position of the agent as a tuple (x, y, z).
        """
        pass

    @abstractmethod
    def set_state(self):
        """
        Update the agent's position and orientation.
        Action should be a 3d direction tuple. It will be normalized to a unit vector and multiplied by the agent's speed.

        Action can also be a dictionary with 'direction' key to specify the direction and have 'quat' key to specify the quaternion orientation.

        Raises:
            KeyError: If action is not found in the agent's state.
            TypeError: If action is not a tuple or dict.
            ValueError: If action dict does not contain 'direction' key
            ValueError: If quat is not a tuple of (w, x, y, z).
        """
        pass

orientation abstractmethod property

Get the orientation of the agent in the x-y plane.

Returns:

Type Description
Tuple[float, float]

Tuple[float, float]: The current orientation of the agent as a tuple (sin, cos).

position abstractmethod property

Get the current position of the agent in 3D space.

Returns:

Type Description
Tuple[float, float, float]

Tuple[float, float, float]: The current position of the agent as a tuple (x, y, z).

quat abstractmethod property

Get the quaternion representation of the agent's orientation. Formatted as (w, x, y, z).

Returns:

Type Description
Tuple[float, float, float, float]

Tuple[float, float, float, float]: The quaternion representing the agent's orientation.

set_state() abstractmethod

Update the agent's position and orientation. Action should be a 3d direction tuple. It will be normalized to a unit vector and multiplied by the agent's speed.

Action can also be a dictionary with 'direction' key to specify the direction and have 'quat' key to specify the quaternion orientation.

Raises:

Type Description
KeyError

If action is not found in the agent's state.

TypeError

If action is not a tuple or dict.

ValueError

If action dict does not contain 'direction' key

ValueError

If quat is not a tuple of (w, x, y, z).

Source code in gamms/typing/agent_engine.py
@abstractmethod
def set_state(self):
    """
    Update the agent's position and orientation.
    Action should be a 3d direction tuple. It will be normalized to a unit vector and multiplied by the agent's speed.

    Action can also be a dictionary with 'direction' key to specify the direction and have 'quat' key to specify the quaternion orientation.

    Raises:
        KeyError: If action is not found in the agent's state.
        TypeError: If action is not a tuple or dict.
        ValueError: If action dict does not contain 'direction' key
        ValueError: If quat is not a tuple of (w, x, y, z).
    """
    pass

IAgent

Bases: ABC

Abstract base class representing an agent in the system.

Source code in gamms/typing/agent_engine.py
class IAgent(ABC):
    """
    Abstract base class representing an agent in the system.

    """

    @property
    @abstractmethod
    def name(self) -> str:
        """
        Get the name identifier of the agent.
        """
        pass

    @property
    @abstractmethod
    def current_node_id(self) -> int:
        """
        Get the current node ID of the agent.
        """
        pass

    @property
    @abstractmethod
    def prev_node_id(self) -> int:
        """
        Get the previous node ID of the agent.
        """
        pass

    @property
    @abstractmethod
    def type(self) -> AgentType:
        """
        Get the type of the agent.

        Returns:
            AgentType: The type of the agent (e.g., BASIC, AERIAL).
        """
        pass

    @property
    @abstractmethod
    def orientation(self) -> Tuple[float, float]:
        """
        Get the orientation of the agent.

        Returns:
            Tuple[float, float]: The current orientation of the agent.
        """
        pass

    @property
    @abstractmethod
    def state(self) -> Dict[str, Any]:
        """
        Get the current state of the agent.

        Returns:
            Dict[str, Any]: The current state data of the agent.
        """
        pass

    @property
    @abstractmethod
    def strategy(self) -> Optional[Callable[[Dict[str, Any]], None]]:
        """
        Get the current strategy of the agent.

        Returns:
            Optional[Callable[[Dict[str, Any]], None]]: The current strategy function or None if no strategy is set.
        """
        pass

    @abstractmethod
    def step(self):
        """
        Execute a single operational step of the agent.

        This method should contain the logic that defines the agent's behavior
        during one iteration or time step in the system.
        """
        pass

    @abstractmethod
    def get_state(self) -> Dict[str, Any]:
        """
        Retrieve the current state of the agent.

        Returns:
            Dict[str, Any]: The current state data of the agent, structure depends on implementation.
        """
        pass

    @abstractmethod
    def set_state(self):
        """
        Update the agent's state.

        Raises:
            KeyError: If action is not found in the agent's state.
            KeyError: If action is an int but not a valid node ID.
            TypeError: If action is not an int or dict.
            ValueError: If action dict does not contain 'node_id' key
            ValueError: If orientation is not a tuple of (sin, cos).
        """
        pass

    @abstractmethod
    def register_sensor(self, name: str, sensor: ISensor):
        """
        Register a sensor with the agent.

        Sensors can be used by the agent to perceive the environment or gather data.

        Args:
            name (str): The unique name identifier for the sensor.
            sensor (ISensor): The sensor instance or object to be registered.
        """
        pass

    @abstractmethod
    def deregister_sensor(self, name: str):
        """
        Deregister a sensor from the agent.

        Args:
            name (str): The unique name identifier for the sensor to be deregistered.
        """
        pass

    @abstractmethod
    def register_strategy(self, strategy: Callable[[Dict[str, Any]], None]):
        """
        Register a strategy with the agent.

        Strategies define the decision-making or action-planning mechanisms for the agent.

        Args:
            strategy (Callable[[Dict[str, Any]], None]): The strategy instance or object to be registered.
        """
        pass

current_node_id abstractmethod property

Get the current node ID of the agent.

name abstractmethod property

Get the name identifier of the agent.

orientation abstractmethod property

Get the orientation of the agent.

Returns:

Type Description
Tuple[float, float]

Tuple[float, float]: The current orientation of the agent.

prev_node_id abstractmethod property

Get the previous node ID of the agent.

state abstractmethod property

Get the current state of the agent.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The current state data of the agent.

strategy abstractmethod property

Get the current strategy of the agent.

Returns:

Type Description
Optional[Callable[[Dict[str, Any]], None]]

Optional[Callable[[Dict[str, Any]], None]]: The current strategy function or None if no strategy is set.

type abstractmethod property

Get the type of the agent.

Returns:

Name Type Description
AgentType AgentType

The type of the agent (e.g., BASIC, AERIAL).

deregister_sensor(name) abstractmethod

Deregister a sensor from the agent.

Parameters:

Name Type Description Default
name str

The unique name identifier for the sensor to be deregistered.

required
Source code in gamms/typing/agent_engine.py
@abstractmethod
def deregister_sensor(self, name: str):
    """
    Deregister a sensor from the agent.

    Args:
        name (str): The unique name identifier for the sensor to be deregistered.
    """
    pass

get_state() abstractmethod

Retrieve the current state of the agent.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The current state data of the agent, structure depends on implementation.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def get_state(self) -> Dict[str, Any]:
    """
    Retrieve the current state of the agent.

    Returns:
        Dict[str, Any]: The current state data of the agent, structure depends on implementation.
    """
    pass

register_sensor(name, sensor) abstractmethod

Register a sensor with the agent.

Sensors can be used by the agent to perceive the environment or gather data.

Parameters:

Name Type Description Default
name str

The unique name identifier for the sensor.

required
sensor ISensor

The sensor instance or object to be registered.

required
Source code in gamms/typing/agent_engine.py
@abstractmethod
def register_sensor(self, name: str, sensor: ISensor):
    """
    Register a sensor with the agent.

    Sensors can be used by the agent to perceive the environment or gather data.

    Args:
        name (str): The unique name identifier for the sensor.
        sensor (ISensor): The sensor instance or object to be registered.
    """
    pass

register_strategy(strategy) abstractmethod

Register a strategy with the agent.

Strategies define the decision-making or action-planning mechanisms for the agent.

Parameters:

Name Type Description Default
strategy Callable[[Dict[str, Any]], None]

The strategy instance or object to be registered.

required
Source code in gamms/typing/agent_engine.py
@abstractmethod
def register_strategy(self, strategy: Callable[[Dict[str, Any]], None]):
    """
    Register a strategy with the agent.

    Strategies define the decision-making or action-planning mechanisms for the agent.

    Args:
        strategy (Callable[[Dict[str, Any]], None]): The strategy instance or object to be registered.
    """
    pass

set_state() abstractmethod

Update the agent's state.

Raises:

Type Description
KeyError

If action is not found in the agent's state.

KeyError

If action is an int but not a valid node ID.

TypeError

If action is not an int or dict.

ValueError

If action dict does not contain 'node_id' key

ValueError

If orientation is not a tuple of (sin, cos).

Source code in gamms/typing/agent_engine.py
@abstractmethod
def set_state(self):
    """
    Update the agent's state.

    Raises:
        KeyError: If action is not found in the agent's state.
        KeyError: If action is an int but not a valid node ID.
        TypeError: If action is not an int or dict.
        ValueError: If action dict does not contain 'node_id' key
        ValueError: If orientation is not a tuple of (sin, cos).
    """
    pass

step() abstractmethod

Execute a single operational step of the agent.

This method should contain the logic that defines the agent's behavior during one iteration or time step in the system.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def step(self):
    """
    Execute a single operational step of the agent.

    This method should contain the logic that defines the agent's behavior
    during one iteration or time step in the system.
    """
    pass

IAgentEngine

Bases: ABC

Abstract base class representing the engine that manages agents.

The engine is responsible for creating, managing, and terminating agents, as well as facilitating interactions between them.

Source code in gamms/typing/agent_engine.py
class IAgentEngine(ABC):
    """
    Abstract base class representing the engine that manages agents.

    The engine is responsible for creating, managing, and terminating agents,
    as well as facilitating interactions between them.
    """

    @abstractmethod
    def create_iter(self) -> Iterable[IAgent]:
        """
        Create an iterator for processing agent steps.

        Returns:
            Iterable[IAgent]: An iterator object over all agents.
        """
        pass

    @abstractmethod
    def create_agent(self, name:str, **kwargs: Dict[str, Any]) -> IAgent:
        """
        Instantiate a new agent within the engine.

        Args:
            name (str): The unique name identifier for the agent.
            **kwargs: Additional keyword arguments for agent initialization.

        Returns:
            IAgent: The newly created agent instance.

        Raises:
            ValueError: If an agent with the same name already exists.
            KeyError: If start_node_id is not provided in kwargs.
        """
        pass

    @abstractmethod
    def terminate(self):
        """
        Terminate the agent engine and perform necessary cleanup operations.

        This method should ensure that all resources are properly released and
        that agents are gracefully shut down.
        """
        pass

    @abstractmethod
    def get_agent(self, name: str) -> IAgent:
        """
        Retrieve an agent by its name.

        Args:
            name (str): The unique name identifier of the agent to retrieve.

        Returns:
            IAgent: The agent instance corresponding to the provided name.

        Raises:
            KeyError: If no agent with the specified name exists.
        """
        pass

    @abstractmethod
    def delete_agent(self, name: str) -> None:
        """
        Delete an agent by its name.

        Args:
            name (str): The unique name identifier of the agent to retrieve.

        Returns:
            None

        Raises:
            Logs a warning and does nothing if no agent with the specified name exists.
        """
        pass

create_agent(name, **kwargs) abstractmethod

Instantiate a new agent within the engine.

Parameters:

Name Type Description Default
name str

The unique name identifier for the agent.

required
**kwargs Dict[str, Any]

Additional keyword arguments for agent initialization.

{}

Returns:

Name Type Description
IAgent IAgent

The newly created agent instance.

Raises:

Type Description
ValueError

If an agent with the same name already exists.

KeyError

If start_node_id is not provided in kwargs.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def create_agent(self, name:str, **kwargs: Dict[str, Any]) -> IAgent:
    """
    Instantiate a new agent within the engine.

    Args:
        name (str): The unique name identifier for the agent.
        **kwargs: Additional keyword arguments for agent initialization.

    Returns:
        IAgent: The newly created agent instance.

    Raises:
        ValueError: If an agent with the same name already exists.
        KeyError: If start_node_id is not provided in kwargs.
    """
    pass

create_iter() abstractmethod

Create an iterator for processing agent steps.

Returns:

Type Description
Iterable[IAgent]

Iterable[IAgent]: An iterator object over all agents.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def create_iter(self) -> Iterable[IAgent]:
    """
    Create an iterator for processing agent steps.

    Returns:
        Iterable[IAgent]: An iterator object over all agents.
    """
    pass

delete_agent(name) abstractmethod

Delete an agent by its name.

Parameters:

Name Type Description Default
name str

The unique name identifier of the agent to retrieve.

required

Returns:

Type Description
None

None

Source code in gamms/typing/agent_engine.py
@abstractmethod
def delete_agent(self, name: str) -> None:
    """
    Delete an agent by its name.

    Args:
        name (str): The unique name identifier of the agent to retrieve.

    Returns:
        None

    Raises:
        Logs a warning and does nothing if no agent with the specified name exists.
    """
    pass

get_agent(name) abstractmethod

Retrieve an agent by its name.

Parameters:

Name Type Description Default
name str

The unique name identifier of the agent to retrieve.

required

Returns:

Name Type Description
IAgent IAgent

The agent instance corresponding to the provided name.

Raises:

Type Description
KeyError

If no agent with the specified name exists.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def get_agent(self, name: str) -> IAgent:
    """
    Retrieve an agent by its name.

    Args:
        name (str): The unique name identifier of the agent to retrieve.

    Returns:
        IAgent: The agent instance corresponding to the provided name.

    Raises:
        KeyError: If no agent with the specified name exists.
    """
    pass

terminate() abstractmethod

Terminate the agent engine and perform necessary cleanup operations.

This method should ensure that all resources are properly released and that agents are gracefully shut down.

Source code in gamms/typing/agent_engine.py
@abstractmethod
def terminate(self):
    """
    Terminate the agent engine and perform necessary cleanup operations.

    This method should ensure that all resources are properly released and
    that agents are gracefully shut down.
    """
    pass

ISensor

Bases: ABC

Abstract base class representing a generic sensor.

Sensors are responsible for collecting data related to specific aspects of the system. Each sensor has a type and maintains its own data state.

Attributes:

Name Type Description
type SensorType

The type of the sensor.

data Dict[str, Any]

The data collected or maintained by the sensor.

Source code in gamms/typing/sensor_engine.py
class ISensor(ABC):
    """
    Abstract base class representing a generic sensor.

    Sensors are responsible for collecting data related to specific aspects of the system.
    Each sensor has a type and maintains its own data state.

    Attributes:
        type (SensorType): The type of the sensor.
        data (Dict[str, Any]): The data collected or maintained by the sensor.
    """

    @property
    @abstractmethod
    def sensor_id(self) -> str:
        """
        Get the unique identifier of the sensor.

        Returns:
            str: The unique identifier of the sensor.
        """
        pass

    @property
    @abstractmethod
    def type(self) -> SensorType:
        """
        Get the type of the sensor.

        Returns:
            SensorType: The type of the sensor.
        """
        pass

    @property
    @abstractmethod
    def data(self) -> Union[Dict[str, Any], List[int]]:
        """
        Get the current data maintained by the sensor.

        Returns:
            Dict[str, Any]: The data maintained by the sensor.
            List[int]: A list of node identifiers for the NEIGHBOR sensor type.
        """
        pass

    @abstractmethod
    def sense(self, node_id: int) -> None:
        """
        Perform sensing operations for a given node.

        This method collects data related to the specified node and returns the sensed information.

        Args:
            node_id (int): The unique identifier of the node to sense.

        Sensed Data type:
            Dict[str, Any]: A dictionary containing the sensed data.
            Only Neigbor sensor returns a list of node ids. List[int]

        Raises:
            ValueError: If the provided node_id is invalid.
            RuntimeError: If the sensing operation fails due to system constraints.
        """
        pass

    @abstractmethod
    def update(self, data: Dict[str, Any]) -> None:
        """
        Update the sensor's data.

        This method modifies the sensor's internal data based on the provided information.

        Args:
            data (Dict[str, Any]): A dictionary containing the data to update the sensor with.

        Raises:
            KeyError: If the provided data contains invalid keys.
            ValueError: If the provided data is malformed or incomplete.
        """
        pass

    @abstractmethod
    def set_owner(self, owner: Union[str, None]) -> None:
        """
        Set the owner of the sensor. Owner is a string that identifies the entity responsible for the sensor.
        Used for setting the owning agent.

        This method assigns a specific owner to the sensor, which can be used for identification
        or management purposes.

        Args:
            owner (str or None): The name of the owner to assign to the sensor.

        Raises:
            TypeError: If the provided owner is not a string.
            ValueError: If the provided owner is invalid or empty.
        """
        pass

data abstractmethod property

Get the current data maintained by the sensor.

Returns:

Type Description
Union[Dict[str, Any], List[int]]

Dict[str, Any]: The data maintained by the sensor.

Union[Dict[str, Any], List[int]]

List[int]: A list of node identifiers for the NEIGHBOR sensor type.

sensor_id abstractmethod property

Get the unique identifier of the sensor.

Returns:

Name Type Description
str str

The unique identifier of the sensor.

type abstractmethod property

Get the type of the sensor.

Returns:

Name Type Description
SensorType SensorType

The type of the sensor.

sense(node_id) abstractmethod

Perform sensing operations for a given node.

This method collects data related to the specified node and returns the sensed information.

Parameters:

Name Type Description Default
node_id int

The unique identifier of the node to sense.

required
Sensed Data type

Dict[str, Any]: A dictionary containing the sensed data. Only Neigbor sensor returns a list of node ids. List[int]

Raises:

Type Description
ValueError

If the provided node_id is invalid.

RuntimeError

If the sensing operation fails due to system constraints.

Source code in gamms/typing/sensor_engine.py
@abstractmethod
def sense(self, node_id: int) -> None:
    """
    Perform sensing operations for a given node.

    This method collects data related to the specified node and returns the sensed information.

    Args:
        node_id (int): The unique identifier of the node to sense.

    Sensed Data type:
        Dict[str, Any]: A dictionary containing the sensed data.
        Only Neigbor sensor returns a list of node ids. List[int]

    Raises:
        ValueError: If the provided node_id is invalid.
        RuntimeError: If the sensing operation fails due to system constraints.
    """
    pass

set_owner(owner) abstractmethod

Set the owner of the sensor. Owner is a string that identifies the entity responsible for the sensor. Used for setting the owning agent.

This method assigns a specific owner to the sensor, which can be used for identification or management purposes.

Parameters:

Name Type Description Default
owner str or None

The name of the owner to assign to the sensor.

required

Raises:

Type Description
TypeError

If the provided owner is not a string.

ValueError

If the provided owner is invalid or empty.

Source code in gamms/typing/sensor_engine.py
@abstractmethod
def set_owner(self, owner: Union[str, None]) -> None:
    """
    Set the owner of the sensor. Owner is a string that identifies the entity responsible for the sensor.
    Used for setting the owning agent.

    This method assigns a specific owner to the sensor, which can be used for identification
    or management purposes.

    Args:
        owner (str or None): The name of the owner to assign to the sensor.

    Raises:
        TypeError: If the provided owner is not a string.
        ValueError: If the provided owner is invalid or empty.
    """
    pass

update(data) abstractmethod

Update the sensor's data.

This method modifies the sensor's internal data based on the provided information.

Parameters:

Name Type Description Default
data Dict[str, Any]

A dictionary containing the data to update the sensor with.

required

Raises:

Type Description
KeyError

If the provided data contains invalid keys.

ValueError

If the provided data is malformed or incomplete.

Source code in gamms/typing/sensor_engine.py
@abstractmethod
def update(self, data: Dict[str, Any]) -> None:
    """
    Update the sensor's data.

    This method modifies the sensor's internal data based on the provided information.

    Args:
        data (Dict[str, Any]): A dictionary containing the data to update the sensor with.

    Raises:
        KeyError: If the provided data contains invalid keys.
        ValueError: If the provided data is malformed or incomplete.
    """
    pass