tbp.monty.experiment

• tbp.monty.experiment.sensor_modules
  • tbp.monty.experiment.sensor_modules.telemetry
    • Snapshot
      • Snapshot.reset()

tbp.monty.experiment.environment

class Interface(env: SimulatedObjectEnvironment, rng: RandomState, seed: int, experiment_mode: ExperimentMode, transform: Transform | Sequence[Transform] | None = None)

Bases: object

Provides an interface to an environment.

Observations and proprioceptive state are returned from the environment based on the actions taken.

env

An instance of a class that implements SimulatedObjectEnvironment.

rng

Random number generator to use.

seed

The configured random seed.

experiment_mode

The experiment mode that this environment interface is used in.

transform

Callable used to transform the observations returned by the environment.

Note

This one on its own won’t work.

__init__(env: SimulatedObjectEnvironment, rng: RandomState, seed: int, experiment_mode: ExperimentMode, transform: Transform | Sequence[Transform] | None = None)

apply_transform(observations: Observations, state: ProprioceptiveState) → Observations

Return type:

Observations

post_episode()

post_epoch()

pre_episode(rng: RandomState)

pre_epoch()

reset(rng: RandomState)

step(actions: Sequence[Action]) → tuple[Observations, ProprioceptiveState]

Take actions in the environment and apply the transform to the observations.

Parameters:

actions (Sequence[Action]) – The actions to take in the environment.

Return type:

tuple[Observations, ProprioceptiveState]

Returns:

The observations and proprioceptive state.

class OmniglotInterface(alphabets: Sequence[int], characters: Sequence[int], versions: Sequence[int], env: OmniglotEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Bases: OneObjectPerEpisodeInterface

Environment interface for Omniglot dataset.

__init__(alphabets: Sequence[int], characters: Sequence[int], versions: Sequence[int], env: OmniglotEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Initialize environment interface.

Parameters:

• alphabets (Sequence[int]) – List of alphabets.

• characters (Sequence[int]) – List of characters.

• versions (Sequence[int]) – List of versions.

• env (OmniglotEnvironment) – An instance of a class that implements OmniglotEnvironment.

• rng (RandomState) – Random number generator to use.

• transform (Transform | Sequence[Transform] | None) – Callable used to transform the observations returned by the environment.

• positioning_procedures (Sequence[PositioningProcedureFactory] | None) – Sequence of positioning procedures to apply prior to each episode.

• *args – Unused?

• **kwargs – Unused?

change_object_by_idx(idx: int)

Update the object in the scene given the idx of it in the object params.

Parameters:

idx (int) – Index of the new object and ints parameters in object params

Raises:

IndexError – If idx is outside the range [0, self.n_objects).

cycle_object()

Switch to the next character image.

post_episode()

post_epoch()

class OneObjectPerEpisodeInterface(object_names: list[str] | ListConfig | MultiObjectNames, object_init_sampler: Default | Predefined | RandomRotation, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *args, **kwargs)

Bases: Interface

Interface for testing in an environment with one “primary target” object.

Interface for testing in an environment where we load one “primary target” object at a time; in addition, we can optionally add other “distractor” objects to the environment.

Has a list of primary target objects, swapping these objects in and out for episodes without resetting the environment. The objects are initialized with parameters such that we can vary their location, rotation, and scale.

After the primary target is added to the environment, other distractor objects, sampled from the same object list, can be added.

__init__(object_names: list[str] | ListConfig | MultiObjectNames, object_init_sampler: Default | Predefined | RandomRotation, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *args, **kwargs)

Initialize environment interface.

Parameters:

• object_names (list[str] | ListConfig | MultiObjectNames) –

  plain list of objects, or Hydra ListConfig, if doing a simple experiment with primary target objects only; mapping typed as MultiObjectNames for experiments with multiple objects, corresponding to –> targets_list : the list of primary target objects source_object_list : the original object list from which the primary

  target objects were sampled; used to sample distractor objects

  num_distractorsthe number of distractor objects to add to the

  environment

• object_init_sampler (Default | Predefined | RandomRotation) – Function that returns dict with position, rotation, and scale of objects when re-initializing.

• positioning_procedures (Sequence[PositioningProcedureFactory] | None) – Sequence of positioning procedures to apply prior to each episode.

• *args – passed to super() call

• **kwargs – passed to super() call

Raises:

TypeError – If object_names is not a list, ListConfig, or a mapping

add_distractor_objects(primary_target_obj: ObjectID, primary_target_name: str)

Add arbitrarily many “distractor” objects to the environment.

Parameters:

• primary_target_obj (ObjectID (int)) – The ID of the object which is the primary target in the scene.

• primary_target_name (str) – name of the primary target object

change_object_by_idx(idx: int)

Update the primary target object in the scene based on the given index.

The given idx is the index of the object in the self.object_names list, which should correspond to the index of the object in the self.object_params list.

Also add any distractor objects if required.

Parameters:

idx (int) – Index of the new object and its parameters in object_params

Raises:

IndexError – If idx is outside the range [0, self.n_objects).

create_semantic_mapping()

Create a unique semantic ID (positive integer) for each object.

Used by Habitat for the semantic sensor.

In addition, create a dictionary mapping back and forth between these IDs and the corresponding name of the object

cycle_object()

Remove the previous object(s) from the scene and add a new primary target.

Also add any potential distractor objects.

post_episode()

post_epoch()

pre_episode(rng: RandomState)

pre_epoch()

class SaccadeOnImageFromStreamInterface(env: SaccadeOnImageFromStreamEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Bases: SaccadeOnImageInterface

Environment interface for moving over a 2D image with depth channel.

__init__(env: SaccadeOnImageFromStreamEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Initialize environment interface.

Parameters:

• env (SaccadeOnImageFromStreamEnvironment) – An instance of a class that implements SaccadeOnImageFromStreamEnvironment.

• rng (RandomState) – Random number generator to use.

• transform (Transform | Sequence[Transform] | None) – Callable used to transform the observations returned by the environment.

• positioning_procedures (Sequence[PositioningProcedureFactory] | None) – Sequence of positioning procedures to apply prior to each episode.

• *args – Unused?

• **kwargs – Unused?

change_scene_by_idx(idx: int)

Update the object in the scene given the idx of it in the object params.

Parameters:

idx (int) – Index of the new object and ints parameters in object params

cycle_scene()

Switch to the next scene image.

post_episode()

post_epoch()

pre_epoch()

class SaccadeOnImageInterface(scenes: Sequence[int], versions: Sequence[int], env: SaccadeOnImageEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Bases: OneObjectPerEpisodeInterface

Environment interface for moving over a 2D image with depth channel.

__init__(scenes: Sequence[int], versions: Sequence[int], env: SaccadeOnImageEnvironment, rng: RandomState, transform: Transform | Sequence[Transform] | None = None, positioning_procedures: Sequence[PositioningProcedureFactory] | None = None, *_args, **_kwargs)

Initialize environment interface.

Parameters:

• scenes (Sequence[int]) – List of scenes

• versions (Sequence[int]) – List of versions

• env (SaccadeOnImageEnvironment) – An instance of a class that implements SaccadeOnImageEnvironment.

• rng (RandomState) – Random number generator to use.

• transform (Transform | Sequence[Transform] | None) – Callable used to transform the observations returned by the environment.

• positioning_procedures (Sequence[PositioningProcedureFactory] | None) – Sequence of positioning procedures to apply prior to each episode.

• *args – Unused?

• **kwargs – Unused?

change_object_by_idx(idx: int)

Update the object in the scene given the idx of it in the object params.

Parameters:

idx (int) – Index of the new object and ints parameters in object params

Raises:

IndexError – If idx is outside the range [0, self.n_versions).

cycle_object()

Switch to the next scene image.

post_episode()

post_epoch()

tbp.monty.experiment.learning_module

class ExperimentLearningModule(*args, **kwargs)

Bases: Protocol

Experiment interface to a Learning Module.

__init__(*args, **kwargs)

fixme_reset_ground_truth(primary_target=None) → None

Reset internal state based on ground truth.

Parameters:

primary_target – The primary target for the learning module to recognize.

Return type:

None

fixme_update_ground_truth() → None

Update internal state based on ground truth.

Return type:

None

reset_stm() → None

Reset short-term memory buffer.

Do things like reset buffers or possible_matches before training.

Return type:

None

set_experiment_mode(mode: ExperimentMode) → None

Set the experiment mode.

Update state variables based on which method (train or evaluate) is being called at the experiment level.

Parameters:

mode (ExperimentMode) – The experiment mode.

Return type:

None

update_ltm_from_stm() → None

Update long-term memory from short-term memory buffer.

Return type:

None

property recognition_status: RecognitionStatus

tbp.monty.experiment.match_criteria

class AnyLMsMatch(count: int) → None

Bases: MatchCriterion

Satisifed once any count of learning modules have reached “match”.

__init__(count: int) → None

Initialize the criterion.

Parameters:

count (int) – The number of learning modules that must reach “match” for the criterion to be satisfied.

Raises:

ValueError – If count is not positive.

class MatchCriterion(*args, **kwargs)

Bases: Protocol

Decides whether learning modules have collectively matched the target.

An individual learning module reaches its own terminal state independently of the others. The match criterion turns those per-LM terminal states into the single system-level decision of whether Monty has recognized the object.

__init__(*args, **kwargs)

class NamedLMsMatch(ids: list[str]) → None

Bases: MatchCriterion

Satisifed once all learning modules with the given IDs have reached “match”.

__init__(ids: list[str]) → None

Initialize the criterion.

Parameters:

ids (list[str]) – The IDs of the learning modules that must reach “match” for the criterion to be satisfied.

Raises:

ValueError – If ids is empty.

tbp.monty.experiment.monty

class ExperimentMonty(*args, **kwargs)

Bases: Protocol

Experiment interface to Monty model.

__init__(*args, **kwargs)

fixme_set_ground_truth(primary_target: dict[str, Any] | None = None, semantic_id_to_label: dict[SemanticID, str] | None = None) → None

Provide ground truth from experiment supervision.

Parameters:

• primary_target (dict[str, Any] | None) – Optional primary target to recognize.

• semantic_id_to_label (dict[SemanticID (int), str] | None) – Optional mapping from IDs to labels.

Return type:

None

is_done() → bool

Return True if the model has reached a terminal condition.

Return type:

bool

reset() → None

Reset the internal state of this Monty model.

Return type:

None

set_experiment_mode(mode: ExperimentMode) → None

Set the experiment mode.

Update state variables based on which method (train or evaluate) is being called at the experiment level.

Parameters:

mode (ExperimentMode) – The experiment mode.

Return type:

None

update_ltm() → None

Transfer short-term buffer to long-term memory.

Return type:

None

tbp.monty.experiment.motor_system

class ExperimentMotorPolicy(*args, **kwargs)

Bases: Protocol

Experiment interface to a Motor Policy.

__init__(*args, **kwargs)

fixme_provide_motor_system(motor_system: ExperimentMotorSystem) → None

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

reset() → None

Reset the internal state of this Motor Policy.

Return type:

None

class ExperimentMotorPolicySelector(*args, **kwargs)

Bases: Protocol

Experiment interface to a Motor Policy Selector.

__init__(*args, **kwargs)

fixme_provide_motor_system(motor_system: ExperimentMotorSystem) → None

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

reset() → None

Reset the internal state of this Motor Policy Selector.

Return type:

None

class ExperimentMotorSystem(*args, **kwargs)

Bases: Protocol

Experiment interface to a Motor System.

__init__(*args, **kwargs)

reset() → None

Reset the internal state of this Motor System.

Return type:

None

property motor_only_step: bool

When True, suppress Learning Module processing.

tbp.monty.experiment.recognition_policy

class AnyPolicy(policies: Sequence[RecognitionPolicy]) → None

Bases: RecognitionPolicy

Combine multiple terminal conditions for Experiments.

Terminal condition is reached if _any_ RecognitionPolicy says so.

__init__(policies: Sequence[RecognitionPolicy]) → None

Initialize the policy.

Parameters:

policies (Sequence[RecognitionPolicy]) – The policies to check (in order).

Raises:

ValueError – If len(policies) < 1.

class MaxTotalSteps(max_total_steps: int) → None

Bases: RecognitionPolicy

step >= max_total_steps.

__init__(max_total_steps: int) → None

Initialize the policy.

Parameters:

max_total_steps (int) – The maximum number of steps before terminating the episode.

Raises:

ValueError – If max_total_steps is not positive.

class MaximumSteps(max_train_steps: int, max_eval_steps: int) → None

Bases: RecognitionPolicy

model.matching_steps >= {max_train_steps | max_eval_steps}.

__init__(max_train_steps: int, max_eval_steps: int) → None

Initialize the policy.

Parameters:

• max_train_steps (int) – The maximum steps to take in training mode.

• max_eval_steps (int) – The maximum steps to take in evaluation mode.

Raises:

ValueError – If max_train_steps, or max_eval_steps are not positive.

class MinEvalMaxTotalSteps(wrapped: RecognitionPolicy, min_eval_steps: int = 0, max_total_steps: int = 1) → None

Bases: RecognitionPolicy

Is not done while matching_steps < min_eval_steps.

The primary purpose of this policy is to return is_done = False until at least min_eval_steps have been taken. It includes a step >= max_total_steps check to return is_done = True when matching_steps fails to increment sufficiently over max_total_steps steps.

__init__(wrapped: RecognitionPolicy, min_eval_steps: int = 0, max_total_steps: int = 1) → None

Initialize the policy.

Parameters:

• wrapped (RecognitionPolicy) – The policy _wrapped_ and possibly overridden by this policy.

• min_eval_steps (int) – The minimum steps to take in matching mode.

• max_total_steps (int) – The maximum number of steps before forcing termination.

Raises:

ValueError –

• If min_eval_steps is negative. - If max_total_steps is not positive. - If min_eval_steps > max_total_steps.

class MinimumLMs(min_lms: int) → None

Bases: RecognitionPolicy

min_lms have reached a conclusion.

__init__(min_lms: int) → None

Initialize the policy.

Parameters:

min_lms (int) – The number of Learning Modules that must reach a conclusion for the policy to be satisfied.

Raises:

ValueError – If min_lms is not positive.

class MontyIsDone(*args, **kwargs)

Bases: RecognitionPolicy

Legacy (default) policy.

class NaiveScan(fixed_amount: int) → None

Bases: RecognitionPolicy

steps >= step_limit, with step_limit derived from fixed_amount.

The step_limit is the number of steps the Naive Scan motor policy takes before its spiral completes.

__init__(fixed_amount: int) → None

Initialize the policy.

Parameters:

fixed_amount (int) – The Naive Scan step size.

Raises:

ValueError – If fixed_amount is not positive.

class ObjectRecognition(max_train_steps: int, max_eval_steps: int, max_total_steps: int) → None

Bases: RecognitionPolicy

Determine terminal conditions for object recognition experiments.

Terminal conditions include: - model.matching_steps >= {max_train_steps | max_eval_steps} - count.step >= max_total_steps - model.is_done

__init__(max_train_steps: int, max_eval_steps: int, max_total_steps: int) → None

Initialize the policy.

Parameters:

• max_train_steps (int) – The maximum steps to take in training mode.

• max_eval_steps (int) – The maximum steps to take in evaluation mode.

• max_total_steps (int) – The maximum total number of steps before terminating.

Raises:

ValueError – If max_train_steps, max_eval_steps, or max_total_steps are not positive.

class RecognitionCounter(step: int = 0, matching_steps: int = 0, exploring_steps: int = 0, mode: ExperimentMode = ExperimentMode.EVAL) → None

Bases: object

Experiment counters and limits.

__init__(step: int = 0, matching_steps: int = 0, exploring_steps: int = 0, mode: ExperimentMode = ExperimentMode.EVAL) → None

exploring_steps: int = 0

Count of exploring steps taken.

matching_steps: int = 0

Count of matching steps taken.

mode: ExperimentMode = 'eval'

The stepping mode (traning or evaluation).

step: int = 0

The current step number.

class RecognitionPolicy(*args, **kwargs)

Bases: Protocol

Decides what constitutes “recognition” in an Experiment.

Each Learning Module determines its own Recognition Status independently of the others. The Recognition Policy turns the per-LM status into the single decision of whether Monty has recognized the object.

__init__(*args, **kwargs)

class RecognitionResult(is_done: bool = False, time_out_lms: bool = False, start_exploring: bool = False) → None

Bases: object

Aggregated result from the Recognition Policy.

__init__(is_done: bool = False, time_out_lms: bool = False, start_exploring: bool = False) → None

is_done: bool = False

A terminal condition has been reached.

start_exploring: bool = False

Request for transition to ‘exploring’ mode.

time_out_lms: bool = False

Request to time-out LMs that are not done yet.

class StepLimit(min_train_steps: int = 0, num_exploring_steps: int = 0, max_train_steps: int = 1, max_eval_steps: int = 1) → None

Bases: RecognitionPolicy

Check step counter limits.

Terminal conditions include: - matching_steps >= {max_train_steps | max_eval_steps} - exploring_steps >= num_exploring_steps

__init__(min_train_steps: int = 0, num_exploring_steps: int = 0, max_train_steps: int = 1, max_eval_steps: int = 1) → None

Initialize the policy.

Parameters:

• min_train_steps (int) – The minimum steps to take in training mode.

• num_exploring_steps (int) – The number of steps to take in exploring mode.

• max_train_steps (int) – The maximum steps to take in training mode.

• max_eval_steps (int) – The maximum steps to take in evaluation mode.

Raises:

ValueError –

• If min_train_steps is negative. - If num_exploring_steps is negative. - If max_train_steps is not positive. - If max_eval_steps is not positive.

tbp.monty.experiment.recognition_status

class RecognitionConclusion(*values)

Bases: Enum

Label for the terminal state of a Learning Module.

MATCH = 'match'

NO_MATCH = 'no_match'

TIME_OUT = 'time_out'

class RecognitionStatus(conclusion: RecognitionConclusion | None = None) → None

Bases: object

Recognition Status from each Learning Module.

__init__(conclusion: RecognitionConclusion | None = None) → None

conclusion: RecognitionConclusion | None = None

tbp.monty.experiment.sensor_module

class ExperimentSensorModule(sensor_module_id: str, sensor_id: SensorID, transforms: Annotated[Sequence[ExperimentTransform], BeforeValidator(func=tuple, json_schema_input_type=PydanticUndefined)]) → None

Bases: SensorModule, ResettableSensorModule

An experimental wrapper around a Sensor Module with experiment affordances.

Allows for resetting the internal state of the transforms in the transform pipeline.

__init__(sensor_module_id: str, sensor_id: SensorID, transforms: Annotated[Sequence[ExperimentTransform], BeforeValidator(func=tuple, json_schema_input_type=PydanticUndefined)]) → None

Initialize the sensor module.

Parameters:

• sensor_module_id (str) – The unique identifier for this sensor module.

• sensor_id (SensorID (str)) – The unique identifier for the sensor associated with this module.

• transforms (Sequence[ExperimentTransform]) – A sequence of transforms to process the Payload containing the sensor observation and produce sensor module outputs.

reset() → None

Reset the internal state of this Sensor Module.

Return type:

None

class ExperimentTransform(*args, **kwargs)

Bases: Transform, Protocol

An experimental transform with experiment affordances.

Allows for resetting the internal state of the transform.

reset() → None

Reset the internal state of this transform.

The protocol offers a default implementation that does nothing.

Return type:

None

class ResettableSensorModule(*args, **kwargs)

Bases: Protocol

A Sensor Module that can be reset.

__init__(*args, **kwargs)

reset() → None

Reset the internal state of this Sensor Module.

Return type:

None