snnlab.sim.simulation_inputs

Complete declared API of the simulation_inputs module, with signatures, data fields, validation and source.

Deterministically realize authenticated SNNLang simulation recipes.

The signatures, defaults, fields, docstrings and implementation excerpts below are generated from the Python source. Annotations are shown as declared; unannotated means the source supplies no type annotation. These pages document callable surfaces, including legacy support utilities, without promising backend support for every declaration.

Symbol	Kind
RealizedInputs	class
realize_simulation_inputs	function

RealizedInputs

View source

Materialized spike inputs, excitatory/inhibitory conductance arrays and the protocol that records their physical timing and generation identity.

Class decorators: dataclass(frozen=True).

Dataclass constructor parameters. Factory defaults are shown as field declarations; omit these arguments to create fresh values per instance:

RealizedInputs(input_spikes: torch.Tensor | None, input_spikes_i: torch.Tensor | None, excitatory_e: torch.Tensor | None, inhibitory_e: torch.Tensor | None, excitatory_i: torch.Tensor | None, inhibitory_i: torch.Tensor | None, retained: dict[str, np.ndarray])

Declared fields, including fields inherited from local data classes:

Field	Annotation	Default	Meaning
`input_spikes`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`input_spikes_i`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`excitatory_e`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`inhibitory_e`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`excitatory_i`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`inhibitory_i`	`torch.Tensor \| None`	required	Stored member of this data contract; see the class docstring and serialization methods.
`retained`	`dict[str, np.ndarray]`	required	Stored member of this data contract; see the class docstring and serialization methods.

Complete class implementation

class RealizedInputs:
    input_spikes: torch.Tensor | None
    input_spikes_i: torch.Tensor | None
    excitatory_e: torch.Tensor | None
    inhibitory_e: torch.Tensor | None
    excitatory_i: torch.Tensor | None
    inhibitory_i: torch.Tensor | None
    retained: dict[str, np.ndarray]

realize_simulation_inputs

View source

def realize_simulation_inputs(recipe: dict[str, Any], *, seed: int, dt: float, t_steps: int, e_id: str, i_id: str, n_e: int, n_i: int, input_size: int | None=None) -> RealizedInputs

Source docstring:

Generate mutually independent private/shared streams with stable seeds.

Parameter	Annotation	Default	Meaning
`recipe`	`dict[str, Any]`	required	Defined by the source contract and implementation below.
`seed`	`int`	required	Seed controlling this operation’s random stream.
`dt`	`float`	required	Timestep; authoring uses a Quantity and legacy simulation uses milliseconds.
`t_steps`	`int`	required	Defined by the source contract and implementation below.
`e_id`	`str`	required	Defined by the source contract and implementation below.
`i_id`	`str`	required	Defined by the source contract and implementation below.
`n_e`	`int`	required	Excitatory population size.
`n_i`	`int`	required	Inhibitory population size.
`input_size`	`int \| None`	`None`	Defined by the source contract and implementation below.

Return annotation: RealizedInputs.

Return expressions (branch-dependent; names refer to the linked implementation):

RealizedInputs(input_spikes=input_spikes, input_spikes_i=input_spikes_i, excitatory_e=resolved.get((e_id, 'excitatory')), inhibitory_e=resolved.get((e_id, 'inhibitory')), excitatory_i=resolved.get((i_id, 'excitatory')), inhibitory_i=resolved.get((i_id, 'inhibitory')), retained=retained)

Explicit exceptions in this implementation; called helpers may raise additional errors:

Explicit exception expression
`ValueError('legacy execution supports one structured spike source')`
`ValueError(f'simulation background targets unsupported population {target!r}')`

Implementation

def realize_simulation_inputs(
    recipe: dict[str, Any],
    *,
    seed: int,
    dt: float,
    t_steps: int,
    e_id: str,
    i_id: str,
    n_e: int,
    n_i: int,
    input_size: int | None = None,
) -> RealizedInputs:
    """Generate mutually independent private/shared streams with stable seeds."""
    retained: dict[str, np.ndarray] = {}
    weather = _stationary_weather(recipe, seed=seed, dt=dt, t_steps=t_steps)
    retained["input_weather_scale"] = weather.numpy()
    afferent_wave = _afferent_wave(recipe, dt=dt, t_steps=t_steps)
    shared_afferent_wave = _afferent_wave(
        recipe, dt=dt, t_steps=t_steps, scale_key="shared_peak_scale"
    )
    retained["input_afferent_scale"] = afferent_wave.numpy()
    retained["input_afferent_shared_scale"] = shared_afferent_wave.numpy()
    afferent_weather = weather * afferent_wave
    shared_afferent_weather = weather * shared_afferent_wave
    spike_sources = recipe.get("spike_sources", [])
    if len(spike_sources) > 1:
        raise ValueError("legacy execution supports one structured spike source")
    input_spikes = None
    input_spikes_i = None
    if spike_sources:
        source = spike_sources[0]
        source_size = n_e if input_size is None else int(input_size)
        if source.get("kind") == "correlated_poisson_afferents":
            shared = _poisson_events(
                source["shared_rate_hz"],
                weather=shared_afferent_weather,
                dt=dt,
                size=source_size,
                generator=torch.Generator().manual_seed(int(seed)),
            )
            e_private = _poisson_events(
                source["e_private_rate_hz"],
                weather=afferent_weather,
                dt=dt,
                size=source_size,
                generator=torch.Generator().manual_seed(int(seed) + 1),
            )
            i_private = _poisson_events(
                source["i_private_rate_hz"],
                weather=afferent_weather,
                dt=dt,
                size=source_size,
                generator=torch.Generator().manual_seed(int(seed) + 2),
            )
            input_spikes = torch.maximum(shared, e_private)
            input_spikes_i = torch.maximum(shared, i_private)
            retained["input_afferent_shared"] = shared.numpy()
            retained["input_afferent_e_private"] = e_private.numpy()
            retained["input_afferent_i_private"] = i_private.numpy()
            retained["input_structured_spikes_e"] = input_spikes.numpy()
            retained["input_structured_spikes_i"] = input_spikes_i.numpy()
        else:
            input_spikes = _poisson_events(
                source["rate_hz"],
                weather=afferent_weather,
                dt=dt,
                size=source_size,
                generator=torch.Generator().manual_seed(int(seed)),
            )
            retained["input_structured_spikes"] = input_spikes.numpy()

    resolved: dict[tuple[str, str], torch.Tensor] = {}
    channel_index = 100
    sizes = {e_id: n_e, i_id: n_i}
    names = {e_id: "e", i_id: "i"}
    for background in recipe.get("backgrounds", []):
        target = background["target"]
        if target not in sizes:
            raise ValueError(
                f"simulation background targets unsupported population {target!r}"
            )
        size = sizes[target]
        modulation = _smoothstep(recipe, target, t_steps, dt)
        retained[f"input_modulation_{names[target]}"] = modulation.numpy()
        for polarity in ("excitatory", "inhibitory"):
            channel = background[polarity]
            heterogeneity = channel["heterogeneity"]
            private_gen = torch.Generator().manual_seed(int(seed) + channel_index)
            channel_index += 1
            shared_gen = torch.Generator().manual_seed(int(seed) + channel_index)
            channel_index += 1
            rate_scale = _distribution(heterogeneity["rate"], size, private_gen)
            amplitude_scale = _distribution(
                heterogeneity["amplitude"], size, private_gen
            )
            private = channel["private"]
            private_p = (
                float(private["rate_hz"])
                * float(dt)
                / 1000.0
                * weather.unsqueeze(1)
                * rate_scale.unsqueeze(0)
            ).clamp(max=1)
            private_events = (
                torch.rand(t_steps, size, generator=private_gen) < private_p
            ).float() * (float(private["amplitude"]) * amplitude_scale).unsqueeze(0)
            shared = channel["shared"]
            shared_p = (float(shared["rate_hz"]) * float(dt) / 1000.0 * weather).clamp(
                max=1
            )
            if shared.get("kind") == "grouped_shot_noise":
                group_size = int(shared["group_size"])
                groups = (size + group_size - 1) // group_size
                group_events = (
                    torch.rand(t_steps, groups, generator=shared_gen)
                    < shared_p.unsqueeze(1)
                ).float() * float(shared["amplitude"])
                shared_cells = group_events.repeat_interleave(group_size, dim=1)[
                    :, :size
                ]
            else:
                shared_events = (
                    torch.rand(t_steps, 1, generator=shared_gen) < shared_p.unsqueeze(1)
                ).float() * float(shared["amplitude"])
                shared_cells = shared_events.expand(-1, size)
            total = private_events + shared_cells
            executed = total * modulation.unsqueeze(1)
            stem = f"input_{polarity}_{names[target]}"
            retained[f"{stem}_private"] = private_events.numpy()
            retained[f"{stem}_shared"] = shared_cells.numpy()
            retained[f"{stem}_total"] = total.numpy()
            retained[f"{stem}_executed"] = executed.numpy()
            retained[f"{stem}_rate_scale"] = rate_scale.numpy()
            retained[f"{stem}_amplitude_scale"] = amplitude_scale.numpy()
            resolved[(target, polarity)] = executed

    return RealizedInputs(
        input_spikes=input_spikes,
        input_spikes_i=input_spikes_i,
        excitatory_e=resolved.get((e_id, "excitatory")),
        inhibitory_e=resolved.get((e_id, "inhibitory")),
        excitatory_i=resolved.get((i_id, "excitatory")),
        inhibitory_i=resolved.get((i_id, "inhibitory")),
        retained=retained,
    )

snnlab.sim.simulation_inputs

RealizedInputsCopy Anchor Link

realize_simulation_inputsCopy Anchor Link

On this page

RealizedInputs

realize_simulation_inputs