mne_rt.protocols.OperantProtocol#

class mne_rt.protocols.OperantProtocol(base_protocol: Any, schedule: str = 'FR', ratio: int = 5, interval: float = 30.0, rng_seed: int | None = None)#

Bases: object

Operant conditioning reinforcement-schedule wrapper.

Wraps any inner NF protocol and applies a classical operant conditioning schedule to determine whether a reward is actually delivered. The inner protocol is always evaluated so that its internal state (running statistics, adaptive threshold, etc.) advances correctly; the schedule only decides whether to pass the reward through to the caller.

Supported schedules

"FR" (Fixed Ratio): Deliver reward on every ratio-th hit (hit = inner protocol returned crossed=True).
"VR" (Variable Ratio): Deliver reward with probability 1 / ratio on each hit (geometric distribution; same average number of hits per reward as FR).
"FI" (Fixed Interval): Deliver reward for the first hit that occurs after interval seconds have elapsed since the last reward.
"VI" (Variable Interval): Deliver reward for the first hit that occurs after a random interval sampled from an exponential distribution with mean interval seconds.

Parameters:

base_protocolany protocol with evaluate(value) -> (bool, float): The inner NF protocol whose output is filtered by the schedule.
schedule{“FR”, “VR”, “FI”, “VI”}: Reinforcement schedule identifier.
ratioint: Required for "FR" and "VR" schedules. For FR: reward is delivered on every ratio-th hit. For VR: reward probability per hit is 1 / ratio. Must be >= 1. Default is 5.
intervalfloat: Required for "FI" and "VI" schedules, in seconds. For FI: minimum fixed interval between rewards. For VI: mean interval (exponential distribution). Must be > 0. Default is 30.0.
rng_seedint | None: Seed for the NumPy random generator used by "VR" and "VI" schedules. Default is None (non-deterministic).

Raises:

ValueError: If schedule is not one of the valid identifiers, ratio < 1, or interval <= 0.

Notes

The internal clock is started on the first call to evaluate() using time.monotonic(), not at construction time. This avoids artificially burning interval time between object creation and the start of the session.

reset() resets all internal counters, hit/reward tallies, and the clock. It also calls base_protocol.reset() if that method exists.

Examples

Fixed-ratio schedule (reward every 5th hit):

from mne_rt.protocols import ZScoreProtocol
from mne_rt.protocols.operant import OperantProtocol

inner = ZScoreProtocol(direction="up")
proto = OperantProtocol(inner, schedule="FR", ratio=5)
for value in nf_stream:
    crossed, magnitude = proto.evaluate(value)
    if crossed:
        deliver_reward(magnitude)

Variable-interval schedule (reward at most once per ~30 s on average):

proto = OperantProtocol(inner, schedule="VI", interval=30.0, rng_seed=0)

Added in version 1.0.0.

__init__(base_protocol: Any, schedule: str = 'FR', ratio: int = 5, interval: float = 30.0, rng_seed: int | None = None) → None[source]#

Methods

`__init__`(base_protocol[, schedule, ratio, ...])
`evaluate`(value)	Evaluate one NF value and apply the reinforcement schedule.
`reset`()	Reset all internal state including counters and the clock.

Attributes

`current_threshold`	Pass-through threshold from the base protocol, if available.
`n_hits`	Total hits (crossed=True from base protocol) since init or reset.
`n_rewards`	Total rewards delivered by the schedule since init or reset.
`reward_rate`	Fraction of hits that resulted in a reward (0–1).

property current_threshold: float | None#

Pass-through threshold from the base protocol, if available.

Returns None if the base protocol does not expose a threshold attribute.

evaluate(value: float) → tuple[bool, float][source]#

Evaluate one NF value and apply the reinforcement schedule.

Delegates to the inner protocol unconditionally, then applies the schedule logic to decide whether to pass the reward through.

Parameters:

valuefloat: Current NF feature value.

Returns:

crossedbool: True if the schedule releases a reward on this evaluation.
magnitudefloat: Reward magnitude from the inner protocol when crossed is True; 0.0 otherwise.

Notes

For interval schedules ("FI", "VI") the clock is initialised on the first call.

property n_hits: int#: Total hits (crossed=True from base protocol) since init or reset.

property n_rewards: int#: Total rewards delivered by the schedule since init or reset.

reset() → None[source]#

Reset all internal state including counters and the clock.

Also calls base_protocol.reset() if that method exists. All constructor parameters are preserved.

property reward_rate: float#

Fraction of hits that resulted in a reward (0–1).

Returns 0.0 before any hits are recorded.