Interaction

pamiq_core.interaction.Agent

Agent(agents: Mapping[str, Agent[Any, Any]] | None = None)

Bases: ABC, InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Base agent class for decision making.

An agent receives observations from an environment and decides on actions to take in response. This abstract class defines the interface that all agent implementations must follow.

Agents can contain child agents that will inherit the parent's inference models and data collectors. State persistence and Thread Event is also propagated to all child agents.

Initialize the agent.

PARAMETER	DESCRIPTION
agents	Optional mapping of names to child agents. Child agents will inherit inference models and data collectors from the parent, and their states will be saved and loaded together with the parent. TYPE: Mapping[str, Agent[Any, Any]] | None DEFAULT: None

Source code in src/pamiq_core/interaction/agent.py

def __init__(self, agents: Mapping[str, Agent[Any, Any]] | None = None) -> None:
    """Initialize the agent.

    Args:
        agents: Optional mapping of names to child agents. Child agents will inherit
            inference models and data collectors from the parent, and their states
            will be saved and loaded together with the parent.
    """
    self._agents: Mapping[str, Agent[Any, Any]] = {}
    if agents is not None:
        self._agents.update(agents)

step abstractmethod

step(observation: ObsType) -> ActType

Processes an observation and determines the next action.

PARAMETER	DESCRIPTION
observation	The current observation from the environment. TYPE: ObsType

RETURNS	DESCRIPTION
ActType	The action to take in response to the observation.

Source code in src/pamiq_core/interaction/agent.py

@abstractmethod
def step(self, observation: ObsType) -> ActType:
    """Processes an observation and determines the next action.

    Args:
        observation: The current observation from the environment.

    Returns:
        The action to take in response to the observation.
    """
    pass

attach_inference_models

attach_inference_models(inference_models: InferenceModelsDict) -> None

Attaches inference models dictionary to this agent.

This method is called to provide the agent with access to inference models for decision making. After attaching, the callback method on_inference_models_attached is called. If the agent has child agents, the models are also attached to them.

PARAMETER	DESCRIPTION
inference_models	Dictionary of inference models to attach. TYPE: InferenceModelsDict

Source code in src/pamiq_core/interaction/agent.py

def attach_inference_models(self, inference_models: InferenceModelsDict) -> None:
    """Attaches inference models dictionary to this agent.

    This method is called to provide the agent with access to inference models
    for decision making. After attaching, the callback method
    `on_inference_models_attached` is called. If the agent has child agents,
    the models are also attached to them.

    Args:
        inference_models: Dictionary of inference models to attach.
    """
    self._inference_models = inference_models
    self.on_inference_models_attached()
    for agent in self._agents.values():
        agent.attach_inference_models(inference_models)

on_inference_models_attached

on_inference_models_attached() -> None

Callback method for when inference models are attached to the agent.

Override this method to retrieve DNN models for inference. Use get_inference_model to retrieve a model.

Source code in src/pamiq_core/interaction/agent.py

def on_inference_models_attached(self) -> None:
    """Callback method for when inference models are attached to the agent.

    Override this method to retrieve DNN models for inference. Use
    `get_inference_model` to retrieve a model.
    """
    pass

attach_data_collectors

attach_data_collectors(data_collectors: DataCollectorsDict) -> None

Attaches data collectors dictionary to this agent.

This method is called to provide the agent with access to data collectors for collecting experience data. After attaching, the callback method on_data_collectors_attached is called. If the agent has child agents, the collectors are also attached to them.

PARAMETER	DESCRIPTION
data_collectors	Dictionary of data collectors to attach. TYPE: DataCollectorsDict

Source code in src/pamiq_core/interaction/agent.py

def attach_data_collectors(self, data_collectors: DataCollectorsDict) -> None:
    """Attaches data collectors dictionary to this agent.

    This method is called to provide the agent with access to data collectors
    for collecting experience data. After attaching, the callback method
    `on_data_collectors_attached` is called. If the agent has child agents,
    the collectors are also attached to them.

    Args:
        data_collectors: Dictionary of data collectors to attach.
    """
    self._data_collectors = data_collectors
    self.on_data_collectors_attached()
    for agent in self._agents.values():
        agent.attach_data_collectors(data_collectors)

on_data_collectors_attached

on_data_collectors_attached() -> None

Callback method for when data collectors are attached to this agent.

Override this method to set up data collection for the agent. Use get_data_collector to acquire a collector.

Source code in src/pamiq_core/interaction/agent.py

def on_data_collectors_attached(self) -> None:
    """Callback method for when data collectors are attached to this agent.

    Override this method to set up data collection for the agent. Use
    `get_data_collector` to acquire a collector.
    """
    pass

get_inference_model

get_inference_model(name: str) -> InferenceModel

Retrieves an inference model by name.

PARAMETER	DESCRIPTION
name	Name of the inference model to retrieve. TYPE: str

RETURNS	DESCRIPTION
InferenceModel	The requested inference model.

RAISES	DESCRIPTION
KeyError	If the model with the specified name does not exist.

Source code in src/pamiq_core/interaction/agent.py

def get_inference_model(self, name: str) -> InferenceModel:
    """Retrieves an inference model by name.

    Args:
        name: Name of the inference model to retrieve.

    Returns:
        The requested inference model.

    Raises:
        KeyError: If the model with the specified name does not exist.
    """
    return self._inference_models[name]

get_data_collector

get_data_collector(name: str) -> DataCollector[Any]

Acquires a data collector by name for exclusive use.

This method acquires a data collector for exclusive use within the current step. The collector can only be acquired once at a time.

PARAMETER	DESCRIPTION
name	Name of the data collector to acquire. TYPE: str

RETURNS	DESCRIPTION
DataCollector[Any]	The requested data collector.

RAISES	DESCRIPTION
KeyError	If the collector is already acquired or not found.

Source code in src/pamiq_core/interaction/agent.py

def get_data_collector(self, name: str) -> DataCollector[Any]:
    """Acquires a data collector by name for exclusive use.

    This method acquires a data collector for exclusive use within
    the current step. The collector can only be acquired once at a time.

    Args:
        name: Name of the data collector to acquire.

    Returns:
        The requested data collector.

    Raises:
        KeyError: If the collector is already acquired or not found.
    """
    return self._data_collectors.acquire(name)

setup

setup() -> None

Handle interaction setup event.

Propagates the event to all child agents.

Source code in src/pamiq_core/interaction/agent.py

@override
def setup(self) -> None:
    """Handle interaction setup event.

    Propagates the event to all child agents.
    """
    super().setup()
    for agent in self._agents.values():
        agent.setup()

teardown

teardown() -> None

Handle interaction teardown event.

Propagates the event to all child agents.

Source code in src/pamiq_core/interaction/agent.py

@override
def teardown(self) -> None:
    """Handle interaction teardown event.

    Propagates the event to all child agents.
    """
    super().teardown()
    for agent in self._agents.values():
        agent.teardown()

save_state

save_state(path: Path) -> None

Save the agent's state and the states of any child agents.

PARAMETER	DESCRIPTION
path	Directory path where to save the states. TYPE: Path

Source code in src/pamiq_core/interaction/agent.py

@override
def save_state(self, path: Path) -> None:
    """Save the agent's state and the states of any child agents.

    Args:
        path: Directory path where to save the states.
    """
    super().save_state(path)
    if len(self._agents) == 0:
        return
    path.mkdir(exist_ok=True)
    for name, agent in self._agents.items():
        agent.save_state(path / name)

load_state

load_state(path: Path) -> None

Load the agent's state and the states of any child agents.

PARAMETER	DESCRIPTION
path	Directory path from where to load the states. TYPE: Path

Source code in src/pamiq_core/interaction/agent.py

@override
def load_state(self, path: Path) -> None:
    """Load the agent's state and the states of any child agents.

    Args:
        path: Directory path from where to load the states.
    """
    super().load_state(path)
    for name, agent in self._agents.items():
        agent.load_state(path / name)

on_paused

on_paused() -> None

Handle system pause event.

Propagates the pause event to all child agents.

Source code in src/pamiq_core/interaction/agent.py

@override
def on_paused(self) -> None:
    """Handle system pause event.

    Propagates the pause event to all child agents.
    """
    super().on_paused()
    for agent in self._agents.values():
        agent.on_paused()

on_resumed

on_resumed() -> None

Handle system resume event.

Propagates the resume event to all child agents.

Source code in src/pamiq_core/interaction/agent.py

@override
def on_resumed(self) -> None:
    """Handle system resume event.

    Propagates the resume event to all child agents.
    """
    super().on_resumed()
    for agent in self._agents.values():
        agent.on_resumed()

pamiq_core.interaction.Environment

Bases: ABC, InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Base environment class for agent interaction.

observe abstractmethod

observe() -> ObsType

Gets and returns observation from the environment.

RETURNS	DESCRIPTION
ObsType	Observation from the environment.

Source code in src/pamiq_core/interaction/env.py

@abstractmethod
def observe(self) -> ObsType:
    """Gets and returns observation from the environment.

    Returns:
        Observation from the environment.
    """
    pass

affect abstractmethod

affect(action: ActType) -> None

Applies an action to the environment.

PARAMETER	DESCRIPTION
action	Action to apply to the environment. TYPE: ActType

Source code in src/pamiq_core/interaction/env.py

@abstractmethod
def affect(self, action: ActType) -> None:
    """Applies an action to the environment.

    Args:
        action: Action to apply to the environment.
    """
    pass

pamiq_core.interaction.Interaction

Interaction(agent: Agent[ObsType, ActType], environment: Environment[ObsType, ActType])

Bases: InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Class that combines an agent and environment to create an interaction loop.

This class manages the interaction between an agent and its environment, implementing a basic observe-decide-act loop. It also handles state persistence and lifecycle management for both components.

Initialize interaction with an agent and environment.

PARAMETER	DESCRIPTION
agent	The agent that makes decisions based on observations. TYPE: Agent[ObsType, ActType]
environment	The environment that provides observations and receives actions. TYPE: Environment[ObsType, ActType]

Source code in src/pamiq_core/interaction/interactions.py

def __init__(
    self, agent: Agent[ObsType, ActType], environment: Environment[ObsType, ActType]
) -> None:
    """Initialize interaction with an agent and environment.

    Args:
        agent: The agent that makes decisions based on observations.
        environment: The environment that provides observations and receives actions.
    """
    self.agent = agent
    self.environment = environment

step

step() -> None

Execute one step of the agent-environment interaction loop.

Gets an observation from the environment, passes it to the agent to get an action, and then applies that action to the environment.

Source code in src/pamiq_core/interaction/interactions.py

def step(self) -> None:
    """Execute one step of the agent-environment interaction loop.

    Gets an observation from the environment, passes it to the agent
    to get an action, and then applies that action to the
    environment.
    """
    obs = self.environment.observe()
    action = self.agent.step(obs)
    self.environment.affect(action)

setup

setup() -> None

Initialize the interaction by setting up agent and environment.

Calls the setup methods of both the agent and environment.

Source code in src/pamiq_core/interaction/interactions.py

@override
def setup(self) -> None:
    """Initialize the interaction by setting up agent and environment.

    Calls the setup methods of both the agent and environment.
    """
    super().setup()
    self.agent.setup()
    self.environment.setup()

teardown

teardown() -> None

Clean up the interaction by tearing down agent and environment.

Calls the teardown methods of both the agent and environment.

Source code in src/pamiq_core/interaction/interactions.py

@override
def teardown(self) -> None:
    """Clean up the interaction by tearing down agent and environment.

    Calls the teardown methods of both the agent and environment.
    """
    super().teardown()
    self.agent.teardown()
    self.environment.teardown()

save_state

save_state(path: Path) -> None

Save the current state of the interaction to the specified path.

Creates a directory at the given path and saves the states of both the agent and environment in subdirectories.

PARAMETER	DESCRIPTION
path	Directory path where to save the interaction state. TYPE: Path

Source code in src/pamiq_core/interaction/interactions.py

@override
def save_state(self, path: Path) -> None:
    """Save the current state of the interaction to the specified path.

    Creates a directory at the given path and saves the states of both
    the agent and environment in subdirectories.

    Args:
        path: Directory path where to save the interaction state.
    """
    path.mkdir()
    self.agent.save_state(path / "agent")
    self.environment.save_state(path / "environment")

load_state

load_state(path: Path) -> None

Load the interaction state from the specified path.

Loads the states of both the agent and environment from subdirectories at the given path.

PARAMETER	DESCRIPTION
path	Directory path from where to load the interaction state. TYPE: Path

Source code in src/pamiq_core/interaction/interactions.py

@override
def load_state(self, path: Path) -> None:
    """Load the interaction state from the specified path.

    Loads the states of both the agent and environment from subdirectories
    at the given path.

    Args:
        path: Directory path from where to load the interaction state.
    """
    self.agent.load_state(path / "agent")
    self.environment.load_state(path / "environment")

on_paused

on_paused() -> None

The method to be called when the thread is paused.

Source code in src/pamiq_core/interaction/interactions.py

@override
def on_paused(self) -> None:
    """The method to be called when the thread is paused."""
    super().on_paused()
    self.agent.on_paused()
    self.environment.on_paused()

on_resumed

on_resumed() -> None

The method to be called when the thread is resumed.

Source code in src/pamiq_core/interaction/interactions.py

@override
def on_resumed(self) -> None:
    """The method to be called when the thread is resumed."""
    super().on_resumed()
    self.agent.on_resumed()
    self.environment.on_resumed()

pamiq_core.interaction.FixedIntervalInteraction

FixedIntervalInteraction(
    agent: Agent[ObsType, ActType],
    environment: Environment[ObsType, ActType],
    adjustor: IntervalAdjustor,
)

Bases: Interaction[ObsType, ActType]

Interaction class that executes steps at fixed time intervals.

This class extends the base Interaction to maintain a consistent timing between steps using an IntervalAdjustor. It ensures the agent-environment interaction loop runs at a specified frequency regardless of how long individual steps take to compute.

Initialize the fixed interval interaction.

PARAMETER	DESCRIPTION
agent	The agent that makes decisions based on observations. TYPE: Agent[ObsType, ActType]
environment	The environment that provides observations and receives actions. TYPE: Environment[ObsType, ActType]
adjustor	The interval adjustor that maintains consistent timing between steps. TYPE: IntervalAdjustor

Source code in src/pamiq_core/interaction/interactions.py

@override
def __init__(
    self,
    agent: Agent[ObsType, ActType],
    environment: Environment[ObsType, ActType],
    adjustor: IntervalAdjustor,
) -> None:
    """Initialize the fixed interval interaction.

    Args:
        agent: The agent that makes decisions based on observations.
        environment: The environment that provides observations and receives actions.
        adjustor: The interval adjustor that maintains consistent timing between steps.
    """
    super().__init__(agent, environment)
    self._adjustor = adjustor

setup

setup() -> None

Initialize the interaction and reset the interval adjustor.

Source code in src/pamiq_core/interaction/interactions.py

@override
def setup(self) -> None:
    """Initialize the interaction and reset the interval adjustor."""
    super().setup()
    self._adjustor.reset()

step

step() -> None

Execute one step of the interaction and adjust timing.

Source code in src/pamiq_core/interaction/interactions.py

@override
def step(self) -> None:
    """Execute one step of the interaction and adjust timing."""
    super().step()
    self._adjustor.adjust()

with_sleep_adjustor classmethod

with_sleep_adjustor(
    agent: Agent[ObsType, ActType],
    environment: Environment[ObsType, ActType],
    interval: float,
    offset: float = 0.0,
) -> Self

Create a FixedIntervalInteraction with a SleepIntervalAdjustor.

PARAMETER	DESCRIPTION
agent	The agent that makes decisions based on observations. TYPE: Agent[ObsType, ActType]
environment	The environment that provides observations and receives actions. TYPE: Environment[ObsType, ActType]
interval	The desired time between each step in seconds. TYPE: float
offset	Optional initial time offset to adjust for system-specific timing differences. Defaults to 0.0. TYPE: float DEFAULT: 0.0

RETURNS	DESCRIPTION
Self	A new FixedIntervalInteraction instance configured with a SleepIntervalAdjustor.

Source code in src/pamiq_core/interaction/interactions.py

@classmethod
def with_sleep_adjustor(
    cls,
    agent: Agent[ObsType, ActType],
    environment: Environment[ObsType, ActType],
    interval: float,
    offset: float = 0.0,
) -> Self:
    """Create a FixedIntervalInteraction with a SleepIntervalAdjustor.

    Args:
        agent: The agent that makes decisions based on observations.
        environment: The environment that provides observations and receives actions.
        interval: The desired time between each step in seconds.
        offset: Optional initial time offset to adjust for system-specific timing differences.
            Defaults to 0.0.

    Returns:
        A new FixedIntervalInteraction instance configured with a SleepIntervalAdjustor.
    """
    return cls(agent, environment, SleepIntervalAdjustor(interval, offset))

pamiq_core.interaction.interval_adjustors

IntervalAdjustor

IntervalAdjustor(interval: float, offset: float = 0.0)

Bases: ABC

Adjusts the loop to run at a fixed interval.

Initial reset time is set to negative infinity for mathematical reasons.

Constructs the IntervalAdjustor.

PARAMETER	DESCRIPTION
interval	The desired time between each invocation in seconds. TYPE: float
offset	The initial time offset to subtract from the interval on the first loop iteration, allowing adjustment of the actual interval for each computer using this parameter. TYPE: float DEFAULT: 0.0

Source code in src/pamiq_core/interaction/interval_adjustors.py

def __init__(self, interval: float, offset: float = 0.0) -> None:
    """Constructs the IntervalAdjustor.

    Args:
        interval: The desired time between each invocation in seconds.
        offset: The initial time offset to subtract from the interval on the first loop iteration,
            allowing adjustment of the actual interval for each computer using this parameter.
    """
    self._interval = interval
    self._offset = offset
    self._time_to_wait = interval - offset

reset

reset() -> float

Resets the start time of this adjustor to the current time.

RETURNS	DESCRIPTION
float	The start time after resetting the timer, as a high precision time counter. TYPE: float

Source code in src/pamiq_core/interaction/interval_adjustors.py

def reset(self) -> float:
    """Resets the start time of this adjustor to the current time.

    Returns:
        float: The start time after resetting the timer, as a high precision time counter.
    """
    self._last_reset_time = time.perf_counter()
    return self._last_reset_time

adjust_impl abstractmethod

adjust_impl() -> None

The actual implementation of the interval adjustor, which is wrapped by the public method adjust.

Source code in src/pamiq_core/interaction/interval_adjustors.py

@abstractmethod
def adjust_impl(self) -> None:
    """The actual implementation of the interval adjustor, which is wrapped
    by the public method `adjust`."""
    pass

adjust

adjust() -> float

Waits until the interval has elapsed since the last adjust or reset call.

RETURNS	DESCRIPTION
float	The elapsed time since the last call, ensuring the loop runs at the specified interval. TYPE: float

Source code in src/pamiq_core/interaction/interval_adjustors.py

def adjust(self) -> float:
    """Waits until the interval has elapsed since the last `adjust` or
    `reset` call.

    Returns:
        float: The elapsed time since the last call, ensuring the loop runs at the specified interval.
    """
    self.adjust_impl()
    delta_time = time.perf_counter() - self._last_reset_time
    self.reset()
    return delta_time

SleepIntervalAdjustor

SleepIntervalAdjustor(interval: float, offset: float = 0.0)

Bases: IntervalAdjustor

Adjusts the interval using time.sleep to pause execution until the next interval begins.

Source code in src/pamiq_core/interaction/interval_adjustors.py

def __init__(self, interval: float, offset: float = 0.0) -> None:
    """Constructs the IntervalAdjustor.

    Args:
        interval: The desired time between each invocation in seconds.
        offset: The initial time offset to subtract from the interval on the first loop iteration,
            allowing adjustment of the actual interval for each computer using this parameter.
    """
    self._interval = interval
    self._offset = offset
    self._time_to_wait = interval - offset

pamiq_core.interaction.modular_env

Sensor

Bases: ABC, InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Abstract base class for sensors that read data from the environment.

This class provides an interface for reading observations from various sensors. Implementations should handle the specific logic for acquiring sensor readings.

read abstractmethod

read() -> T

Read data from the sensor.

RETURNS	DESCRIPTION
T	Sensor reading/observation data.

Source code in src/pamiq_core/interaction/modular_env.py

@abstractmethod
def read(self) -> T:
    """Read data from the sensor.

    Returns:
        Sensor reading/observation data.
    """
    pass

Actuator

Bases: ABC, InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Abstract base class for actuators that affect the environment.

This class provides an interface for operating actuators based on action commands. Implementations should handle the specific logic for executing actions through hardware or simulation interfaces.

operate abstractmethod

operate(action: T) -> None

Execute the specified action through the actuator.

PARAMETER	DESCRIPTION
action	The action to be executed. TYPE: T

Source code in src/pamiq_core/interaction/modular_env.py

@abstractmethod
def operate(self, action: T) -> None:
    """Execute the specified action through the actuator.

    Args:
        action: The action to be executed.
    """
    pass

ModularEnvironment

ModularEnvironment(sensor: Sensor[ObsType], actuator: Actuator[ActType])

Bases: Environment[ObsType, ActType]

Environment implementation that uses a Sensor and Actuator.

This class provides a modular approach to environment implementation by separating the sensing (observation) and actuation components.

Initialize with a sensor and actuator.

PARAMETER	DESCRIPTION
sensor	Component to read observations from the environment. TYPE: Sensor[ObsType]
actuator	Component to execute actions in the environment. TYPE: Actuator[ActType]

Source code in src/pamiq_core/interaction/modular_env.py

@override
def __init__(self, sensor: Sensor[ObsType], actuator: Actuator[ActType]) -> None:
    """Initialize with a sensor and actuator.

    Args:
        sensor: Component to read observations from the environment.
        actuator: Component to execute actions in the environment.
    """
    self.sensor = sensor
    self.actuator = actuator

observe

observe() -> ObsType

Get observations from the environment using the sensor.

RETURNS	DESCRIPTION
ObsType	Current observation from the sensor.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def observe(self) -> ObsType:
    """Get observations from the environment using the sensor.

    Returns:
        Current observation from the sensor.
    """
    return self.sensor.read()

affect

affect(action: ActType) -> None

Apply an action to the environment using the actuator.

PARAMETER	DESCRIPTION
action	The action to apply to the environment. TYPE: ActType

Source code in src/pamiq_core/interaction/modular_env.py

@override
def affect(self, action: ActType) -> None:
    """Apply an action to the environment using the actuator.

    Args:
        action: The action to apply to the environment.
    """
    self.actuator.operate(action)

setup

setup() -> None

Set up the environment by initializing sensor and actuator.

This method is called before starting interaction with the environment.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def setup(self) -> None:
    """Set up the environment by initializing sensor and actuator.

    This method is called before starting interaction with the
    environment.
    """
    self.sensor.setup()
    self.actuator.setup()

teardown

teardown() -> None

Clean up the environment by finalizing sensor and actuator.

This method is called after finishing interaction with the environment.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def teardown(self) -> None:
    """Clean up the environment by finalizing sensor and actuator.

    This method is called after finishing interaction with the
    environment.
    """
    self.sensor.teardown()
    self.actuator.teardown()

save_state

save_state(path: Path) -> None

Save the state of the environment to the specified path.

Creates a directory at the given path and saves the states of the sensor and actuator in subdirectories.

PARAMETER	DESCRIPTION
path	Directory path where to save the environment state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def save_state(self, path: Path) -> None:
    """Save the state of the environment to the specified path.

    Creates a directory at the given path and saves the states of
    the sensor and actuator in subdirectories.

    Args:
        path: Directory path where to save the environment state.
    """
    path.mkdir()
    self.sensor.save_state(path / "sensor")
    self.actuator.save_state(path / "actuator")

load_state

load_state(path: Path) -> None

Load the state of the environment from the specified path.

Loads the states of the sensor and actuator from subdirectories at the given path.

PARAMETER	DESCRIPTION
path	Directory path from where to load the environment state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def load_state(self, path: Path) -> None:
    """Load the state of the environment from the specified path.

    Loads the states of the sensor and actuator from subdirectories
    at the given path.

    Args:
        path: Directory path from where to load the environment state.
    """
    self.sensor.load_state(path / "sensor")
    self.actuator.load_state(path / "actuator")

on_paused

on_paused() -> None

The method to be called when the thread is paused.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_paused(self) -> None:
    """The method to be called when the thread is paused."""
    super().on_paused()
    self.sensor.on_paused()
    self.actuator.on_paused()

on_resumed

on_resumed() -> None

The method to be called when the thread is resumed.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_resumed(self) -> None:
    """The method to be called when the thread is resumed."""
    super().on_resumed()
    self.sensor.on_resumed()
    self.actuator.on_resumed()

from_dict staticmethod

from_dict(
    sensors: Mapping[str, Sensor[Any]], actuators: Mapping[str, Actuator[Any]]
) -> ModularEnvironment[Mapping[str, Any], Mapping[str, Any]]

Create a modular environment from dictionaries of sensors and actuators.

PARAMETER	DESCRIPTION
sensors	A mapping of sensor names to sensor instances. TYPE: Mapping[str, Sensor[Any]]
actuators	A mapping of actuator names to actuator instances. TYPE: Mapping[str, Actuator[Any]]

RETURNS	DESCRIPTION
ModularEnvironment[Mapping[str, Any], Mapping[str, Any]]	A modular environment that uses composite sensors and actuators.

Example

env = ModularEnvironment.from_dict(
    sensors={"camera": CameraSensor(), "lidar": LidarSensor()},
    actuators={"motor": MotorActuator(), "gripper": GripperActuator()}
)

Source code in src/pamiq_core/interaction/modular_env.py

@staticmethod
def from_dict(
    sensors: Mapping[str, Sensor[Any]], actuators: Mapping[str, Actuator[Any]]
) -> ModularEnvironment[Mapping[str, Any], Mapping[str, Any]]:
    """Create a modular environment from dictionaries of sensors and
    actuators.

    Args:
        sensors: A mapping of sensor names to sensor instances.
        actuators: A mapping of actuator names to actuator instances.

    Returns:
        A modular environment that uses composite sensors and actuators.

    Example:
        ```python
        env = ModularEnvironment.from_dict(
            sensors={"camera": CameraSensor(), "lidar": LidarSensor()},
            actuators={"motor": MotorActuator(), "gripper": GripperActuator()}
        )
        ```
    """
    return ModularEnvironment(SensorsDict(sensors), ActuatorsDict(actuators))

SensorsDict

Bases: Sensor[Mapping[str, Any]], UserDict[str, Sensor[Any]]

Dictionary of sensors that acts as a composite sensor.

This class allows grouping multiple sensors together under a single interface. When read, it returns a mapping of sensor names to their readings, providing a way to collect data from multiple sensors simultaneously.

All lifecycle events (setup, teardown, pause, resume) and state persistence operations are propagated to all contained sensors.

read

read() -> Mapping[str, Any]

Read data from all contained sensors.

RETURNS	DESCRIPTION
Mapping[str, Any]	A mapping from sensor names to their respective readings.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def read(self) -> Mapping[str, Any]:
    """Read data from all contained sensors.

    Returns:
        A mapping from sensor names to their respective readings.
    """
    return {k: v.read() for k, v in self.items()}

setup

setup() -> None

Set up all contained sensors.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def setup(self) -> None:
    """Set up all contained sensors."""
    super().setup()
    for v in self.values():
        v.setup()

teardown

teardown() -> None

Clean up resources for all contained sensors.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def teardown(self) -> None:
    """Clean up resources for all contained sensors."""
    super().teardown()
    for v in self.values():
        v.teardown()

save_state

save_state(path: Path) -> None

Save the state of all contained sensors.

Creates a directory at the given path and saves each sensor's state in a subdirectory named after its key in this dictionary.

PARAMETER	DESCRIPTION
path	Directory path where to save the state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def save_state(self, path: Path) -> None:
    """Save the state of all contained sensors.

    Creates a directory at the given path and saves each sensor's state
    in a subdirectory named after its key in this dictionary.

    Args:
        path: Directory path where to save the state.
    """
    super().save_state(path)
    path.mkdir()
    for k, v in self.items():
        v.save_state(path / k)

load_state

load_state(path: Path) -> None

Load the state of all contained sensors.

Loads each sensor's state from a subdirectory named after its key in this dictionary.

PARAMETER	DESCRIPTION
path	Directory path from where to load the state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def load_state(self, path: Path) -> None:
    """Load the state of all contained sensors.

    Loads each sensor's state from a subdirectory named after its key
    in this dictionary.

    Args:
        path: Directory path from where to load the state.
    """
    super().load_state(path)
    for k, v in self.items():
        v.load_state(path / k)

on_paused

on_paused() -> None

Handle system pause event for all contained sensors.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_paused(self) -> None:
    """Handle system pause event for all contained sensors."""
    super().on_paused()
    for v in self.values():
        v.on_paused()

on_resumed

on_resumed() -> None

Handle system resume event for all contained sensors.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_resumed(self) -> None:
    """Handle system resume event for all contained sensors."""
    super().on_resumed()
    for v in self.values():
        v.on_resumed()

ActuatorsDict

Bases: Actuator[Mapping[str, Any]], UserDict[str, Actuator[Any]]

Dictionary of actuators that acts as a composite actuator.

This class allows grouping multiple actuators together under a single interface. When operated, it distributes actions to individual actuators based on their keys.

All lifecycle events (setup, teardown, pause, resume) and state persistence operations are propagated to all contained actuators.

operate

operate(action: Mapping[str, Any]) -> None

Operate all contained actuators with their respective actions.

Distributes the actions to the appropriate actuators based on their keys.

PARAMETER	DESCRIPTION
action	A mapping from actuator names to their respective actions. Each action will be passed to the corresponding actuator. TYPE: Mapping[str, Any]

RAISES	DESCRIPTION
KeyError	If action doesn't contain a required actuator key.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def operate(self, action: Mapping[str, Any]) -> None:
    """Operate all contained actuators with their respective actions.

    Distributes the actions to the appropriate actuators based on their keys.

    Args:
        action: A mapping from actuator names to their respective actions.
               Each action will be passed to the corresponding actuator.

    Raises:
        KeyError: If action doesn't contain a required actuator key.
    """
    for k, v in self.items():
        v.operate(action[k])

setup

setup() -> None

Set up all contained actuators.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def setup(self) -> None:
    """Set up all contained actuators."""
    super().setup()
    for v in self.values():
        v.setup()

teardown

teardown() -> None

Clean up resources for all contained actuators.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def teardown(self) -> None:
    """Clean up resources for all contained actuators."""
    super().teardown()
    for v in self.values():
        v.teardown()

save_state

save_state(path: Path) -> None

Save the state of all contained actuators.

Creates a directory at the given path and saves each actuator's state in a subdirectory named after its key in this dictionary.

PARAMETER	DESCRIPTION
path	Directory path where to save the state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def save_state(self, path: Path) -> None:
    """Save the state of all contained actuators.

    Creates a directory at the given path and saves each actuator's state
    in a subdirectory named after its key in this dictionary.

    Args:
        path: Directory path where to save the state.
    """
    super().save_state(path)
    path.mkdir()
    for k, v in self.items():
        v.save_state(path / k)

load_state

load_state(path: Path) -> None

Load the state of all contained actuators.

Loads each actuator's state from a subdirectory named after its key in this dictionary.

PARAMETER	DESCRIPTION
path	Directory path from where to load the state. TYPE: Path

Source code in src/pamiq_core/interaction/modular_env.py

@override
def load_state(self, path: Path) -> None:
    """Load the state of all contained actuators.

    Loads each actuator's state from a subdirectory named after its key
    in this dictionary.

    Args:
        path: Directory path from where to load the state.
    """
    super().load_state(path)
    for k, v in self.items():
        v.load_state(path / k)

on_paused

on_paused() -> None

Handle system pause event for all contained actuators.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_paused(self) -> None:
    """Handle system pause event for all contained actuators."""
    super().on_paused()
    for v in self.values():
        v.on_paused()

on_resumed

on_resumed() -> None

Handle system resume event for all contained actuators.

Source code in src/pamiq_core/interaction/modular_env.py

@override
def on_resumed(self) -> None:
    """Handle system resume event for all contained actuators."""
    super().on_resumed()
    for v in self.values():
        v.on_resumed()

pamiq_core.interaction.wrappers

Wrapper

Bases: ABC, InteractionEventMixin, PersistentStateMixin, ThreadEventMixin

Base wrapper class for transforming values.

This abstract class provides an interface for wrapping and transforming values from one type to another. Subclasses should implement the wrap method to define the specific transformation.

wrap abstractmethod

wrap(value: T) -> W

Transform the input value to another value.

PARAMETER	DESCRIPTION
value	Input value to be transformed. TYPE: T

RETURNS	DESCRIPTION
W	Transformed value.

Source code in src/pamiq_core/interaction/wrappers.py

@abstractmethod
def wrap(self, value: T) -> W:
    """Transform the input value to another value.

    Args:
        value: Input value to be transformed.

    Returns:
        Transformed value.
    """
    pass

__call__

__call__(value: T) -> W

Enable calling the wrapper as a function.

PARAMETER	DESCRIPTION
value	Input value to be transformed. TYPE: T

RETURNS	DESCRIPTION
W	Transformed value.

Source code in src/pamiq_core/interaction/wrappers.py

def __call__(self, value: T) -> W:
    """Enable calling the wrapper as a function.

    Args:
        value: Input value to be transformed.

    Returns:
        Transformed value.
    """
    return self.wrap(value)

wrap_sensor

wrap_sensor(sensor: Sensor[T]) -> SensorWrapper[T, W]

Create a SensorWrapper that combines this wrapper with the provided sensor.

This is a convenience method that applies the current wrapper to a sensor, creating a SensorWrapper that will transform the sensor's readings using this wrapper's transformation logic.

PARAMETER	DESCRIPTION
sensor	The sensor to wrap with this wrapper. TYPE: Sensor[T]

RETURNS	DESCRIPTION
SensorWrapper[T, W]	A new SensorWrapper that applies this wrapper's transformation to the sensor.

Source code in src/pamiq_core/interaction/wrappers.py

def wrap_sensor(self, sensor: Sensor[T]) -> SensorWrapper[T, W]:
    """Create a SensorWrapper that combines this wrapper with the provided
    sensor.

    This is a convenience method that applies the current wrapper to a sensor,
    creating a SensorWrapper that will transform the sensor's readings using
    this wrapper's transformation logic.

    Args:
        sensor: The sensor to wrap with this wrapper.

    Returns:
        A new SensorWrapper that applies this wrapper's transformation to the sensor.
    """
    return SensorWrapper(sensor, self)

wrap_actuator

wrap_actuator(actuator: Actuator[W]) -> ActuatorWrapper[W, T]

Create an ActuatorWrapper that combines this wrapper with the provided actuator.

This is a convenience method that applies the current wrapper to an actuator, creating an ActuatorWrapper that will transform actions before passing them to the actuator using this wrapper's transformation logic.

PARAMETER	DESCRIPTION
actuator	The actuator to wrap with this wrapper. TYPE: Actuator[W]

RETURNS	DESCRIPTION
ActuatorWrapper[W, T]	A new ActuatorWrapper that applies this wrapper's transformation to actions.

Source code in src/pamiq_core/interaction/wrappers.py

def wrap_actuator(self, actuator: Actuator[W]) -> ActuatorWrapper[W, T]:
    """Create an ActuatorWrapper that combines this wrapper with the
    provided actuator.

    This is a convenience method that applies the current wrapper to an actuator,
    creating an ActuatorWrapper that will transform actions before passing them
    to the actuator using this wrapper's transformation logic.

    Args:
        actuator: The actuator to wrap with this wrapper.

    Returns:
        A new ActuatorWrapper that applies this wrapper's transformation to actions.
    """
    return ActuatorWrapper(actuator, self)

LambdaWrapper

LambdaWrapper(func: Callable[[T], W])

Bases: Wrapper[T, W]

Wrapper that uses a callable function to transform values.

This wrapper enables using lambda functions or any callable to perform the transformation.

Initialize with a transformation function.

PARAMETER	DESCRIPTION
func	Function that transforms input values of type T to type W. TYPE: Callable[[T], W]

Source code in src/pamiq_core/interaction/wrappers.py

def __init__(self, func: Callable[[T], W]) -> None:
    """Initialize with a transformation function.

    Args:
        func: Function that transforms input values of type T to type W.
    """
    super().__init__()
    self._func = func

wrap

wrap(value: T) -> W

Transform the value using the provided function.

PARAMETER	DESCRIPTION
value	Input value to be transformed. TYPE: T

RETURNS	DESCRIPTION
W	Transformed value.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def wrap(self, value: T) -> W:
    """Transform the value using the provided function.

    Args:
        value: Input value to be transformed.

    Returns:
        Transformed value.
    """
    return self._func(value)

EnvironmentWrapper

EnvironmentWrapper(
    env: Environment[ObsType, ActType],
    obs_wrapper: Wrapper[ObsType, WrappedObsType] | Callable[[ObsType], WrappedObsType],
    act_wrapper: Wrapper[WrappedActType, ActType] | Callable[[WrappedActType], ActType],
)

Bases: Environment[WrappedObsType, WrappedActType]

Wrapper for Environment that transforms observations and actions.

This wrapper transforms the observations returned by the wrapped environment and the actions passed to it.

Initialize with an environment and wrappers.

PARAMETER	DESCRIPTION
env	The environment to wrap. TYPE: Environment[ObsType, ActType]
obs_wrapper	Wrapper for transforming observations from the environment. TYPE: Wrapper[ObsType, WrappedObsType] | Callable[[ObsType], WrappedObsType]
act_wrapper	Wrapper for transforming actions before passing to the environment. TYPE: Wrapper[WrappedActType, ActType] | Callable[[WrappedActType], ActType]

Source code in src/pamiq_core/interaction/wrappers.py

def __init__(
    self,
    env: Environment[ObsType, ActType],
    obs_wrapper: Wrapper[ObsType, WrappedObsType]
    | Callable[[ObsType], WrappedObsType],
    act_wrapper: Wrapper[WrappedActType, ActType]
    | Callable[[WrappedActType], ActType],
) -> None:
    """Initialize with an environment and wrappers.

    Args:
        env: The environment to wrap.
        obs_wrapper: Wrapper for transforming observations from the environment.
        act_wrapper: Wrapper for transforming actions before passing to the environment.
    """
    self.env = env
    self._obs_wrapper = _ensure_wrapper(obs_wrapper)
    self._act_wrapper = _ensure_wrapper(act_wrapper)

observe

observe() -> WrappedObsType

Get wrapped observation from the environment.

RETURNS	DESCRIPTION
WrappedObsType	Transformed observation.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def observe(self) -> WrappedObsType:
    """Get wrapped observation from the environment.

    Returns:
        Transformed observation.
    """
    return self._obs_wrapper(self.env.observe())

affect

affect(action: WrappedActType) -> None

Apply wrapped action to the environment.

PARAMETER	DESCRIPTION
action	Action to be transformed and applied. TYPE: WrappedActType

Source code in src/pamiq_core/interaction/wrappers.py

@override
def affect(self, action: WrappedActType) -> None:
    """Apply wrapped action to the environment.

    Args:
        action: Action to be transformed and applied.
    """
    self.env.affect(self._act_wrapper(action))

setup

setup() -> None

Set up the wrapped environment and wrappers.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def setup(self) -> None:
    """Set up the wrapped environment and wrappers."""
    self.env.setup()
    self._obs_wrapper.setup()
    self._act_wrapper.setup()

teardown

teardown() -> None

Clean up the wrapped environment and wrappers.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def teardown(self) -> None:
    """Clean up the wrapped environment and wrappers."""
    self.env.teardown()
    self._obs_wrapper.teardown()
    self._act_wrapper.teardown()

save_state

save_state(path: Path) -> None

Save state of the wrapped environment and wrappers.

PARAMETER	DESCRIPTION
path	Directory path where to save the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def save_state(self, path: Path) -> None:
    """Save state of the wrapped environment and wrappers.

    Args:
        path: Directory path where to save the state.
    """
    path.mkdir()
    self.env.save_state(path / "env")
    self._obs_wrapper.save_state(path / "obs_wrapper")
    self._act_wrapper.save_state(path / "act_wrapper")

load_state

load_state(path: Path) -> None

Load state of the wrapped environment and wrappers.

PARAMETER	DESCRIPTION
path	Directory path from where to load the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def load_state(self, path: Path) -> None:
    """Load state of the wrapped environment and wrappers.

    Args:
        path: Directory path from where to load the state.
    """
    self.env.load_state(path / "env")
    self._obs_wrapper.load_state(path / "obs_wrapper")
    self._act_wrapper.load_state(path / "act_wrapper")

on_paused

on_paused() -> None

The method to be called when the thread is paused.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_paused(self) -> None:
    """The method to be called when the thread is paused."""
    super().on_paused()
    self.env.on_paused()
    self._obs_wrapper.on_paused()
    self._act_wrapper.on_paused()

on_resumed

on_resumed() -> None

The method to be called when the thread is on_resumed.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_resumed(self) -> None:
    """The method to be called when the thread is on_resumed."""
    super().on_resumed()
    self.env.on_resumed()
    self._obs_wrapper.on_resumed()
    self._act_wrapper.on_resumed()

SensorWrapper

SensorWrapper(sensor: Sensor[T], wrapper: Wrapper[T, W] | Callable[[T], W])

Bases: Sensor[W]

Wrapper for Sensor that transforms sensor readings.

This wrapper applies a transformation to the readings from the wrapped sensor.

Initialize with a sensor and a wrapper.

PARAMETER	DESCRIPTION
sensor	The sensor to wrap. TYPE: Sensor[T]
wrapper	Wrapper for transforming sensor readings. TYPE: Wrapper[T, W] | Callable[[T], W]

Source code in src/pamiq_core/interaction/wrappers.py

def __init__(
    self, sensor: Sensor[T], wrapper: Wrapper[T, W] | Callable[[T], W]
) -> None:
    """Initialize with a sensor and a wrapper.

    Args:
        sensor: The sensor to wrap.
        wrapper: Wrapper for transforming sensor readings.
    """
    self.sensor = sensor
    self._wrapper = _ensure_wrapper(wrapper)

read

read() -> W

Get transformed reading from the sensor.

RETURNS	DESCRIPTION
W	Transformed sensor reading.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def read(self) -> W:
    """Get transformed reading from the sensor.

    Returns:
        Transformed sensor reading.
    """
    return self._wrapper(self.sensor.read())

setup

setup() -> None

Set up the wrapped sensor and wrapper.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def setup(self) -> None:
    """Set up the wrapped sensor and wrapper."""
    self.sensor.setup()
    self._wrapper.setup()

teardown

teardown() -> None

Clean up the wrapped sensor and wrapper.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def teardown(self) -> None:
    """Clean up the wrapped sensor and wrapper."""
    self.sensor.teardown()
    self._wrapper.teardown()

save_state

save_state(path: Path) -> None

Save state of the wrapped sensor and wrapper.

PARAMETER	DESCRIPTION
path	Directory path where to save the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def save_state(self, path: Path) -> None:
    """Save state of the wrapped sensor and wrapper.

    Args:
        path: Directory path where to save the state.
    """
    path.mkdir()
    self.sensor.save_state(path / "sensor")
    self._wrapper.save_state(path / "wrapper")

load_state

load_state(path: Path) -> None

Load state of the wrapped sensor and wrapper.

PARAMETER	DESCRIPTION
path	Directory path from where to load the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def load_state(self, path: Path) -> None:
    """Load state of the wrapped sensor and wrapper.

    Args:
        path: Directory path from where to load the state.
    """
    self.sensor.load_state(path / "sensor")
    self._wrapper.load_state(path / "wrapper")

on_paused

on_paused() -> None

The method to be called when the thread is paused.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_paused(self) -> None:
    """The method to be called when the thread is paused."""
    super().on_paused()
    self.sensor.on_paused()
    self._wrapper.on_paused()

on_resumed

on_resumed() -> None

The method to be called when the thread is resumed.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_resumed(self) -> None:
    """The method to be called when the thread is resumed."""
    super().on_resumed()
    self.sensor.on_resumed()
    self._wrapper.on_resumed()

ActuatorWrapper

ActuatorWrapper(actuator: Actuator[T], wrapper: Wrapper[W, T] | Callable[[W], T])

Bases: Actuator[W]

Wrapper for Actuator that transforms actions.

This wrapper applies a transformation to actions before passing them to the wrapped actuator.

Initialize with an actuator and a wrapper.

PARAMETER	DESCRIPTION
actuator	The actuator to wrap. TYPE: Actuator[T]
wrapper	Wrapper for transforming actions. TYPE: Wrapper[W, T] | Callable[[W], T]

Source code in src/pamiq_core/interaction/wrappers.py

def __init__(
    self, actuator: Actuator[T], wrapper: Wrapper[W, T] | Callable[[W], T]
) -> None:
    """Initialize with an actuator and a wrapper.

    Args:
        actuator: The actuator to wrap.
        wrapper: Wrapper for transforming actions.
    """
    self.actuator = actuator
    self._wrapper = _ensure_wrapper(wrapper)

operate

operate(action: W) -> None

Apply transformed action to the actuator.

PARAMETER	DESCRIPTION
action	Action to be transformed and applied. TYPE: W

Source code in src/pamiq_core/interaction/wrappers.py

@override
def operate(self, action: W) -> None:
    """Apply transformed action to the actuator.

    Args:
        action: Action to be transformed and applied.
    """
    self.actuator.operate(self._wrapper(action))

setup

setup() -> None

Set up the wrapped actuator and wrapper.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def setup(self) -> None:
    """Set up the wrapped actuator and wrapper."""
    self.actuator.setup()
    self._wrapper.setup()

teardown

teardown() -> None

Clean up the wrapped actuator and wrapper.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def teardown(self) -> None:
    """Clean up the wrapped actuator and wrapper."""
    self.actuator.teardown()
    self._wrapper.teardown()

save_state

save_state(path: Path) -> None

Save state of the wrapped actuator and wrapper.

PARAMETER	DESCRIPTION
path	Directory path where to save the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def save_state(self, path: Path) -> None:
    """Save state of the wrapped actuator and wrapper.

    Args:
        path: Directory path where to save the state.
    """
    path.mkdir()
    self.actuator.save_state(path / "actuator")
    self._wrapper.save_state(path / "wrapper")

load_state

load_state(path: Path) -> None

Load state of the wrapped actuator and wrapper.

PARAMETER	DESCRIPTION
path	Directory path from where to load the state. TYPE: Path

Source code in src/pamiq_core/interaction/wrappers.py

@override
def load_state(self, path: Path) -> None:
    """Load state of the wrapped actuator and wrapper.

    Args:
        path: Directory path from where to load the state.
    """
    self.actuator.load_state(path / "actuator")
    self._wrapper.load_state(path / "wrapper")

on_paused

on_paused() -> None

The method to be called when the thread is paused.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_paused(self) -> None:
    """The method to be called when the thread is paused."""
    super().on_paused()
    self.actuator.on_paused()
    self._wrapper.on_paused()

on_resumed

on_resumed() -> None

The method to be called when the thread is resumed.

Source code in src/pamiq_core/interaction/wrappers.py

@override
def on_resumed(self) -> None:
    """The method to be called when the thread is resumed."""
    super().on_resumed()
    self.actuator.on_resumed()
    self._wrapper.on_resumed()