Skip to content

Sensor Engine

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 
 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
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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@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

ISensorEngine

Bases: ABC

Abstract base class representing a sensor engine.

The sensor engine manages the lifecycle of sensors, including their creation, retrieval, and termination. It serves as a central point for interacting with various sensors within the system.

Source code in gamms/typing/sensor_engine.py 
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
class ISensorEngine(ABC):
    """
    Abstract base class representing a sensor engine.

    The sensor engine manages the lifecycle of sensors, including their creation, retrieval,
    and termination. It serves as a central point for interacting with various sensors
    within the system.
    """

    @abstractmethod
    def create_sensor(self, sensor_id: str, sensor_type: SensorType, **kwargs: Dict[str, Any]) -> ISensor:
        """
        Create a new sensor of the specified type.

        This method initializes a sensor based on the provided type and data, and registers
        it within the sensor engine for management.

        Args:
            sensor_id (str): The unique identifier for the sensor to be created.
            sensor_type (SensorType): The type of sensor to create.

        Kwargs:
            **kwargs: Additional keyword arguments for sensor initialization.

        Returns:
            ISensor: The newly created sensor instance.

        Raises:
            ValueError: If the sensor_type is unsupported or sensor_data is invalid.
            RuntimeError: If the sensor cannot be created due to system constraints.
        """
        pass

    @abstractmethod
    def get_sensor(self, sensor_id: Any) -> ISensor:
        """
        Retrieve a sensor by its unique identifier.

        This method fetches the sensor instance corresponding to the provided sensor_id.

        Args:
            sensor_id (Any): The unique identifier of the sensor to retrieve.

        Returns:
            ISensor: The sensor instance associated with the provided sensor_id.

        Raises:
            KeyError: If no sensor with the specified sensor_id exists.
            TypeError: If the sensor_id is of an incorrect type.
        """
        pass

    @abstractmethod
    def add_sensor(self, sensor: ISensor) -> None:
        """
        Add a sensor to the sensor engine.

        This method registers an existing sensor instance within the sensor engine for management.

        Args:
            sensor (ISensor): The sensor instance to add.

        Raises:
            TypeError: If the provided sensor is not an instance of ISensor.
            ValueError: If the sensor is already registered.
        """
        pass

    @abstractmethod
    def terminate(self) -> None:
        """
        Terminate the sensor engine and perform necessary cleanup operations.

        This method gracefully shuts down the sensor engine, ensuring that all sensors
        are properly terminated and that resources are released.

        Raises:
            RuntimeError: If the sensor engine fails to terminate gracefully.
            IOError: If there are issues during the cleanup process.
        """
        pass

    @abstractmethod
    def custom(self, name: str) -> Callable[[ISensor], ISensor]:
        """
        Decorator for custom sensor
        For example, if the user create a new custom sensor, add a new type to SensorType enum
        and implement the new sensor class, the user can add the custom sensor to the sensor engine
        using this method.
        """
        pass

add_sensor(sensor) abstractmethod

Add a sensor to the sensor engine.

This method registers an existing sensor instance within the sensor engine for management.

Parameters:

Name Type Description Default
sensor ISensor

The sensor instance to add.

 required

Raises:

Type Description
TypeError

If the provided sensor is not an instance of ISensor.
ValueError

If the sensor is already registered.
Source code in gamms/typing/sensor_engine.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@abstractmethod
def add_sensor(self, sensor: ISensor) -> None:
    """
    Add a sensor to the sensor engine.

    This method registers an existing sensor instance within the sensor engine for management.

    Args:
        sensor (ISensor): The sensor instance to add.

    Raises:
        TypeError: If the provided sensor is not an instance of ISensor.
        ValueError: If the sensor is already registered.
    """
    pass

create_sensor(sensor_id, sensor_type, **kwargs) abstractmethod

Create a new sensor of the specified type.

This method initializes a sensor based on the provided type and data, and registers it within the sensor engine for management.

Parameters:

Name Type Description Default
sensor_id str

The unique identifier for the sensor to be created.

 required
sensor_type SensorType

The type of sensor to create.

 required
Kwargs

**kwargs: Additional keyword arguments for sensor initialization.

Returns:

Name Type Description
ISensor ISensor

The newly created sensor instance.

Raises:

Type Description
ValueError

If the sensor_type is unsupported or sensor_data is invalid.
RuntimeError

If the sensor cannot be created due to system constraints.
Source code in gamms/typing/sensor_engine.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@abstractmethod
def create_sensor(self, sensor_id: str, sensor_type: SensorType, **kwargs: Dict[str, Any]) -> ISensor:
    """
    Create a new sensor of the specified type.

    This method initializes a sensor based on the provided type and data, and registers
    it within the sensor engine for management.

    Args:
        sensor_id (str): The unique identifier for the sensor to be created.
        sensor_type (SensorType): The type of sensor to create.

    Kwargs:
        **kwargs: Additional keyword arguments for sensor initialization.

    Returns:
        ISensor: The newly created sensor instance.

    Raises:
        ValueError: If the sensor_type is unsupported or sensor_data is invalid.
        RuntimeError: If the sensor cannot be created due to system constraints.
    """
    pass

custom(name) abstractmethod

Decorator for custom sensor For example, if the user create a new custom sensor, add a new type to SensorType enum and implement the new sensor class, the user can add the custom sensor to the sensor engine using this method.

Source code in gamms/typing/sensor_engine.py 
228
229
230
231
232
233
234
235
236
@abstractmethod
def custom(self, name: str) -> Callable[[ISensor], ISensor]:
    """
    Decorator for custom sensor
    For example, if the user create a new custom sensor, add a new type to SensorType enum
    and implement the new sensor class, the user can add the custom sensor to the sensor engine
    using this method.
    """
    pass

get_sensor(sensor_id) abstractmethod

Retrieve a sensor by its unique identifier.

This method fetches the sensor instance corresponding to the provided sensor_id.

Parameters:

Name Type Description Default
sensor_id Any

The unique identifier of the sensor to retrieve.

 required

Returns:

Name Type Description
ISensor ISensor

The sensor instance associated with the provided sensor_id.

Raises:

Type Description
KeyError

If no sensor with the specified sensor_id exists.
TypeError

If the sensor_id is of an incorrect type.
Source code in gamms/typing/sensor_engine.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@abstractmethod
def get_sensor(self, sensor_id: Any) -> ISensor:
    """
    Retrieve a sensor by its unique identifier.

    This method fetches the sensor instance corresponding to the provided sensor_id.

    Args:
        sensor_id (Any): The unique identifier of the sensor to retrieve.

    Returns:
        ISensor: The sensor instance associated with the provided sensor_id.

    Raises:
        KeyError: If no sensor with the specified sensor_id exists.
        TypeError: If the sensor_id is of an incorrect type.
    """
    pass

terminate() abstractmethod

Terminate the sensor engine and perform necessary cleanup operations.

This method gracefully shuts down the sensor engine, ensuring that all sensors are properly terminated and that resources are released.

Raises:

Type Description
RuntimeError

If the sensor engine fails to terminate gracefully.
IOError

If there are issues during the cleanup process.
Source code in gamms/typing/sensor_engine.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
@abstractmethod
def terminate(self) -> None:
    """
    Terminate the sensor engine and perform necessary cleanup operations.

    This method gracefully shuts down the sensor engine, ensuring that all sensors
    are properly terminated and that resources are released.

    Raises:
        RuntimeError: If the sensor engine fails to terminate gracefully.
        IOError: If there are issues during the cleanup process.
    """
    pass

SensorType

Bases: Enum

Enumeration of different sensor types.

Attributes:

Name Type Description
CUSTOM Enum

Dummy Sensor type when the user create a custom sensor. Data Representation depends on the custom sensor implementation.
NEIGHBOR Enum

Sensor type for detecting neighboring entities. Data Representation (List[int]): List of node identifiers representing neighbors.
MAP Enum

Sensor type for map-related data. Data Representation (Dict[str, Dict[int, Union[Node, OSMEdge]]): Keys nodes and edges give respective node and edge data.
AGENT Enum

Sensor type for agent locations. Data Representation (Dict[str, int]): Dictionary mapping agent names to node identifiers.
RANGE Enum

Sensor type for range-based data. Data Representation (Dict[str, Dict[int, Union[Node, OSMEdge]]): Keys nodes and edges give respective node and edge data. Range only version of MAP.
ARC Enum

Sensor type for arc-based data. Data Representation (Dict[str, Dict[int, Union[Node, OSMEdge]]): Keys nodes and edges give respective node and edge data. Range and Fov version of MAP.
AGENT_RANGE Enum

Sensor type for agent range data. Data Representation (Dict[str, int]): Dictionary mapping agent names to node identifiers. Range only version of AGENT.
AGENT_ARC Enum

Sensor type for agent arc data. Data Representation (Dict[str, int]): Dictionary mapping agent names to node identifiers. Range and Fov version of AGENT.
AERIAL Enum

Sensor type for aerial map data. Data Representation (Dict[str, Union[Dict[int, Node], List[OSMEdge]]]): Keys nodes and edges give respective node and edge data. Range and Fov version of MAP for aerial agents.
AERIAL_AGENT Enum

Sensor type for aerial agent data. Data Representation (Dict[str, Tuple[AgentType, Tuple[float, float, float]]]): Dictionary mapping agent names to (x, y, z) coordinates.
Source code in gamms/typing/sensor_engine.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
class SensorType(Enum):
    """
    Enumeration of different sensor types.

    Attributes:
        CUSTOM (Enum): Dummy Sensor type when the user create a custom sensor.
            Data Representation depends on the custom sensor implementation.
        NEIGHBOR (Enum): Sensor type for detecting neighboring entities.
            Data Representation (`List[int]`): List of node identifiers representing neighbors.
        MAP (Enum): Sensor type for map-related data.
            Data Representation (`Dict[str, Dict[int, Union[Node, OSMEdge]]`): Keys nodes and edges give respective node and edge data.
        AGENT (Enum): Sensor type for agent locations.
            Data Representation (`Dict[str, int]`): Dictionary mapping agent names to node identifiers.
        RANGE (Enum): Sensor type for range-based data.
            Data Representation (`Dict[str, Dict[int, Union[Node, OSMEdge]]`): Keys nodes and edges give respective node and edge data. Range only version of MAP.
        ARC (Enum): Sensor type for arc-based data.
            Data Representation (`Dict[str, Dict[int, Union[Node, OSMEdge]]`): Keys nodes and edges give respective node and edge data. Range and Fov version of MAP.
        AGENT_RANGE (Enum): Sensor type for agent range data.
            Data Representation (`Dict[str, int]`): Dictionary mapping agent names to node identifiers. Range only version of AGENT.
        AGENT_ARC (Enum): Sensor type for agent arc data.
            Data Representation (`Dict[str, int]`): Dictionary mapping agent names to node identifiers. Range and Fov version of AGENT.
        AERIAL (Enum): Sensor type for aerial map data.
            Data Representation (`Dict[str, Union[Dict[int, Node], List[OSMEdge]]]`): Keys nodes and edges give respective node and edge data. Range and Fov version of MAP for aerial agents.
        AERIAL_AGENT (Enum): Sensor type for aerial agent data.
            Data Representation (`Dict[str, Tuple[AgentType, Tuple[float, float, float]]]`): Dictionary mapping agent names to (x, y, z) coordinates.

    """
    CUSTOM = 0
    NEIGHBOR = 1
    MAP = 2
    AGENT = 3
    RANGE = 4
    ARC = 5
    AGENT_RANGE = 6
    AGENT_ARC = 7
    AERIAL = 8
    AERIAL_AGENT = 9