Skip to content

Pipeline

Data ingestion and training orchestration for SNN workflows.

  • DataIngestor — Validated multimodal dataset preparation: min-max normalizes each modality to [0, 1], preserves the reserved labels field as labels, and rejects empty, scalar, non-finite, or mismatched sample axes.
  • SCTrainingLoop — Standard and RL training orchestration with logging, checkpointing, and early stopping
Python
from sc_neurocore.pipeline import DataIngestor, SCTrainingLoop

dataset = DataIngestor().prepare_dataset(
    {"vision": [[0.0, 1.0], [2.0, 3.0]], "labels": [0, 1]}
)
sample = dataset.get_sample(0)

sc_neurocore.pipeline

sc_neurocore.pipeline -- Tier: research (experimental / research).

DataIngestor

Normalize raw multimodal arrays into a MultimodalDataset.

Parameters

label_key: Reserved key used to extract labels from the raw input mapping.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class DataIngestor:
    """Normalize raw multimodal arrays into a `MultimodalDataset`.

    Parameters
    ----------
    label_key:
        Reserved key used to extract labels from the raw input mapping.
    """

    def __init__(self, label_key: str = DEFAULT_LABEL_KEY) -> None:
        """Initialize the ingestor with the reserved label key."""
        if not label_key:
            raise ValueError("label_key must be a non-empty string")
        self.label_key = label_key

    def prepare_dataset(self, raw_data: Mapping[str, Any]) -> MultimodalDataset:
        """Normalize and package raw multimodal data."""
        processed_data: dict[str, Array] = {}
        raw_labels = raw_data.get(self.label_key)

        for name, values in raw_data.items():
            if name == self.label_key:
                continue
            processed_data[name] = _normalize_modality(name, values)

        if not processed_data:
            raise ValueError("raw_data must contain at least one modality")

        first_array = next(iter(processed_data.values()))
        sample_count = int(first_array.shape[0])
        labels: Array = np.zeros(sample_count, dtype=int)
        if raw_labels is not None:
            labels = np.asarray(raw_labels)

        return MultimodalDataset(data=processed_data, labels=labels)

__init__(label_key=DEFAULT_LABEL_KEY)

Initialize the ingestor with the reserved label key.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
 96
 97
 98
 99
100
def __init__(self, label_key: str = DEFAULT_LABEL_KEY) -> None:
    """Initialize the ingestor with the reserved label key."""
    if not label_key:
        raise ValueError("label_key must be a non-empty string")
    self.label_key = label_key

prepare_dataset(raw_data)

Normalize and package raw multimodal data.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def prepare_dataset(self, raw_data: Mapping[str, Any]) -> MultimodalDataset:
    """Normalize and package raw multimodal data."""
    processed_data: dict[str, Array] = {}
    raw_labels = raw_data.get(self.label_key)

    for name, values in raw_data.items():
        if name == self.label_key:
            continue
        processed_data[name] = _normalize_modality(name, values)

    if not processed_data:
        raise ValueError("raw_data must contain at least one modality")

    first_array = next(iter(processed_data.values()))
    sample_count = int(first_array.shape[0])
    labels: Array = np.zeros(sample_count, dtype=int)
    if raw_labels is not None:
        labels = np.asarray(raw_labels)

    return MultimodalDataset(data=processed_data, labels=labels)

MultimodalDataset dataclass

Validated multimodal training dataset.

Parameters

data: Mapping from modality names to normalized arrays. The first axis is the sample axis and must have the same length for every modality. labels: Label array whose first axis matches the modality sample count.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass
class MultimodalDataset:
    """Validated multimodal training dataset.

    Parameters
    ----------
    data:
        Mapping from modality names to normalized arrays. The first axis is the
        sample axis and must have the same length for every modality.
    labels:
        Label array whose first axis matches the modality sample count.
    """

    data: dict[str, Array]
    labels: Array

    def __post_init__(self) -> None:
        """Validate dataset shape invariants after construction."""
        _validate_dataset_shapes(self.data, self.labels)

    def get_sample(self, idx: int) -> dict[str, Array]:
        """Return the per-modality arrays for one sample index."""
        return {k: v[idx] for k, v in self.data.items()}

__post_init__()

Validate dataset shape invariants after construction.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
78
79
80
def __post_init__(self) -> None:
    """Validate dataset shape invariants after construction."""
    _validate_dataset_shapes(self.data, self.labels)

get_sample(idx)

Return the per-modality arrays for one sample index.

Source code in src/sc_neurocore/pipeline/ingestion.py
Python
82
83
84
def get_sample(self, idx: int) -> dict[str, Array]:
    """Return the per-modality arrays for one sample index."""
    return {k: v[idx] for k, v in self.data.items()}

SCTrainingLoop

Standard and Reinforcement Learning loops for SC Networks.

Source code in src/sc_neurocore/pipeline/training.py
Python
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class SCTrainingLoop:
    """
    Standard and Reinforcement Learning loops for SC Networks.
    """

    @staticmethod
    def run_rl_epoch(
        agent: SCLearningLayer,
        env_step_func: Callable[[np.ndarray[Any, Any]], float],
        input_data: np.ndarray[Any, Any],
        generations: int = 10,
    ) -> None:
        """
        Runs a reinforcement learning epoch.
        Uses RewardModulatedSTDPSynapse logic.
        """
        for gen in range(generations):
            # 1. Run forward pass
            spikes = agent.run_epoch(input_data)  # type: ignore[arg-type]

            # 2. Get reward from environment
            reward = env_step_func(spikes)

            # 3. Apply reward to all synapses
            for i in range(agent.n_neurons):
                for j in range(agent.n_inputs):
                    syn = agent.synapses[i][j]
                    if isinstance(syn, RewardModulatedSTDPSynapse):
                        syn.apply_reward(reward)

            logger.info("RL Epoch %d: Reward = %.4f", gen, reward)

    @staticmethod
    def train_multimodal_fusion(fusion_layer: Any, dataset: Any, epochs: int = 5) -> None:
        """Train weights in a multimodal fusion layer via per-sample updates.

        Iterates over the dataset for ``epochs`` rounds, calling
        ``fusion_layer.train_step(sample)`` on each sample returned by
        ``dataset.get_sample(i)``.  The fusion layer is responsible for
        its own weight update rule (Hebbian, LMS, etc.).
        """
        n_samples = getattr(dataset, "n_samples", len(getattr(dataset, "labels", [])))
        for epoch in range(epochs):
            total_loss = 0.0
            for i in range(n_samples):
                sample = dataset.get_sample(i)
                output = fusion_layer.train_step(sample)
                if output is not None:
                    total_loss += float(np.sum(np.abs(output)))
            avg_loss = total_loss / max(n_samples, 1)
            logger.info("Fusion Epoch %d/%d: avg_loss=%.4f", epoch + 1, epochs, avg_loss)

run_rl_epoch(agent, env_step_func, input_data, generations=10) staticmethod

Runs a reinforcement learning epoch. Uses RewardModulatedSTDPSynapse logic.

Source code in src/sc_neurocore/pipeline/training.py
Python
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@staticmethod
def run_rl_epoch(
    agent: SCLearningLayer,
    env_step_func: Callable[[np.ndarray[Any, Any]], float],
    input_data: np.ndarray[Any, Any],
    generations: int = 10,
) -> None:
    """
    Runs a reinforcement learning epoch.
    Uses RewardModulatedSTDPSynapse logic.
    """
    for gen in range(generations):
        # 1. Run forward pass
        spikes = agent.run_epoch(input_data)  # type: ignore[arg-type]

        # 2. Get reward from environment
        reward = env_step_func(spikes)

        # 3. Apply reward to all synapses
        for i in range(agent.n_neurons):
            for j in range(agent.n_inputs):
                syn = agent.synapses[i][j]
                if isinstance(syn, RewardModulatedSTDPSynapse):
                    syn.apply_reward(reward)

        logger.info("RL Epoch %d: Reward = %.4f", gen, reward)

train_multimodal_fusion(fusion_layer, dataset, epochs=5) staticmethod

Train weights in a multimodal fusion layer via per-sample updates.

Iterates over the dataset for epochs rounds, calling fusion_layer.train_step(sample) on each sample returned by dataset.get_sample(i). The fusion layer is responsible for its own weight update rule (Hebbian, LMS, etc.).

Source code in src/sc_neurocore/pipeline/training.py
Python
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def train_multimodal_fusion(fusion_layer: Any, dataset: Any, epochs: int = 5) -> None:
    """Train weights in a multimodal fusion layer via per-sample updates.

    Iterates over the dataset for ``epochs`` rounds, calling
    ``fusion_layer.train_step(sample)`` on each sample returned by
    ``dataset.get_sample(i)``.  The fusion layer is responsible for
    its own weight update rule (Hebbian, LMS, etc.).
    """
    n_samples = getattr(dataset, "n_samples", len(getattr(dataset, "labels", [])))
    for epoch in range(epochs):
        total_loss = 0.0
        for i in range(n_samples):
            sample = dataset.get_sample(i)
            output = fusion_layer.train_step(sample)
            if output is not None:
                total_loss += float(np.sum(np.abs(output)))
        avg_loss = total_loss / max(n_samples, 1)
        logger.info("Fusion Epoch %d/%d: avg_loss=%.4f", epoch + 1, epochs, avg_loss)