events

Event generation and configuration for simulations.

toric_spines_sim.events

Event generation and configuration for simulations.

EventGenerator

Bases: ABC

Pure-interface base class for all event stream generators.

Subclasses must implement :meth:generate. This class provides the shared IO methods :meth:generate_to_file and :meth:load_from_file.

generate(labels=None) abstractmethod

Generate event times for all streams.

Args: labels: Optional override for stream labels.

Returns: TsGroup mapping stream indices to Ts objects (timestamps in ms).

Source code in toric_spines_sim/events/base.py

@abstractmethod
def generate(self, labels: Optional[Sequence[str]] = None) -> nap.TsGroup:
    """Generate event times for all streams.

    Args:
        labels: Optional override for stream labels.

    Returns:
        TsGroup mapping stream indices to Ts objects (timestamps in ms).
    """
    pass

generate_to_file(path, labels=None)

Generate events and save to file with times in ms.

Source code in toric_spines_sim/events/base.py

def generate_to_file(
    self, path: str, labels: Optional[Sequence[str]] = None
) -> nap.TsGroup:
    """Generate events and save to file with times in ms."""
    tsgroup = self.generate(labels=labels)
    with open(path, "w", encoding="utf-8") as f:
        for idx, ts in tsgroup.items():
            lab = (
                tsgroup.get_info("label")[idx]
                if "label" in tsgroup.metadata
                else str(idx)
            )
            # Pynapple stores times in seconds; convert to ms for file output
            times_ms = ts.as_units("ms").index.values
            if len(times_ms) > 0:
                f.write(lab + " " + " ".join(str(t) for t in times_ms) + "\n")
            else:
                f.write(lab + "\n")
    return tsgroup

load_from_file(path) staticmethod

Load event times from file as TsGroup.

Source code in toric_spines_sim/events/base.py

@staticmethod
def load_from_file(path: str) -> nap.TsGroup:
    """Load event times from file as TsGroup."""
    with open(path, "r", encoding="utf-8") as f:
        lines = f.readlines()

    ts_dict = {}
    labels = []
    all_times = []
    idx = 0
    for line in lines:
        line = line.strip()
        if not line:
            continue
        parts = line.split()
        label = parts[0]
        labels.append(label)
        if len(parts) == 1:
            ts_dict[idx] = nap.Ts(t=[], time_units="ms")
        else:
            times = [float(t) for t in parts[1:]]
            all_times.extend(times)
            ts_dict[idx] = nap.Ts(t=times, time_units="ms")
        idx += 1

    # Create time support from data range or default
    if all_times:
        max_t = max(all_times)
        time_support = nap.IntervalSet(
            start=[0], end=[max(max_t, 1.0)], time_units="ms"
        )
    else:
        time_support = nap.IntervalSet(start=[0], end=[1.0], time_units="ms")

    return nap.TsGroup(ts_dict, label=labels, time_support=time_support)

DeterministicEventGenerator(rate_curves, n_synapses_per_axon, T_ms, delay_ms=0.0, labels=None, routing_mode='roundrobin')

Bases: EventGenerator

Deterministic event generator using rate curves.

Generates events at times determined by integrating the rate curve. For a flat rate curve, this produces periodic (regularly spaced) events. For varying rate curves, event density follows the rate profile.

Can operate in shared mode (single rate curve routed to multiple axons) or independent mode (per-axon rate curves).

Args: rate_curves: Single RateCurve for shared mode, or sequence of RateCurves (one per axon) for independent mode. n_synapses_per_axon: Number of synapses attached to each axon. len(n_synapses_per_axon) is the number of axons (N_A). The total number of output channels is sum(n_synapses_per_axon). T_ms: Total simulation time in milliseconds. delay_ms: Time before events start (ms). Default: 0.0. labels: Optional labels for each output channel. Length must equal sum(n_synapses_per_axon). Default labels are A{a}S{s}. TSRecipe / TSSimulator ignore those labels and map by TsGroup index to syn_0, syn_1, … — use remap_axon_channel_events_to_synapses when axon assignments are not point-file order. routing_mode: How to route master events to axons in shared mode. "broadcast": every master event goes to all axons. "roundrobin": events distributed cyclically among axons. Only used in shared mode. Defaults to "roundrobin".

Source code in toric_spines_sim/events/generators.py

def __init__(
    self,
    rate_curves: Union[RateCurve, Sequence[RateCurve]],
    n_synapses_per_axon: Sequence[int],
    T_ms: float,
    delay_ms: float = 0.0,
    labels: Optional[Sequence[str]] = None,
    routing_mode: str = "roundrobin",
):
    self._n_synapses_per_axon: List[int] = list(n_synapses_per_axon)
    self._n_axons: int = len(self._n_synapses_per_axon)
    self._n_channels: int = sum(self._n_synapses_per_axon)
    self._T_ms = float(T_ms)
    self._delay_ms = float(delay_ms)
    self._labels: Optional[List[str]] = list(labels) if labels is not None else None

    # Determine shared vs independent mode
    if isinstance(rate_curves, RateCurve):
        # Shared mode: one curve for all axons
        self._shared_mode = True
        self._shared_curve: Optional[RateCurve] = rate_curves
        self._axon_curves: List[RateCurve] = []

        # Validate routing_mode
        if routing_mode not in ("broadcast", "roundrobin"):
            raise ValueError(
                f"routing_mode must be 'broadcast' or 'roundrobin', got {routing_mode}"
            )
        self._routing_mode: str = routing_mode
    else:
        # Independent mode: per-axon curves
        self._shared_mode = False
        self._shared_curve = None
        self._axon_curves: List[RateCurve] = list(rate_curves)

        if len(self._axon_curves) != self._n_axons:
            raise ValueError(
                f"Number of rate_curves ({len(self._axon_curves)}) must match "
                f"number of axons ({self._n_axons})"
            )
        self._routing_mode: str = routing_mode

generate(labels=None)

Generate deterministic event times for all output channels.

Source code in toric_spines_sim/events/generators.py

def generate(self, labels: Optional[Sequence[str]] = None) -> nap.TsGroup:
    """Generate deterministic event times for all output channels."""
    labs = self._resolve_labels(labels)

    if self._shared_mode:
        # Shared mode: generate master times from single curve
        assert self._shared_curve is not None
        master_isis = self._shared_curve.get_isis(self._T_ms, self._delay_ms)
        master_times = self._cumulative_to_times(master_isis)

        # Route events to axons (deterministic routing)
        axon_times = self._route_shared_times(master_times)
    else:
        # Independent mode: each axon has its own times
        axon_times = []
        for curve in self._axon_curves:
            isis = curve.get_isis(self._T_ms, self._delay_ms)
            times = self._cumulative_to_times(isis)
            axon_times.append(times)

    # Fan-out: assign axon event times to each synapse channel
    ts_dict: dict = {}
    channel_idx = 0
    for axon_idx, n_syn in enumerate(self._n_synapses_per_axon):
        times = axon_times[axon_idx]
        for _ in range(n_syn):
            ts_dict[channel_idx] = nap.Ts(t=times, time_units="ms")
            channel_idx += 1

    time_support = nap.IntervalSet(
        start=[0], end=[max(self._T_ms, 1.0)], time_units="ms"
    )
    return nap.TsGroup(ts_dict, label=labs, time_support=time_support)

StochasticEventGenerator(rate_curves, n_synapses_per_axon, T_ms, delay_ms=0.0, seed=None, labels=None, routing_weights=None, arp_ms=0.0)

Bases: EventGenerator

Stochastic event generator using inhomogeneous Poisson process.

Generates events via thinning algorithm applied to rate curves. Supports absolute refractory period (ARP) to enforce minimum ISI.

Can operate in shared mode (single rate curve routed to multiple axons) or independent mode (per-axon rate curves with independent processes).

Args: rate_curves: Single RateCurve for shared mode, or sequence of RateCurves (one per axon) for independent mode. n_synapses_per_axon: Number of synapses attached to each axon. len(n_synapses_per_axon) is the number of axons (N_A). The total number of output channels is sum(n_synapses_per_axon). T_ms: Total simulation time in milliseconds. delay_ms: Time before events start (ms). Default: 0.0. seed: Random seed for reproducibility. Default: None. labels: Optional labels for each output channel. Length must equal sum(n_synapses_per_axon). Default labels are A{a}S{s}. TSRecipe / TSSimulator map by TsGroup index to syn_i; use remap_axon_channel_events_to_synapses when axon assignments are not point-file order. routing_weights: Probability of routing each master event to each axon. Only used in shared mode. Must sum to 1. Defaults to uniform. arp_ms: Absolute refractory period in ms. Can be scalar (same for all axons) or per-axon list. Default: 0.0.

Source code in toric_spines_sim/events/generators.py

def __init__(
    self,
    rate_curves: Union[RateCurve, Sequence[RateCurve]],
    n_synapses_per_axon: Sequence[int],
    T_ms: float,
    delay_ms: float = 0.0,
    seed: Optional[int] = None,
    labels: Optional[Sequence[str]] = None,
    routing_weights: Optional[Sequence[float]] = None,
    arp_ms: Union[float, Sequence[float]] = 0.0,
):
    self._n_synapses_per_axon: List[int] = list(n_synapses_per_axon)
    self._n_axons: int = len(self._n_synapses_per_axon)
    self._n_channels: int = sum(self._n_synapses_per_axon)
    self._T_ms = float(T_ms)
    self._delay_ms = float(delay_ms)
    self._seed = seed
    self._labels: Optional[List[str]] = list(labels) if labels is not None else None

    # Resolve ARP
    if isinstance(arp_ms, (int, float)):
        self._arp_ms: List[float] = [float(arp_ms)] * self._n_axons
    else:
        self._arp_ms = [float(a) for a in arp_ms]
        if len(self._arp_ms) != self._n_axons:
            raise ValueError(
                f"arp_ms length {len(self._arp_ms)} must match number of axons {self._n_axons}"
            )

    # Determine shared vs independent mode
    if isinstance(rate_curves, RateCurve):
        # Shared mode: one curve for all axons
        self._shared_mode = True
        self._shared_curve: Optional[RateCurve] = rate_curves
        self._axon_curves: List[RateCurve] = []

        # Validate routing_weights
        if routing_weights is not None:
            if len(routing_weights) != self._n_axons:
                raise ValueError(
                    f"routing_weights length {len(routing_weights)} must match "
                    f"number of axons {self._n_axons}"
                )
            weight_sum = sum(routing_weights)
            if not np.isclose(weight_sum, 1.0):
                raise ValueError(
                    f"routing_weights must sum to 1.0, got {weight_sum}"
                )
            self._routing_weights: Optional[List[float]] = list(routing_weights)
        else:
            self._routing_weights = None
    else:
        # Independent mode: per-axon curves
        self._shared_mode = False
        self._shared_curve = None
        self._axon_curves: List[RateCurve] = list(rate_curves)

        if len(self._axon_curves) != self._n_axons:
            raise ValueError(
                f"Number of rate_curves ({len(self._axon_curves)}) must match "
                f"number of axons ({self._n_axons})"
            )
        self._routing_weights = None

generate(labels=None)

Generate stochastic event times for all output channels.

Source code in toric_spines_sim/events/generators.py

def generate(self, labels: Optional[Sequence[str]] = None) -> nap.TsGroup:
    """Generate stochastic event times for all output channels."""
    labs = self._resolve_labels(labels)

    # Set up RNGs
    if self._seed is None:
        rng = np.random.default_rng()
    else:
        rng = np.random.default_rng(self._seed)

    if self._shared_mode:
        # Shared mode: generate master times, then route
        assert self._shared_curve is not None
        master_times = self._generate_thinning(self._shared_curve, rng)

        # Route events to axons
        axon_times = self._route_shared_times_stochastic(master_times, rng)
    else:
        # Independent mode: each axon generates its own times
        axon_times = []
        for curve in self._axon_curves:
            times = self._generate_thinning(curve, rng)
            axon_times.append(times)

    # Apply ARP filter per axon
    axon_times = [
        self._apply_arp_filter(times, self._arp_ms[i])
        for i, times in enumerate(axon_times)
    ]

    # Fan-out: assign axon event times to each synapse channel
    ts_dict: dict = {}
    channel_idx = 0
    for axon_idx, n_syn in enumerate(self._n_synapses_per_axon):
        times = axon_times[axon_idx]
        for _ in range(n_syn):
            ts_dict[channel_idx] = nap.Ts(t=times, time_units="ms")
            channel_idx += 1

    time_support = nap.IntervalSet(
        start=[0], end=[max(self._T_ms, 1.0)], time_units="ms"
    )
    metadata: dict = {"label": labs}
    if self._seed is not None:
        metadata["seed"] = [self._seed] * self._n_channels
    return nap.TsGroup(ts_dict, time_support=time_support, **metadata)

RateCurve

Bases: ABC

Abstract base class for rate curves.

Subclasses must implement rate_at for getting instantaneous rate, max_rate for the upper bound (used in thinning), and get_isis for deterministic event generation.

rate_at(t_ms, T_ms=None) abstractmethod

Return the instantaneous rate at time t_ms (Hz).

Parameters:

Name	Type	Description	Default
t_ms	float	Time in milliseconds.	required
T_ms	float	Total duration in milliseconds. Required by curves whose shape is defined over a finite window (currently LinearRateCurve).	None

Source code in toric_spines_sim/events/rate_curves.py

@abstractmethod
def rate_at(self, t_ms: float, T_ms: float | None = None) -> float:
    """Return the instantaneous rate at time ``t_ms`` (Hz).

    Parameters
    ----------
    t_ms : float
        Time in milliseconds.
    T_ms : float, optional
        Total duration in milliseconds. Required by curves whose shape
        is defined over a finite window (currently ``LinearRateCurve``).
    """
    pass

max_rate() abstractmethod

Return the maximum rate over all time (for thinning algorithm).

Source code in toric_spines_sim/events/rate_curves.py

@abstractmethod
def max_rate(self) -> float:
    """Return the maximum rate over all time (for thinning algorithm)."""
    pass

get_isis(T_ms, delay_ms) abstractmethod

Compute inter-spike intervals for deterministic generation.

Parameters:

Name	Type	Description	Default
T_ms	float	Total simulation time in milliseconds.	required
delay_ms	float	Initial delay before events start.	required

Returns:

Type	Description
list of float	ISIs in ms. Cumulative sum plus delay gives event times.

Source code in toric_spines_sim/events/rate_curves.py

@abstractmethod
def get_isis(self, T_ms: float, delay_ms: float) -> List[float]:
    """Compute inter-spike intervals for deterministic generation.

    Parameters
    ----------
    T_ms : float
        Total simulation time in milliseconds.
    delay_ms : float
        Initial delay before events start.

    Returns
    -------
    list of float
        ISIs in ms. Cumulative sum plus delay gives event times.
    """
    pass

FlatRateCurve(rate_hz)

Bases: RateCurve

Constant (flat) rate curve.

Args: rate_hz: Constant rate in Hz.

Source code in toric_spines_sim/events/rate_curves.py

def __init__(self, rate_hz: float):
    self._rate_hz = float(rate_hz)

LinearRateCurve(rate_start_hz, rate_end_hz)

Bases: RateCurve

Linear ramp from rate_start_hz at t=0 to rate_end_hz at t=T_ms.

Stochastic generators pass T_ms into rate_at so Poisson thinning follows the same ramp as deterministic get_isis.

Parameters:

Name	Type	Description	Default
rate_start_hz	float	Rate at t=0 (Hz).	required
rate_end_hz	float	Rate at t=T_ms (Hz).	required

Examples:

>>> curve = LinearRateCurve(10.0, 20.0)
>>> curve.rate_at(500.0, T_ms=1000.0)
15.0

Source code in toric_spines_sim/events/rate_curves.py

def __init__(self, rate_start_hz: float, rate_end_hz: float):
    self._rate_start_hz = float(rate_start_hz)
    self._rate_end_hz = float(rate_end_hz)

StepRateCurve(rates_hz, step_duration_ms)

Bases: RateCurve

Piecewise-constant (step) rate curve.

Args: rates_hz: List of rates for each step phase (in Hz). step_duration_ms: Duration of each step phase in ms.

Source code in toric_spines_sim/events/rate_curves.py

def __init__(self, rates_hz: List[float], step_duration_ms: float):
    self._rates_hz = [float(r) for r in rates_hz]
    self._step_duration_ms = float(step_duration_ms)
    self._n_steps = len(self._rates_hz)

SineRateCurve(peak_rate_hz, freq_hz, baseline=0.0, phase_rad=None, phase_deg=None)

Bases: RateCurve

Sinusoidally modulated rate curve.

Rate follows: r(t) = peak_rate_hz * max(0, sin(2π * freq_hz * t_ms/1000 + phase) + baseline)

Args: peak_rate_hz: Peak amplitude of the sine wave in Hz. freq_hz: Frequency of modulation in Hz. baseline: Offset added to sine wave (default: 0.0). phase_rad: Phase offset in radians (mutually exclusive with phase_deg). phase_deg: Phase offset in degrees (mutually exclusive with phase_rad).

Raises: ValueError: If both phase_rad and phase_deg are provided.

Source code in toric_spines_sim/events/rate_curves.py

def __init__(
    self,
    peak_rate_hz: float,
    freq_hz: float,
    baseline: float = 0.0,
    phase_rad: float | None = None,
    phase_deg: float | None = None,
):
    if phase_rad is not None and phase_deg is not None:
        raise ValueError("Cannot specify both phase_rad and phase_deg")

    self._peak_rate_hz = float(peak_rate_hz)
    self._freq_hz = float(freq_hz)
    self._baseline = float(baseline)

    if phase_rad is not None:
        self._phase = float(phase_rad)
    elif phase_deg is not None:
        self._phase = float(phase_deg) * np.pi / 180.0
    else:
        self._phase = 0.0

    # Precompute angular frequency (rad per ms)
    self._omega = 2.0 * np.pi * self._freq_hz / 1000.0

to_sine_v1_params()

Return compact parameters for JSON serialization and plot reconstruction.

Source code in toric_spines_sim/events/rate_curves.py

def to_sine_v1_params(self) -> dict[str, float]:
    """Return compact parameters for JSON serialization and plot reconstruction."""
    return {
        "peak_rate_hz": self._peak_rate_hz,
        "freq_hz": self._freq_hz,
        "phase_rad": self._phase,
        "baseline": self._baseline,
    }