volume_transmitter

volume_transmitter#

class brainpy.state.volume_transmitter(in_size=1, deliver_interval=1, min_delay=Quantity(1., 'ms'), name=None)#

NEST-compatible volume_transmitter support device.

volume_transmitter collects neuromodulatory spikes and periodically exposes their cumulative spike history to dopamine-modulated synapses (e.g. stdp_dopamine_synapse). It reproduces the NEST ring-buffer scheduling, trigger logic, and pseudo-spike reset conventions while exposing a Python batch-update API.

1. Discrete-Time State

Let simulation resolution be \(\Delta t\) (ms) and define the on-grid delivery stamp for the current step index \(n = \mathrm{round}(t / \Delta t)\) as \(s = n + 1\). Internal mutable state consists of:

\(P[s]\) — pending multiplicity map, dict[int, float], accumulating contributions scheduled for future delivery stamp s.
\(H\) — ordered spike-history list of spikecounter entries \((t_i, m_i)\), with time in ms and non-negative multiplicity.
Delivery metadata: last_delivery_spikes, last_delivery_time_ms, and delivery_count.

Immediately after init_state(), \(H = [(0.0,\; 0.0)]\) (a NEST-compatible pseudo-spike with zero multiplicity).

2. Update Equations and Trigger Rule

For each input item \(i\) with spike indicator \(x_i > 0\), an effective count \(c_i \ge 0\) is accumulated into \(P[s_i]\), where \(s_i\) is either stamp_steps[i] or the current stamp \(s\).

At each update() call, the pending multiplicity for stamp \(s\) is consumed:

\[m_s = P[s] \quad (\text{or } 0 \text{ if absent}), \qquad t_s = s \cdot \Delta t.\]

If \(m_s > 0\), append \((t_s,\, m_s)\) to \(H\).

The delivery period in steps is

\[T_s = k \cdot d_{\min,s}, \qquad d_{\min,s} = \mathrm{round}\!\left(\frac{d_{\min}}{\Delta t}\right),\]

where \(k = \texttt{deliver\_interval}\). A delivery trigger fires when \(s \bmod T_s = 0\).

On trigger:

Capture \(D = H\) as the delivered history (spikes at stamp \(s\) are included before the reset).
Store \(D\) and trigger time \(t_s\) in delivery metadata.
Increment the delivery counter.
Reset \(H = [(t_s,\; 0.0)]\) (new pseudo-spike).

3. Multiplicity Inference

Incoming arrays are flattened to one-dimensional vectors of length \(N\). Let \(x_j\) denote spikes[j]:

If multiplicities is None and all \(x_j\) are integer-like (within 1e-12 tolerance): \(c_j = \max(\mathrm{round}(x_j),\; 0)\).
If multiplicities is None and \(x_j\) contains non-integer values: \(c_j = \mathbf{1}[x_j > 0]\).
If multiplicities is provided with non-negative integers \(m_j\): \(c_j = m_j \,\mathbf{1}[x_j > 0]\).

4. Assumptions and Constraints

deliver_interval must be a scalar integer >= 1.
min_delay must be scalar, strictly positive, and an integer multiple of dt.
Simulation time t must be aligned to the simulation grid (enforced at each update() call).
If stamp_steps is provided, every entry must satisfy stamp_steps[i] >= current_stamp.

5. Computational Implications

For \(N\) incoming items per call, scheduling is \(O(N)\) via dictionary accumulation by target stamp. History memory grows linearly with the number of unique stamped events between consecutive triggers. Each trigger resets the history to a single pseudo-spike, bounding worst-case growth to one trigger period.

Parameters:

in_size (Size, optional) – Shape/size argument consumed by brainstate.nn.Dynamics. Stored for API compatibility with other device models; it does not affect transmitter state-update logic. Default is 1.
deliver_interval (ArrayLike, optional) – Scalar integer-like value (unitless) specifying the trigger period in units of min_delay. Converted via nearest-integer rounding and validated to be >= 1. Increasing this value reduces how often connected synapses receive delivered spike histories. Default is 1.
min_delay (saiunit.Quantity or float, optional) – Scalar effective global minimal synaptic delay. Unitful values are converted to ms; plain floats are interpreted as ms. Must be strictly positive and an integer multiple of the simulation dt at the time update() is called. Default is 1.0 * u.ms.
name (str or None, optional) – Optional node name forwarded to brainstate.nn.Dynamics. Default is None.

Parameter Mapping

Table 45 Mapping of constructor parameters to model symbols#
Parameter	Default	Math symbol	Semantics
`deliver_interval`	`1`	\(k\)	Number of minimal-delay intervals per delivery trigger.
`min_delay`	`1.0 * u.ms`	\(d_{\min}\)	Effective global minimal synaptic delay for trigger period.
`dt` (environment)	runtime	\(\Delta t\)	Simulation resolution for stamp conversion and ms time computation.
`delivery_period_steps`	runtime	\(T_s\)	\(k \cdot \mathrm{round}(d_{\min} / \Delta t)\).

Raises:

ValueError – If deliver_interval is non-scalar, not integer-like, or < 1 (raised during __init__).
ValueError – At update() time: if dt <= 0; if min_delay is not a positive integer multiple of dt; if time t is not grid-aligned; if multiplicities contains negative values; if stamp_steps contains past stamps; or if payload arrays are non-integer where integer values are required or have mismatched flattened sizes.
TypeError – If provided scalar/array inputs cannot be converted by saiunit or NumPy conversion paths.
KeyError – At update() time, if the simulation context is missing the required 't' or dt entries (depends on brainstate.environ behaviour).

Notes

deliver_spikes() returns the current (undelivered) history vector.
last_delivery_spikes stores the history snapshot delivered at the most recent trigger.
last_delivery_time stores the most recent trigger time in ms.
update() aggregates multiplicities exactly by delivery step, mirroring NEST’s internal ring-buffer logic.
handles_test_event() accepts only receptor type 0, matching the NEST volume_transmitter interface.
set_local_device_id() / get_local_device_id() are provided for compatibility with NEST’s device duplication logic.

References

Examples

Instantiate a transmitter with a two-step delivery period, inject two simultaneous spikes, and advance one more step:

>>> import brainstate
>>> import saiunit as u
>>> import numpy as np
>>> from brainpy.state import volume_transmitter
>>> with brainstate.environ.context(dt=0.1 * u.ms):
...     vt = volume_transmitter(deliver_interval=2, min_delay=0.3 * u.ms)
...     with brainstate.environ.context(t=0.0 * u.ms):
...         y0 = vt.update(
...             spikes=np.array([1.0, 1.0]),
...             multiplicities=np.array([1, 2]),
...         )
...     with brainstate.environ.context(t=0.5 * u.ms):
...         y1 = vt.update()
...     _ = (y0['triggered'], y1['triggered'])

Query transmitter state and delivery metadata via get():

>>> import brainstate
>>> import saiunit as u
>>> from brainpy.state import volume_transmitter
>>> with brainstate.environ.context(dt=0.1 * u.ms):
...     vt = volume_transmitter(deliver_interval=1, min_delay=0.1 * u.ms)
...     _ = vt.get('deliver_interval')
...     _ = vt.get('spike_history')

connect()[source]#

No-op compatibility hook matching the NEST device interface.

Provided so that code that calls connect() on NEST devices works without modification when targeting volume_transmitter.

deliver_spikes()[source]#

Return the current (undelivered) spike-history vector.

The history always contains at least one entry: the pseudo-spike spikecounter(t, 0.0) inserted by init_state() or the most recent trigger reset.

Returns:: Immutable copy of the current internal history list \(H\), ordered chronologically by delivery stamp.
Return type:: tuple[spikecounter, ]

flush()[source]#

Return a non-triggering snapshot of the current state.

Unlike update(), this method does not advance the simulation step, consume pending multiplicities, or reset the history. It is useful for inspecting state between update() calls (e.g. at the end of a simulation run).

Returns:

Dictionary with the following keys:

'triggered' — False (no trigger occurs on flush).
't_trig' — None (no trigger time).
'delivered_spikes' — empty tuple (no delivery).
'spike_history' — tuple[spikecounter, ...], the current history returned by deliver_spikes().

Return type:

dict

get(key='deliver_interval')[source]#

Query transmitter parameters and mutable state by string key.

Parameters:

key (str, optional) –

Selector string. Supported values:

'deliver_interval' — returns int.
'min_delay' — returns the stored min_delay value as passed to the constructor (saiunit.Quantity or float).
'local_device_id' — returns int.
'spike_history' — returns tuple[spikecounter, ...] (same as deliver_spikes()).
'last_delivery_spikes' — returns tuple[spikecounter, ...] (same as last_delivery_spikes).
'last_delivery_time' — returns float ms.
'n_deliveries' — returns int.

Default is 'deliver_interval'.

Returns:

The selected value. Type depends on key as described above.

Return type:

int or float or saiunit.Quantity or tuple[spikecounter, ]

Raises:

KeyError – If key is not one of the supported selector strings.

get_local_device_id()[source]#

Return the current local device ID as a Python int.

Returns:: Current value of the local device ID.
Return type:: int

handles_test_event(receptor_type)[source]#

Validate a receptor type identifier and return the accepted index.

Mirrors the NEST volume_transmitter::handles_test_event method. Only receptor type 0 is accepted; all other values raise ValueError.

Parameters:

receptor_type (ArrayLike) – Scalar integer-like receptor identifier to validate.

Returns:

Always 0 when the receptor type is valid.

Return type:

int

Raises:

ValueError – If receptor_type is non-scalar, not integer-like, or not equal to 0.
TypeError – If receptor_type cannot be converted to a numeric array.

init_state(batch_size=None, **kwargs)[source]#

Reset all queue and history state to NEST-compatible initial conditions.

Clears the pending-multiplicity map \(P\), resets the spike-history vector \(H\) to the single pseudo-spike spikecounter(0.0, 0.0), and resets all delivery metadata (last_delivery_spikes, last_delivery_time_ms, delivery_count) to their initial values.

Parameters:

batch_size (int or None, optional) – Unused placeholder required by the brainstate.nn.Dynamics API. Ignored. Default is None.
**kwargs – Additional keyword arguments accepted for API compatibility and silently ignored.

property last_delivery_spikes: tuple[spikecounter, ...]#

Spike-history tuple delivered at the most recent trigger.

Returns an empty tuple before the first trigger.

Returns:: Immutable copy of the history vector \(H\) that was captured at the most recent delivery trigger, including the pseudo-spike at the trigger stamp. Empty tuple if no trigger has fired yet.
Return type:: tuple[spikecounter, ]

property last_delivery_time: float#

Most recent delivery trigger time in milliseconds.

Returns 0.0 if no trigger has occurred since init_state().

Returns:: Trigger time \(t_s = s \cdot \Delta t\) in ms at the most recent trigger step, or 0.0 before the first trigger.
Return type:: float

property local_device_id: int#

Local device ID used for NEST-compatible device-duplication logic.

Returns:: Current local device ID as a Python int.
Return type:: int

property n_deliveries: int#

Number of completed delivery triggers since initialization.

Returns:: Count of trigger events that have fired since the last init_state() call. Increments only when \(s \bmod T_s = 0\) and the history vector is non-empty.
Return type:: int

set_local_device_id(ldid)[source]#

Set the local device ID from a scalar integer-like value.

Parameters:

ldid (ArrayLike) – Scalar integer-like value for the new local device ID. Converted via nearest-integer rounding; non-integer values raise ValueError.

Raises:

ValueError – If ldid is non-scalar or not integer-like.
TypeError – If ldid cannot be converted to a numeric array.

update(spikes=None, multiplicities=None, stamp_steps=None)[source]#

Advance transmitter state by one simulation step.

Reads the current simulation time t and resolution dt from brainstate.environ, schedules incoming spike contributions into the pending map \(P\), consumes the current-stamp entry, optionally appends a new history entry, and evaluates the delivery trigger.

Parameters:

spikes (ArrayLike or None, optional) –
Scalar or 1-D array of spike event indicators/counts for the current call, shape (N,) after flattening. Unitful inputs are accepted; only the mantissa is used. Multiplicity inference rules:
- multiplicities is None and all values are integer-like (within 1e-12): \(c_j = \max(\mathrm{round}(x_j),\; 0)\).
- multiplicities is None and values contain non-integers: \(c_j = \mathbf{1}[x_j > 0]\) (binary threshold).
- multiplicities provided: \(c_j = m_j \,\mathbf{1}[x_j > 0]\).
None means no incoming events for this step.
multiplicities (ArrayLike or None, optional) – Scalar or 1-D integer-like array, shape (N,) matching the flattened size of spikes. Each value must be non-negative. Applied only where spikes[j] > 0; non-positive spike indicators force zero contribution regardless of multiplicities[j]. None enables implicit inference from spikes.
stamp_steps (ArrayLike or None, optional) – Scalar or 1-D integer-like array, shape (N,) matching the flattened size of spikes. Values are absolute delivery-stamp indices in step-space and must satisfy stamp_steps[j] >= current_stamp (past stamps raise ValueError). None assigns all contributions to the current stamp \(s\).

Returns:

Dictionary with the following keys:

'triggered' — bool: whether the current stamp fires the delivery trigger (\(s \bmod T_s = 0\)).
't_trig' — float ms or None: trigger time \(t_s = s \cdot \Delta t\) if triggered, else None.
'delivered_spikes' — tuple[spikecounter, ...]: history \(H\) captured before the trigger reset, or empty tuple if not triggered.
'spike_history' — tuple[spikecounter, ...]: current history after all processing (post-reset pseudo-spike if triggered).

Return type:

dict

Raises:

ValueError – If dt <= 0; if min_delay is not a positive integer multiple of dt; if t is not grid-aligned to dt; if multiplicities contains negative values; if stamp_steps contains stamps earlier than the current stamp; or if any array payload is non-integer where integer values are required.
ValueError – If spikes, multiplicities, or stamp_steps are not scalar or 1-D, or have mismatched flattened sizes.
TypeError – If numeric or unit conversion fails for any payload or environment time value.
KeyError – If required environment values ('t' or dt) are unavailable from brainstate.environ.

Notes

Trigger evaluation uses stamp \(s = n + 1\) (one ahead of the step index) and period \(T_s = k \cdot d_{\min,s}\). Spikes stamped exactly at the trigger stamp are included in 'delivered_spikes' before the history reset, matching NEST ordering semantics.

Examples

Schedule spikes at a future stamp and confirm delivery:

>>> import brainstate
>>> import saiunit as u
>>> import numpy as np
>>> from brainpy.state import volume_transmitter
>>> with brainstate.environ.context(dt=0.1 * u.ms):
...     vt = volume_transmitter(deliver_interval=1, min_delay=0.2 * u.ms)
...     with brainstate.environ.context(t=0.0 * u.ms):
...         out0 = vt.update(
...             spikes=np.array([1.0, 1.0, 0.0]),
...             multiplicities=np.array([2, 3, 7]),
...             stamp_steps=np.array([2, 2, 2]),
...         )
...     with brainstate.environ.context(t=0.1 * u.ms):
...         out1 = vt.update()
...     _ = (out0['triggered'], out1['delivered_spikes'])

volume_transmitter

Contents

volume_transmitter#