api.task_logic¶

Task¶

Trial Generators¶

Warmup¶

Coupled¶

Composite¶

Integration Test¶

Trial¶

class aind_behavior_dynamic_foraging.task_logic.AindDynamicForagingTaskLogic(*, name: Literal['AindDynamicForaging'] = 'AindDynamicForaging', description: str = '', task_parameters: AindDynamicForagingTaskParameters, version: Literal['0.0.2-rc33'] = '0.0.2-rc33', stage_name: str | None = None)[source]¶

Bases: Task

Main task logic model for the AIND Dynamic Foraging task.

This is the top-level class that encapsulates the complete task logic specification for the dynamic foraging behavioral experiment. It includes all task parameters, environment specifications, and control settings.

model_config = {'extra': 'forbid', 'str_strip_whitespace': True, 'strict': True, 'validate_assignment': True, 'validate_default': True}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: Literal['AindDynamicForaging'][source]¶

task_parameters: AindDynamicForagingTaskParameters[source]¶

version: Literal['0.0.2-rc33'][source]¶

class aind_behavior_dynamic_foraging.task_logic.AindDynamicForagingTaskParameters(*, rng_seed: float | None = None, aind_behavior_services_pkg_version: Annotated[Literal['0.13.5'], _PydanticGeneralMetadata(pattern='^(0|[1-9]\\d*)\\.(0|[1-9]\\d*)\\.(0|[1-9]\\d*)(?:-((?:0|[1-9]\\d*|\\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\\.(?:0|[1-9]\\d*|\\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\\+([0-9a-zA-Z-]+(?:\\.[0-9a-zA-Z-]+)*))?$')] = '0.13.5', reward_size: RewardSize = RewardSize(right_value_volume=3.0, left_value_volume=3.0), trial_generator: TrialGeneratorSpec = IntegrationTestTrialGeneratorSpec(type='IntegrationTestTrialGenerator'), **extra_data: Any)[source]¶

Bases: TaskParameters

Complete parameter specification for the AIND Dynamic Foraging task.

This class contains all configurable parameters for the Dynamic Foraging task, including environment structure, task mode settings, operation control, and numerical updaters for dynamic parameter modification.

model_config = {'extra': 'allow', 'str_strip_whitespace': True, 'strict': True, 'validate_assignment': True, 'validate_default': True}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reward_size: RewardSize[source]¶

trial_generator: TrialGeneratorSpec[source]¶

class aind_behavior_dynamic_foraging.task_logic.RewardSize(*, right_value_volume: float, left_value_volume: float)[source]¶

Bases: BaseModel

left_value_volume: float[source]¶

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

right_value_volume: float[source]¶

class aind_behavior_dynamic_foraging.task_logic.trial_models.AuditorySecondaryReinforcer(*, type: Literal['Auditory'] = 'Auditory')[source]¶

Bases: BaseModel

Represents an auditory secondary reinforcer.

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['Auditory'][source]¶

class aind_behavior_dynamic_foraging.task_logic.trial_models.Metadata(*, p_reward_left: Annotated[float | None, Ge(ge=0), Le(le=1)] = None, p_reward_right: Annotated[float | None, Ge(ge=0), Le(le=1)] = None, extra: Annotated[Any, SerializeAsAny()] | None = None)[source]¶

Bases: BaseModel

Metadata for trial. These fields will NOT be used by the task engine.

extra: Annotated[Any, SerializeAsAny()] | None[source]¶

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_reward_left: float | None[source]¶

p_reward_right: float | None[source]¶

class aind_behavior_dynamic_foraging.task_logic.trial_models.QuickRetractSettings(*, enable_during_quiescence: bool = False, time_to_reset_during_quiescence: Distribution, <aind_behavior_services.schema._SgenTypenameAnnotation object at 0x7fa23dbb2660>]=Scalar(family=<DistributionFamily.SCALAR: 'Scalar'>, distribution_parameters=ScalarDistributionParameter(family=<DistributionFamily.SCALAR: 'Scalar'>, value=1.0), truncation_parameters=None, scaling_parameters=None), enable_on_response: bool = False)[source]¶

Bases: BaseModel

Settings for the quick retract feature.

enable_during_quiescence: bool[source]¶

enable_on_response: bool[source]¶

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

time_to_reset_during_quiescence: _SgenTypenameAnnotation object at 0x7fa23dbb2660>][source]¶

type aind_behavior_dynamic_foraging.task_logic.trial_models.SecondaryReinforcer = Annotated[AuditorySecondaryReinforcer, FieldInfo(annotation=NoneType, required=True, description='Type of secondary reinforcer', discriminator='type')][source]¶

class aind_behavior_dynamic_foraging.task_logic.trial_models.Trial(*, p_reward_left: Annotated[float, Ge(ge=0), Le(le=1)] = 1.0, p_reward_right: Annotated[float, Ge(ge=0), Le(le=1)] = 1.0, reward_consumption_duration: Annotated[float, Ge(ge=0)] = 5.0, reward_delay_duration: Annotated[float, Ge(ge=0)] = 0.0, secondary_reinforcer: SecondaryReinforcer | None = None, response_deadline_duration: Annotated[float, Ge(ge=0)] = 5.0, quick_retract_settings: QuickRetractSettings | None = None, quiescence_period_duration: Annotated[float, Ge(ge=0)] = 0.5, inter_trial_interval_duration: Annotated[float, Ge(ge=0.5)] = 5.0, is_auto_response_right: bool | None = None, lickspout_offset_delta: float = 0.0, metadata: Metadata | None = None)[source]¶

Bases: BaseModel

Represents a single trial that can be instantiated by the Bonsai state machine.

inter_trial_interval_duration: float[source]¶

is_auto_response_right: bool | None[source]¶

lickspout_offset_delta: float[source]¶

metadata: Metadata | None[source]¶

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_reward_left: float[source]¶

p_reward_right: float[source]¶

quick_retract_settings: QuickRetractSettings | None[source]¶

quiescence_period_duration: float[source]¶

response_deadline_duration: float[source]¶

reward_consumption_duration: float[source]¶

reward_delay_duration: float[source]¶

secondary_reinforcer: SecondaryReinforcer | None[source]¶

class aind_behavior_dynamic_foraging.task_logic.trial_models.TrialOutcome(*, trial: Trial, is_right_choice: bool | None, is_rewarded: bool)[source]¶

Bases: BaseModel

Represents the outcome of a single trial.

is_rewarded: bool[source]¶

is_right_choice: bool | None[source]¶

model_config = {}[source]¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

trial: Trial[source]¶

aind_behavior_dynamic_foraging.task_logic.trial_models.scalar(value: float) → Scalar[source]¶: Helper function to create a Scalar distribution.

type aind_behavior_dynamic_foraging.task_logic.trial_generators.TrialGeneratorSpec = Annotated[UncoupledTrialGeneratorSpec | BlockBasedTrialGeneratorSpec | BaseCoupledTrialGeneratorSpec | CoupledWarmupTrialGeneratorSpec | CoupledTrialGeneratorSpec | IntegrationTestTrialGeneratorSpec | TrialGeneratorCompositeSpec[TrialGeneratorSpec], FieldInfo(annotation=NoneType, required=True, description='Type of trial generator', discriminator='type')][source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators._base.BaseTrialGeneratorSpecModel(*, type: str)[source]

Bases: BaseModel, ABC

Base model for trial generator specifications.

abstractmethod create_generator() → ITrialGenerator[source]: Create a trial generator instance from the specification.

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: str[source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators._base.ITrialGenerator(*args, **kwargs)[source]

Bases: Protocol

Interface for trial generators.

next() → Trial | None[source]: Return the next trial to run. Return None if there are no more trials to run.

update(outcome: TrialOutcome | str) → None[source]: Update the trial generator with the outcome of the previous trial.

class aind_behavior_dynamic_foraging.task_logic.trial_generators.composite_trial_generator.TrialGeneratorComposite(spec: TrialGeneratorCompositeSpec[BaseTrialGeneratorSpecModel])[source]

Bases: ITrialGenerator

A composite trial generator that concatenates multiple trial generators.

When the current generator’s next() method returns None, the composite automatically moves to the next generator in the list.

next() → Trial | None[source]

Get the next trial from the current generator.

If the current generator returns None, automatically advance to the next generator in the list. Returns None only when all generators are exhausted.

Returns:: The next Trial, or None if all generators are exhausted

update(outcome: TrialOutcome | str) → None[source]

Update the current active generator with the trial outcome.

Parameters:: outcome – The outcome of the last trial

class aind_behavior_dynamic_foraging.task_logic.trial_generators.composite_trial_generator.TrialGeneratorCompositeSpec(*, type: Literal['TrialGeneratorComposite'] = 'TrialGeneratorComposite', generators: Annotated[list[Annotated[_TSpec, SerializeAsAny()]], MinLen(min_length=1)])[source]

Bases: BaseTrialGeneratorSpecModel, Generic[_TSpec]

Specification for a composite trial generator that concatenates multiple generators.

create_generator() → TrialGeneratorComposite[source]: Create a trial generator instance from the specification.

generators: list[Annotated[_TSpec, SerializeAsAny()]][source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['TrialGeneratorComposite'][source]

Bases: BaseTrialGeneratorSpecModel, Generic[_TSpec]

Specification for a composite trial generator that concatenates multiple generators.

create_generator() → TrialGeneratorComposite[source]: Create a trial generator instance from the specification.

generators: list[Annotated[_TSpec, SerializeAsAny()]][source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['TrialGeneratorComposite'][source]

type composite_trial_generator.TrialGeneratorSpec = Annotated[UncoupledTrialGeneratorSpec | BlockBasedTrialGeneratorSpec | BaseCoupledTrialGeneratorSpec | CoupledWarmupTrialGeneratorSpec | CoupledTrialGeneratorSpec | IntegrationTestTrialGeneratorSpec | TrialGeneratorCompositeSpec[TrialGeneratorSpec], FieldInfo(annotation=NoneType, required=True, description='Type of trial generator', discriminator='type')][source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator.AutoWaterParameters(*, min_ignored_trials: Annotated[int, Ge(ge=0)] = 3, min_unrewarded_trials: Annotated[int, Ge(ge=0)] = 3, reward_fraction: Annotated[float, Ge(ge=0), Le(le=1)] = 0.8)[source]

Bases: BaseModel

min_ignored_trials: int[source]

min_unrewarded_trials: int[source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reward_fraction: float[source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator.Block(*, p_right_reward: Annotated[float, Ge(ge=0), Le(le=1)], p_left_reward: Annotated[float, Ge(ge=0), Le(le=1)], right_length: Annotated[int, Ge(ge=0)], left_length: Annotated[int, Ge(ge=0)])[source]

Bases: BaseModel

left_length: int[source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_left_reward: float[source]

p_right_reward: float[source]

right_length: int[source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator.BlockBasedTrialGenerator(spec: BlockBasedTrialGeneratorSpec)[source]

Bases: ITrialGenerator, ABC

Abstract trial generator for block-based dynamic foraging tasks.

Manages block transitions, baiting logic, and trial generation. Subclasses must implement _are_end_conditions_met to define session termination logic.

spec[source]: The specification used to configure this generator.

is_right_choice_history[source]: Record of whether each trial was a right choice. None indicates no choice was made (e.g. missed trial).

reward_history[source]: Record of whether each trial resulted in a reward.

is_left_baited[source]: Whether the left port currently has a baited reward.

is_right_baited[source]: Whether the right port currently has a baited reward.

block: Block[source]

is_left_baited: bool[source]

is_right_baited: bool[source]

is_right_choice_history: list[bool | None][source]

next() → Trial | None[source]

Generates the next trial in the session.

Checks end conditions, samples timing parameters, and applies baiting logic if enabled. Returns None if the session should end.

Returns:: The next Trial, or None if end conditions are met.

reward_history: list[bool][source]

update(outcome: TrialOutcome | str)[source]: Updates generator state from the previous trial outcome. Records choice and reward history and manages baiting state. :param outcome: The TrialOutcome from the most recently completed trial.

class aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator.BlockBasedTrialGeneratorSpec(*, type: Literal['BlockBasedTrialGenerator'] = 'BlockBasedTrialGenerator', quiescent_duration: Distribution, <aind_behavior_services.schema._SgenTypenameAnnotation object at 0x7fa23dbb2660>]=ExponentialDistribution(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, distribution_parameters=ExponentialDistributionParameters(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, rate=1.0), truncation_parameters=TruncationParameters(truncation_mode='exclude', min=0.0, max=1.0), scaling_parameters=None), response_duration: Annotated[float, ~annotated_types.Ge(ge=0)] = 1.0, reward_consumption_duration: Annotated[float, ~annotated_types.Ge(ge=0)] = 3.0, inter_trial_interval_duration: Distribution, <aind_behavior_services.schema._SgenTypenameAnnotation object at 0x7fa23dbb2660>]=ExponentialDistribution(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, distribution_parameters=ExponentialDistributionParameters(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, rate=0.5), truncation_parameters=TruncationParameters(truncation_mode='exclude', min=1.0, max=8.0), scaling_parameters=None), block_length: Distribution, <aind_behavior_services.schema._SgenTypenameAnnotation object at 0x7fa23dbb2660>]=ExponentialDistribution(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, distribution_parameters=ExponentialDistributionParameters(family=<DistributionFamily.EXPONENTIAL: 'Exponential'>, rate=0.05), truncation_parameters=TruncationParameters(truncation_mode='exclude', min=20.0, max=60.0), scaling_parameters=None), autowater_parameters: AutoWaterParameters | None = AutoWaterParameters(min_ignored_trials=3, min_unrewarded_trials=3, reward_fraction=0.8), is_baiting: bool = False)[source]

Bases: BaseTrialGeneratorSpecModel

autowater_parameters: AutoWaterParameters | None[source]

block_length: _SgenTypenameAnnotation object at 0x7fa23dbb2660>][source]

create_generator() → BlockBasedTrialGenerator[source]: Create a trial generator instance from the specification.

inter_trial_interval_duration: _SgenTypenameAnnotation object at 0x7fa23dbb2660>][source]

is_baiting: bool[source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

quiescent_duration: _SgenTypenameAnnotation object at 0x7fa23dbb2660>][source]

response_duration: float[source]

reward_consumption_duration: float[source]

type: Literal['BlockBasedTrialGenerator'][source]

class aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator.BlockBasedTrialMetadata(*, is_autowater: bool = False)[source]

Bases: BaseModel

Metadata for block based trial. These fields will NOT be used by the task engine.

is_autowater: bool[source]

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aind_behavior_dynamic_foraging.task_logic.trial_generators.integration_test_trial_generator.IntegrationTestTrialGenerator(spec: IntegrationTestTrialGeneratorSpec)[source]

Bases: ITrialGenerator

next() → Trial | None[source]: Return the next trial to run. Return None if there are no more trials to run.

update(outcome: TrialOutcome | str) → None[source]: Update the trial generator with the outcome of the previous trial.

class aind_behavior_dynamic_foraging.task_logic.trial_generators.integration_test_trial_generator.IntegrationTestTrialGeneratorSpec(*, type: Literal['IntegrationTestTrialGenerator'] = 'IntegrationTestTrialGenerator')[source]

Bases: BaseTrialGeneratorSpecModel

create_generator() → IntegrationTestTrialGenerator[source]: Create a trial generator instance from the specification.

model_config = {}[source]: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['IntegrationTestTrialGenerator'][source]