Bioware interface

sc_neurocore.bioware is an experimental research interface for moving a finite multi-electrode-array (MEA) frame through spike detection, Address-Event Representation (AER), stochastic bitstream encoding, and an optogenetic pulse proposal. It is not a medical device, a clinical controller, or a tissue-safety certification surface.

The maintained implementation is Python-only. Earlier generated Go, Julia, Mojo, and Rust files were non-executable placeholders with no dispatch path and have been removed. The separately maintained Julia plasticity solvers remain documented in Julia solvers; they are not a second implementation of this closed-loop orchestration.

Python

from sc_neurocore.bioware import (
    AERToSCConverter,
    BioHybridSession,
    CultureHealth,
    MEAConfig,
    MEAToAERTranscoder,
    SCToOptoEncoder,
    SpikeDetector,
)

Maintained boundaries

The historical module sc_neurocore.bioware.bioware is a compatibility facade. Implementations are owned by focused modules:

Module	Responsibility
bioware_contracts.py	Stable MEA, spike, AER, pulse, and frame-result records
bioware_validation.py	Shared finite-value, shape, integer, and bitstream guards
bioware_acquisition.py	MAD noise estimation, threshold detection, sorting, artifact blanking
bioware_encoding.py	MEA→AER, AER→SC, SC→opto, and rate decoding
bioware_plasticity.py	Pair-STDP, BCM, and homeostatic Q8.8 adapters
bioware_analysis.py	Culture heuristic, LFP bands, latency, and network bursts
bioware_experiment.py	Pharmacology prototype and multi-well metadata
bioware_audit.py	Deterministic in-memory audit records and checksum
bioware_session.py	One-frame closed-loop orchestration
bioware_fitness.py	Legacy evolutionary fitness adapter

Package-root objects, historical qualified names, and pickle lookup paths are preserved. The facade contains no implementation definitions.

Signal contract

MEA frame and spike detection

Input must be a non-empty, finite numeric NumPy matrix with shape (samples, channels). Its channel count must exactly match MEAConfig.

For channel (c), the detector estimates a robust noise scale

[ \hat\sigma_c = \frac{\operatorname{median}t |V ]}|}{0.6745

and detects crossings of

[ |V_{t,c}| > \alpha \hat\sigma_c, ]

where spike_threshold_sigma is (alpha). The 0.6745 scaling is the standard Gaussian MAD conversion used in extracellular spike detection; see the primary-culture methods discussion in Maccione et al.. The refractory interval is enforced independently per channel. Waveform snippets are edge-padded to a fixed length.

SpikeSorter is a deliberately small PCA + K-Means research adapter. It uses a fixed random_state=0 by default, validates uniform waveform lengths, and requires the optional scikit-learn dependency only when enough waveforms are available to fit clusters.

AER epoch

MEAToAERTranscoder converts a spike timestamp and explicit origin to a clock tick:

[ \tau = \left\lfloor (t_{spike} - t_0) f_{hw} \right\rfloor. ]

The maintained packet has a 16-bit unsigned timestamp. Therefore (0 \le \tau \le 65535); negative relative time and overflow are errors. The implementation never silently wraps timestamps. At the default 1 MHz clock, one frame must fit within approximately 65.536 ms. Longer recordings must be split into explicit epochs by the caller.

BioHybridSession treats detector timestamps as frame-relative. Its t_start_s argument is non-negative experiment time for optional experiment models; it does not change the frame-local AER origin.

AER to stochastic bitstreams

For valid events inside window_ticks, the converter counts events per neuron. If (k_n) is the count for neuron (n), it uses

[ p_n = \frac{k_n}{\max_j k_j}. ]

p_n is encoded by a deterministic 16-bit LFSR into a bitstream of length bitstream_length (default 256). This is not an IID Bernoulli sampler and does not use a smoothing constant. The configured neuron count and window are hard bounds rather than unused metadata.

Stochastic bitstreams to optical pulses

For bitstream density (d_n), the encoder proposes

[ I_n = d_n I_{max}, \qquad T_n = T_{min} + d_n(T_{max} - T_{min}). ]

intensity_mw_mm2 is irradiance. Optical power is computed with the illuminated area:

[ P_n\,[\mathrm{mW}] = I_n\,[\mathrm{mW/mm^2}]\,A_n\,[\mathrm{mm^2}]. ]

The encoder accumulates (P_n) and omits a channel if including it would exceed max_total_power_mw. This software budget is a consistency guard, not a biological safety limit. Real optical safety depends on wavelength, duty cycle, geometry, absorption, scattering, thermal transport, and experimental calibration.

Plasticity adapters

Pair STDP

BiologicalSTDP implements the parameterised exponential pair rule

[ \Delta w = \begin{cases} A_+e^{-\Delta t/\tau_+}, & \Delta t > 0,\ -A_-e^{\Delta t/\tau_-}, & \Delta t < 0,\ 0, & \Delta t = 0. \end{cases} ]

Defaults are tau_plus_ms = tau_minus_ms = 20, a_plus = 0.01, and a_minus = 0.012. These are configurable model parameters, not claimed fits to one specific preparation. The biological timing reference is Bi and Poo (1998).

BCM and homeostasis

BCMPlasticity uses

[ \Delta w = \eta x y(y-\theta), \qquad \theta \leftarrow \theta + \frac{\Delta t}{\tau_\theta}(y^2-\theta), ]

following the sliding-threshold family introduced by Bienenstock, Cooper, and Munro.

HomeostaticPlasticity applies a bounded Q8.8 proportional update:

[ \theta_{next}^{q88} = \operatorname{clip}\left( \theta^{q88} + \left\lfloor \frac{\Delta t}{\tau_h}(r-r^*)\,256 \right\rfloor,\theta_{min}^{q88},\theta_{max}^{q88}\right). ]

BioHybridSession stores stdp and optional homeostatic policies for caller coordination, but process_frame does not update plasticity implicitly.

Session semantics

BioHybridSession.process_frame executes these stages synchronously:

1. validate the frame, experiment time, AER epoch, and converter window;
2. optionally blank stimulus artifacts;
3. detect and optionally sort spikes;
4. optionally apply the pharmacology rate prototype;
5. transcode to AER and deterministic SC bitstreams;
6. optionally pass decoded rates to an ArcaneZenith object;
7. propose optogenetic pulses and calculate a culture-health snapshot;
8. create a typed BioHybridFrameResult, record optional latency, and only then advance round_count.

An exception before completion leaves round_count unchanged. The detector may still retain its internal noise estimate, and an external Zenith object may have its own state; callers needing distributed transactions must manage those resources explicitly.

Python

import numpy as np

config = MEAConfig(num_channels=8, sample_rate_hz=20_000.0)
session = BioHybridSession(
    mea_config=config,
    detector=SpikeDetector(config),
    transcoder=MEAToAERTranscoder(hw_clock_hz=1e6),
    sc_converter=AERToSCConverter(
        window_ticks=0x10000,
        bitstream_length=512,
        num_neurons=config.num_channels,
    ),
    opto_encoder=SCToOptoEncoder(
        illuminated_area_mm2=1.0,
        max_total_power_mw=50.0,
    ),
)

# 1,000 / 20,000 Hz = 50 ms, inside one 16-bit epoch at 1 MHz.
frame = np.zeros((1_000, config.num_channels), dtype=np.float64)
result = session.process_frame(frame, t_start_s=0.0)
assert result.round == result["round"] == 1

Analysis, experiment, and audit limits

• CultureHealth returns a bounded aggregate heuristic from channel rates. It is not a viability assay or clinical endpoint.
• extract_lfp_power is an FFT band-power helper, not a full spectral estimator with windowing, leakage correction, or uncertainty intervals.
• detect_network_bursts is a binned threshold heuristic.
• PharmModel implements application time, onset interpolation, and a firing gain. wash_time_s is reserved configuration and is not yet applied as a washout curve.
• BioAuditLog is an ordered, tamper-evident in-memory record. Its canonical SHA-256 includes schema name, experiment identity, and entries. It does not provide durable storage, signatures, access control, or regulatory compliance.
• The legacy fitness key energy_mw equals 0.5 * spike_count for compatibility. It is a dimensionless optimisation proxy, not measured power or energy.

Verification and benchmark evidence

Focused verification executes 200 tests and covers all 1,052 production statements and 384 branches in the Bioware package. Source and tests are responsibility-sized, the module import graph is acyclic, the facade and package objects are identical, and historical pickle paths remain valid.

benchmarks/results/bench_bioware.json is a 30-sample, interleaved comparison of parent c4e492ff5 and the modular working tree. Both produced exactly 6,865 canonical bytes with SHA-256 2491dc73a2de93a45a1cc944539c170b151403e42b973b18806143f318b7d669:

Local diagnostic metric	Parent median	Modular candidate median	Delta
Pipeline	2.801 ms	3.345 ms	+19.41%
Import	34.771 ms	50.265 ms	+44.56%
Subprocess wall	579.937 ms	535.945 ms	-7.59%
Maximum RSS	37,076 KiB	37,154 KiB	+0.21%

The capture used taskset affinity but no exclusive isolated core, the CPU governor was powersave, and host load was high. These numbers are local regression context only. Rerun on reserved isolated hardware before publishing performance claims.

Bash

PYTHONPATH=src .venv/bin/pytest tests/test_bioware -q

.venv/bin/python benchmarks/bench_bioware.py \
  --baseline-root <clean-parent-tree> \
  --baseline-ref c4e492ff5 \
  --candidate-root . \
  --candidate-ref working-tree \
  --iterations 30 \
  --warmups 2 \
  --output benchmarks/results/bench_bioware.json

The closed-loop biohybrid research context is exemplified by Kagan et al. (2022); that work does not validate this software implementation or its safety.

API reference

sc_neurocore.bioware.bioware

Historical import facade for the biological-hardware interface.

Implementations live in responsibility-specific sibling modules. This module retains the established import, qualified-name, object-identity, and pickle contracts for callers of :mod:sc_neurocore.bioware.bioware.

ArtifactRejector dataclass

Blanks stimulation artifacts from voltage data.

Zeros the voltage trace in a window around each stimulation onset.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

@dataclass
class ArtifactRejector:
    """Blanks stimulation artifacts from voltage data.

    Zeros the voltage trace in a window around each stimulation onset.
    """

    blanking_pre_ms: float = 0.5
    blanking_post_ms: float = 2.0

    def __post_init__(self) -> None:
        """Validate non-negative artifact-blanking intervals."""
        require_nonnegative(self.blanking_pre_ms, "blanking_pre_ms")
        require_nonnegative(self.blanking_post_ms, "blanking_post_ms")

    def blank(
        self,
        voltage_data: np.ndarray[Any, Any],
        stim_times_s: List[float],
        sample_rate_hz: float,
    ) -> np.ndarray[Any, Any]:
        """Return voltage data with stimulus artifacts blanked."""
        validate_voltage_matrix(voltage_data)
        require_positive(sample_rate_hz, "sample_rate_hz")
        result = voltage_data.copy()
        pre_samples = int(self.blanking_pre_ms * sample_rate_hz / 1000.0)
        post_samples = int(self.blanking_post_ms * sample_rate_hz / 1000.0)

        duration_s = result.shape[0] / sample_rate_hz
        for t_s in stim_times_s:
            require_nonnegative(t_s, "stimulus time")
            if t_s >= duration_s:
                raise ValueError("stimulus time must fall inside the voltage frame")
            center = int(t_s * sample_rate_hz)
            start = max(0, center - pre_samples)
            end = min(result.shape[0], center + post_samples)
            result[start:end, :] = 0.0
        return result

__post_init__()

Validate non-negative artifact-blanking intervals.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def __post_init__(self) -> None:
    """Validate non-negative artifact-blanking intervals."""
    require_nonnegative(self.blanking_pre_ms, "blanking_pre_ms")
    require_nonnegative(self.blanking_post_ms, "blanking_post_ms")

blank(voltage_data, stim_times_s, sample_rate_hz)

Return voltage data with stimulus artifacts blanked.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def blank(
    self,
    voltage_data: np.ndarray[Any, Any],
    stim_times_s: List[float],
    sample_rate_hz: float,
) -> np.ndarray[Any, Any]:
    """Return voltage data with stimulus artifacts blanked."""
    validate_voltage_matrix(voltage_data)
    require_positive(sample_rate_hz, "sample_rate_hz")
    result = voltage_data.copy()
    pre_samples = int(self.blanking_pre_ms * sample_rate_hz / 1000.0)
    post_samples = int(self.blanking_post_ms * sample_rate_hz / 1000.0)

    duration_s = result.shape[0] / sample_rate_hz
    for t_s in stim_times_s:
        require_nonnegative(t_s, "stimulus time")
        if t_s >= duration_s:
            raise ValueError("stimulus time must fall inside the voltage frame")
        center = int(t_s * sample_rate_hz)
        start = max(0, center - pre_samples)
        end = min(result.shape[0], center + post_samples)
        result[start:end, :] = 0.0
    return result

SpikeDetector dataclass

Threshold-based spike detector for MEA voltage traces.

Uses adaptive threshold: threshold = mean ± sigma * noise_estimate where noise_estimate = median(|x|) / 0.6745 (robust RMS). Supports configurable refractory period to prevent double-counting.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

@dataclass
class SpikeDetector:
    """Threshold-based spike detector for MEA voltage traces.

    Uses adaptive threshold: threshold = mean ± sigma * noise_estimate
    where noise_estimate = median(|x|) / 0.6745 (robust RMS).
    Supports configurable refractory period to prevent double-counting.
    """

    config: MEAConfig
    refractory_samples: int = 30
    _noise_estimates: Optional[np.ndarray[Any, Any]] = field(default=None, repr=False)

    def __post_init__(self) -> None:
        """Validate detector configuration and refractory interval."""
        if not isinstance(self.config, MEAConfig):
            raise TypeError("config must be an MEAConfig")
        require_nonnegative_int(self.refractory_samples, "refractory_samples")

    def estimate_noise(self, voltage_data: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
        """Estimate per-channel noise from voltage data.

        Uses median absolute deviation (MAD) for robustness against spikes.
        voltage_data: shape (num_samples, num_channels)
        """
        validate_voltage_matrix(voltage_data, expected_channels=self.config.num_channels)
        mad: np.ndarray[Any, Any] = np.median(np.abs(voltage_data), axis=0) / 0.6745
        self._noise_estimates = mad
        return mad

    def detect(
        self, voltage_data: np.ndarray[Any, Any], snippet_ms: float = 2.0
    ) -> List[DetectedSpike]:
        """Detect spikes in multi-channel voltage data.

        voltage_data: shape (num_samples, num_channels)
        Returns list of DetectedSpike events.
        """
        validate_voltage_matrix(voltage_data, expected_channels=self.config.num_channels)
        require_positive(snippet_ms, "snippet_ms")
        n_samples, n_channels = voltage_data.shape
        if self._noise_estimates is None:
            self.estimate_noise(voltage_data)
        noise_estimates = cast(np.ndarray[Any, Any], self._noise_estimates)

        spikes = []
        dt = 1.0 / self.config.sample_rate_hz
        sigma = self.config.spike_threshold_sigma
        half = int(snippet_ms * self.config.sample_rate_hz / 2000.0)
        if half < 1:
            raise ValueError("snippet_ms is shorter than one sample on each side")

        for ch in range(n_channels):
            threshold = sigma * noise_estimates[ch]
            above = np.abs(voltage_data[:, ch]) > threshold
            crossings = np.where(np.diff(above.astype(int)) == 1)[0]
            last_spike_idx = -self.refractory_samples - 1
            for idx in crossings:
                if idx - last_spike_idx < self.refractory_samples:
                    continue
                last_spike_idx = idx
                amp = float(voltage_data[idx, ch])
                ts = idx * dt

                # Extract waveform snippet
                start = max(0, idx - half)
                end = min(n_samples, idx + half)

                # Pad if too close to edges. The raw slice length is at most
                # 2*half == target_len (it is min(n, idx+half) - max(0, idx-half)),
                # and for an edge spike (idx < half) the slice starts at 0 so
                # pad_before = half - idx exactly closes the gap: pad_after is
                # never negative and the padded waveform is exactly target_len.
                raw_wave = voltage_data[start:end, ch].copy()
                target_len = int(2 * half)
                if len(raw_wave) < target_len:
                    pad_before = max(0, half - idx)
                    pad_after = max(0, target_len - len(raw_wave) - pad_before)
                    raw_wave = np.pad(raw_wave, (pad_before, pad_after), "constant")

                spikes.append(
                    DetectedSpike(
                        channel=ch,
                        timestamp_s=ts,
                        amplitude_uv=amp,
                        unit_id=ch,
                        waveform=raw_wave,
                    )
                )
        return spikes

__post_init__()

Validate detector configuration and refractory interval.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def __post_init__(self) -> None:
    """Validate detector configuration and refractory interval."""
    if not isinstance(self.config, MEAConfig):
        raise TypeError("config must be an MEAConfig")
    require_nonnegative_int(self.refractory_samples, "refractory_samples")

estimate_noise(voltage_data)

Estimate per-channel noise from voltage data.

Uses median absolute deviation (MAD) for robustness against spikes. voltage_data: shape (num_samples, num_channels)

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def estimate_noise(self, voltage_data: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
    """Estimate per-channel noise from voltage data.

    Uses median absolute deviation (MAD) for robustness against spikes.
    voltage_data: shape (num_samples, num_channels)
    """
    validate_voltage_matrix(voltage_data, expected_channels=self.config.num_channels)
    mad: np.ndarray[Any, Any] = np.median(np.abs(voltage_data), axis=0) / 0.6745
    self._noise_estimates = mad
    return mad

detect(voltage_data, snippet_ms=2.0)

Detect spikes in multi-channel voltage data.

voltage_data: shape (num_samples, num_channels) Returns list of DetectedSpike events.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def detect(
    self, voltage_data: np.ndarray[Any, Any], snippet_ms: float = 2.0
) -> List[DetectedSpike]:
    """Detect spikes in multi-channel voltage data.

    voltage_data: shape (num_samples, num_channels)
    Returns list of DetectedSpike events.
    """
    validate_voltage_matrix(voltage_data, expected_channels=self.config.num_channels)
    require_positive(snippet_ms, "snippet_ms")
    n_samples, n_channels = voltage_data.shape
    if self._noise_estimates is None:
        self.estimate_noise(voltage_data)
    noise_estimates = cast(np.ndarray[Any, Any], self._noise_estimates)

    spikes = []
    dt = 1.0 / self.config.sample_rate_hz
    sigma = self.config.spike_threshold_sigma
    half = int(snippet_ms * self.config.sample_rate_hz / 2000.0)
    if half < 1:
        raise ValueError("snippet_ms is shorter than one sample on each side")

    for ch in range(n_channels):
        threshold = sigma * noise_estimates[ch]
        above = np.abs(voltage_data[:, ch]) > threshold
        crossings = np.where(np.diff(above.astype(int)) == 1)[0]
        last_spike_idx = -self.refractory_samples - 1
        for idx in crossings:
            if idx - last_spike_idx < self.refractory_samples:
                continue
            last_spike_idx = idx
            amp = float(voltage_data[idx, ch])
            ts = idx * dt

            # Extract waveform snippet
            start = max(0, idx - half)
            end = min(n_samples, idx + half)

            # Pad if too close to edges. The raw slice length is at most
            # 2*half == target_len (it is min(n, idx+half) - max(0, idx-half)),
            # and for an edge spike (idx < half) the slice starts at 0 so
            # pad_before = half - idx exactly closes the gap: pad_after is
            # never negative and the padded waveform is exactly target_len.
            raw_wave = voltage_data[start:end, ch].copy()
            target_len = int(2 * half)
            if len(raw_wave) < target_len:
                pad_before = max(0, half - idx)
                pad_after = max(0, target_len - len(raw_wave) - pad_before)
                raw_wave = np.pad(raw_wave, (pad_before, pad_after), "constant")

            spikes.append(
                DetectedSpike(
                    channel=ch,
                    timestamp_s=ts,
                    amplitude_uv=amp,
                    unit_id=ch,
                    waveform=raw_wave,
                )
            )
    return spikes

SpikeSorter dataclass

Research spike sorter using PCA feature extraction and K-Means clustering.

Projects uniform waveforms onto their dominant principal components before clustering them into units. Fitting requires the optional scikit-learn dependency; incomplete waveform sets remain explicitly unassigned.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

@dataclass
class SpikeSorter:
    """Research spike sorter using PCA feature extraction and K-Means clustering.

    Projects uniform waveforms onto their dominant principal components before
    clustering them into units. Fitting requires the optional ``scikit-learn``
    dependency; incomplete waveform sets remain explicitly unassigned.
    """

    num_units: int = 4
    n_components: int = 3
    random_state: int = 0
    _pca: Any = field(default=None, repr=False)
    _kmeans: Any = field(default=None, repr=False)

    def __post_init__(self) -> None:
        """Validate cluster count, projection width, and deterministic seed."""
        require_positive_int(self.num_units, "num_units")
        require_positive_int(self.n_components, "n_components")
        require_nonnegative_int(self.random_state, "random_state")

    def fit(self, spikes: List[DetectedSpike]) -> None:
        """Fit PCA and KMeans models sequentially on available waveforms.

        Silently no-ops (leaves ``_pca``/``_kmeans`` as ``None``) when
        fewer than ``num_units`` waveforms are present — sklearn is only
        imported in the path that actually needs it, so empty or
        amplitude-only spike lists don't require scikit-learn.
        """
        waveforms = [s.waveform for s in spikes if s.waveform is not None]
        if len(waveforms) < self.num_units:
            self._pca = None
            self._kmeans = None
            return

        try:
            from sklearn.cluster import KMeans
            from sklearn.decomposition import PCA
        except ImportError as exc:
            raise ImportError(
                "SpikeSorter.fit requires scikit-learn to cluster waveforms. "
                "Install with `pip install scikit-learn` or "
                "`pip install 'sc-neurocore[bioware]'`."
            ) from exc

        waveform_lengths = {waveform.shape for waveform in waveforms}
        if len(waveform_lengths) != 1:
            raise ValueError("all spike waveforms must have the same shape")
        waves_array = np.vstack(waveforms)
        self._pca = PCA(n_components=min(self.n_components, len(waveforms), waves_array.shape[1]))
        features = self._pca.fit_transform(waves_array)

        self._kmeans = KMeans(
            n_clusters=self.num_units,
            n_init=10,
            random_state=self.random_state,
        )
        self._kmeans.fit(features)

    def assign(self, spikes: List[DetectedSpike]) -> List[DetectedSpike]:
        """Assign cluster IDs based on PCA feature projections."""
        if self._pca is None or self._kmeans is None:
            return spikes

        result = []
        for s in spikes:
            if s.waveform is None:
                result.append(s)
                continue

            expected_features = int(self._pca.n_features_in_)
            if s.waveform.size != expected_features:
                raise ValueError(
                    f"spike waveform has {s.waveform.size} samples; expected {expected_features}"
                )

            features = self._pca.transform(s.waveform.reshape(1, -1))
            unit = int(self._kmeans.predict(features)[0])

            result.append(
                DetectedSpike(
                    channel=s.channel,
                    timestamp_s=s.timestamp_s,
                    amplitude_uv=s.amplitude_uv,
                    unit_id=unit,
                    waveform=s.waveform,
                )
            )
        return result

__post_init__()

Validate cluster count, projection width, and deterministic seed.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def __post_init__(self) -> None:
    """Validate cluster count, projection width, and deterministic seed."""
    require_positive_int(self.num_units, "num_units")
    require_positive_int(self.n_components, "n_components")
    require_nonnegative_int(self.random_state, "random_state")

fit(spikes)

Fit PCA and KMeans models sequentially on available waveforms.

Silently no-ops (leaves _pca/_kmeans as None) when fewer than num_units waveforms are present — sklearn is only imported in the path that actually needs it, so empty or amplitude-only spike lists don't require scikit-learn.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def fit(self, spikes: List[DetectedSpike]) -> None:
    """Fit PCA and KMeans models sequentially on available waveforms.

    Silently no-ops (leaves ``_pca``/``_kmeans`` as ``None``) when
    fewer than ``num_units`` waveforms are present — sklearn is only
    imported in the path that actually needs it, so empty or
    amplitude-only spike lists don't require scikit-learn.
    """
    waveforms = [s.waveform for s in spikes if s.waveform is not None]
    if len(waveforms) < self.num_units:
        self._pca = None
        self._kmeans = None
        return

    try:
        from sklearn.cluster import KMeans
        from sklearn.decomposition import PCA
    except ImportError as exc:
        raise ImportError(
            "SpikeSorter.fit requires scikit-learn to cluster waveforms. "
            "Install with `pip install scikit-learn` or "
            "`pip install 'sc-neurocore[bioware]'`."
        ) from exc

    waveform_lengths = {waveform.shape for waveform in waveforms}
    if len(waveform_lengths) != 1:
        raise ValueError("all spike waveforms must have the same shape")
    waves_array = np.vstack(waveforms)
    self._pca = PCA(n_components=min(self.n_components, len(waveforms), waves_array.shape[1]))
    features = self._pca.fit_transform(waves_array)

    self._kmeans = KMeans(
        n_clusters=self.num_units,
        n_init=10,
        random_state=self.random_state,
    )
    self._kmeans.fit(features)

assign(spikes)

Assign cluster IDs based on PCA feature projections.

Source code in src/sc_neurocore/bioware/bioware_acquisition.py

def assign(self, spikes: List[DetectedSpike]) -> List[DetectedSpike]:
    """Assign cluster IDs based on PCA feature projections."""
    if self._pca is None or self._kmeans is None:
        return spikes

    result = []
    for s in spikes:
        if s.waveform is None:
            result.append(s)
            continue

        expected_features = int(self._pca.n_features_in_)
        if s.waveform.size != expected_features:
            raise ValueError(
                f"spike waveform has {s.waveform.size} samples; expected {expected_features}"
            )

        features = self._pca.transform(s.waveform.reshape(1, -1))
        unit = int(self._kmeans.predict(features)[0])

        result.append(
            DetectedSpike(
                channel=s.channel,
                timestamp_s=s.timestamp_s,
                amplitude_uv=s.amplitude_uv,
                unit_id=unit,
                waveform=s.waveform,
            )
        )
    return result

CultureHealth dataclass

Monitor organoid/culture viability from MEA activity.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

@dataclass
class CultureHealth:
    """Monitor organoid/culture viability from MEA activity."""

    min_active_channels: int = 5
    min_firing_rate_hz: float = 0.1
    max_firing_rate_hz: float = 100.0
    burst_threshold_hz: float = 50.0

    def __post_init__(self) -> None:
        """Validate rate thresholds used by the aggregate health heuristic."""
        require_positive_int(self.min_active_channels, "min_active_channels")
        require_nonnegative(self.min_firing_rate_hz, "min_firing_rate_hz")
        require_positive(self.max_firing_rate_hz, "max_firing_rate_hz")
        if self.max_firing_rate_hz <= self.min_firing_rate_hz:
            raise ValueError("max_firing_rate_hz must exceed min_firing_rate_hz")
        require_nonnegative(self.burst_threshold_hz, "burst_threshold_hz")

    def assess(self, spike_counts: np.ndarray[Any, Any], duration_s: float) -> Dict[str, float]:
        """Assess culture health from spike activity.

        spike_counts: per-channel spike counts over duration_s
        """
        if not isinstance(spike_counts, np.ndarray):
            raise TypeError("spike_counts must be a NumPy array")
        if spike_counts.ndim != 1 or spike_counts.size == 0:
            raise ValueError("spike_counts must be a non-empty one-dimensional array")
        if not np.issubdtype(spike_counts.dtype, np.number):
            raise TypeError("spike_counts must have a numeric dtype")
        if not np.all(np.isfinite(spike_counts)) or np.any(spike_counts < 0):
            raise ValueError("spike_counts must contain finite non-negative values")
        require_positive(duration_s, "duration_s")
        rates = spike_counts / duration_s
        active = np.sum(rates > self.min_firing_rate_hz)
        mean_rate = float(np.mean(rates[rates > 0])) if np.any(rates > 0) else 0.0
        bursting = np.sum(rates > self.burst_threshold_hz)

        health_score = 1.0
        if active < self.min_active_channels:
            health_score *= active / self.min_active_channels
        if mean_rate > self.max_firing_rate_hz:
            health_score *= self.max_firing_rate_hz / mean_rate

        return {
            "active_channels": int(active),
            "mean_firing_rate_hz": mean_rate,
            "bursting_channels": int(bursting),
            "health_score": float(np.clip(health_score, 0.0, 1.0)),
            "is_viable": bool(health_score > 0.5),
        }

__post_init__()

Validate rate thresholds used by the aggregate health heuristic.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def __post_init__(self) -> None:
    """Validate rate thresholds used by the aggregate health heuristic."""
    require_positive_int(self.min_active_channels, "min_active_channels")
    require_nonnegative(self.min_firing_rate_hz, "min_firing_rate_hz")
    require_positive(self.max_firing_rate_hz, "max_firing_rate_hz")
    if self.max_firing_rate_hz <= self.min_firing_rate_hz:
        raise ValueError("max_firing_rate_hz must exceed min_firing_rate_hz")
    require_nonnegative(self.burst_threshold_hz, "burst_threshold_hz")

assess(spike_counts, duration_s)

Assess culture health from spike activity.

spike_counts: per-channel spike counts over duration_s

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def assess(self, spike_counts: np.ndarray[Any, Any], duration_s: float) -> Dict[str, float]:
    """Assess culture health from spike activity.

    spike_counts: per-channel spike counts over duration_s
    """
    if not isinstance(spike_counts, np.ndarray):
        raise TypeError("spike_counts must be a NumPy array")
    if spike_counts.ndim != 1 or spike_counts.size == 0:
        raise ValueError("spike_counts must be a non-empty one-dimensional array")
    if not np.issubdtype(spike_counts.dtype, np.number):
        raise TypeError("spike_counts must have a numeric dtype")
    if not np.all(np.isfinite(spike_counts)) or np.any(spike_counts < 0):
        raise ValueError("spike_counts must contain finite non-negative values")
    require_positive(duration_s, "duration_s")
    rates = spike_counts / duration_s
    active = np.sum(rates > self.min_firing_rate_hz)
    mean_rate = float(np.mean(rates[rates > 0])) if np.any(rates > 0) else 0.0
    bursting = np.sum(rates > self.burst_threshold_hz)

    health_score = 1.0
    if active < self.min_active_channels:
        health_score *= active / self.min_active_channels
    if mean_rate > self.max_firing_rate_hz:
        health_score *= self.max_firing_rate_hz / mean_rate

    return {
        "active_channels": int(active),
        "mean_firing_rate_hz": mean_rate,
        "bursting_channels": int(bursting),
        "health_score": float(np.clip(health_score, 0.0, 1.0)),
        "is_viable": bool(health_score > 0.5),
    }

LFPBand dataclass

Frequency band definition for LFP extraction.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

@dataclass
class LFPBand:
    """Frequency band definition for LFP extraction."""

    name: str
    low_hz: float
    high_hz: float

    def __post_init__(self) -> None:
        """Validate a named half-open frequency interval."""
        if not self.name or not self.name.strip():
            raise ValueError("LFP band name must not be empty")
        require_nonnegative(self.low_hz, "low_hz")
        require_positive(self.high_hz, "high_hz")
        if self.high_hz <= self.low_hz:
            raise ValueError("high_hz must exceed low_hz")

__post_init__()

Validate a named half-open frequency interval.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def __post_init__(self) -> None:
    """Validate a named half-open frequency interval."""
    if not self.name or not self.name.strip():
        raise ValueError("LFP band name must not be empty")
    require_nonnegative(self.low_hz, "low_hz")
    require_positive(self.high_hz, "high_hz")
    if self.high_hz <= self.low_hz:
        raise ValueError("high_hz must exceed low_hz")

LatencyBudget dataclass

Tracks and enforces closed-loop latency requirements.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

@dataclass
class LatencyBudget:
    """Tracks and enforces closed-loop latency requirements."""

    max_latency_us: float = 1000.0  # 1 ms default
    history: List[float] = field(default_factory=list)
    violations: int = 0

    def __post_init__(self) -> None:
        """Validate budget, history, and violation accounting."""
        require_positive(self.max_latency_us, "max_latency_us")
        for latency_us in self.history:
            require_nonnegative(latency_us, "history latency_us")
        require_nonnegative_int(self.violations, "violations")
        if self.violations > len(self.history):
            raise ValueError("violations cannot exceed the number of history samples")

    def record(self, latency_us: float) -> bool:
        """Record a latency measurement. Returns True if within budget."""
        require_nonnegative(latency_us, "latency_us")
        self.history.append(latency_us)
        if latency_us > self.max_latency_us:
            self.violations += 1
            return False
        return True

    @property
    def mean_latency_us(self) -> float:
        """Return the arithmetic mean of recorded loop latencies.

        Returns
        -------
        float
            Mean latency in microseconds, or ``0.0`` before any samples exist.
        """
        return float(np.mean(self.history)) if self.history else 0.0

    @property
    def p99_latency_us(self) -> float:
        """Return the 99th percentile closed-loop latency.

        Returns
        -------
        float
            99th percentile latency in microseconds, or ``0.0`` for an empty
            history.
        """
        return float(np.percentile(self.history, 99)) if self.history else 0.0

    @property
    def compliance_ratio(self) -> float:
        """Return the fraction of samples inside the latency budget.

        Returns
        -------
        float
            Ratio in ``[0.0, 1.0]``; an empty history is defined as fully
            compliant.
        """
        if not self.history:
            return 1.0
        return 1.0 - self.violations / len(self.history)

mean_latency_us property

Return the arithmetic mean of recorded loop latencies.

Returns

float Mean latency in microseconds, or 0.0 before any samples exist.

p99_latency_us property

Return the 99th percentile closed-loop latency.

Returns

float 99th percentile latency in microseconds, or 0.0 for an empty history.

compliance_ratio property

Return the fraction of samples inside the latency budget.

Returns

float Ratio in [0.0, 1.0]; an empty history is defined as fully compliant.

__post_init__()

Validate budget, history, and violation accounting.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def __post_init__(self) -> None:
    """Validate budget, history, and violation accounting."""
    require_positive(self.max_latency_us, "max_latency_us")
    for latency_us in self.history:
        require_nonnegative(latency_us, "history latency_us")
    require_nonnegative_int(self.violations, "violations")
    if self.violations > len(self.history):
        raise ValueError("violations cannot exceed the number of history samples")

record(latency_us)

Record a latency measurement. Returns True if within budget.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def record(self, latency_us: float) -> bool:
    """Record a latency measurement. Returns True if within budget."""
    require_nonnegative(latency_us, "latency_us")
    self.history.append(latency_us)
    if latency_us > self.max_latency_us:
        self.violations += 1
        return False
    return True

NetworkBurst dataclass

Detected network-wide synchronised burst event.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

@dataclass
class NetworkBurst:
    """Detected network-wide synchronised burst event."""

    onset_s: float
    duration_s: float
    participating_channels: int
    total_spikes: int

    def __post_init__(self) -> None:
        """Validate a detected network-burst summary."""
        require_nonnegative(self.onset_s, "onset_s")
        require_positive(self.duration_s, "duration_s")
        require_positive_int(self.participating_channels, "participating_channels")
        require_positive_int(self.total_spikes, "total_spikes")

__post_init__()

Validate a detected network-burst summary.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def __post_init__(self) -> None:
    """Validate a detected network-burst summary."""
    require_nonnegative(self.onset_s, "onset_s")
    require_positive(self.duration_s, "duration_s")
    require_positive_int(self.participating_channels, "participating_channels")
    require_positive_int(self.total_spikes, "total_spikes")

BioAuditEntry dataclass

One audit entry for a bio-hybrid session.

Source code in src/sc_neurocore/bioware/bioware_audit.py

@dataclass
class BioAuditEntry:
    """One audit entry for a bio-hybrid session."""

    round_number: int
    timestamp_iso: str
    num_spikes: int
    num_opto_pulses: int
    latency_us: float
    health_score: float
    notes: str = ""

    def __post_init__(self) -> None:
        """Validate one timestamped session-audit record."""
        require_nonnegative_int(self.round_number, "round_number")
        if not self.timestamp_iso or not self.timestamp_iso.strip():
            raise ValueError("timestamp_iso must not be empty")
        try:
            datetime.fromisoformat(self.timestamp_iso)
        except ValueError as exc:
            raise ValueError("timestamp_iso must be an ISO-8601 date or datetime") from exc
        require_nonnegative_int(self.num_spikes, "num_spikes")
        require_nonnegative_int(self.num_opto_pulses, "num_opto_pulses")
        require_nonnegative(self.latency_us, "latency_us")
        require_nonnegative(self.health_score, "health_score")
        if self.health_score > 1.0:
            raise ValueError("health_score must be <= 1")
        if not isinstance(self.notes, str):
            raise TypeError("notes must be a string")

__post_init__()

Validate one timestamped session-audit record.

Source code in src/sc_neurocore/bioware/bioware_audit.py

def __post_init__(self) -> None:
    """Validate one timestamped session-audit record."""
    require_nonnegative_int(self.round_number, "round_number")
    if not self.timestamp_iso or not self.timestamp_iso.strip():
        raise ValueError("timestamp_iso must not be empty")
    try:
        datetime.fromisoformat(self.timestamp_iso)
    except ValueError as exc:
        raise ValueError("timestamp_iso must be an ISO-8601 date or datetime") from exc
    require_nonnegative_int(self.num_spikes, "num_spikes")
    require_nonnegative_int(self.num_opto_pulses, "num_opto_pulses")
    require_nonnegative(self.latency_us, "latency_us")
    require_nonnegative(self.health_score, "health_score")
    if self.health_score > 1.0:
        raise ValueError("health_score must be <= 1")
    if not isinstance(self.notes, str):
        raise TypeError("notes must be a string")

BioAuditLog dataclass

Tamper-evident in-memory audit log for bio-hybrid experiments.

Source code in src/sc_neurocore/bioware/bioware_audit.py

@dataclass
class BioAuditLog:
    """Tamper-evident in-memory audit log for bio-hybrid experiments."""

    entries: List[BioAuditEntry] = field(default_factory=list)
    experiment_id: str = ""

    def __post_init__(self) -> None:
        """Validate experiment identity and strictly ordered audit entries."""
        if not isinstance(self.experiment_id, str):
            raise TypeError("experiment_id must be a string")
        if self.experiment_id and not self.experiment_id.strip():
            raise ValueError("experiment_id must not be whitespace only")
        previous_round = -1
        for entry in self.entries:
            if not isinstance(entry, BioAuditEntry):
                raise TypeError("entries must contain BioAuditEntry instances")
            if entry.round_number <= previous_round:
                raise ValueError("audit entry round numbers must increase strictly")
            previous_round = entry.round_number

    def log(self, entry: BioAuditEntry) -> None:
        """Append one audit entry to the session log.

        Parameters
        ----------
        entry:
            Timestamped closed-loop session summary to retain in append order.
        """
        if not isinstance(entry, BioAuditEntry):
            raise TypeError("entry must be a BioAuditEntry")
        if self.entries and entry.round_number <= self.entries[-1].round_number:
            raise ValueError("audit entry round numbers must increase strictly")
        self.entries.append(entry)

    @property
    def total_rounds(self) -> int:
        """Return the number of recorded audit entries.

        Returns
        -------
        int
            Count of entries currently stored in the log.
        """
        return len(self.entries)

    def to_list(self) -> List[Dict[str, Any]]:
        """Serialise audit entries to deterministic dictionaries.

        Returns
        -------
        list[dict[str, Any]]
            JSON-compatible records used by ``checksum`` and external evidence
            sinks.
        """
        return [
            {
                "round": e.round_number,
                "timestamp": e.timestamp_iso,
                "spikes": e.num_spikes,
                "opto_pulses": e.num_opto_pulses,
                "latency_us": e.latency_us,
                "health_score": e.health_score,
                "notes": e.notes,
            }
            for e in self.entries
        ]

    def checksum(self) -> str:
        """Return a cross-environment SHA-256 over identity and log contents."""
        payload = {
            "schema": "sc-neurocore.bioware-audit.v1",
            "experiment_id": self.experiment_id,
            "entries": self.to_list(),
        }
        data = json.dumps(
            payload,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        ).encode("utf-8")
        return hashlib.sha256(data).hexdigest()

total_rounds property

Return the number of recorded audit entries.

Returns

int Count of entries currently stored in the log.

__post_init__()

Validate experiment identity and strictly ordered audit entries.

Source code in src/sc_neurocore/bioware/bioware_audit.py

def __post_init__(self) -> None:
    """Validate experiment identity and strictly ordered audit entries."""
    if not isinstance(self.experiment_id, str):
        raise TypeError("experiment_id must be a string")
    if self.experiment_id and not self.experiment_id.strip():
        raise ValueError("experiment_id must not be whitespace only")
    previous_round = -1
    for entry in self.entries:
        if not isinstance(entry, BioAuditEntry):
            raise TypeError("entries must contain BioAuditEntry instances")
        if entry.round_number <= previous_round:
            raise ValueError("audit entry round numbers must increase strictly")
        previous_round = entry.round_number

log(entry)

Append one audit entry to the session log.

Parameters

entry: Timestamped closed-loop session summary to retain in append order.

Source code in src/sc_neurocore/bioware/bioware_audit.py

def log(self, entry: BioAuditEntry) -> None:
    """Append one audit entry to the session log.

    Parameters
    ----------
    entry:
        Timestamped closed-loop session summary to retain in append order.
    """
    if not isinstance(entry, BioAuditEntry):
        raise TypeError("entry must be a BioAuditEntry")
    if self.entries and entry.round_number <= self.entries[-1].round_number:
        raise ValueError("audit entry round numbers must increase strictly")
    self.entries.append(entry)

to_list()

Serialise audit entries to deterministic dictionaries.

Returns

list[dict[str, Any]] JSON-compatible records used by checksum and external evidence sinks.

Source code in src/sc_neurocore/bioware/bioware_audit.py

def to_list(self) -> List[Dict[str, Any]]:
    """Serialise audit entries to deterministic dictionaries.

    Returns
    -------
    list[dict[str, Any]]
        JSON-compatible records used by ``checksum`` and external evidence
        sinks.
    """
    return [
        {
            "round": e.round_number,
            "timestamp": e.timestamp_iso,
            "spikes": e.num_spikes,
            "opto_pulses": e.num_opto_pulses,
            "latency_us": e.latency_us,
            "health_score": e.health_score,
            "notes": e.notes,
        }
        for e in self.entries
    ]

checksum()

Return a cross-environment SHA-256 over identity and log contents.

Source code in src/sc_neurocore/bioware/bioware_audit.py

def checksum(self) -> str:
    """Return a cross-environment SHA-256 over identity and log contents."""
    payload = {
        "schema": "sc-neurocore.bioware-audit.v1",
        "experiment_id": self.experiment_id,
        "entries": self.to_list(),
    }
    data = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode("utf-8")
    return hashlib.sha256(data).hexdigest()

AEREvent dataclass

Address-Event Representation packet.

Compatible with sc_aer_encoder.v format:

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@dataclass
class AEREvent:
    """Address-Event Representation packet.

    Compatible with sc_aer_encoder.v format:
    {valid, neuron_id, timestamp}
    """

    neuron_id: int
    timestamp: int  # clock ticks (not real time)
    valid: bool = True
    weight: int = 256  # Q8.8 = 1.0

    def __post_init__(self) -> None:
        """Validate the maintained unsigned 16-bit AER packet fields."""
        require_nonnegative_int(self.neuron_id, "neuron_id")
        require_nonnegative_int(self.timestamp, "timestamp")
        if self.timestamp > 0xFFFF:
            raise ValueError("timestamp must fit the 16-bit AER field")
        if not isinstance(self.valid, bool):
            raise TypeError("valid must be a bool")
        require_nonnegative_int(self.weight, "weight")
        if self.weight > 0xFFFF:
            raise ValueError("weight must fit an unsigned 16-bit Q8.8 field")

__post_init__()

Validate the maintained unsigned 16-bit AER packet fields.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __post_init__(self) -> None:
    """Validate the maintained unsigned 16-bit AER packet fields."""
    require_nonnegative_int(self.neuron_id, "neuron_id")
    require_nonnegative_int(self.timestamp, "timestamp")
    if self.timestamp > 0xFFFF:
        raise ValueError("timestamp must fit the 16-bit AER field")
    if not isinstance(self.valid, bool):
        raise TypeError("valid must be a bool")
    require_nonnegative_int(self.weight, "weight")
    if self.weight > 0xFFFF:
        raise ValueError("weight must fit an unsigned 16-bit Q8.8 field")

BioHybridFrameResult dataclass

Strictly typed output packet detailing a full closed-loop step.

Behaves both as a dataclass (result.round) and, for backward compatibility with pre-dataclass callers, as a mapping view of its fields (result["round"], "latency_us" in result, dict(result)). The mapping surface is read-only.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@dataclass
class BioHybridFrameResult:
    """Strictly typed output packet detailing a full closed-loop step.

    Behaves both as a dataclass (``result.round``) and, for backward
    compatibility with pre-dataclass callers, as a mapping view of its
    fields (``result["round"]``, ``"latency_us" in result``,
    ``dict(result)``). The mapping surface is read-only.
    """

    round: int
    num_spikes: int
    num_aer_events: int
    num_bitstreams: int
    num_opto_pulses: int
    latency_us: float
    health: Dict[str, Any]
    spikes: List[DetectedSpike]
    aer_events: List[AEREvent]
    bitstreams: Dict[int, np.ndarray[Any, Any]]
    opto_pulses: List[OptogeneticPulse]

    def __post_init__(self) -> None:
        """Validate counts and payload cardinalities for one closed-loop frame."""
        require_nonnegative_int(self.round, "round")
        for name, value in (
            ("num_spikes", self.num_spikes),
            ("num_aer_events", self.num_aer_events),
            ("num_bitstreams", self.num_bitstreams),
            ("num_opto_pulses", self.num_opto_pulses),
        ):
            require_nonnegative_int(value, name)
        require_nonnegative(self.latency_us, "latency_us")
        if self.num_spikes != len(self.spikes):
            raise ValueError("num_spikes must equal len(spikes)")
        if self.num_aer_events != len(self.aer_events):
            raise ValueError("num_aer_events must equal len(aer_events)")
        if self.num_bitstreams != len(self.bitstreams):
            raise ValueError("num_bitstreams must equal len(bitstreams)")
        if self.num_opto_pulses != len(self.opto_pulses):
            raise ValueError("num_opto_pulses must equal len(opto_pulses)")
        for neuron_id, bitstream in self.bitstreams.items():
            require_nonnegative_int(neuron_id, "bitstream neuron_id")
            validate_binary_bitstream(
                bitstream,
                name=f"bitstreams[{neuron_id}]",
                allow_empty=True,
            )

    def __getitem__(self, key: str) -> Any:
        """Return a dataclass field through the legacy mapping interface.

        Parameters
        ----------
        key:
            Public dataclass field name to read.

        Returns
        -------
        Any
            The underlying field value, preserving object identity for mutable
            payloads such as ``health`` and ``bitstreams``.

        Raises
        ------
        KeyError
            If ``key`` is not a public field name.
        """
        if not isinstance(key, str) or key.startswith("_"):
            raise KeyError(key)
        try:
            return getattr(self, key)
        except AttributeError as exc:
            raise KeyError(key) from exc

    def __contains__(self, key: object) -> bool:
        """Return whether ``key`` names a public result field.

        Parameters
        ----------
        key:
            Candidate mapping key.

        Returns
        -------
        bool
            ``True`` only for string keys matching declared dataclass fields.
        """
        if not isinstance(key, str):
            return False
        return key in {f.name for f in fields(self)}

    def keys(self) -> List[str]:
        """Return the mapping-view field names in dataclass declaration order.

        Returns
        -------
        list[str]
            Public field names accepted by ``__getitem__`` and ``__contains__``.
        """
        return [f.name for f in fields(self)]

__post_init__()

Validate counts and payload cardinalities for one closed-loop frame.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __post_init__(self) -> None:
    """Validate counts and payload cardinalities for one closed-loop frame."""
    require_nonnegative_int(self.round, "round")
    for name, value in (
        ("num_spikes", self.num_spikes),
        ("num_aer_events", self.num_aer_events),
        ("num_bitstreams", self.num_bitstreams),
        ("num_opto_pulses", self.num_opto_pulses),
    ):
        require_nonnegative_int(value, name)
    require_nonnegative(self.latency_us, "latency_us")
    if self.num_spikes != len(self.spikes):
        raise ValueError("num_spikes must equal len(spikes)")
    if self.num_aer_events != len(self.aer_events):
        raise ValueError("num_aer_events must equal len(aer_events)")
    if self.num_bitstreams != len(self.bitstreams):
        raise ValueError("num_bitstreams must equal len(bitstreams)")
    if self.num_opto_pulses != len(self.opto_pulses):
        raise ValueError("num_opto_pulses must equal len(opto_pulses)")
    for neuron_id, bitstream in self.bitstreams.items():
        require_nonnegative_int(neuron_id, "bitstream neuron_id")
        validate_binary_bitstream(
            bitstream,
            name=f"bitstreams[{neuron_id}]",
            allow_empty=True,
        )

__getitem__(key)

Return a dataclass field through the legacy mapping interface.

Parameters

key: Public dataclass field name to read.

Returns

Any The underlying field value, preserving object identity for mutable payloads such as health and bitstreams.

Raises

KeyError If key is not a public field name.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __getitem__(self, key: str) -> Any:
    """Return a dataclass field through the legacy mapping interface.

    Parameters
    ----------
    key:
        Public dataclass field name to read.

    Returns
    -------
    Any
        The underlying field value, preserving object identity for mutable
        payloads such as ``health`` and ``bitstreams``.

    Raises
    ------
    KeyError
        If ``key`` is not a public field name.
    """
    if not isinstance(key, str) or key.startswith("_"):
        raise KeyError(key)
    try:
        return getattr(self, key)
    except AttributeError as exc:
        raise KeyError(key) from exc

__contains__(key)

Return whether key names a public result field.

Parameters

key: Candidate mapping key.

Returns

bool True only for string keys matching declared dataclass fields.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __contains__(self, key: object) -> bool:
    """Return whether ``key`` names a public result field.

    Parameters
    ----------
    key:
        Candidate mapping key.

    Returns
    -------
    bool
        ``True`` only for string keys matching declared dataclass fields.
    """
    if not isinstance(key, str):
        return False
    return key in {f.name for f in fields(self)}

keys()

Return the mapping-view field names in dataclass declaration order.

Returns

list[str] Public field names accepted by __getitem__ and __contains__.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def keys(self) -> List[str]:
    """Return the mapping-view field names in dataclass declaration order.

    Returns
    -------
    list[str]
        Public field names accepted by ``__getitem__`` and ``__contains__``.
    """
    return [f.name for f in fields(self)]

DetectedSpike dataclass

One detected spike event from MEA data.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@dataclass
class DetectedSpike:
    """One detected spike event from MEA data."""

    channel: int
    timestamp_s: float
    amplitude_uv: float
    unit_id: int = 0  # cluster assignment
    waveform: Optional[np.ndarray[Any, Any]] = None

    def __post_init__(self) -> None:
        """Validate spike identity, timing, amplitude, and optional waveform."""
        require_nonnegative_int(self.channel, "channel")
        require_nonnegative(self.timestamp_s, "timestamp_s")
        require_finite(self.amplitude_uv, "amplitude_uv")
        require_nonnegative_int(self.unit_id, "unit_id")
        if self.waveform is None:
            return
        if not isinstance(self.waveform, np.ndarray):
            raise TypeError("waveform must be a NumPy array when provided")
        if self.waveform.ndim != 1 or self.waveform.size == 0:
            raise ValueError("waveform must be a non-empty one-dimensional array")
        if not np.issubdtype(self.waveform.dtype, np.number):
            raise TypeError("waveform must have a numeric dtype")
        if not np.all(np.isfinite(self.waveform)):
            raise ValueError("waveform must contain only finite values")

__post_init__()

Validate spike identity, timing, amplitude, and optional waveform.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __post_init__(self) -> None:
    """Validate spike identity, timing, amplitude, and optional waveform."""
    require_nonnegative_int(self.channel, "channel")
    require_nonnegative(self.timestamp_s, "timestamp_s")
    require_finite(self.amplitude_uv, "amplitude_uv")
    require_nonnegative_int(self.unit_id, "unit_id")
    if self.waveform is None:
        return
    if not isinstance(self.waveform, np.ndarray):
        raise TypeError("waveform must be a NumPy array when provided")
    if self.waveform.ndim != 1 or self.waveform.size == 0:
        raise ValueError("waveform must be a non-empty one-dimensional array")
    if not np.issubdtype(self.waveform.dtype, np.number):
        raise TypeError("waveform must have a numeric dtype")
    if not np.all(np.isfinite(self.waveform)):
        raise ValueError("waveform must contain only finite values")

MEAConfig dataclass

Multi-electrode array configuration.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@dataclass
class MEAConfig:
    """Multi-electrode array configuration."""

    layout: MEALayout = MEALayout.MEA_60
    num_channels: int = 60
    sample_rate_hz: float = 20_000.0
    voltage_gain: float = 1000.0
    noise_floor_uv: float = 5.0
    spike_threshold_sigma: float = 5.0
    electrode_pitch_um: float = 200.0

    def __post_init__(self) -> None:
        """Validate physical and acquisition configuration boundaries."""
        if not isinstance(self.layout, MEALayout):
            raise TypeError("layout must be a MEALayout")
        require_positive_int(self.num_channels, "num_channels")
        require_positive(self.sample_rate_hz, "sample_rate_hz")
        require_positive(self.voltage_gain, "voltage_gain")
        require_nonnegative(self.noise_floor_uv, "noise_floor_uv")
        require_positive(self.spike_threshold_sigma, "spike_threshold_sigma")
        require_positive(self.electrode_pitch_um, "electrode_pitch_um")

    @classmethod
    def from_layout(cls, layout: MEALayout) -> MEAConfig:
        """Create a configuration preset for a standard MEA layout.

        Parameters
        ----------
        layout:
            Standard electrode layout whose channel count and pitch should seed
            the returned configuration.

        Returns
        -------
        MEAConfig
            Configuration with the requested layout and canonical channel/pitch
            preset while retaining the default sampling and detector gains.
        """
        if not isinstance(layout, MEALayout):
            raise TypeError("layout must be a MEALayout")
        presets: Dict[MEALayout, Dict[str, Any]] = {
            MEALayout.MEA_60: dict(num_channels=60, electrode_pitch_um=200.0),
            MEALayout.MEA_120: dict(num_channels=120, electrode_pitch_um=100.0),
            MEALayout.MEA_256: dict(num_channels=256, electrode_pitch_um=100.0),
            MEALayout.MEA_4096: dict(num_channels=4096, electrode_pitch_um=17.5),
            MEALayout.CUSTOM: dict(num_channels=60, electrode_pitch_um=200.0),
        }
        return cls(layout=layout, **presets[layout])

__post_init__()

Validate physical and acquisition configuration boundaries.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __post_init__(self) -> None:
    """Validate physical and acquisition configuration boundaries."""
    if not isinstance(self.layout, MEALayout):
        raise TypeError("layout must be a MEALayout")
    require_positive_int(self.num_channels, "num_channels")
    require_positive(self.sample_rate_hz, "sample_rate_hz")
    require_positive(self.voltage_gain, "voltage_gain")
    require_nonnegative(self.noise_floor_uv, "noise_floor_uv")
    require_positive(self.spike_threshold_sigma, "spike_threshold_sigma")
    require_positive(self.electrode_pitch_um, "electrode_pitch_um")

from_layout(layout) classmethod

Create a configuration preset for a standard MEA layout.

Parameters

layout: Standard electrode layout whose channel count and pitch should seed the returned configuration.

Returns

MEAConfig Configuration with the requested layout and canonical channel/pitch preset while retaining the default sampling and detector gains.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@classmethod
def from_layout(cls, layout: MEALayout) -> MEAConfig:
    """Create a configuration preset for a standard MEA layout.

    Parameters
    ----------
    layout:
        Standard electrode layout whose channel count and pitch should seed
        the returned configuration.

    Returns
    -------
    MEAConfig
        Configuration with the requested layout and canonical channel/pitch
        preset while retaining the default sampling and detector gains.
    """
    if not isinstance(layout, MEALayout):
        raise TypeError("layout must be a MEALayout")
    presets: Dict[MEALayout, Dict[str, Any]] = {
        MEALayout.MEA_60: dict(num_channels=60, electrode_pitch_um=200.0),
        MEALayout.MEA_120: dict(num_channels=120, electrode_pitch_um=100.0),
        MEALayout.MEA_256: dict(num_channels=256, electrode_pitch_um=100.0),
        MEALayout.MEA_4096: dict(num_channels=4096, electrode_pitch_um=17.5),
        MEALayout.CUSTOM: dict(num_channels=60, electrode_pitch_um=200.0),
    }
    return cls(layout=layout, **presets[layout])

MEALayout

Bases: Enum

Standard MEA electrode layouts.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

class MEALayout(Enum):
    """Standard MEA electrode layouts."""

    MEA_60 = "60ch"
    MEA_120 = "120ch"
    MEA_256 = "256ch"
    MEA_4096 = "4096ch"
    CUSTOM = "custom"

OptogeneticPulse dataclass

One optical stimulation pulse.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

@dataclass
class OptogeneticPulse:
    """One optical stimulation pulse."""

    channel: int
    onset_ms: float
    duration_ms: float
    intensity_mw_mm2: float = 1.0
    wavelength_nm: int = 470  # blue (ChR2)
    illuminated_area_mm2: float = 1.0

    def __post_init__(self) -> None:
        """Validate timing, irradiance, wavelength, and illuminated area."""
        require_nonnegative_int(self.channel, "channel")
        require_nonnegative(self.onset_ms, "onset_ms")
        require_positive(self.duration_ms, "duration_ms")
        require_nonnegative(self.intensity_mw_mm2, "intensity_mw_mm2")
        require_positive_int(self.wavelength_nm, "wavelength_nm")
        require_positive(self.illuminated_area_mm2, "illuminated_area_mm2")

    @property
    def power_mw(self) -> float:
        """Return optical power as irradiance multiplied by illuminated area."""
        return self.intensity_mw_mm2 * self.illuminated_area_mm2

power_mw property

Return optical power as irradiance multiplied by illuminated area.

__post_init__()

Validate timing, irradiance, wavelength, and illuminated area.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

def __post_init__(self) -> None:
    """Validate timing, irradiance, wavelength, and illuminated area."""
    require_nonnegative_int(self.channel, "channel")
    require_nonnegative(self.onset_ms, "onset_ms")
    require_positive(self.duration_ms, "duration_ms")
    require_nonnegative(self.intensity_mw_mm2, "intensity_mw_mm2")
    require_positive_int(self.wavelength_nm, "wavelength_nm")
    require_positive(self.illuminated_area_mm2, "illuminated_area_mm2")

StimProtocol

Bases: Enum

Optogenetic stimulation protocols.

Source code in src/sc_neurocore/bioware/bioware_contracts.py

class StimProtocol(Enum):
    """Optogenetic stimulation protocols."""

    CONSTANT = "constant"
    PULSED = "pulsed"
    GRADED = "graded"
    PATTERN = "pattern"

AERToSCConverter dataclass

Converts AER event streams to SC bitstreams.

Uses a time-windowed rate code: count events per neuron per window, then LFSR-encode the resulting firing probabilities.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

@dataclass
class AERToSCConverter:
    """Converts AER event streams to SC bitstreams.

    Uses a time-windowed rate code: count events per neuron per window,
    then LFSR-encode the resulting firing probabilities.
    """

    window_ticks: int = 0x10000
    bitstream_length: int = 256
    num_neurons: int = 128
    lfsr_seed: int = 0xACE1

    def __post_init__(self) -> None:
        """Validate window, bitstream, neuron, and LFSR boundaries."""
        require_positive_int(self.window_ticks, "window_ticks")
        require_positive_int(self.bitstream_length, "bitstream_length")
        require_positive_int(self.num_neurons, "num_neurons")
        require_nonnegative_int(self.lfsr_seed, "lfsr_seed")
        if self.lfsr_seed > 0xFFFF:
            raise ValueError("lfsr_seed must fit 16 bits")

    def convert(self, events: List[AEREvent]) -> Dict[int, np.ndarray[Any, Any]]:
        """Convert AER events to per-neuron SC bitstreams."""
        # Count events per neuron in the window
        counts: Dict[int, int] = {}
        for e in events:
            if e.valid:
                if e.neuron_id >= self.num_neurons:
                    raise ValueError(
                        f"AER neuron_id {e.neuron_id} is outside num_neurons={self.num_neurons}"
                    )
                if e.timestamp >= self.window_ticks:
                    raise ValueError(
                        f"AER timestamp {e.timestamp} is outside window_ticks={self.window_ticks}"
                    )
                counts[e.neuron_id] = counts.get(e.neuron_id, 0) + 1

        max_count = max(counts.values()) if counts else 1
        bitstreams = {}
        for nid, count in counts.items():
            prob = count / max_count
            bitstreams[nid] = self._lfsr_encode(prob, nid)
        return bitstreams

    def _lfsr_encode(self, probability: float, neuron_id: int) -> np.ndarray[Any, Any]:
        """LFSR-16 encoding (bit-compatible with core_engine)."""
        require_finite(probability, "probability")
        if not 0.0 <= probability <= 1.0:
            raise ValueError("probability must be in [0, 1]")
        require_nonnegative_int(neuron_id, "neuron_id")
        if neuron_id >= self.num_neurons:
            raise ValueError("neuron_id must be smaller than num_neurons")
        threshold = int(np.clip(probability, 0.0, 1.0) * 65535)
        seed = (self.lfsr_seed + neuron_id * 7919) & 0xFFFF
        if seed == 0:
            seed = 1
        reg = seed
        bits = np.zeros(self.bitstream_length, dtype=np.uint8)
        for i in range(self.bitstream_length):
            bits[i] = 1 if reg < threshold else 0
            feedback = ((reg >> 15) ^ (reg >> 13) ^ (reg >> 12) ^ (reg >> 10)) & 1
            reg = ((reg << 1) | feedback) & 0xFFFF
        return bits

__post_init__()

Validate window, bitstream, neuron, and LFSR boundaries.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def __post_init__(self) -> None:
    """Validate window, bitstream, neuron, and LFSR boundaries."""
    require_positive_int(self.window_ticks, "window_ticks")
    require_positive_int(self.bitstream_length, "bitstream_length")
    require_positive_int(self.num_neurons, "num_neurons")
    require_nonnegative_int(self.lfsr_seed, "lfsr_seed")
    if self.lfsr_seed > 0xFFFF:
        raise ValueError("lfsr_seed must fit 16 bits")

convert(events)

Convert AER events to per-neuron SC bitstreams.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def convert(self, events: List[AEREvent]) -> Dict[int, np.ndarray[Any, Any]]:
    """Convert AER events to per-neuron SC bitstreams."""
    # Count events per neuron in the window
    counts: Dict[int, int] = {}
    for e in events:
        if e.valid:
            if e.neuron_id >= self.num_neurons:
                raise ValueError(
                    f"AER neuron_id {e.neuron_id} is outside num_neurons={self.num_neurons}"
                )
            if e.timestamp >= self.window_ticks:
                raise ValueError(
                    f"AER timestamp {e.timestamp} is outside window_ticks={self.window_ticks}"
                )
            counts[e.neuron_id] = counts.get(e.neuron_id, 0) + 1

    max_count = max(counts.values()) if counts else 1
    bitstreams = {}
    for nid, count in counts.items():
        prob = count / max_count
        bitstreams[nid] = self._lfsr_encode(prob, nid)
    return bitstreams

MEAToAERTranscoder dataclass

Converts MEA spike events to AER events for hardware.

Maps biological electrode channels to AER neuron IDs, converting real-time timestamps to hardware clock ticks.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

@dataclass
class MEAToAERTranscoder:
    """Converts MEA spike events to AER events for hardware.

    Maps biological electrode channels to AER neuron IDs,
    converting real-time timestamps to hardware clock ticks.
    """

    hw_clock_hz: float = 1e6  # 1 MHz default AER clock
    channel_map: Optional[Dict[int, int]] = None  # electrode → neuron_id

    def __post_init__(self) -> None:
        """Validate the hardware clock and optional channel mapping."""
        require_positive(self.hw_clock_hz, "hw_clock_hz")
        if self.channel_map is None:
            return
        for channel, neuron_id in self.channel_map.items():
            require_nonnegative_int(channel, "channel_map channel")
            require_nonnegative_int(neuron_id, "channel_map neuron_id")

    def transcode(
        self,
        spikes: List[DetectedSpike],
        t_start_s: float = 0.0,
    ) -> List[AEREvent]:
        """Convert spikes in one 16-bit hardware-clock epoch to AER events.

        ``DetectedSpike.timestamp_s`` and ``t_start_s`` must use the same time
        origin. The method rejects events outside the representable epoch;
        callers must split longer recordings instead of accepting timestamp
        wraparound and the resulting loss of temporal ordering.
        """
        require_nonnegative(t_start_s, "t_start_s")
        events = []
        for spike in spikes:
            neuron_id = self._map_channel(spike.channel)
            relative_s = spike.timestamp_s - t_start_s
            if relative_s < 0.0:
                raise ValueError("spike timestamp precedes t_start_s")
            ticks = relative_s * self.hw_clock_hz
            if not np.isfinite(ticks) or ticks > 0xFFFF:
                raise ValueError("spike timestamp does not fit the 16-bit AER window")
            ts_hw = int(ticks)
            events.append(
                AEREvent(
                    neuron_id=neuron_id,
                    timestamp=ts_hw,
                    valid=True,
                )
            )
        # Sort by timestamp (AER is time-ordered)
        events.sort(key=lambda e: e.timestamp)
        return events

    def _map_channel(self, channel: int) -> int:
        if self.channel_map is not None:
            return self.channel_map.get(channel, channel)
        return channel

__post_init__()

Validate the hardware clock and optional channel mapping.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def __post_init__(self) -> None:
    """Validate the hardware clock and optional channel mapping."""
    require_positive(self.hw_clock_hz, "hw_clock_hz")
    if self.channel_map is None:
        return
    for channel, neuron_id in self.channel_map.items():
        require_nonnegative_int(channel, "channel_map channel")
        require_nonnegative_int(neuron_id, "channel_map neuron_id")

transcode(spikes, t_start_s=0.0)

Convert spikes in one 16-bit hardware-clock epoch to AER events.

DetectedSpike.timestamp_s and t_start_s must use the same time origin. The method rejects events outside the representable epoch; callers must split longer recordings instead of accepting timestamp wraparound and the resulting loss of temporal ordering.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def transcode(
    self,
    spikes: List[DetectedSpike],
    t_start_s: float = 0.0,
) -> List[AEREvent]:
    """Convert spikes in one 16-bit hardware-clock epoch to AER events.

    ``DetectedSpike.timestamp_s`` and ``t_start_s`` must use the same time
    origin. The method rejects events outside the representable epoch;
    callers must split longer recordings instead of accepting timestamp
    wraparound and the resulting loss of temporal ordering.
    """
    require_nonnegative(t_start_s, "t_start_s")
    events = []
    for spike in spikes:
        neuron_id = self._map_channel(spike.channel)
        relative_s = spike.timestamp_s - t_start_s
        if relative_s < 0.0:
            raise ValueError("spike timestamp precedes t_start_s")
        ticks = relative_s * self.hw_clock_hz
        if not np.isfinite(ticks) or ticks > 0xFFFF:
            raise ValueError("spike timestamp does not fit the 16-bit AER window")
        ts_hw = int(ticks)
        events.append(
            AEREvent(
                neuron_id=neuron_id,
                timestamp=ts_hw,
                valid=True,
            )
        )
    # Sort by timestamp (AER is time-ordered)
    events.sort(key=lambda e: e.timestamp)
    return events

SCToOptoEncoder dataclass

Encodes SC bitstream output as optogenetic pulse sequences.

Maps SC bitstream density to optical stimulation intensity, enabling closed-loop feedback from in-silico → biological. Enforces total power budget for tissue safety.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

@dataclass
class SCToOptoEncoder:
    """Encodes SC bitstream output as optogenetic pulse sequences.

    Maps SC bitstream density to optical stimulation intensity,
    enabling closed-loop feedback from in-silico → biological.
    Enforces total power budget for tissue safety.
    """

    max_intensity_mw_mm2: float = 5.0
    min_pulse_ms: float = 1.0
    max_pulse_ms: float = 50.0
    wavelength_nm: int = 470
    clock_period_ms: float = 0.001  # 1 MHz
    max_total_power_mw: float = 50.0
    illuminated_area_mm2: float = 1.0

    def __post_init__(self) -> None:
        """Validate optical timing, irradiance, area, and power limits."""
        require_nonnegative(self.max_intensity_mw_mm2, "max_intensity_mw_mm2")
        require_positive(self.min_pulse_ms, "min_pulse_ms")
        require_positive(self.max_pulse_ms, "max_pulse_ms")
        if self.max_pulse_ms < self.min_pulse_ms:
            raise ValueError("max_pulse_ms must be >= min_pulse_ms")
        require_positive_int(self.wavelength_nm, "wavelength_nm")
        require_nonnegative(self.clock_period_ms, "clock_period_ms")
        require_positive(self.max_total_power_mw, "max_total_power_mw")
        require_positive(self.illuminated_area_mm2, "illuminated_area_mm2")

    def encode(
        self,
        bitstreams: Dict[int, np.ndarray[Any, Any]],
        t_start_ms: float = 0.0,
    ) -> List[OptogeneticPulse]:
        """Convert SC bitstreams to optogenetic pulses."""
        require_nonnegative(t_start_ms, "t_start_ms")
        pulses = []
        total_power = 0.0
        for nid, bs in sorted(bitstreams.items()):
            require_nonnegative_int(nid, "bitstream neuron_id")
            validate_binary_bitstream(bs, name=f"bitstreams[{nid}]", allow_empty=True)
            density = float(np.sum(bs)) / len(bs) if len(bs) > 0 else 0.0
            if density < 0.01:
                continue

            intensity = density * self.max_intensity_mw_mm2
            power_mw = intensity * self.illuminated_area_mm2
            if total_power + power_mw > self.max_total_power_mw:
                continue
            total_power += power_mw

            duration = self.min_pulse_ms + density * (self.max_pulse_ms - self.min_pulse_ms)
            onset = t_start_ms + nid * self.clock_period_ms

            pulses.append(
                OptogeneticPulse(
                    channel=nid,
                    onset_ms=onset,
                    duration_ms=duration,
                    intensity_mw_mm2=intensity,
                    wavelength_nm=self.wavelength_nm,
                    illuminated_area_mm2=self.illuminated_area_mm2,
                )
            )
        return pulses

__post_init__()

Validate optical timing, irradiance, area, and power limits.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def __post_init__(self) -> None:
    """Validate optical timing, irradiance, area, and power limits."""
    require_nonnegative(self.max_intensity_mw_mm2, "max_intensity_mw_mm2")
    require_positive(self.min_pulse_ms, "min_pulse_ms")
    require_positive(self.max_pulse_ms, "max_pulse_ms")
    if self.max_pulse_ms < self.min_pulse_ms:
        raise ValueError("max_pulse_ms must be >= min_pulse_ms")
    require_positive_int(self.wavelength_nm, "wavelength_nm")
    require_nonnegative(self.clock_period_ms, "clock_period_ms")
    require_positive(self.max_total_power_mw, "max_total_power_mw")
    require_positive(self.illuminated_area_mm2, "illuminated_area_mm2")

encode(bitstreams, t_start_ms=0.0)

Convert SC bitstreams to optogenetic pulses.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def encode(
    self,
    bitstreams: Dict[int, np.ndarray[Any, Any]],
    t_start_ms: float = 0.0,
) -> List[OptogeneticPulse]:
    """Convert SC bitstreams to optogenetic pulses."""
    require_nonnegative(t_start_ms, "t_start_ms")
    pulses = []
    total_power = 0.0
    for nid, bs in sorted(bitstreams.items()):
        require_nonnegative_int(nid, "bitstream neuron_id")
        validate_binary_bitstream(bs, name=f"bitstreams[{nid}]", allow_empty=True)
        density = float(np.sum(bs)) / len(bs) if len(bs) > 0 else 0.0
        if density < 0.01:
            continue

        intensity = density * self.max_intensity_mw_mm2
        power_mw = intensity * self.illuminated_area_mm2
        if total_power + power_mw > self.max_total_power_mw:
            continue
        total_power += power_mw

        duration = self.min_pulse_ms + density * (self.max_pulse_ms - self.min_pulse_ms)
        onset = t_start_ms + nid * self.clock_period_ms

        pulses.append(
            OptogeneticPulse(
                channel=nid,
                onset_ms=onset,
                duration_ms=duration,
                intensity_mw_mm2=intensity,
                wavelength_nm=self.wavelength_nm,
                illuminated_area_mm2=self.illuminated_area_mm2,
            )
        )
    return pulses

MultiWellPlate dataclass

Multi-well plate (e.g., 6/24/48/96-well MEA plate).

Source code in src/sc_neurocore/bioware/bioware_experiment.py

@dataclass
class MultiWellPlate:
    """Multi-well plate (e.g., 6/24/48/96-well MEA plate)."""

    wells: List[WellConfig] = field(default_factory=list)

    def __post_init__(self) -> None:
        """Validate well types and unique well identifiers."""
        seen: set[str] = set()
        for well in self.wells:
            if not isinstance(well, WellConfig):
                raise TypeError("wells must contain WellConfig instances")
            if well.well_id in seen:
                raise ValueError(f"duplicate well_id: {well.well_id}")
            seen.add(well.well_id)

    def add_well(self, well: WellConfig) -> None:
        """Append a well configuration to the plate.

        Parameters
        ----------
        well:
            Well metadata and MEA configuration to append.
        """
        if not isinstance(well, WellConfig):
            raise TypeError("well must be a WellConfig")
        if self.get_well(well.well_id) is not None:
            raise ValueError(f"duplicate well_id: {well.well_id}")
        self.wells.append(well)

    @classmethod
    def standard_6_well(cls, layout: MEALayout = MEALayout.MEA_60) -> MultiWellPlate:
        """Construct a six-well plate with uniform MEA layout presets.

        Parameters
        ----------
        layout:
            MEA layout preset used for each generated well.

        Returns
        -------
        MultiWellPlate
            Plate containing wells ``W1`` through ``W6``.
        """
        if not isinstance(layout, MEALayout):
            raise TypeError("layout must be a MEALayout")
        plate = cls()
        for i in range(6):
            plate.add_well(
                WellConfig(
                    well_id=f"W{i + 1}",
                    mea_config=MEAConfig.from_layout(layout),
                )
            )
        return plate

    @property
    def num_wells(self) -> int:
        """Return the number of configured wells.

        Returns
        -------
        int
            Length of the plate's well list.
        """
        return len(self.wells)

    def get_well(self, well_id: str) -> Optional[WellConfig]:
        """Return a well by identifier.

        Parameters
        ----------
        well_id:
            Identifier such as ``"W1"``.

        Returns
        -------
        WellConfig | None
            Matching well configuration, or ``None`` when the plate does not
            contain ``well_id``.
        """
        if not well_id or not well_id.strip():
            raise ValueError("well_id must not be empty")
        return next((w for w in self.wells if w.well_id == well_id), None)

num_wells property

Return the number of configured wells.

Returns

int Length of the plate's well list.

__post_init__()

Validate well types and unique well identifiers.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def __post_init__(self) -> None:
    """Validate well types and unique well identifiers."""
    seen: set[str] = set()
    for well in self.wells:
        if not isinstance(well, WellConfig):
            raise TypeError("wells must contain WellConfig instances")
        if well.well_id in seen:
            raise ValueError(f"duplicate well_id: {well.well_id}")
        seen.add(well.well_id)

add_well(well)

Append a well configuration to the plate.

Parameters

well: Well metadata and MEA configuration to append.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def add_well(self, well: WellConfig) -> None:
    """Append a well configuration to the plate.

    Parameters
    ----------
    well:
        Well metadata and MEA configuration to append.
    """
    if not isinstance(well, WellConfig):
        raise TypeError("well must be a WellConfig")
    if self.get_well(well.well_id) is not None:
        raise ValueError(f"duplicate well_id: {well.well_id}")
    self.wells.append(well)

standard_6_well(layout=MEALayout.MEA_60) classmethod

Construct a six-well plate with uniform MEA layout presets.

Parameters

layout: MEA layout preset used for each generated well.

Returns

MultiWellPlate Plate containing wells W1 through W6.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

@classmethod
def standard_6_well(cls, layout: MEALayout = MEALayout.MEA_60) -> MultiWellPlate:
    """Construct a six-well plate with uniform MEA layout presets.

    Parameters
    ----------
    layout:
        MEA layout preset used for each generated well.

    Returns
    -------
    MultiWellPlate
        Plate containing wells ``W1`` through ``W6``.
    """
    if not isinstance(layout, MEALayout):
        raise TypeError("layout must be a MEALayout")
    plate = cls()
    for i in range(6):
        plate.add_well(
            WellConfig(
                well_id=f"W{i + 1}",
                mea_config=MEAConfig.from_layout(layout),
            )
        )
    return plate

get_well(well_id)

Return a well by identifier.

Parameters

well_id: Identifier such as "W1".

Returns

WellConfig | None Matching well configuration, or None when the plate does not contain well_id.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def get_well(self, well_id: str) -> Optional[WellConfig]:
    """Return a well by identifier.

    Parameters
    ----------
    well_id:
        Identifier such as ``"W1"``.

    Returns
    -------
    WellConfig | None
        Matching well configuration, or ``None`` when the plate does not
        contain ``well_id``.
    """
    if not well_id or not well_id.strip():
        raise ValueError("well_id must not be empty")
    return next((w for w in self.wells if w.well_id == well_id), None)

PharmModel dataclass

Simulates effect of pharmacological agents on spike rate.

Models excitatory (e.g., bicuculline) or inhibitory (e.g., TTX) agents as gain factors on firing rate.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

@dataclass
class PharmModel:
    """Simulates effect of pharmacological agents on spike rate.

    Models excitatory (e.g., bicuculline) or inhibitory (e.g., TTX) agents
    as gain factors on firing rate.
    """

    agent_name: str = "none"
    gain: float = 1.0  # >1 = excitatory, <1 = inhibitory, 0 = silencing
    onset_delay_s: float = 30.0
    wash_time_s: float = 120.0
    _applied_at: float = -1.0

    def __post_init__(self) -> None:
        """Validate the pharmacological gain and experiment-time constants."""
        if not self.agent_name or not self.agent_name.strip():
            raise ValueError("agent_name must not be empty")
        require_nonnegative(self.gain, "gain")
        require_nonnegative(self.onset_delay_s, "onset_delay_s")
        require_positive(self.wash_time_s, "wash_time_s")
        require_finite(self._applied_at, "_applied_at")
        if self._applied_at < -1.0:
            raise ValueError("_applied_at must be -1 or a non-negative timestamp")

    def apply(self, t_current_s: float) -> None:
        """Mark the pharmacological agent as applied at the current time.

        Parameters
        ----------
        t_current_s:
            Experiment time in seconds used as the onset reference for
            subsequent gain interpolation.
        """
        require_nonnegative(t_current_s, "t_current_s")
        self._applied_at = t_current_s

    def effective_gain(self, t_current_s: float) -> float:
        """Return the active firing-rate gain at an experiment timestamp.

        Parameters
        ----------
        t_current_s:
            Experiment time in seconds.

        Returns
        -------
        float
            ``1.0`` before application, a linearly interpolated onset gain
            during ``onset_delay_s``, or the configured steady-state gain after
            onset.
        """
        require_nonnegative(t_current_s, "t_current_s")
        if self._applied_at < 0:
            return 1.0
        elapsed = t_current_s - self._applied_at
        if elapsed < 0.0:
            raise ValueError("t_current_s must not precede the application time")
        if elapsed < self.onset_delay_s:
            frac = elapsed / self.onset_delay_s
            return 1.0 + frac * (self.gain - 1.0)
        return self.gain

    def modulate_spikes(
        self, spike_counts: np.ndarray[Any, Any], t_current_s: float
    ) -> np.ndarray[Any, Any]:
        """Modulate spike counts by pharmacological gain."""
        if not isinstance(spike_counts, np.ndarray):
            raise TypeError("spike_counts must be a NumPy array")
        if spike_counts.ndim != 1:
            raise ValueError("spike_counts must be one-dimensional")
        if not np.issubdtype(spike_counts.dtype, np.number):
            raise TypeError("spike_counts must have a numeric dtype")
        if not np.all(np.isfinite(spike_counts)) or np.any(spike_counts < 0):
            raise ValueError("spike_counts must contain finite non-negative values")
        g = self.effective_gain(t_current_s)
        return np.round(spike_counts * g).astype(int)

    def modulate_spike_events(
        self,
        spikes: List[DetectedSpike],
        t_current_s: float,
    ) -> List[DetectedSpike]:
        """Apply pharmacological rate gain to spike events.

        Inhibitory gains deterministically thin events across the observed
        response span instead of truncating the head of the frame. Excitatory
        gains preserve observed events and insert synthetic events inside the
        observed temporal support, using nearest observed spikes as channel,
        unit, amplitude, and waveform templates.
        """
        require_nonnegative(t_current_s, "t_current_s")
        if not spikes:
            return []

        gain = self.effective_gain(t_current_s)
        if not math.isfinite(gain) or gain < 0.0:
            raise ValueError("pharmacological gain must be finite and >= 0")

        ordered = sorted(spikes, key=lambda s: (s.timestamp_s, s.channel, s.unit_id))
        target_count = int(round(len(ordered) * gain))
        if target_count <= 0:
            return []
        if target_count == len(ordered):
            return list(ordered)
        if target_count < len(ordered):
            indices = _quantile_indices(len(ordered), target_count)
            return [ordered[i] for i in indices]

        extra = target_count - len(ordered)
        timestamps = np.array([s.timestamp_s for s in ordered], dtype=float)
        if not np.all(np.isfinite(timestamps)):
            raise ValueError("detected spike timestamps must be finite")

        if len(ordered) == 1 or timestamps[-1] <= timestamps[0]:
            synthetic = [
                _clone_spike(ordered[0], timestamp_s=float(timestamps[0])) for _ in range(extra)
            ]
        else:
            insert_times = np.linspace(timestamps[0], timestamps[-1], extra + 2)[1:-1]
            synthetic = []
            for t in insert_times:
                # insert_times are strictly interior to (timestamps[0],
                # timestamps[-1]) — the linspace endpoints are dropped — so for
                # t < timestamps[-1] a left-side searchsorted always yields
                # idx <= len(ordered) - 1; no upper clamp is reachable.
                idx = int(np.searchsorted(timestamps, t, side="left"))
                if idx > 0 and abs(timestamps[idx - 1] - t) <= abs(timestamps[idx] - t):
                    idx -= 1
                synthetic.append(_clone_spike(ordered[idx], timestamp_s=float(t)))

        return sorted([*ordered, *synthetic], key=lambda s: (s.timestamp_s, s.channel, s.unit_id))

__post_init__()

Validate the pharmacological gain and experiment-time constants.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def __post_init__(self) -> None:
    """Validate the pharmacological gain and experiment-time constants."""
    if not self.agent_name or not self.agent_name.strip():
        raise ValueError("agent_name must not be empty")
    require_nonnegative(self.gain, "gain")
    require_nonnegative(self.onset_delay_s, "onset_delay_s")
    require_positive(self.wash_time_s, "wash_time_s")
    require_finite(self._applied_at, "_applied_at")
    if self._applied_at < -1.0:
        raise ValueError("_applied_at must be -1 or a non-negative timestamp")

apply(t_current_s)

Mark the pharmacological agent as applied at the current time.

Parameters

t_current_s: Experiment time in seconds used as the onset reference for subsequent gain interpolation.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def apply(self, t_current_s: float) -> None:
    """Mark the pharmacological agent as applied at the current time.

    Parameters
    ----------
    t_current_s:
        Experiment time in seconds used as the onset reference for
        subsequent gain interpolation.
    """
    require_nonnegative(t_current_s, "t_current_s")
    self._applied_at = t_current_s

effective_gain(t_current_s)

Return the active firing-rate gain at an experiment timestamp.

Parameters

t_current_s: Experiment time in seconds.

Returns

float 1.0 before application, a linearly interpolated onset gain during onset_delay_s, or the configured steady-state gain after onset.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def effective_gain(self, t_current_s: float) -> float:
    """Return the active firing-rate gain at an experiment timestamp.

    Parameters
    ----------
    t_current_s:
        Experiment time in seconds.

    Returns
    -------
    float
        ``1.0`` before application, a linearly interpolated onset gain
        during ``onset_delay_s``, or the configured steady-state gain after
        onset.
    """
    require_nonnegative(t_current_s, "t_current_s")
    if self._applied_at < 0:
        return 1.0
    elapsed = t_current_s - self._applied_at
    if elapsed < 0.0:
        raise ValueError("t_current_s must not precede the application time")
    if elapsed < self.onset_delay_s:
        frac = elapsed / self.onset_delay_s
        return 1.0 + frac * (self.gain - 1.0)
    return self.gain

modulate_spikes(spike_counts, t_current_s)

Modulate spike counts by pharmacological gain.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def modulate_spikes(
    self, spike_counts: np.ndarray[Any, Any], t_current_s: float
) -> np.ndarray[Any, Any]:
    """Modulate spike counts by pharmacological gain."""
    if not isinstance(spike_counts, np.ndarray):
        raise TypeError("spike_counts must be a NumPy array")
    if spike_counts.ndim != 1:
        raise ValueError("spike_counts must be one-dimensional")
    if not np.issubdtype(spike_counts.dtype, np.number):
        raise TypeError("spike_counts must have a numeric dtype")
    if not np.all(np.isfinite(spike_counts)) or np.any(spike_counts < 0):
        raise ValueError("spike_counts must contain finite non-negative values")
    g = self.effective_gain(t_current_s)
    return np.round(spike_counts * g).astype(int)

modulate_spike_events(spikes, t_current_s)

Apply pharmacological rate gain to spike events.

Inhibitory gains deterministically thin events across the observed response span instead of truncating the head of the frame. Excitatory gains preserve observed events and insert synthetic events inside the observed temporal support, using nearest observed spikes as channel, unit, amplitude, and waveform templates.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def modulate_spike_events(
    self,
    spikes: List[DetectedSpike],
    t_current_s: float,
) -> List[DetectedSpike]:
    """Apply pharmacological rate gain to spike events.

    Inhibitory gains deterministically thin events across the observed
    response span instead of truncating the head of the frame. Excitatory
    gains preserve observed events and insert synthetic events inside the
    observed temporal support, using nearest observed spikes as channel,
    unit, amplitude, and waveform templates.
    """
    require_nonnegative(t_current_s, "t_current_s")
    if not spikes:
        return []

    gain = self.effective_gain(t_current_s)
    if not math.isfinite(gain) or gain < 0.0:
        raise ValueError("pharmacological gain must be finite and >= 0")

    ordered = sorted(spikes, key=lambda s: (s.timestamp_s, s.channel, s.unit_id))
    target_count = int(round(len(ordered) * gain))
    if target_count <= 0:
        return []
    if target_count == len(ordered):
        return list(ordered)
    if target_count < len(ordered):
        indices = _quantile_indices(len(ordered), target_count)
        return [ordered[i] for i in indices]

    extra = target_count - len(ordered)
    timestamps = np.array([s.timestamp_s for s in ordered], dtype=float)
    if not np.all(np.isfinite(timestamps)):
        raise ValueError("detected spike timestamps must be finite")

    if len(ordered) == 1 or timestamps[-1] <= timestamps[0]:
        synthetic = [
            _clone_spike(ordered[0], timestamp_s=float(timestamps[0])) for _ in range(extra)
        ]
    else:
        insert_times = np.linspace(timestamps[0], timestamps[-1], extra + 2)[1:-1]
        synthetic = []
        for t in insert_times:
            # insert_times are strictly interior to (timestamps[0],
            # timestamps[-1]) — the linspace endpoints are dropped — so for
            # t < timestamps[-1] a left-side searchsorted always yields
            # idx <= len(ordered) - 1; no upper clamp is reachable.
            idx = int(np.searchsorted(timestamps, t, side="left"))
            if idx > 0 and abs(timestamps[idx - 1] - t) <= abs(timestamps[idx] - t):
                idx -= 1
            synthetic.append(_clone_spike(ordered[idx], timestamp_s=float(t)))

    return sorted([*ordered, *synthetic], key=lambda s: (s.timestamp_s, s.channel, s.unit_id))

WellConfig dataclass

One well in a multi-well MEA plate.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

@dataclass
class WellConfig:
    """One well in a multi-well MEA plate."""

    well_id: str
    mea_config: MEAConfig
    culture_type: str = "cortical"
    passage_number: int = 0

    def __post_init__(self) -> None:
        """Validate well identity, culture label, passage, and MEA config."""
        if not self.well_id or not self.well_id.strip():
            raise ValueError("well_id must not be empty")
        if not isinstance(self.mea_config, MEAConfig):
            raise TypeError("mea_config must be an MEAConfig")
        if not self.culture_type or not self.culture_type.strip():
            raise ValueError("culture_type must not be empty")
        require_nonnegative_int(self.passage_number, "passage_number")

    @property
    def label(self) -> str:
        """Return the stable plate label for this well.

        Returns
        -------
        str
            Identifier combining well ID, culture type, and passage number.
        """
        return f"{self.well_id}_{self.culture_type}_P{self.passage_number}"

label property

Return the stable plate label for this well.

Returns

str Identifier combining well ID, culture type, and passage number.

__post_init__()

Validate well identity, culture label, passage, and MEA config.

Source code in src/sc_neurocore/bioware/bioware_experiment.py

def __post_init__(self) -> None:
    """Validate well identity, culture label, passage, and MEA config."""
    if not self.well_id or not self.well_id.strip():
        raise ValueError("well_id must not be empty")
    if not isinstance(self.mea_config, MEAConfig):
        raise TypeError("mea_config must be an MEAConfig")
    if not self.culture_type or not self.culture_type.strip():
        raise ValueError("culture_type must not be empty")
    require_nonnegative_int(self.passage_number, "passage_number")

BCMPlasticity dataclass

Bienenstock-Cooper-Munro plasticity adapter.

Implements sliding-threshold BCM rule where the modification threshold θ tracks the postsynaptic firing rate. Converts biological firing rates to Q8.8 weight deltas.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

@dataclass
class BCMPlasticity:
    """Bienenstock-Cooper-Munro plasticity adapter.

    Implements sliding-threshold BCM rule where the modification
    threshold θ tracks the postsynaptic firing rate. Converts
    biological firing rates to Q8.8 weight deltas.
    """

    tau_theta_ms: float = 1000.0  # threshold adaptation time constant
    learning_rate: float = 0.001
    theta: float = 0.0  # sliding threshold (internal state)
    w_max_q88: int = 512
    w_min_q88: int = 0

    def __post_init__(self) -> None:
        """Validate BCM dynamics and Q8.8 weight bounds."""
        require_positive(self.tau_theta_ms, "tau_theta_ms")
        require_nonnegative(self.learning_rate, "learning_rate")
        require_nonnegative(self.theta, "theta")
        require_nonnegative_int(self.w_min_q88, "w_min_q88")
        require_nonnegative_int(self.w_max_q88, "w_max_q88")
        if self.w_max_q88 < self.w_min_q88:
            raise ValueError("w_max_q88 must be >= w_min_q88")

    def update_theta(self, post_rate_hz: float, dt_ms: float) -> float:
        """Update the sliding threshold from postsynaptic activity."""
        require_nonnegative(post_rate_hz, "post_rate_hz")
        require_nonnegative(dt_ms, "dt_ms")
        alpha = dt_ms / self.tau_theta_ms
        target = post_rate_hz**2
        self.theta += alpha * (target - self.theta)
        return self.theta

    def compute_dw(self, pre_rate_hz: float, post_rate_hz: float) -> float:
        """BCM weight change: ΔW = η * x * y * (y - θ)."""
        require_nonnegative(pre_rate_hz, "pre_rate_hz")
        require_nonnegative(post_rate_hz, "post_rate_hz")
        require_nonnegative(self.theta, "theta")
        return self.learning_rate * pre_rate_hz * post_rate_hz * (post_rate_hz - self.theta)

    def update_weight(self, current_q88: int, pre_rate: float, post_rate: float) -> int:
        """Apply the BCM update to a saturated Q8.8 synaptic weight.

        Parameters
        ----------
        current_q88:
            Current synaptic weight encoded as Q8.8.
        pre_rate:
            Presynaptic firing rate in hertz.
        post_rate:
            Postsynaptic firing rate in hertz.

        Returns
        -------
        int
            Updated Q8.8 weight clamped to ``[w_min_q88, w_max_q88]``.
        """
        require_nonnegative_int(current_q88, "current_q88")
        if not self.w_min_q88 <= current_q88 <= self.w_max_q88:
            raise ValueError("current_q88 must be inside configured weight bounds")
        dw = self.compute_dw(pre_rate, post_rate)
        dw_q88 = int(dw * 256)
        new_w = current_q88 + dw_q88
        return max(self.w_min_q88, min(self.w_max_q88, new_w))

__post_init__()

Validate BCM dynamics and Q8.8 weight bounds.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def __post_init__(self) -> None:
    """Validate BCM dynamics and Q8.8 weight bounds."""
    require_positive(self.tau_theta_ms, "tau_theta_ms")
    require_nonnegative(self.learning_rate, "learning_rate")
    require_nonnegative(self.theta, "theta")
    require_nonnegative_int(self.w_min_q88, "w_min_q88")
    require_nonnegative_int(self.w_max_q88, "w_max_q88")
    if self.w_max_q88 < self.w_min_q88:
        raise ValueError("w_max_q88 must be >= w_min_q88")

update_theta(post_rate_hz, dt_ms)

Update the sliding threshold from postsynaptic activity.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def update_theta(self, post_rate_hz: float, dt_ms: float) -> float:
    """Update the sliding threshold from postsynaptic activity."""
    require_nonnegative(post_rate_hz, "post_rate_hz")
    require_nonnegative(dt_ms, "dt_ms")
    alpha = dt_ms / self.tau_theta_ms
    target = post_rate_hz**2
    self.theta += alpha * (target - self.theta)
    return self.theta

compute_dw(pre_rate_hz, post_rate_hz)

BCM weight change: ΔW = η * x * y * (y - θ).

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def compute_dw(self, pre_rate_hz: float, post_rate_hz: float) -> float:
    """BCM weight change: ΔW = η * x * y * (y - θ)."""
    require_nonnegative(pre_rate_hz, "pre_rate_hz")
    require_nonnegative(post_rate_hz, "post_rate_hz")
    require_nonnegative(self.theta, "theta")
    return self.learning_rate * pre_rate_hz * post_rate_hz * (post_rate_hz - self.theta)

update_weight(current_q88, pre_rate, post_rate)

Apply the BCM update to a saturated Q8.8 synaptic weight.

Parameters

current_q88: Current synaptic weight encoded as Q8.8. pre_rate: Presynaptic firing rate in hertz. post_rate: Postsynaptic firing rate in hertz.

Returns

int Updated Q8.8 weight clamped to [w_min_q88, w_max_q88].

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def update_weight(self, current_q88: int, pre_rate: float, post_rate: float) -> int:
    """Apply the BCM update to a saturated Q8.8 synaptic weight.

    Parameters
    ----------
    current_q88:
        Current synaptic weight encoded as Q8.8.
    pre_rate:
        Presynaptic firing rate in hertz.
    post_rate:
        Postsynaptic firing rate in hertz.

    Returns
    -------
    int
        Updated Q8.8 weight clamped to ``[w_min_q88, w_max_q88]``.
    """
    require_nonnegative_int(current_q88, "current_q88")
    if not self.w_min_q88 <= current_q88 <= self.w_max_q88:
        raise ValueError("current_q88 must be inside configured weight bounds")
    dw = self.compute_dw(pre_rate, post_rate)
    dw_q88 = int(dw * 256)
    new_w = current_q88 + dw_q88
    return max(self.w_min_q88, min(self.w_max_q88, new_w))

BiologicalSTDP dataclass

Spike-Timing-Dependent Plasticity adapter for bio-hybrid loops.

Bridges biological STDP time constants (∼20 ms) to SC clock rates (MHz) via a time-scaling factor. Computes ΔW from pre/post spike timing in biological time, then converts to Q8.8 weight updates for the SC domain.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

@dataclass
class BiologicalSTDP:
    """Spike-Timing-Dependent Plasticity adapter for bio-hybrid loops.

    Bridges biological STDP time constants (∼20 ms) to SC clock
    rates (MHz) via a time-scaling factor. Computes ΔW from
    pre/post spike timing in biological time, then converts to
    Q8.8 weight updates for the SC domain.
    """

    tau_plus_ms: float = 20.0  # potentiation time constant
    tau_minus_ms: float = 20.0  # depression time constant
    a_plus: float = 0.01  # potentiation amplitude
    a_minus: float = 0.012  # depression amplitude (slightly > a_plus)
    w_max_q88: int = 512  # Q8.8 = 2.0
    w_min_q88: int = 0

    def __post_init__(self) -> None:
        """Validate time constants, amplitudes, and Q8.8 bounds."""
        require_positive(self.tau_plus_ms, "tau_plus_ms")
        require_positive(self.tau_minus_ms, "tau_minus_ms")
        require_nonnegative(self.a_plus, "a_plus")
        require_nonnegative(self.a_minus, "a_minus")
        require_nonnegative_int(self.w_min_q88, "w_min_q88")
        require_nonnegative_int(self.w_max_q88, "w_max_q88")
        if self.w_max_q88 < self.w_min_q88:
            raise ValueError("w_max_q88 must be >= w_min_q88")

    def compute_dw(self, dt_ms: float) -> float:
        """Compute weight change from spike timing difference.

        dt_ms = t_post - t_pre (positive = potentiation, negative = depression)
        """
        require_finite(dt_ms, "dt_ms")
        if dt_ms > 0:
            return float(self.a_plus * np.exp(-dt_ms / self.tau_plus_ms))
        elif dt_ms < 0:
            return float(-self.a_minus * np.exp(dt_ms / self.tau_minus_ms))
        return 0.0

    def update_weight(self, current_q88: int, dt_ms: float) -> int:
        """Update Q8.8 weight from spike timing."""
        require_nonnegative_int(current_q88, "current_q88")
        if not self.w_min_q88 <= current_q88 <= self.w_max_q88:
            raise ValueError("current_q88 must be inside configured weight bounds")
        dw = self.compute_dw(dt_ms)
        dw_q88 = int(dw * 256)  # Convert to Q8.8
        new_w = current_q88 + dw_q88
        return max(self.w_min_q88, min(self.w_max_q88, new_w))

__post_init__()

Validate time constants, amplitudes, and Q8.8 bounds.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def __post_init__(self) -> None:
    """Validate time constants, amplitudes, and Q8.8 bounds."""
    require_positive(self.tau_plus_ms, "tau_plus_ms")
    require_positive(self.tau_minus_ms, "tau_minus_ms")
    require_nonnegative(self.a_plus, "a_plus")
    require_nonnegative(self.a_minus, "a_minus")
    require_nonnegative_int(self.w_min_q88, "w_min_q88")
    require_nonnegative_int(self.w_max_q88, "w_max_q88")
    if self.w_max_q88 < self.w_min_q88:
        raise ValueError("w_max_q88 must be >= w_min_q88")

compute_dw(dt_ms)

Compute weight change from spike timing difference.

dt_ms = t_post - t_pre (positive = potentiation, negative = depression)

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def compute_dw(self, dt_ms: float) -> float:
    """Compute weight change from spike timing difference.

    dt_ms = t_post - t_pre (positive = potentiation, negative = depression)
    """
    require_finite(dt_ms, "dt_ms")
    if dt_ms > 0:
        return float(self.a_plus * np.exp(-dt_ms / self.tau_plus_ms))
    elif dt_ms < 0:
        return float(-self.a_minus * np.exp(dt_ms / self.tau_minus_ms))
    return 0.0

update_weight(current_q88, dt_ms)

Update Q8.8 weight from spike timing.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def update_weight(self, current_q88: int, dt_ms: float) -> int:
    """Update Q8.8 weight from spike timing."""
    require_nonnegative_int(current_q88, "current_q88")
    if not self.w_min_q88 <= current_q88 <= self.w_max_q88:
        raise ValueError("current_q88 must be inside configured weight bounds")
    dw = self.compute_dw(dt_ms)
    dw_q88 = int(dw * 256)  # Convert to Q8.8
    new_w = current_q88 + dw_q88
    return max(self.w_min_q88, min(self.w_max_q88, new_w))

HomeostaticPlasticity dataclass

Intrinsic excitability scaling to maintain target firing rate.

Implements homeostatic plasticity: if a neuron fires too fast, reduce its excitability (threshold up); too slow, increase it. Operates on Q8.8 threshold values.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

@dataclass
class HomeostaticPlasticity:
    """Intrinsic excitability scaling to maintain target firing rate.

    Implements homeostatic plasticity: if a neuron fires too fast,
    reduce its excitability (threshold up); too slow, increase it.
    Operates on Q8.8 threshold values.
    """

    target_rate_hz: float = 10.0
    tau_homeo_ms: float = 10000.0  # slow timescale (seconds)
    max_threshold_q88: int = 512  # Q8.8 = 2.0
    min_threshold_q88: int = 64  # Q8.8 = 0.25

    def __post_init__(self) -> None:
        """Validate target dynamics and Q8.8 threshold bounds."""
        require_nonnegative(self.target_rate_hz, "target_rate_hz")
        require_positive(self.tau_homeo_ms, "tau_homeo_ms")
        require_nonnegative_int(self.min_threshold_q88, "min_threshold_q88")
        require_nonnegative_int(self.max_threshold_q88, "max_threshold_q88")
        if self.max_threshold_q88 < self.min_threshold_q88:
            raise ValueError("max_threshold_q88 must be >= min_threshold_q88")

    def update_threshold(
        self,
        current_q88: int,
        observed_rate_hz: float,
        dt_ms: float,
    ) -> int:
        """Adjust threshold to drive firing rate toward target.

        Proportional homeostatic controller on a Q8.8 fixed-point
        threshold. ``alpha = dt_ms / tau_homeo_ms`` is the integration
        weight over the time step; the rate error (``observed − target``)
        is scaled by ``alpha·256`` so that a 1 Hz error integrated over
        one full time-constant shifts the threshold by 1.0 Q8.8 unit
        (i.e. by ``256`` in integer representation). Result clamped to
        ``[min_threshold_q88, max_threshold_q88]``.
        """
        require_nonnegative_int(current_q88, "current_q88")
        if not self.min_threshold_q88 <= current_q88 <= self.max_threshold_q88:
            raise ValueError("current_q88 must be inside configured threshold bounds")
        require_nonnegative(observed_rate_hz, "observed_rate_hz")
        require_nonnegative(dt_ms, "dt_ms")
        error = observed_rate_hz - self.target_rate_hz
        alpha = dt_ms / self.tau_homeo_ms
        delta_q88 = int(alpha * error * 256.0)
        new_q88 = current_q88 + delta_q88
        return max(self.min_threshold_q88, min(self.max_threshold_q88, new_q88))

__post_init__()

Validate target dynamics and Q8.8 threshold bounds.

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def __post_init__(self) -> None:
    """Validate target dynamics and Q8.8 threshold bounds."""
    require_nonnegative(self.target_rate_hz, "target_rate_hz")
    require_positive(self.tau_homeo_ms, "tau_homeo_ms")
    require_nonnegative_int(self.min_threshold_q88, "min_threshold_q88")
    require_nonnegative_int(self.max_threshold_q88, "max_threshold_q88")
    if self.max_threshold_q88 < self.min_threshold_q88:
        raise ValueError("max_threshold_q88 must be >= min_threshold_q88")

update_threshold(current_q88, observed_rate_hz, dt_ms)

Adjust threshold to drive firing rate toward target.

Proportional homeostatic controller on a Q8.8 fixed-point threshold. alpha = dt_ms / tau_homeo_ms is the integration weight over the time step; the rate error (observed − target) is scaled by alpha·256 so that a 1 Hz error integrated over one full time-constant shifts the threshold by 1.0 Q8.8 unit (i.e. by 256 in integer representation). Result clamped to [min_threshold_q88, max_threshold_q88].

Source code in src/sc_neurocore/bioware/bioware_plasticity.py

def update_threshold(
    self,
    current_q88: int,
    observed_rate_hz: float,
    dt_ms: float,
) -> int:
    """Adjust threshold to drive firing rate toward target.

    Proportional homeostatic controller on a Q8.8 fixed-point
    threshold. ``alpha = dt_ms / tau_homeo_ms`` is the integration
    weight over the time step; the rate error (``observed − target``)
    is scaled by ``alpha·256`` so that a 1 Hz error integrated over
    one full time-constant shifts the threshold by 1.0 Q8.8 unit
    (i.e. by ``256`` in integer representation). Result clamped to
    ``[min_threshold_q88, max_threshold_q88]``.
    """
    require_nonnegative_int(current_q88, "current_q88")
    if not self.min_threshold_q88 <= current_q88 <= self.max_threshold_q88:
        raise ValueError("current_q88 must be inside configured threshold bounds")
    require_nonnegative(observed_rate_hz, "observed_rate_hz")
    require_nonnegative(dt_ms, "dt_ms")
    error = observed_rate_hz - self.target_rate_hz
    alpha = dt_ms / self.tau_homeo_ms
    delta_q88 = int(alpha * error * 256.0)
    new_q88 = current_q88 + delta_q88
    return max(self.min_threshold_q88, min(self.max_threshold_q88, new_q88))

BioHybridSession dataclass

Manages a complete bio-hybrid experiment session.

Orchestrates MEA recording, spike detection, AER transcoding, stochastic conversion, optogenetic feedback, and culture-health assessment. The stdp and homeostatic policies are retained for caller-managed updates; process_frame does not mutate plasticity state implicitly.

Source code in src/sc_neurocore/bioware/bioware_session.py

@dataclass
class BioHybridSession:
    """Manages a complete bio-hybrid experiment session.

    Orchestrates MEA recording, spike detection, AER transcoding, stochastic
    conversion, optogenetic feedback, and culture-health assessment. The
    ``stdp`` and ``homeostatic`` policies are retained for caller-managed
    updates; ``process_frame`` does not mutate plasticity state implicitly.
    """

    mea_config: MEAConfig
    detector: SpikeDetector
    transcoder: MEAToAERTranscoder
    sc_converter: AERToSCConverter
    opto_encoder: SCToOptoEncoder
    stdp: BiologicalSTDP = field(default_factory=BiologicalSTDP)
    health_monitor: CultureHealth = field(default_factory=CultureHealth)
    artifact_rejector: Optional["ArtifactRejector"] = None
    pharm_model: Optional["PharmModel"] = None
    latency_budget: Optional["LatencyBudget"] = None
    homeostatic: Optional["HomeostaticPlasticity"] = None
    sorter: Optional["SpikeSorter"] = None
    zenith_core: Optional["ArcaneZenithCognitiveCore"] = None
    round_count: int = 0

    def __post_init__(self) -> None:
        """Validate component compatibility before processing live data."""
        if not isinstance(self.mea_config, MEAConfig):
            raise TypeError("mea_config must be an MEAConfig")
        if not isinstance(self.detector, SpikeDetector):
            raise TypeError("detector must be a SpikeDetector")
        if self.detector.config != self.mea_config:
            raise ValueError("detector.config must match mea_config")
        if not isinstance(self.transcoder, MEAToAERTranscoder):
            raise TypeError("transcoder must be a MEAToAERTranscoder")
        if not isinstance(self.sc_converter, AERToSCConverter):
            raise TypeError("sc_converter must be an AERToSCConverter")
        if not isinstance(self.opto_encoder, SCToOptoEncoder):
            raise TypeError("opto_encoder must be an SCToOptoEncoder")
        if not isinstance(self.stdp, BiologicalSTDP):
            raise TypeError("stdp must be a BiologicalSTDP")
        if not isinstance(self.health_monitor, CultureHealth):
            raise TypeError("health_monitor must be a CultureHealth")
        if self.artifact_rejector is not None and not isinstance(
            self.artifact_rejector, ArtifactRejector
        ):
            raise TypeError("artifact_rejector must be an ArtifactRejector or None")
        if self.pharm_model is not None and not isinstance(self.pharm_model, PharmModel):
            raise TypeError("pharm_model must be a PharmModel or None")
        if self.latency_budget is not None and not isinstance(self.latency_budget, LatencyBudget):
            raise TypeError("latency_budget must be a LatencyBudget or None")
        if self.homeostatic is not None and not isinstance(self.homeostatic, HomeostaticPlasticity):
            raise TypeError("homeostatic must be a HomeostaticPlasticity or None")
        if self.sorter is not None and not isinstance(self.sorter, SpikeSorter):
            raise TypeError("sorter must be a SpikeSorter or None")
        if self.zenith_core is not None and not callable(
            getattr(self.zenith_core, "step_from_bio_rates", None)
        ):
            raise TypeError("zenith_core must provide step_from_bio_rates or be None")
        if self.sc_converter.num_neurons < self.mea_config.num_channels:
            raise ValueError("sc_converter.num_neurons must cover every MEA channel")
        if self.transcoder.channel_map is not None:
            for neuron_id in self.transcoder.channel_map.values():
                if neuron_id >= self.sc_converter.num_neurons:
                    raise ValueError("channel_map targets must fit sc_converter.num_neurons")
        require_nonnegative_int(self.round_count, "round_count")

    def process_frame(
        self,
        voltage_data: np.ndarray[Any, Any],
        t_start_s: float = 0.0,
        stim_times_s: Optional[List[float]] = None,
    ) -> BioHybridFrameResult:
        """Process one frame whose timestamps fit one AER counter epoch.

        Detector timestamps are frame-relative. ``t_start_s`` is the
        non-negative experiment time used by optional experiment models; it
        does not change the frame-local AER timestamp origin.
        """
        validate_voltage_matrix(
            voltage_data,
            expected_channels=self.mea_config.num_channels,
        )
        require_nonnegative(t_start_s, "t_start_s")
        max_frame_tick = int(
            ((voltage_data.shape[0] - 1) / self.mea_config.sample_rate_hz)
            * self.transcoder.hw_clock_hz
        )
        if max_frame_tick > 0xFFFF:
            raise ValueError("voltage frame exceeds the 16-bit AER timestamp epoch")
        if max_frame_tick >= self.sc_converter.window_ticks:
            raise ValueError("voltage frame exceeds sc_converter.window_ticks")

        t0 = time.perf_counter_ns()
        next_round = self.round_count + 1

        if self.artifact_rejector is not None and stim_times_s is not None:
            voltage_data = self.artifact_rejector.blank(
                voltage_data, stim_times_s, self.mea_config.sample_rate_hz
            )

        # 1. Detect spikes
        spikes = self.detector.detect(voltage_data)

        # 1.5 Core primitive wiring
        if self.sorter is not None:
            spikes = self.sorter.assign(spikes)

        if self.pharm_model is not None:
            spikes = self.pharm_model.modulate_spike_events(spikes, t_start_s)

        # 2. Transcode to AER
        aer_events = self.transcoder.transcode(spikes)

        # 3. Convert to SC bitstreams
        bitstreams = self.sc_converter.convert(aer_events)

        # 3.5 Zenith integration!
        if self.zenith_core is not None:
            rates = decode_bitstream_rate(bitstreams)
            self.zenith_core.step_from_bio_rates(rates)

        # 4. Generate optogenetic pulses
        opto_pulses = self.opto_encoder.encode(bitstreams)

        # 5. Health assessment
        n_channels = voltage_data.shape[1]
        spike_counts = np.zeros(n_channels)
        for s in spikes:
            if s.channel < n_channels:
                spike_counts[s.channel] += 1
        duration = voltage_data.shape[0] / self.mea_config.sample_rate_hz
        health = self.health_monitor.assess(spike_counts, duration_s=duration)

        latency_us = (time.perf_counter_ns() - t0) / 1000.0

        result = BioHybridFrameResult(
            round=next_round,
            num_spikes=len(spikes),
            num_aer_events=len(aer_events),
            num_bitstreams=len(bitstreams),
            num_opto_pulses=len(opto_pulses),
            latency_us=latency_us,
            health=health,
            spikes=spikes,
            aer_events=aer_events,
            bitstreams=bitstreams,
            opto_pulses=opto_pulses,
        )
        if self.latency_budget is not None:
            self.latency_budget.record(latency_us)
        self.round_count = next_round
        return result

__post_init__()

Validate component compatibility before processing live data.

Source code in src/sc_neurocore/bioware/bioware_session.py

def __post_init__(self) -> None:
    """Validate component compatibility before processing live data."""
    if not isinstance(self.mea_config, MEAConfig):
        raise TypeError("mea_config must be an MEAConfig")
    if not isinstance(self.detector, SpikeDetector):
        raise TypeError("detector must be a SpikeDetector")
    if self.detector.config != self.mea_config:
        raise ValueError("detector.config must match mea_config")
    if not isinstance(self.transcoder, MEAToAERTranscoder):
        raise TypeError("transcoder must be a MEAToAERTranscoder")
    if not isinstance(self.sc_converter, AERToSCConverter):
        raise TypeError("sc_converter must be an AERToSCConverter")
    if not isinstance(self.opto_encoder, SCToOptoEncoder):
        raise TypeError("opto_encoder must be an SCToOptoEncoder")
    if not isinstance(self.stdp, BiologicalSTDP):
        raise TypeError("stdp must be a BiologicalSTDP")
    if not isinstance(self.health_monitor, CultureHealth):
        raise TypeError("health_monitor must be a CultureHealth")
    if self.artifact_rejector is not None and not isinstance(
        self.artifact_rejector, ArtifactRejector
    ):
        raise TypeError("artifact_rejector must be an ArtifactRejector or None")
    if self.pharm_model is not None and not isinstance(self.pharm_model, PharmModel):
        raise TypeError("pharm_model must be a PharmModel or None")
    if self.latency_budget is not None and not isinstance(self.latency_budget, LatencyBudget):
        raise TypeError("latency_budget must be a LatencyBudget or None")
    if self.homeostatic is not None and not isinstance(self.homeostatic, HomeostaticPlasticity):
        raise TypeError("homeostatic must be a HomeostaticPlasticity or None")
    if self.sorter is not None and not isinstance(self.sorter, SpikeSorter):
        raise TypeError("sorter must be a SpikeSorter or None")
    if self.zenith_core is not None and not callable(
        getattr(self.zenith_core, "step_from_bio_rates", None)
    ):
        raise TypeError("zenith_core must provide step_from_bio_rates or be None")
    if self.sc_converter.num_neurons < self.mea_config.num_channels:
        raise ValueError("sc_converter.num_neurons must cover every MEA channel")
    if self.transcoder.channel_map is not None:
        for neuron_id in self.transcoder.channel_map.values():
            if neuron_id >= self.sc_converter.num_neurons:
                raise ValueError("channel_map targets must fit sc_converter.num_neurons")
    require_nonnegative_int(self.round_count, "round_count")

process_frame(voltage_data, t_start_s=0.0, stim_times_s=None)

Process one frame whose timestamps fit one AER counter epoch.

Detector timestamps are frame-relative. t_start_s is the non-negative experiment time used by optional experiment models; it does not change the frame-local AER timestamp origin.

Source code in src/sc_neurocore/bioware/bioware_session.py

def process_frame(
    self,
    voltage_data: np.ndarray[Any, Any],
    t_start_s: float = 0.0,
    stim_times_s: Optional[List[float]] = None,
) -> BioHybridFrameResult:
    """Process one frame whose timestamps fit one AER counter epoch.

    Detector timestamps are frame-relative. ``t_start_s`` is the
    non-negative experiment time used by optional experiment models; it
    does not change the frame-local AER timestamp origin.
    """
    validate_voltage_matrix(
        voltage_data,
        expected_channels=self.mea_config.num_channels,
    )
    require_nonnegative(t_start_s, "t_start_s")
    max_frame_tick = int(
        ((voltage_data.shape[0] - 1) / self.mea_config.sample_rate_hz)
        * self.transcoder.hw_clock_hz
    )
    if max_frame_tick > 0xFFFF:
        raise ValueError("voltage frame exceeds the 16-bit AER timestamp epoch")
    if max_frame_tick >= self.sc_converter.window_ticks:
        raise ValueError("voltage frame exceeds sc_converter.window_ticks")

    t0 = time.perf_counter_ns()
    next_round = self.round_count + 1

    if self.artifact_rejector is not None and stim_times_s is not None:
        voltage_data = self.artifact_rejector.blank(
            voltage_data, stim_times_s, self.mea_config.sample_rate_hz
        )

    # 1. Detect spikes
    spikes = self.detector.detect(voltage_data)

    # 1.5 Core primitive wiring
    if self.sorter is not None:
        spikes = self.sorter.assign(spikes)

    if self.pharm_model is not None:
        spikes = self.pharm_model.modulate_spike_events(spikes, t_start_s)

    # 2. Transcode to AER
    aer_events = self.transcoder.transcode(spikes)

    # 3. Convert to SC bitstreams
    bitstreams = self.sc_converter.convert(aer_events)

    # 3.5 Zenith integration!
    if self.zenith_core is not None:
        rates = decode_bitstream_rate(bitstreams)
        self.zenith_core.step_from_bio_rates(rates)

    # 4. Generate optogenetic pulses
    opto_pulses = self.opto_encoder.encode(bitstreams)

    # 5. Health assessment
    n_channels = voltage_data.shape[1]
    spike_counts = np.zeros(n_channels)
    for s in spikes:
        if s.channel < n_channels:
            spike_counts[s.channel] += 1
    duration = voltage_data.shape[0] / self.mea_config.sample_rate_hz
    health = self.health_monitor.assess(spike_counts, duration_s=duration)

    latency_us = (time.perf_counter_ns() - t0) / 1000.0

    result = BioHybridFrameResult(
        round=next_round,
        num_spikes=len(spikes),
        num_aer_events=len(aer_events),
        num_bitstreams=len(bitstreams),
        num_opto_pulses=len(opto_pulses),
        latency_us=latency_us,
        health=health,
        spikes=spikes,
        aer_events=aer_events,
        bitstreams=bitstreams,
        opto_pulses=opto_pulses,
    )
    if self.latency_budget is not None:
        self.latency_budget.record(latency_us)
    self.round_count = next_round
    return result

detect_network_bursts(spikes, bin_width_s=0.01, threshold_sigma=3.0, min_channels=3)

Detect network-wide synchronised bursts.

Bins spikes in time, detects bins with activity > threshold_sigma above the mean, and requires participation from ≥ min_channels.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def detect_network_bursts(
    spikes: List[DetectedSpike],
    bin_width_s: float = 0.01,
    threshold_sigma: float = 3.0,
    min_channels: int = 3,
) -> List[NetworkBurst]:
    """Detect network-wide synchronised bursts.

    Bins spikes in time, detects bins with activity > threshold_sigma
    above the mean, and requires participation from ≥ min_channels.
    """
    require_positive(bin_width_s, "bin_width_s")
    require_nonnegative(threshold_sigma, "threshold_sigma")
    require_positive_int(min_channels, "min_channels")
    if not spikes:
        return []

    timestamps = np.array([s.timestamp_s for s in spikes])
    t_start, t_end = timestamps.min(), timestamps.max()
    if t_end <= t_start:
        return []

    n_bins = max(1, int((t_end - t_start) / bin_width_s) + 1)
    bin_counts = np.zeros(n_bins)
    bin_channels: List[set[int]] = [set() for _ in range(n_bins)]

    for s in spikes:
        idx = min(int((s.timestamp_s - t_start) / bin_width_s), n_bins - 1)
        bin_counts[idx] += 1
        bin_channels[idx].add(s.channel)

    mean_count = np.mean(bin_counts)
    std_count = np.std(bin_counts)
    if std_count == 0:
        return []
    threshold = mean_count + threshold_sigma * std_count

    bursts = []
    for i in range(n_bins):
        if bin_counts[i] >= threshold and len(bin_channels[i]) >= min_channels:
            bursts.append(
                NetworkBurst(
                    onset_s=t_start + i * bin_width_s,
                    duration_s=bin_width_s,
                    participating_channels=len(bin_channels[i]),
                    total_spikes=int(bin_counts[i]),
                )
            )
    return bursts

extract_lfp_power(voltage_data, sample_rate_hz, bands=None)

Extract per-channel power in each LFP band.

Uses FFT-based power spectral density estimation. Returns dict of band_name → per-channel power array.

Source code in src/sc_neurocore/bioware/bioware_analysis.py

def extract_lfp_power(
    voltage_data: np.ndarray[Any, Any],
    sample_rate_hz: float,
    bands: Optional[List[LFPBand]] = None,
) -> Dict[str, np.ndarray[Any, Any]]:
    """Extract per-channel power in each LFP band.

    Uses FFT-based power spectral density estimation.
    Returns dict of band_name → per-channel power array.
    """
    if bands is None:
        bands = DEFAULT_LFP_BANDS

    validate_voltage_matrix(voltage_data)
    require_positive(sample_rate_hz, "sample_rate_hz")
    if not bands:
        raise ValueError("bands must not be empty")
    names: set[str] = set()
    for band in bands:
        if not isinstance(band, LFPBand):
            raise TypeError("bands must contain LFPBand instances")
        if band.name in names:
            raise ValueError(f"duplicate LFP band name: {band.name}")
        names.add(band.name)

    n_samples, n_channels = voltage_data.shape
    freqs = np.fft.rfftfreq(n_samples, d=1.0 / sample_rate_hz)
    fft_mag = np.abs(np.fft.rfft(voltage_data, axis=0)) ** 2

    result = {}
    for band in bands:
        mask = (freqs >= band.low_hz) & (freqs < band.high_hz)
        power = np.sum(fft_mag[mask, :], axis=0) if mask.any() else np.zeros(n_channels)
        result[band.name] = power
    return result

decode_bitstream_rate(bitstreams, sc_clock_hz=1000000.0)

Decode SC bitstreams back to biological firing rates (Hz).

Interprets popcount/length as probability, scales by SC clock to get equivalent biological firing rate.

Source code in src/sc_neurocore/bioware/bioware_encoding.py

def decode_bitstream_rate(
    bitstreams: Dict[int, np.ndarray[Any, Any]],
    sc_clock_hz: float = 1e6,
) -> Dict[int, float]:
    """Decode SC bitstreams back to biological firing rates (Hz).

    Interprets popcount/length as probability, scales by SC clock
    to get equivalent biological firing rate.
    """
    require_positive(sc_clock_hz, "sc_clock_hz")
    rates = {}
    for nid, bs in bitstreams.items():
        require_nonnegative_int(nid, "bitstream neuron_id")
        validate_binary_bitstream(bs, name=f"bitstreams[{nid}]", allow_empty=True)
        if len(bs) == 0:
            rates[nid] = 0.0
            continue
        prob = float(np.sum(bs)) / len(bs)
        rates[nid] = prob * sc_clock_hz
    return rates

mea_fitness_hook(detected_spikes, target_rate=10.0, *, duration_s=None, stimulus_time_s=None, measured_latency_ms=None)

Organism fitness metrics derived from MEA response dynamics.

Designed to plug into the evo_substrate ReplicationEngine(metrics_fn=mea_fitness_hook) — returns the {"accuracy", "energy_mw", "latency_ms"} triple the engine scores.

Accuracy is a bounded distance to the target mean per-channel firing rate when duration_s is supplied, or to the legacy per-channel spike count when it is omitted. The legacy energy_mw key is a dimensionless optimisation proxy equal to 0.5 * spike_count; it is not a measured power or energy quantity. latency_ms is either a caller-supplied closed-loop measurement, the first response latency after stimulus_time_s, or the first spike timestamp relative to frame start.

Source code in src/sc_neurocore/bioware/bioware_fitness.py

def mea_fitness_hook(
    detected_spikes: List[DetectedSpike],
    target_rate: float = 10.0,
    *,
    duration_s: Optional[float] = None,
    stimulus_time_s: Optional[float] = None,
    measured_latency_ms: Optional[float] = None,
) -> Dict[str, float]:
    """Organism fitness metrics derived from MEA response dynamics.

    Designed to plug into the evo_substrate
    ``ReplicationEngine(metrics_fn=mea_fitness_hook)`` — returns the
    ``{"accuracy", "energy_mw", "latency_ms"}`` triple the engine scores.

    Accuracy is a bounded distance to the target mean per-channel firing
    rate when ``duration_s`` is supplied, or to the legacy per-channel
    spike count when it is omitted. The legacy ``energy_mw`` key is a
    dimensionless optimisation proxy equal to ``0.5 * spike_count``; it is
    not a measured power or energy quantity. ``latency_ms`` is either a
    caller-supplied closed-loop measurement, the first response latency after
    ``stimulus_time_s``, or the first spike timestamp relative to frame start.
    """
    require_nonnegative(target_rate, "target_rate")
    if duration_s is not None and (not math.isfinite(duration_s) or duration_s <= 0.0):
        raise ValueError("duration_s must be finite and > 0 when provided")
    if stimulus_time_s is not None and not math.isfinite(stimulus_time_s):
        raise ValueError("stimulus_time_s must be finite when provided")
    if measured_latency_ms is not None:
        if not math.isfinite(measured_latency_ms) or measured_latency_ms < 0.0:
            raise ValueError("measured_latency_ms must be finite and >= 0 when provided")

    if not detected_spikes:
        latency_ms = _mea_response_latency_ms(
            detected_spikes,
            stimulus_time_s=stimulus_time_s,
            measured_latency_ms=measured_latency_ms,
        )
        return {"accuracy": 0.1, "energy_mw": 0.0, "latency_ms": latency_ms}

    counts: Dict[int, float] = {}
    for s in detected_spikes:
        counts[s.channel] = counts.get(s.channel, 0.0) + 1.0

    per_channel_activity = np.array(list(counts.values()), dtype=float)
    if duration_s is not None:
        per_channel_activity = per_channel_activity / duration_s
    mean_rate = float(np.mean(per_channel_activity)) if per_channel_activity.size else 0.0

    # Normalised distance to target rate → accuracy ∈ [0.1, 0.99].
    if target_rate > 0.0:
        accuracy = 1.0 - min(1.0, abs(mean_rate - target_rate) / target_rate)
    else:
        accuracy = 0.1

    latency_ms = _mea_response_latency_ms(
        detected_spikes,
        stimulus_time_s=stimulus_time_s,
        measured_latency_ms=measured_latency_ms,
    )
    return {
        "accuracy": float(np.clip(accuracy, 0.1, 0.99)),
        "energy_mw": float(len(detected_spikes) * 0.5),
        "latency_ms": latency_ms,
    }