mne_rt.protocols.MultiBandProtocol#

class mne_rt.protocols.MultiBandProtocol(protocol_up: Any, protocol_down: Any, require_both: bool = True, up_label: str = 'up_band', down_label: str = 'down_band')#

Bases: object

Reward protocol for simultaneous two-band control.

Wraps two inner protocols (one for up-regulation, one for down-regulation) and issues a combined reward only when BOTH criteria are met simultaneously (or either, if require_both=False).

The combined magnitude is the geometric mean of the two individual magnitudes to ensure both bands contribute equally. When one magnitude is zero the arithmetic mean is used as a fallback so that partial rewards are still numerically meaningful.

Parameters:

protocol_upprotocol with evaluate(value) -> (bool, float): Protocol applied to the up-regulation value (e.g., alpha power).
protocol_downprotocol with evaluate(value) -> (bool, float): Protocol applied to the down-regulation value (e.g., theta power).
require_bothbool, default True: If True, both criteria must be met for a reward (AND logic). If False, either criterion suffices (OR logic).
up_labelstr, default “up_band”: Human-readable label for the up-regulation band (used in __repr__ and logging).
down_labelstr, default “down_band”: Human-readable label for the down-regulation band (used in __repr__ and logging).

Notes

Call evaluate(up_value, down_value) with TWO positional arguments — one from each modality/band. Configure RTStream with modality=["sensor_power_alpha", "sensor_power_theta"] (or similar) and unpack the two returned values before each call.

Examples

Alpha-up / theta-down simultaneous reward:

from mne_rt.protocols import ZScoreProtocol
from mne_rt.protocols.multiband import MultiBandProtocol

alpha_proto = ZScoreProtocol(direction="up")
theta_proto = ZScoreProtocol(direction="down")

proto = MultiBandProtocol(
    protocol_up=alpha_proto,
    protocol_down=theta_proto,
    up_label="alpha",
    down_label="theta",
)
for alpha_val, theta_val in zip(alpha_stream, theta_stream):
    crossed, magnitude = proto.evaluate(alpha_val, theta_val)
    if crossed:
        send_reward(magnitude)

Added in version 1.0.0.

__init__(protocol_up: Any, protocol_down: Any, require_both: bool = True, up_label: str = 'up_band', down_label: str = 'down_band') → None[source]#

Methods

`__init__`(protocol_up, protocol_down[, ...])
`evaluate`(up_value, down_value)	Evaluate one pair of NF values and return (crossed, magnitude).
`reset`()	Reset both inner protocols and the evaluation counter.

Attributes

n_evaluated

Total number of value-pair evaluations since init or last reset.

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

Evaluate one pair of NF values and return (crossed, magnitude).

Delegates to both inner protocols, then combines the results according to require_both. The combined magnitude is the geometric mean of the two individual magnitudes; when one is zero the arithmetic mean is used as fallback.

Parameters:

up_valuefloat: Current NF feature value for the up-regulation band.
down_valuefloat: Current NF feature value for the down-regulation band.

Returns:

crossedbool: True if the combined criterion is met.
magnitudefloat: Non-negative combined reward magnitude.

property n_evaluated: int#: Total number of value-pair evaluations since init or last reset.

reset() → None[source]#

Reset both inner protocols and the evaluation counter.

Calls reset() on each inner protocol if that method exists.