sc_neurocore_engine/bindings/
predictive_coding.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2// Commercial license available
3// © Concepts 1996–2026 Miroslav Šotek. All rights reserved.
4// © Code 2020–2026 Miroslav Šotek. All rights reserved.
5// ORCID: 0009-0009-3560-0851
6// Contact: www.anulum.li | protoscience@anulum.li
7// SC-NeuroCore — Predictive-coding PyO3 bindings
8
9//! Python bindings for packed prediction error and lossless spike prediction codecs.
10
11use numpy::{PyArray1, PyReadonlyArray1};
12use pyo3::exceptions::PyValueError;
13use pyo3::prelude::*;
14
15/// Register predictive-coding functions with the extension module.
16pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
17    module.add_function(wrap_pyfunction!(py_prediction_error, module)?)?;
18    module.add_function(wrap_pyfunction!(py_predict_xor_ema, module)?)?;
19    module.add_function(wrap_pyfunction!(py_predict_xor_lfsr, module)?)?;
20    module.add_function(wrap_pyfunction!(py_recover_xor_ema, module)?)?;
21    module.add_function(wrap_pyfunction!(py_recover_xor_lfsr, module)?)?;
22    Ok(())
23}
24
25/// Return normalized XOR error between packed prediction and observation streams.
26#[pyfunction]
27fn py_prediction_error(
28    _py: Python<'_>,
29    predicted: PyReadonlyArray1<'_, u64>,
30    actual: PyReadonlyArray1<'_, u64>,
31    length: usize,
32) -> PyResult<f64> {
33    let predicted = predicted.as_slice().map_err(|error| {
34        PyValueError::new_err(format!("predicted array must be contiguous: {error}"))
35    })?;
36    let actual = actual.as_slice().map_err(|error| {
37        PyValueError::new_err(format!("actual array must be contiguous: {error}"))
38    })?;
39    Ok(crate::predictive_coding::prediction_error_packed(
40        predicted, actual, length,
41    ))
42}
43
44/// Encode a spike matrix as EMA-prediction XOR errors.
45#[pyfunction]
46fn py_predict_xor_ema(
47    py: Python<'_>,
48    spikes: PyReadonlyArray1<'_, i8>,
49    n_channels: usize,
50    alpha: f64,
51    threshold: f64,
52) -> PyResult<(Py<PyArray1<i8>>, usize)> {
53    let spikes = spikes
54        .as_slice()
55        .map_err(|error| PyValueError::new_err(format!("spikes must be contiguous: {error}")))?;
56    let (errors, correct) =
57        crate::predictive_coding::predict_and_xor_ema(spikes, n_channels, alpha, threshold);
58    Ok((PyArray1::from_vec(py, errors).into(), correct))
59}
60
61/// Recover a spike matrix from EMA-prediction XOR errors.
62#[pyfunction]
63fn py_recover_xor_ema(
64    py: Python<'_>,
65    errors: PyReadonlyArray1<'_, i8>,
66    n_channels: usize,
67    alpha: f64,
68    threshold: f64,
69) -> PyResult<Py<PyArray1<i8>>> {
70    let errors = errors
71        .as_slice()
72        .map_err(|error| PyValueError::new_err(format!("errors must be contiguous: {error}")))?;
73    let spikes =
74        crate::predictive_coding::xor_and_recover_ema(errors, n_channels, alpha, threshold);
75    Ok(PyArray1::from_vec(py, spikes).into())
76}
77
78/// Encode a spike matrix as seeded LFSR-prediction XOR errors.
79#[pyfunction]
80fn py_predict_xor_lfsr(
81    py: Python<'_>,
82    spikes: PyReadonlyArray1<'_, i8>,
83    n_channels: usize,
84    alpha_q8: i32,
85    seed: u16,
86) -> PyResult<(Py<PyArray1<i8>>, usize)> {
87    let spikes = spikes
88        .as_slice()
89        .map_err(|error| PyValueError::new_err(format!("spikes must be contiguous: {error}")))?;
90    let (errors, correct) =
91        crate::predictive_coding::predict_and_xor_lfsr(spikes, n_channels, alpha_q8, seed);
92    Ok((PyArray1::from_vec(py, errors).into(), correct))
93}
94
95/// Recover a spike matrix from seeded LFSR-prediction XOR errors.
96#[pyfunction]
97fn py_recover_xor_lfsr(
98    py: Python<'_>,
99    errors: PyReadonlyArray1<'_, i8>,
100    n_channels: usize,
101    alpha_q8: i32,
102    seed: u16,
103) -> PyResult<Py<PyArray1<i8>>> {
104    let errors = errors
105        .as_slice()
106        .map_err(|error| PyValueError::new_err(format!("errors must be contiguous: {error}")))?;
107    let spikes = crate::predictive_coding::xor_and_recover_lfsr(errors, n_channels, alpha_q8, seed);
108    Ok(PyArray1::from_vec(py, spikes).into())
109}
sc_neurocore_engine/bindings/predictive_coding.rs

sc_neurocore_engine/bindings/
predictive_coding.rs
