tbp.monty.frameworks.models.salience#

tbp.monty.frameworks.models.salience.motor_policy#

exception GoalCollocatedWithSensor[source]#

Bases: RuntimeError

Raised when a goal is collocated with a sensor.

class LookAtGoal(agent_id: AgentID, sensor_id: SensorID) → None[source]#

Bases: MotorPolicy

A policy that looks at a target.

This class assumes a system similar to a 2-DOF gimbal in which the “outer” part can yaw left/right about the y-axis and the “inner” part can pitch up/down about the x-axis. This setup is typical of our distant agent in which the agent turns left/right and sensor mounted to it looks up/down.

Note that this code only uses TurnLeft and LookUp. Turning right or looking down are performed using negative degrees with TurnLeft and LookUp, respectively.

__init__(agent_id: AgentID, sensor_id: SensorID) → None[source]#

Initialize the look at policy.

Parameters:

agent_id (AgentID (str)) – The agent ID
sensor_id (SensorID (str)) – The sensor ID

fixme_provide_motor_system(motor_system: ExperimentMotorSystem) → None[source]#

Provide access to the Motor System during initialization.

This is part of the work to remove reset() in favor or Hydra instantiation. It is used to provide a reference to the Motor System so the SurfacePolicy and its subclasses can override the motor_only_step property.

TODO: This whole mechanism is a hack for the benefit of SurfacePolicy et. al. What we should be doing is supporting more complex actions, like “follow surface in this direction,” whose details are left to the simulator.

Parameters:: motor_system (ExperimentMotorSystem) – The associated Motor System.
Return type:: None

load_state_dict(memento: Mapping[str, Any]) → None[source]#

Load the internal object state from a previously snapshot memento.

Parameters:: memento (Mapping[str, Any]) – State dict to load.
Return type:: None

reset() → None[source]#

Reset the internal state of this Motor Policy.

Return type:: None

state_dict() → Mapping[str, Any][source]#

Snapshot a memento representing the internal state of this object.

Return type:: Mapping[str, Any]
Returns:: State dict for logging and saving.

tbp.monty.frameworks.models.salience.on_object_observation#

class OnObjectObservation(center_location: npt.NDArray[np.float64] | None, locations: npt.NDArray[np.float64], salience: npt.NDArray[np.float32], on_object_map: npt.NDArray[np.bool_], location_map: npt.NDArray[np.float64]) → None[source]#

Bases: object

__init__(center_location: npt.NDArray[np.float64] | None, locations: npt.NDArray[np.float64], salience: npt.NDArray[np.float32], on_object_map: npt.NDArray[np.bool_], location_map: npt.NDArray[np.float64]) → None#

center_location: npt.NDArray[np.float64] | None#: 3D location corresponding to the central pixel, if it is on an object.

location_map: npt.NDArray[np.float64]#: H x W x 3 array indicating the 3D location of each pixel.

locations: npt.NDArray[np.float64]#: 3D locations corresponding to on-object pixels.

on_object_map: npt.NDArray[np.bool_]#: H x W boolean array indicating on-object pixels.

salience: npt.NDArray[np.float32]#: Salience values corresponding to on-object pixels.

on_object_observation(observation: SensorObservation, salience_map: numpy.ndarray) → OnObjectObservation[source]#

Convert all raw observation data into image format.

This function reformats the arrays in a raw observations dictionary so that they’re all indexable by image row and column indices. It also splits the semantic_3d array into 3D locations and an on-object/surface indicator array.

Parameters:

observation (SensorObservation) – A sensor observation.
salience_map (ndarray) – A salience map.

Return type:

OnObjectObservation

Returns:

The grid/matrix formatted (unraveled) on-object salience and location data, along with the location corresponding to the central pixel.

tbp.monty.frameworks.models.salience.return_inhibitor#

class DecayField(kernel_factory: DecayKernelFactory | None = None)[source]#

Bases: object

Manages a collection of decay kernels.

__init__(kernel_factory: DecayKernelFactory | None = None)[source]#

add(location: numpy.ndarray) → None[source]#

Add a kernel to the field.

Return type:: None

compute_weights(points: numpy.ndarray) → numpy.ndarray[source]#

Return type:: ndarray

reset() → None[source]#

Return type:: None

step() → None[source]#

Step each kernel to increment its counter, and keep only non-expired ones.

Return type:: None

class DecayKernel(location: numpy.ndarray, tau_t: float = 10.0, tau_s: float = 0.01, spatial_cutoff: float | None = 0.02, w_t_min: float = 0.1)[source]#

Bases: object

Decay kernel represents a previously visited location.

Returns the product of time- and space-dependent exponentials.

__init__(location: numpy.ndarray, tau_t: float = 10.0, tau_s: float = 0.01, spatial_cutoff: float | None = 0.02, w_t_min: float = 0.1)[source]#

step() → bool[source]#

Increment the step counter, and check if the kernel is expired.

Return type:: bool
Returns:: True if the kernel is expired, False otherwise.

w_s(points: numpy.ndarray) → numpy.ndarray[source]#

Compute the distance-dependent weight.

The weight is computed as exp(-z / lam), where z is the distance between the kernel’s center and the given point(s), and lam is equal to tau_s / log(2).

Parameters:: points (ndarray) – An (num_points, 3) array of points.
Return type:: ndarray
Returns:: The weight, bounded to [0, 1]. Has shape (num_points,).

w_t() → float[source]#

Compute the time-dependent weight at the current step.

The weight is computed as exp(-t / lam), where t is the number of steps since the kernel was created, and lam is equal to tau_t / log(2).

Return type:: float
Returns:: The weight, bounded to [0, 1].

class DecayKernelFactory(tau_t: float = 10.0, tau_s: float = 0.01, spatial_cutoff: float | None = 0.02, w_t_min: float = 0.1)[source]#

Bases: object

__init__(tau_t: float = 10.0, tau_s: float = 0.01, spatial_cutoff: float | None = 0.02, w_t_min: float = 0.1)[source]#

class ReturnInhibitor(decay_field: DecayField | None = None)[source]#

Bases: object

__init__(decay_field: DecayField | None = None)[source]#

reset() → None[source]#

Return type:: None

tbp.monty.frameworks.models.salience.sensor_module#

class SalienceSM(sensor_module_id: str, salience_strategy: SalienceStrategy | None = None, return_inhibitor: ReturnInhibitor | None = None, snapshot_telemetry: SalienceSMTelemetry | None = None, segmentation_strategy: SegmentationStrategy | None = None) → None[source]#

Bases: SensorModule

A wide field-of-view sensor module that proposes locations to attend to.

Implements inhibition-of-return, and can be configured with different strategies for ranking target locations. It can also be configured with a segmentation strategy to propose ``AttentionRegion``s.

static region_locations_on_object(segmentation_map: NDArray[uint8], on_object_map: NDArray[bool], location_map: NDArray[float64]) → NDArray[float64][source]#

Return the 3D locations of the segmented region that are on the object.

Parameters:

segmentation_map (GenericAlias[ubyte]) – The segmentation map of the image.
on_object_map (GenericAlias[bool]) – The on-object view of the observation as a boolean mask.
location_map (GenericAlias[double]) – The corresponding 3D locations for each pixel in the observation.

Return type:

GenericAlias[double]

Returns:

The 3D locations of the segmented region that are on the object.

__init__(sensor_module_id: str, salience_strategy: SalienceStrategy | None = None, return_inhibitor: ReturnInhibitor | None = None, snapshot_telemetry: SalienceSMTelemetry | None = None, segmentation_strategy: SegmentationStrategy | None = None) → None[source]#

propose_goals() → list[Goal][source]#

Return type:: list[Goal]

propose_region() → AttentionRegion[source]#

Return type:: AttentionRegion

reset() → None[source]#

Reset the internal state of this Sensor Module.

Return type:: None

state_dict() → Mapping[str, Any][source]#

Return type:: Mapping[str, Any]

step(ctx: RuntimeContext, observation: SensorObservation, motor_only_step: bool = False) → None[source]#

Generate goal for the current step.

If motor_only_step is True, this method will return without using the salience strategy, stepping the return inhibitor, or modifying self._goals in any way.

Parameters:

ctx (RuntimeContext) – The runtime context.
observation (SensorObservation) – Sensor observation.
motor_only_step (bool) – Whether the current step is a motor-only step.

Return type:

None

update_state(agent: AgentState) → None[source]#

Update information about the sensor’s location and rotation.

Return type:: None

property sensor_module_id: str#

tbp.monty.frameworks.models.salience.telemetry#

class DetailedSalienceSMTelemetry → None[source]#

Bases: SalienceSMTelemetry

Keeps track of all of SalienceSM’s telemetry.

Records per step: raw observation snapshots with their poses, the 2D salience map, the 2D segmentation mask, the goals proposed and the attention region proposed from the mask.

Everything stored here is JSON-encodable by BufferEncoder, so the state dict rides into the detailed logging stream with no special handling.

__init__() → None[source]#

attention_region(region: AttentionRegion) → None[source]#

Record the attention region proposed this step.

Parameters:: region (AttentionRegion) – The proposed region.
Return type:: None

goals(goals: Sequence[Goal]) → None[source]#

Record one step’s proposed goals.

Parameters:: goals (Sequence[Goal]) – The goals the sensor module proposed this step.
Return type:: None

raw_observation(raw_observation: SensorObservation, rotation: quaternion.quaternion, position: tuple[float, float, float]) → None[source]#

Record a snapshot of a raw observation and its pose information.

Parameters:

raw_observation (SensorObservation) – Raw observation.
rotation (quaternion) – Rotation of the sensor.
position (tuple[float, float, float]) – Position of the sensor.

Return type:

None

reset() → None[source]#

Reset the telemetry.

Return type:: None

salience_map(salience_map: numpy.typing.NDArray.numpy.float64) → None[source]#

Record one step’s salience map.

Parameters:: salience_map (float64) – The 2D salience map.
Return type:: None

segmentation_map(segmentation_map: numpy.typing.NDArray.numpy.uint8) → None[source]#

Record one step’s segmentation mask.

Parameters:: segmentation_map (uint8) – The 2D segmentation mask.
Return type:: None

state_dict() → Mapping[str, Any][source]#

Return all recorded telemetry.

Return type:: Mapping[str, Any]
Returns:: Raw observations in raw_observations with poses in sm_properties, salience maps in salience_maps, segmentation masks in segmentation_maps, goal columns in goals, and the proposed regions in attention_regions.

class NoopSalienceSMTelemetry(*args, **kwargs)[source]#

Bases: SalienceSMTelemetry

attention_region(region: AttentionRegion) → None[source]#

Record the attention region proposed this step.

Parameters:: region (AttentionRegion) – The proposed region.
Return type:: None

goals(goals: Sequence[Goal]) → None[source]#

Record one step’s proposed goals.

Parameters:: goals (Sequence[Goal]) – The goals the sensor module proposed this step.
Return type:: None

raw_observation(raw_observation: SensorObservation, rotation: quaternion.quaternion, position: tuple[float, float, float]) → None[source]#

Record a snapshot of a raw observation and its pose information.

Parameters:

raw_observation (SensorObservation) – Raw observation.
rotation (quaternion) – Rotation of the sensor.
position (tuple[float, float, float]) – Position of the sensor.

Return type:

None

reset() → None[source]#

Return type:: None

salience_map(salience_map: numpy.typing.NDArray.numpy.float64) → None[source]#

Record one step’s salience map.

Parameters:: salience_map (float64) – The 2D salience map.
Return type:: None

segmentation_map(segmentation_map: numpy.typing.NDArray.numpy.uint8) → None[source]#

Record one step’s segmentation mask.

Parameters:: segmentation_map (uint8) – The 2D segmentation mask.
Return type:: None

state_dict() → Mapping[str, Any][source]#

Return all recorded telemetry.

Return type:: Mapping[str, Any]
Returns:: Raw observations in raw_observations with poses in sm_properties, salience maps in salience_maps, segmentation masks in segmentation_maps, goal columns in goals, and the proposed regions in attention_regions.

class SalienceSMTelemetry(*args, **kwargs)[source]#

Bases: Protocol

__init__(*args, **kwargs)#

attention_region(region: AttentionRegion) → None[source]#

Record the attention region proposed this step.

Parameters:: region (AttentionRegion) – The proposed region.
Return type:: None

goals(goals: Sequence[Goal]) → None[source]#

Record one step’s proposed goals.

Parameters:: goals (Sequence[Goal]) – The goals the sensor module proposed this step.
Return type:: None

raw_observation(raw_observation: SensorObservation, rotation: quaternion.quaternion, position: tuple[float, float, float]) → None[source]#

Record a snapshot of a raw observation and its pose information.

Parameters:

raw_observation (SensorObservation) – Raw observation.
rotation (quaternion) – Rotation of the sensor.
position (tuple[float, float, float]) – Position of the sensor.

Return type:

None

reset() → None[source]#

Return type:: None

salience_map(salience_map: numpy.typing.NDArray.numpy.float64) → None[source]#

Record one step’s salience map.

Parameters:: salience_map (float64) – The 2D salience map.
Return type:: None

segmentation_map(segmentation_map: numpy.typing.NDArray.numpy.uint8) → None[source]#

Record one step’s segmentation mask.

Parameters:: segmentation_map (uint8) – The 2D segmentation mask.
Return type:: None

state_dict() → Mapping[str, Any][source]#

Return all recorded telemetry.

Return type:: Mapping[str, Any]
Returns:: Raw observations in raw_observations with poses in sm_properties, salience maps in salience_maps, segmentation masks in segmentation_maps, goal columns in goals, and the proposed regions in attention_regions.