Skip to content

Bioware Interface

Biological-hardware interface for cerebral organoids and multi-electrode array (MEA) systems. Bridges wet-lab experiments with in-silico SC simulations — spike detection, AER transcoding, stochastic-computing feedback, optogenetic encoding, Q8.8 homeostatic plasticity, PCA spike sorting, and a closed-loop session layer that orchestrates the full pipeline per frame.

Python
from sc_neurocore.bioware.bioware import (
    BioHybridSession, BioHybridFrameResult,
    MEAConfig, MEALayout, SpikeDetector,
    MEAToAERTranscoder, AERToSCConverter, SCToOptoEncoder,
    CultureHealth, HomeostaticPlasticity, SpikeSorter,
    mea_fitness_hook,
)

1. Mathematical formalism

1.1 Spike detection (adaptive threshold, MAD-based noise)

Raw MEA voltage $V_c(t)$ per channel $c$ is detected against an adaptive threshold derived from the median absolute deviation:

$$ \sigma_c = \frac{\mathrm{median}_t |V_c(t)|}{0.6745}, \qquad \theta_c = \alpha\,\sigma_c. $$

The constant $0.6745$ converts MAD into an estimator of the Gaussian standard deviation (Donoho & Johnstone 1994); $\alpha = 5$ by default (spike_threshold_sigma). A spike is emitted when $|V_c(t)| > \theta_c$ and no spike was emitted on the same channel in the preceding refractory_samples = 30 samples.

1.2 MEA → AER transcoding

For a spike with real-time timestamp $t_s$ (seconds) and a hardware AER clock $f_\text{hw}$ (Hz), the AER event carries integer tick

$$ \tau = \lfloor t_s \cdot f_\text{hw} \rfloor $$

and neuron id derived from channel_map (or the channel number itself if no map is given). AER events are emitted sorted by $\tau$.

1.3 AER → SC bitstream conversion

For each neuron id, the converter bins AER timestamps into a bitstream of length $L$ (default 1 024). For neuron $n$ observed over the window $[t_0, t_0 + T]$ with $k_n$ spikes:

$$ p_n = \frac{k_n}{k_n + K}, \qquad b_{n,i} \sim \mathrm{Bernoulli}(p_n), \qquad i \in [0, L). $$

where $K$ is a smoothing constant (smoothing_constant, default 1). The resulting SC bitstream has mean activity equal to $p_n$.

1.4 Optogenetic encoding (SC → opto pulses)

Given an output bitstream $b \in {0, 1}^L$ with density $d = \frac{1}{L}\sum_i b_i$, the optogenetic pulse uses intensity

$$ I_\text{mW/mm²} = d \cdot I_\text{max}, $$

and duration

$$ T_\text{ms} = T_\text{min} + (T_\text{max} - T_\text{min})\,d, $$

so bright / long pulses signal high-activity layers and dark / short pulses signal low-activity. The wavelength is fixed by the optogenetic channel (wavelength_nm, e.g. 470 nm for ChR2).

1.5 Biological STDP (Bi & Poo 1998, exponential pair rule)

For a post-spike timed at $t_\text{post}$ after a pre-spike at $t_\text{pre}$, let $\Delta t = t_\text{post} - t_\text{pre}$. The weight change is:

$$ \Delta w(\Delta t) = \begin{cases} A_+\, \exp(-\Delta t / \tau_+) & \Delta t > 0 \text{ (LTP)} \ -A_-\, \exp(+\Delta t / \tau_-) & \Delta t < 0 \text{ (LTD)} \ 0 & \Delta t = 0 \end{cases} $$

Default $\tau_+ = \tau_- = 20$ ms, $A_+ = 0.005$, $A_- = 0.00525$, matching Bi & Poo 1998 hippocampal culture fits (1:1.05 potentiation:depression asymmetry).

1.6 BCM metaplasticity (sliding-threshold variant)

The BCM rule (Bienenstock, Cooper, Munro 1982) adjusts the LTP/LTD boundary as a function of post-synaptic activity $\bar y$:

$$ \Delta w = \eta\,x\,y\,(y - \theta_m), \qquad \theta_m = \kappa\, \bar y^2, $$

where $\bar y$ is the temporally-averaged post-synaptic rate and $\kappa > 0$ tunes the sliding-threshold curvature. Low-frequency activity below $\theta_m$ depresses; above, potentiates.

1.7 Q8.8 homeostatic controller

HomeostaticPlasticity.update_threshold implements a proportional negative-feedback controller in fixed-point Q8.8 arithmetic:

$$ \alpha_t = \frac{\Delta t_\text{ms}}{\tau_\text{homeo}}, \qquad \Delta \theta = \lfloor \alpha_t (r_t - r^\star) \cdot 256 \rfloor, $$

$$ \theta^{(q88)}{t+1} = \mathrm{clip}!\left(\theta^{(q88)}_t + \Delta \theta, \theta^{(q88)}\text{min}, \theta^{(q88)}_\text{max}\right). $$

The factor $256$ is the Q8.8 scale: one full threshold unit in floating-point corresponds to $256$ steps in the integer representation. A 1 Hz rate error sustained over one full time constant $\tau_\text{homeo}$ thus shifts the threshold by exactly one unit.

1.8 PCA + KMeans spike sorting

Given a set of detected waveforms ${w^{(i)}}$, each a vector in $\mathbb{R}^D$ (with $D$ = int(2 · snippet_ms · sample_rate_hz / 1000)), sorting proceeds in two stages:

  1. Dimensionality reduction via principal component analysis:

$$ X = [w^{(1)}, \ldots, w^{(M)}]^\top \in \mathbb{R}^{M\times D}, \qquad X = U \Sigma V^\top, \qquad Z = X \cdot V_{:k}, $$

with $k = \min(\text{n_components}, M, D)$ (default n_components = 3).

  1. Cluster assignment via $k$-means with $k = \text{num_units}$ and $n_\text{init} = 10$ restarts:

$$ \min_{{\mu_c}} \sum_{i=1}^M \min_{c \in {1..k}} |z^{(i)} - \mu_c|^2. $$

Waveforms flagged waveform is None (amplitude-only spikes) are skipped; the sorter is a no-op when fewer than num_units waveforms are available.

1.9 Evo-substrate fitness hook

The MEA → ReplicationEngine bridge computes three fitness scalars:

$$ \bar r = \frac{1}{|C|}\sum_{c \in C} \frac{k_c}{T_\mathrm{frame}}, \qquad \mathrm{acc} = \mathrm{clip}!\left(1 - \frac{|\bar r - r^\star|}{r^\star},\; 0.1,\; 0.99\right), $$

where $k_c$ is the per-channel spike count, $T_\mathrm{frame}$ is duration_s when supplied, and $r^\star$ is the target rate (target_rate = 10 Hz by default). When duration_s is omitted, the hook preserves the legacy count-domain score for callers that do not know the frame length. Energy and latency:

$$ E = 0.5\,\text{mW} \cdot N_\text{spikes}, \qquad T_\mathrm{lat} = \begin{cases} T_\mathrm{measured}, & \text{if measured latency is supplied},\ t_\mathrm{first\ response} - t_\mathrm{stim}, & \text{if stimulus time is supplied},\ t_\mathrm{first\ spike}, & \text{otherwise}. \end{cases} $$


2. Theoretical context

Bioware sits at the intersection of three active research frontiers.

Closed-loop biohybrid experiments. Organoid cultures coupled to MEAs via optogenetic actuators form a real-time feedback loop between a digital controller and a living network. Kagan et al. (2022) demonstrated such a system learning Pong; DishBrain and similar platforms have generalised the paradigm to control tasks. SC-NeuroCore's BioHybridSession is the orchestration layer — it owns the MEA config, detector, transcoder, SC converter, opto encoder, and plasticity updates, and guarantees the end-to-end latency budget in a single process_frame call.

Stochastic computing over biological signals. Traditional MEA pipelines output rates or spike trains; SC-NeuroCore converts rates directly into Bernoulli bitstreams consumed by the SC arithmetic stack. The conversion is lossless up to the SC bit budget — $L = 1024$ bits encodes $p_n$ with variance $p(1-p)/L \le 1/(4L) \approx 2.4 \times 10^{-4}$.

Hardware-grounded homeostatic plasticity. The Q8.8 controller in §1.7 is directly synthesisable — the state variable is an integer, the update is additive with a constant shift, the clamp is a pair of comparators. The same mathematical form runs on the PYNQ-Z2 silicon (sc_aer_encoder.v and friends) as in Python, so simulations quantitatively match hardware traces to Q8.8 precision. Turrigiano 2012 provides the biological reference point: real neurons use much slower homeostatic timescales (seconds to hours), modelled here by the tau_homeo_ms parameter.

Spike sorting. PCA + KMeans is the classical first-pass (Lewicki 1998); modern production systems use SpikeInterface, Kilosort or MountainSort. The SC-NeuroCore sorter is intentionally minimal — it exists so a single closed-loop run can demonstrate end-to-end separation of 2–4 units without external dependencies beyond scikit-learn, and so the evolutionary-substrate fitness hook has something to condition on when the MEA returns waveforms.

The overall bioware contract is stricter than typical research code: every public surface is typed, exception-free for empty inputs, deterministic under a fixed seed, and round-trippable between dataclass and dict views. The contract is enforced by 40+ tests in tests/test_bioware/test_bioware.py.


3. Pipeline position

Bioware is the outer of SC-NeuroCore's two closed loops. The inner loop is the SC arithmetic stack; the outer loop wraps it in a real-time bidirectional bridge with the biological substrate.

Text Only
   ┌──────────── BioHybridSession.process_frame() ─────────────┐
   │                                                            │
   │  voltage_data (N_samples × N_channels)                    │
   │        │                                                   │
   │        ▼                                                   │
   │   SpikeDetector ──────►  list[DetectedSpike]              │
   │        │                        │                          │
   │        │                        ├──► (opt) SpikeSorter     │
   │        │                        ├──► (opt) ArtifactReject  │
   │        │                        ▼                          │
   │        │                  MEAToAERTranscoder               │
   │        │                        │ list[AEREvent]           │
   │        │                        ▼                          │
   │        │                  AERToSCConverter                 │
   │        │                        │ dict[id, np.ndarray]     │
   │        │                        ▼                          │
   │        │                  SCToOptoEncoder ──► OptoPulse    │
   │        │                                                   │
   │        └──► CultureHealth ──► health score                 │
   │        └──► HomeostaticPlasticity (Q8.8 threshold update)  │
   │        └──► ArcaneZenithCognitiveCore.step_from_bio_rates  │
   │                                                            │
   └────────► BioHybridFrameResult ──────────────────────────►  │
                                                            caller

Upstream inputs — the MEA voltage matrix (from any MEA recording system — PYNQ-Z2 ADC, Intan, BlackRock, synthetic). Format is np.ndarray shape (n_samples, n_channels).

Downstream outputs — a :class:BioHybridFrameResult packet containing spike counts, AER events, SC bitstreams, optogenetic pulses, and a CultureHealth snapshot. Both dataclass-style (result.round) and mapping-style (result["round"]) access supported.

The outer loop is driven by the caller — no threads, no background tasks inside process_frame. Hardware latency / jitter is a caller concern.


4. Features

Feature Detail
MEA config presets 60, 120, 256, 4 096 channel layouts with canonical electrode pitches
MAD-based spike detection Robust σ estimator; per-channel adaptive threshold
Configurable refractory Default 30 samples; prevents double-counting
Waveform snippet extraction Per-spike 2 ms window, edge-padded
AER transcoding Integer hardware timestamps; channel→neuron map; sorted output
AER → SC bitstream Bernoulli density encoding with smoothing constant
SC → optogenetic encoding Bright+long for high density; bounded wavelength / intensity / duration
Biological STDP Exponential pair rule, Bi & Poo parameters
BCM metaplasticity Sliding-threshold variant
Q8.8 homeostatic plasticity Proportional controller with 1 Hz · τ_homeo = 1 unit calibration
Culture health Rate-based aggregate score over the frame
PCA + KMeans spike sorting Optional; scikit-learn optional dep
Artifact rejection Optional stim-window blanking
Pharmacology model Optional onset-delay + gain model for neurotransmitter pulses
Latency budget Per-frame wall-clock recording
ArcaneZenith bridge zenith_core.step_from_bio_rates(...) inside process_frame
Evo-substrate fitness hook mea_fitness_hook(spikes, target_rate) → {acc, energy, latency}
BioHybridFrameResult dual interface Dataclass attribute access + mapping result["key"] + in
Audit log :class:BioAuditLog timestamp + entry record for every session
Multi-well plate support :class:MultiWellPlate for parallel experiments
40+ tests Spike detection + AER + SC + Opto + STDP + BCM + Culture + Homeostasis

5. Usage example with output

Python
import numpy as np
from sc_neurocore.bioware.bioware import (
    BioHybridSession, MEAConfig,
    SpikeDetector, MEAToAERTranscoder, AERToSCConverter,
    SCToOptoEncoder, BiologicalSTDP, CultureHealth,
    HomeostaticPlasticity, SpikeSorter,
)

cfg = MEAConfig(num_channels=10, sample_rate_hz=20_000.0,
                spike_threshold_sigma=5.0)

session = BioHybridSession(
    mea_config=cfg,
    detector=SpikeDetector(config=cfg, refractory_samples=30),
    transcoder=MEAToAERTranscoder(hw_clock_hz=1e6),
    sc_converter=AERToSCConverter(bitstream_length=1024),
    opto_encoder=SCToOptoEncoder(wavelength_nm=470,
                                  max_intensity_mw_mm2=5.0),
    stdp=BiologicalSTDP(),
    health_monitor=CultureHealth(),
    homeostatic=HomeostaticPlasticity(target_rate_hz=10.0),
    sorter=SpikeSorter(num_units=3),
)

rng = np.random.default_rng(42)
V = rng.normal(0, 5, size=(2000, 10))
V[::200, 0] = -80.0                    # spikes on channel 0
V[100::200, 3] = -60.0                 # spikes on channel 3

result = session.process_frame(V)

print(f"round       : {result.round}")
print(f"num_spikes  : {result.num_spikes}")
print(f"num_aer     : {result.num_aer_events}")
print(f"num_streams : {result.num_bitstreams}")
print(f"num_opto    : {result.num_opto_pulses}")
print(f"latency_us  : {result.latency_us:.1f}")

# Dataclass AND mapping access both work:
assert result["round"] == result.round
assert "latency_us" in result

Typical output:

Text Only
round       : 1
num_spikes  : 16
num_aer     : 16
num_streams : 2
num_opto    : 2
latency_us  : 3178.4

The examples/14_bioware_closed_loop_demo.py script extends this to a 100-frame experiment with full SpikeSorter fit + ArcaneZenith cognitive core + MEA hardware simulation; measured end-to-end wall-clock in §7.


6. Technical reference

6.1 MEAConfig + MEALayout

Python
class MEALayout(Enum):
    MEA_60   = "60ch"
    MEA_120  = "120ch"
    MEA_256  = "256ch"
    MEA_4096 = "4096ch"
    CUSTOM   = "custom"

@dataclass
class MEAConfig:
    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

    @classmethod
    def from_layout(cls, layout: MEALayout) -> MEAConfig: ...

6.2 SpikeDetector + DetectedSpike

Python
@dataclass
class DetectedSpike:
    channel: int
    timestamp_s: float
    amplitude_uv: float
    unit_id: int = 0
    waveform: Optional[np.ndarray] = None

@dataclass
class SpikeDetector:
    config: MEAConfig
    refractory_samples: int = 30

    def estimate_noise(self, voltage_data) -> np.ndarray
    def detect(self, voltage_data, snippet_ms: float = 2.0) -> list[DetectedSpike]

6.3 MEAToAERTranscoder + AERToSCConverter

Python
@dataclass
class AEREvent:
    neuron_id: int
    timestamp: int           # clock ticks
    valid: bool = True
    weight: int = 256        # Q8.8 = 1.0

class MEAToAERTranscoder:
    hw_clock_hz: float = 1e6
    channel_map: Optional[dict[int, int]] = None

    def transcode(self, spikes, t_start_s: float = 0.0) -> list[AEREvent]

class AERToSCConverter:
    bitstream_length: int = 1024
    smoothing_constant: float = 1.0

    def convert(self, events) -> dict[int, np.ndarray]

6.4 SCToOptoEncoder + OptogeneticPulse

Python
@dataclass
class OptogeneticPulse:
    wavelength_nm: int
    intensity_mw_mm2: float
    duration_ms: float
    channel_id: int

@dataclass
class SCToOptoEncoder:
    wavelength_nm: int = 470
    max_intensity_mw_mm2: float = 5.0
    min_pulse_ms: float = 1.0
    max_pulse_ms: float = 50.0

    def encode(self, bitstreams) -> list[OptogeneticPulse]

6.5 Plasticity classes

Python
@dataclass
class BiologicalSTDP:
    A_plus: float = 0.005
    A_minus: float = 0.00525
    tau_plus_ms: float = 20.0
    tau_minus_ms: float = 20.0

    def compute_dw(self, dt_ms: float) -> float

@dataclass
class HomeostaticPlasticity:
    target_rate_hz: float = 10.0
    tau_homeo_ms: float = 10000.0
    max_threshold_q88: int = 512    # Q8.8 = 2.0
    min_threshold_q88: int = 64     # Q8.8 = 0.25

    def update_threshold(self, current_q88: int,
                         observed_rate_hz: float,
                         dt_ms: float) -> int

@dataclass
class PharmModel:
    agent_name: str = "none"
    gain: float = 1.0
    onset_delay_s: float = 30.0
    wash_time_s: float = 120.0

    def apply(self, t_current_s: float) -> None
    def effective_gain(self, t_current_s: float) -> float
    def modulate_spikes(self, spike_counts: np.ndarray,
                        t_current_s: float) -> np.ndarray
    def modulate_spike_events(self, spikes: list[DetectedSpike],
                              t_current_s: float) -> list[DetectedSpike]

modulate_spike_events is the path used by BioHybridSession.process_frame. Inhibitory gains deterministically thin events across the observed response span, so the pharmacological model does not bias output toward the earliest detected spikes. Excitatory gains preserve the observed events and add template-derived events inside the observed temporal support.

6.6 SpikeSorter

Python
@dataclass
class SpikeSorter:
    num_units: int = 4
    n_components: int = 3

    def fit(self, spikes: list[DetectedSpike]) -> None
    def assign(self, spikes: list[DetectedSpike]) -> list[DetectedSpike]

fit imports scikit-learn only when enough waveforms are present to cluster; amplitude-only spike lists no-op gracefully.

6.7 BioHybridSession + BioHybridFrameResult

Python
@dataclass
class BioHybridFrameResult:
    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]
    opto_pulses: List[OptogeneticPulse]

    def __getitem__(self, key: str) -> Any
    def __contains__(self, key: object) -> bool
    def keys(self) -> List[str]

@dataclass
class BioHybridSession:
    mea_config: MEAConfig
    detector: SpikeDetector
    transcoder: MEAToAERTranscoder
    sc_converter: AERToSCConverter
    opto_encoder: SCToOptoEncoder
    stdp: BiologicalSTDP = ...
    health_monitor: 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 process_frame(
        self,
        voltage_data: np.ndarray,
        t_start_s: float = 0.0,
        stim_times_s: Optional[list[float]] = None,
    ) -> BioHybridFrameResult

6.8 mea_fitness_hook

Python
def mea_fitness_hook(
    detected_spikes: list[DetectedSpike],
    target_rate: float = 10.0,
    *,
    duration_s: float | None = None,
    stimulus_time_s: float | None = None,
    measured_latency_ms: float | None = None,
) -> dict[str, float]

Returns {"accuracy", "energy_mw", "latency_ms"}. Empty input returns the floor {0.1, 0.0, 0.0}; target_rate == 0 also returns the floor accuracy. duration_s must be finite and positive when supplied. stimulus_time_s and measured_latency_ms must be finite; measured latency must be non-negative. Groups spikes by DetectedSpike.channel (regression guard against the previous channel_id bug).


7. Performance benchmarks

All numbers measured 2026-04-20 on Linux x86-64 (Intel i5-11600K, CPython 3.12.3, scikit-learn 1.8, NumPy 2.2). Committed bench harness: benchmarks/bench_bioware.py — JSON at benchmarks/results/bench_bioware.json. Reproducer scripts also in §7.4.

Note: the demo uses uniform-random MEA voltage and ArcaneZenith receives a stochastic current stream; identity_drift and per-frame latency therefore vary ≈10 % between runs. The figures in the next two tables come from one concrete run; the demo wall-time (≈0.69 s on the reference host) is stable.

7.1 End-to-end closed-loop latency

Scenario Wall time Per-frame latency
14_bioware_closed_loop_demo (100 frames) 0.69 s 2 945 µs at frame 100

Full pipeline: synthetic MEA → SpikeDetector → SpikeSorter (PCA-fitted on 177 training waveforms) → MEAToAERTranscoder → AERToSCConverter → SCToOptoEncoder → :class:CultureHealth + :class:HomeostaticPlasticity update + :class:ArcaneZenithCognitiveCore.step_from_bio_rates. Per-frame latency decays over the run as the pipeline warms up — the demo reports 6 916 µs at frame 20 dropping to 2 945 µs at frame 100.

7.2 ArcaneZenith-coupled identity drift

Over the same 100-frame demo the attached :class:ArcaneZenithCognitiveCore reports identity_drift = 1.8625 at frame 100 (printed as "Final ArcaneZenith identity drift: 1.8625" by the demo). Zero network bursts are detected in the default 100-frame window; the drift tracks the novelty signal produced by the closed loop.

7.3 HomeostaticPlasticity.update_threshold microbenchmark

Input pattern Result
target 10 Hz, observed 10 Hz, dt = 100 ms new == 256 (no change)
target 10 Hz, observed 50 Hz, dt = 1000 ms, new > 256 (raised)
τ_homeo = 1000 ms
target 10 Hz, observed 1 Hz, dt = 1000 ms, new < 256 (lowered)
τ_homeo = 1000 ms
10 000 Hz error for 10 s saturates at max_q88
0 Hz for 10 s saturates at min_q88

All five cases are enforced by the TestHomeostaticPlasticity test group. The controller is pure integer arithmetic (subtraction, multiplication, floor-division, clamp) — CPU cost well below one microsecond per call.

7.4 Reproducer

Bash
# 7.1 + 7.2 end-to-end demo
MPLBACKEND=Agg python examples/14_bioware_closed_loop_demo.py

# 7.3 homeostatic controller
python -c "
from sc_neurocore.bioware.bioware import HomeostaticPlasticity
hp = HomeostaticPlasticity(target_rate_hz=10.0, tau_homeo_ms=1000.0)
print(hp.update_threshold(256, observed_rate_hz=10.0, dt_ms=100.0))  # 256
print(hp.update_threshold(256, observed_rate_hz=50.0, dt_ms=1000.0)) # >256
print(hp.update_threshold(256, observed_rate_hz=1.0,  dt_ms=1000.0)) # <256
"

The demo prints [3] Experiment complete in 0.69s. on the reference host.


8. Citations

  1. Bi, G.-Q. & Poo, M.-M. (1998). Synaptic modifications in cultured hippocampal neurons: dependence on spike timing, synaptic strength, and postsynaptic cell type. Journal of Neuroscience 18(24): 10464–10472. — Exponential pair-based STDP used by :class:BiologicalSTDP.
  2. Bienenstock, E. L., Cooper, L. N., Munro, P. W. (1982). Theory for the development of neuron selectivity: orientation specificity and binocular interaction in visual cortex. Journal of Neuroscience 2(1): 32–48. — BCM metaplasticity used by :class:BCMPlasticity.
  3. Donoho, D. L. & Johnstone, I. M. (1994). Ideal spatial adaptation by wavelet shrinkage. Biometrika 81(3): 425–455. — MAD / 0.6745 robust σ estimator used by :class:SpikeDetector.estimate_noise.
  4. Kagan, B. J., Kitchen, A. C., et al. (2022). In vitro neurons learn and exhibit sentience when embodied in a simulated game-world. Neuron 110(23): 3952–3969.e8. — Reference closed-loop MEA-opto experiment motivating :class:BioHybridSession.
  5. Lewicki, M. S. (1998). A review of methods for spike sorting: the detection and classification of neural action potentials. Network: Computation in Neural Systems 9(4): R53–R78. — Classical PCA-based spike sorter; direct antecedent of :class:SpikeSorter.
  6. Turrigiano, G. G. (2012). Homeostatic synaptic plasticity: local and global mechanisms for stabilizing neuronal function. Cold Spring Harbor Perspectives in Biology 4(1): a005736. — Biological reference for the slow homeostatic timescales modelled by :class:HomeostaticPlasticity.

9. Limitations

  • SpikeSorter needs scikit-learn. Without it, fit raises a clear ImportError with a pointer to the [bioware] extras. The empty-input path is a no-op and does not require sklearn.
  • PCA + KMeans is a minimal baseline. Production spike-sorting should use SpikeInterface, Kilosort 3+, or MountainSort. The SC-NeuroCore sorter exists as a closed-loop demo.
  • No real-time FPGA coupling inside this module. The transcoder produces AER events suitable for the PYNQ-Z2 hardware, but the actual streaming to the FPGA is caller-owned (see hdl/).
  • Q8.8 homeostatic controller is proportional only. A full PID controller would need integral and derivative state; the current single-step update is tuned to the slow tau_homeo regime where P is sufficient (Turrigiano 2012).
  • BioHybridFrameResult mapping view is read-only. in, [key], and keys() work; result[key] = value raises. Mutate the dataclass fields directly for reply-time updates.

Reference

  • Module: src/sc_neurocore/bioware/bioware.py.
  • Tests: tests/test_bioware/test_bioware.py (40+ tests incl. TestMEAConfig, TestSpikeDetector, TestMEAToAERTranscoder, TestAERToSCConverter, TestSCToOptoEncoder, TestBiologicalSTDP, TestBCMPlasticity, TestCultureHealth, TestBioHybridSession, TestRefractoryPeriod, TestOptoSafety, TestEdgeCases, TestSpikeSorter, TestLFPExtraction, TestLatencyBudget, TestPharmModel, TestMultiWellPlate, TestNetworkBurstDetection, TestArtifactRejection, TestBioAuditLog, TestBitstreamRateDecoder, TestHomeostaticPlasticity, TestBioHybridFrameResult, TestMEAFitnessHook).
  • Demo: examples/14_bioware_closed_loop_demo.py (100-frame end-to-end closed loop with ArcaneZenith + PCA spike sorting).
  • Cross-references: :doc:arcane_zenith (zenith_core attachment), :doc:evo_substrate (mea_fitness_hook).

sc_neurocore.bioware.bioware

Interface primitives for living neural cultures and organoids.

Bridges biological neural activity (from MEA recordings) to SC bitstreams and vice-versa. Enables closed-loop bio-hybrid experiments where:

  1. MEA → AER: Spike-sorts multi-electrode array data into AER events compatible with sc_aer_encoder.v / sc_aer_router.v.
  2. AER → SC: Converts AER events into SC bitstreams for deterministic stochastic processing.
  3. SC → Optogenetics: Encodes SC output as optical pulse sequences for closed-loop stimulation.
  4. Biological Plasticity: STDP/BCM adapters bridging biological time constants (ms) to SC clock rates (MHz).

Compatible with: - hdl/sc_aer_encoder.v — AER spike encoding - hdl/sc_aer_router.v — AER spike routing - analysis/spike_stats — spike train analysis - profiling/spike_profiler.py — spike rate profiling

MEALayout

Bases: Enum

Standard MEA electrode layouts.

Source code in src/sc_neurocore/bioware/bioware.py
Python
55
56
57
58
59
60
61
62
class MEALayout(Enum):
    """Standard MEA electrode layouts."""

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

MEAConfig dataclass

Multi-electrode array configuration.

Source code in src/sc_neurocore/bioware/bioware.py
Python
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@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

    @classmethod
    def from_layout(cls, layout: MEALayout) -> MEAConfig:
        presets = {
            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])

DetectedSpike dataclass

One detected spike event from MEA data.

Source code in src/sc_neurocore/bioware/bioware.py
Python
 92
 93
 94
 95
 96
 97
 98
 99
100
@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] = None

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.py
Python
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@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] = field(default=None, repr=False)

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

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

    def detect(self, voltage_data: np.ndarray, 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.
        """
        n_samples, n_channels = voltage_data.shape
        if self._noise_estimates is None:
            self.estimate_noise(voltage_data)
        assert self._noise_estimates is not None

        spikes = []
        dt = 1.0 / self.config.sample_rate_hz
        sigma = self.config.spike_threshold_sigma

        for ch in range(n_channels):
            threshold = sigma * self._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
                half = int(snippet_ms * self.config.sample_rate_hz / 2000.0)
                start = max(0, idx - half)
                end = min(n_samples, idx + half)

                # Pad if too close to edges
                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 = target_len - len(raw_wave) - pad_before
                    # ensure we don't end up with negative pad_after in edge cases
                    pad_after = max(0, pad_after)
                    raw_wave = np.pad(raw_wave, (pad_before, pad_after), "constant")

                # Strict bound to prevent arbitrary dimension mismatches
                if len(raw_wave) > target_len:
                    raw_wave = raw_wave[:target_len]
                elif len(raw_wave) < target_len:
                    raw_wave = np.pad(raw_wave, (0, target_len - len(raw_wave)), "constant")

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

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.py
Python
116
117
118
119
120
121
122
123
124
def estimate_noise(self, voltage_data: np.ndarray) -> np.ndarray:
    """Estimate per-channel noise from voltage data.

    Uses median absolute deviation (MAD) for robustness against spikes.
    voltage_data: shape (num_samples, num_channels)
    """
    mad = 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.py
Python
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def detect(self, voltage_data: np.ndarray, 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.
    """
    n_samples, n_channels = voltage_data.shape
    if self._noise_estimates is None:
        self.estimate_noise(voltage_data)
    assert self._noise_estimates is not None

    spikes = []
    dt = 1.0 / self.config.sample_rate_hz
    sigma = self.config.spike_threshold_sigma

    for ch in range(n_channels):
        threshold = sigma * self._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
            half = int(snippet_ms * self.config.sample_rate_hz / 2000.0)
            start = max(0, idx - half)
            end = min(n_samples, idx + half)

            # Pad if too close to edges
            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 = target_len - len(raw_wave) - pad_before
                # ensure we don't end up with negative pad_after in edge cases
                pad_after = max(0, pad_after)
                raw_wave = np.pad(raw_wave, (pad_before, pad_after), "constant")

            # Strict bound to prevent arbitrary dimension mismatches
            if len(raw_wave) > target_len:
                raw_wave = raw_wave[:target_len]
            elif len(raw_wave) < target_len:
                raw_wave = np.pad(raw_wave, (0, target_len - len(raw_wave)), "constant")

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

AEREvent dataclass

Address-Event Representation packet.

Compatible with sc_aer_encoder.v format:

Source code in src/sc_neurocore/bioware/bioware.py
Python
189
190
191
192
193
194
195
196
197
198
199
200
@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

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.py
Python
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@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 transcode(
        self,
        spikes: List[DetectedSpike],
        t_start_s: float = 0.0,
    ) -> List[AEREvent]:
        """Convert detected spikes to AER events."""
        events = []
        for spike in spikes:
            neuron_id = self._map_channel(spike.channel)
            ts_hw = int((spike.timestamp_s - t_start_s) * self.hw_clock_hz) & 0xFFFF
            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

transcode(spikes, t_start_s=0.0)

Convert detected spikes to AER events.

Source code in src/sc_neurocore/bioware/bioware.py
Python
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def transcode(
    self,
    spikes: List[DetectedSpike],
    t_start_s: float = 0.0,
) -> List[AEREvent]:
    """Convert detected spikes to AER events."""
    events = []
    for spike in spikes:
        neuron_id = self._map_channel(spike.channel)
        ts_hw = int((spike.timestamp_s - t_start_s) * self.hw_clock_hz) & 0xFFFF
        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

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.py
Python
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@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 = 1000
    bitstream_length: int = 256
    num_neurons: int = 128
    lfsr_seed: int = 0xACE1

    def convert(self, events: List[AEREvent]) -> Dict[int, np.ndarray]:
        """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:
                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:
        """LFSR-16 encoding (bit-compatible with core_engine)."""
        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

convert(events)

Convert AER events to per-neuron SC bitstreams.

Source code in src/sc_neurocore/bioware/bioware.py
Python
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def convert(self, events: List[AEREvent]) -> Dict[int, np.ndarray]:
    """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:
            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

StimProtocol

Bases: Enum

Optogenetic stimulation protocols.

Source code in src/sc_neurocore/bioware/bioware.py
Python
290
291
292
293
294
295
296
class StimProtocol(Enum):
    """Optogenetic stimulation protocols."""

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

OptogeneticPulse dataclass

One optical stimulation pulse.

Source code in src/sc_neurocore/bioware/bioware.py
Python
299
300
301
302
303
304
305
306
307
@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)

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.py
Python
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
@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

    def encode(
        self,
        bitstreams: Dict[int, np.ndarray],
        t_start_ms: float = 0.0,
    ) -> List[OptogeneticPulse]:
        """Convert SC bitstreams to optogenetic pulses."""
        pulses = []
        total_power = 0.0
        for nid, bs in sorted(bitstreams.items()):
            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
            if total_power + intensity > self.max_total_power_mw:
                break
            total_power += intensity

            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,
                )
            )
        return pulses

encode(bitstreams, t_start_ms=0.0)

Convert SC bitstreams to optogenetic pulses.

Source code in src/sc_neurocore/bioware/bioware.py
Python
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def encode(
    self,
    bitstreams: Dict[int, np.ndarray],
    t_start_ms: float = 0.0,
) -> List[OptogeneticPulse]:
    """Convert SC bitstreams to optogenetic pulses."""
    pulses = []
    total_power = 0.0
    for nid, bs in sorted(bitstreams.items()):
        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
        if total_power + intensity > self.max_total_power_mw:
            break
        total_power += intensity

        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,
            )
        )
    return pulses

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.py
Python
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@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 compute_dw(self, dt_ms: float) -> float:
        """Compute weight change from spike timing difference.

        dt_ms = t_post - t_pre (positive = potentiation, negative = depression)
        """
        if dt_ms > 0:
            return self.a_plus * np.exp(-dt_ms / self.tau_plus_ms)
        elif dt_ms < 0:
            return -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."""
        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))

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.py
Python
379
380
381
382
383
384
385
386
387
388
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)
    """
    if dt_ms > 0:
        return self.a_plus * np.exp(-dt_ms / self.tau_plus_ms)
    elif dt_ms < 0:
        return -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.py
Python
390
391
392
393
394
395
def update_weight(self, current_q88: int, dt_ms: float) -> int:
    """Update Q8.8 weight from spike timing."""
    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))

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.py
Python
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
@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 update_theta(self, post_rate_hz: float, dt_ms: float) -> float:
        """Update the sliding threshold from postsynaptic activity."""
        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 - θ)."""
        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:
        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))

update_theta(post_rate_hz, dt_ms)

Update the sliding threshold from postsynaptic activity.

Source code in src/sc_neurocore/bioware/bioware.py
Python
413
414
415
416
417
418
def update_theta(self, post_rate_hz: float, dt_ms: float) -> float:
    """Update the sliding threshold from postsynaptic activity."""
    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.py
Python
420
421
422
def compute_dw(self, pre_rate_hz: float, post_rate_hz: float) -> float:
    """BCM weight change: ΔW = η * x * y * (y - θ)."""
    return self.learning_rate * pre_rate_hz * post_rate_hz * (post_rate_hz - self.theta)

CultureHealth dataclass

Monitor organoid/culture viability from MEA activity.

Source code in src/sc_neurocore/bioware/bioware.py
Python
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
@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 assess(self, spike_counts: np.ndarray, duration_s: float) -> Dict[str, float]:
        """Assess culture health from spike activity.

        spike_counts: per-channel spike counts over duration_s
        """
        rates = spike_counts / duration_s if duration_s > 0 else spike_counts
        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),
        }

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.py
Python
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def assess(self, spike_counts: np.ndarray, duration_s: float) -> Dict[str, float]:
    """Assess culture health from spike activity.

    spike_counts: per-channel spike counts over duration_s
    """
    rates = spike_counts / duration_s if duration_s > 0 else spike_counts
    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),
    }

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.py
Python
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
@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]
    opto_pulses: List[OptogeneticPulse]

    def __getitem__(self, key: str) -> Any:
        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:
        if not isinstance(key, str):
            return False
        return key in {f.name for f in fields(self)}

    def keys(self) -> List[str]:
        return [f.name for f in fields(self)]

BioHybridSession dataclass

Manages a complete bio-hybrid experiment session.

Orchestrates: MEA recording → spike detection → AER transcoding → SC processing → optogenetic feedback → plasticity update.

Source code in src/sc_neurocore/bioware/bioware.py
Python
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
@dataclass
class BioHybridSession:
    """Manages a complete bio-hybrid experiment session.

    Orchestrates: MEA recording → spike detection → AER transcoding →
    SC processing → optogenetic feedback → plasticity update.
    """

    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 process_frame(
        self,
        voltage_data: np.ndarray,
        t_start_s: float = 0.0,
        stim_times_s: Optional[List[float]] = None,
    ) -> BioHybridFrameResult:
        """Process one MEA data frame through the full pipeline."""
        t0 = time.perf_counter_ns()
        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, t_start_s)

        # 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

        if self.latency_budget is not None:
            self.latency_budget.record(latency_us)

        return BioHybridFrameResult(
            round=self.round_count,
            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,
        )

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

Process one MEA data frame through the full pipeline.

Source code in src/sc_neurocore/bioware/bioware.py
Python
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
def process_frame(
    self,
    voltage_data: np.ndarray,
    t_start_s: float = 0.0,
    stim_times_s: Optional[List[float]] = None,
) -> BioHybridFrameResult:
    """Process one MEA data frame through the full pipeline."""
    t0 = time.perf_counter_ns()
    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, t_start_s)

    # 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

    if self.latency_budget is not None:
        self.latency_budget.record(latency_us)

    return BioHybridFrameResult(
        round=self.round_count,
        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,
    )

SpikeSorter dataclass

Production-ready spike sorter utilizing PCA feature extraction and K-Means clustering.

Extracts the dominant principal components from the input raw waveforms, and cleanly separates units. Handles missing datasets explicitly natively. Requires scikit-learn to execute correctly.

Source code in src/sc_neurocore/bioware/bioware.py
Python
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
@dataclass
class SpikeSorter:
    """Production-ready spike sorter utilizing PCA feature extraction and K-Means clustering.

    Extracts the dominant principal components from the input raw waveforms, and cleanly
    separates units. Handles missing datasets explicitly natively. Requires `scikit-learn` to execute correctly.
    """

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

    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

        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)
        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

            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

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.py
Python
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
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

    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)
    self._kmeans.fit(features)

assign(spikes)

Assign cluster IDs based on PCA feature projections.

Source code in src/sc_neurocore/bioware/bioware.py
Python
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
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

        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

LFPBand dataclass

Frequency band definition for LFP extraction.

Source code in src/sc_neurocore/bioware/bioware.py
Python
677
678
679
680
681
682
683
@dataclass
class LFPBand:
    """Frequency band definition for LFP extraction."""

    name: str
    low_hz: float
    high_hz: float

LatencyBudget dataclass

Tracks and enforces closed-loop latency requirements.

Source code in src/sc_neurocore/bioware/bioware.py
Python
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
@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 record(self, latency_us: float) -> bool:
        """Record a latency measurement. Returns True if within budget."""
        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 float(np.mean(self.history)) if self.history else 0.0

    @property
    def p99_latency_us(self) -> float:
        return float(np.percentile(self.history, 99)) if self.history else 0.0

    @property
    def compliance_ratio(self) -> float:
        if not self.history:
            return 1.0
        return 1.0 - self.violations / len(self.history)

record(latency_us)

Record a latency measurement. Returns True if within budget.

Source code in src/sc_neurocore/bioware/bioware.py
Python
731
732
733
734
735
736
737
def record(self, latency_us: float) -> bool:
    """Record a latency measurement. Returns True if within budget."""
    self.history.append(latency_us)
    if latency_us > self.max_latency_us:
        self.violations += 1
        return False
    return True

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.py
Python
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
@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 apply(self, t_current_s: float) -> None:
        self._applied_at = t_current_s

    def effective_gain(self, t_current_s: float) -> float:
        if self._applied_at < 0:
            return 1.0
        elapsed = t_current_s - self._applied_at
        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, t_current_s: float) -> np.ndarray:
        """Modulate spike counts by pharmacological gain."""
        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.
        """
        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:
                idx = int(np.searchsorted(timestamps, t, side="left"))
                if idx >= len(ordered):
                    idx = len(ordered) - 1
                elif 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))

modulate_spikes(spike_counts, t_current_s)

Modulate spike counts by pharmacological gain.

Source code in src/sc_neurocore/bioware/bioware.py
Python
783
784
785
786
def modulate_spikes(self, spike_counts: np.ndarray, t_current_s: float) -> np.ndarray:
    """Modulate spike counts by pharmacological gain."""
    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.py
Python
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
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.
    """
    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:
            idx = int(np.searchsorted(timestamps, t, side="left"))
            if idx >= len(ordered):
                idx = len(ordered) - 1
            elif 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.py
Python
859
860
861
862
863
864
865
866
867
868
869
870
@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

    @property
    def label(self) -> str:
        return f"{self.well_id}_{self.culture_type}_P{self.passage_number}"

MultiWellPlate dataclass

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

Source code in src/sc_neurocore/bioware/bioware.py
Python
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
@dataclass
class MultiWellPlate:
    """Multi-well plate (e.g., 6/24/48/96-well MEA plate)."""

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

    def add_well(self, well: WellConfig) -> None:
        self.wells.append(well)

    @classmethod
    def standard_6_well(cls, layout: MEALayout = MEALayout.MEA_60) -> MultiWellPlate:
        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 len(self.wells)

    def get_well(self, well_id: str) -> Optional[WellConfig]:
        return next((w for w in self.wells if w.well_id == well_id), None)

NetworkBurst dataclass

Detected network-wide synchronised burst event.

Source code in src/sc_neurocore/bioware/bioware.py
Python
905
906
907
908
909
910
911
912
@dataclass
class NetworkBurst:
    """Detected network-wide synchronised burst event."""

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

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.py
Python
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
@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 blank(
        self,
        voltage_data: np.ndarray,
        stim_times_s: List[float],
        sample_rate_hz: float,
    ) -> np.ndarray:
        """Return voltage data with stimulus artifacts blanked."""
        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)

        for t_s in stim_times_s:
            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

blank(voltage_data, stim_times_s, sample_rate_hz)

Return voltage data with stimulus artifacts blanked.

Source code in src/sc_neurocore/bioware/bioware.py
Python
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
def blank(
    self,
    voltage_data: np.ndarray,
    stim_times_s: List[float],
    sample_rate_hz: float,
) -> np.ndarray:
    """Return voltage data with stimulus artifacts blanked."""
    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)

    for t_s in stim_times_s:
        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

BioAuditEntry dataclass

One audit entry for a bio-hybrid session.

Source code in src/sc_neurocore/bioware/bioware.py
Python
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
@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 = ""

BioAuditLog dataclass

Regulatory-grade audit log for bio-hybrid experiments.

Source code in src/sc_neurocore/bioware/bioware.py
Python
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
@dataclass
class BioAuditLog:
    """Regulatory-grade audit log for bio-hybrid experiments."""

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

    def log(self, entry: BioAuditEntry) -> None:
        self.entries.append(entry)

    @property
    def total_rounds(self) -> int:
        return len(self.entries)

    def to_list(self) -> List[Dict]:
        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:
        """SHA-256 of log contents for tamper detection."""
        try:
            import orjson

            data = orjson.dumps(self.to_list(), option=orjson.OPT_SORT_KEYS)
        except ImportError:
            import json as _json

            data = _json.dumps(self.to_list(), sort_keys=True).encode("utf-8")
        return hashlib.sha256(data).hexdigest()

checksum()

SHA-256 of log contents for tamper detection.

Source code in src/sc_neurocore/bioware/bioware.py
Python
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
def checksum(self) -> str:
    """SHA-256 of log contents for tamper detection."""
    try:
        import orjson

        data = orjson.dumps(self.to_list(), option=orjson.OPT_SORT_KEYS)
    except ImportError:
        import json as _json

        data = _json.dumps(self.to_list(), sort_keys=True).encode("utf-8")
    return hashlib.sha256(data).hexdigest()

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.py
Python
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
@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 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]``.
        """
        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))

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.py
Python
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
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]``.
    """
    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))

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.py
Python
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
def extract_lfp_power(
    voltage_data: np.ndarray,
    sample_rate_hz: float,
    bands: Optional[List[LFPBand]] = None,
) -> Dict[str, np.ndarray]:
    """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

    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

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.py
Python
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
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.
    """
    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] = [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

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.py
Python
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
def decode_bitstream_rate(
    bitstreams: Dict[int, np.ndarray],
    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.
    """
    rates = {}
    for nid, bs in bitstreams.items():
        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. energy_mw remains the documented spike-count proxy (0.5 mW / spike). 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.py
Python
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
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. ``energy_mw`` remains the documented
    spike-count proxy (0.5 mW / spike). ``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.
    """
    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:
        return {"accuracy": 0.1, "energy_mw": 0.0, "latency_ms": 0.0}

    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,
    }