Skip to content

Core Modules

Central infrastructure: model definition language parser, simulation orchestrator, and tensor streaming for spike data.

MDL Parser

sc_neurocore.core.mdl_parser

Mind Description Language helpers for serialising orchestrator state.

MDLSpecification dataclass

Serializable MDL payload containing architecture and state sections.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
22
23
24
25
26
27
28
29
@dataclass
class MDLSpecification:
    """Serializable MDL payload containing architecture and state sections."""

    version: str = "1.0"
    agent_name: str = "Unknown"
    architecture: Dict[str, Any] = field(default_factory=dict)
    state: Dict[str, Any] = field(default_factory=dict)

MindDescriptionLanguage

Parser for the Mind Description Language (MDL).

A universal, substrate-independent format for archiving an agent's architecture and state.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
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
class MindDescriptionLanguage:
    """Parser for the Mind Description Language (MDL).

    A universal, substrate-independent format for archiving an agent's
    architecture and state.
    """

    @staticmethod
    def encode(orchestrator: Any, agent_name: str) -> str:
        """Export the orchestrator state to a YAML MDL string."""
        architecture = {}
        state = {}

        for name, module in orchestrator.modules.items():
            # Abstract representation
            architecture[name] = {"type": module.__class__.__name__, "module": module.__module__}

            if hasattr(module, "get_state"):
                state[name] = module.get_state()
            elif hasattr(module, "weights"):
                # Convert numpy to list for YAML
                state[name] = {"weights": module.weights.tolist()}

        mdl = MDLSpecification(agent_name=agent_name, architecture=architecture, state=state)
        return str(yaml.dump(asdict(mdl), sort_keys=False))

    @staticmethod
    def decode(mdl_string: str) -> Dict[str, Any]:
        """Parse an MDL string back into a dictionary for reconstruction."""
        data = yaml.safe_load(mdl_string)
        if not isinstance(data, dict):
            raise ValueError("MDL string must decode into a mapping")
        decoded: dict[str, Any] = data
        logger.info(
            "MDL: Decoded mind of '%s' (v%s)",
            decoded.get("agent_name", "Unknown"),
            decoded.get("version"),
        )
        return decoded

encode(orchestrator, agent_name) staticmethod

Export the orchestrator state to a YAML MDL string.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@staticmethod
def encode(orchestrator: Any, agent_name: str) -> str:
    """Export the orchestrator state to a YAML MDL string."""
    architecture = {}
    state = {}

    for name, module in orchestrator.modules.items():
        # Abstract representation
        architecture[name] = {"type": module.__class__.__name__, "module": module.__module__}

        if hasattr(module, "get_state"):
            state[name] = module.get_state()
        elif hasattr(module, "weights"):
            # Convert numpy to list for YAML
            state[name] = {"weights": module.weights.tolist()}

    mdl = MDLSpecification(agent_name=agent_name, architecture=architecture, state=state)
    return str(yaml.dump(asdict(mdl), sort_keys=False))

decode(mdl_string) staticmethod

Parse an MDL string back into a dictionary for reconstruction.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
58
59
60
61
62
63
64
65
66
67
68
69
70
@staticmethod
def decode(mdl_string: str) -> Dict[str, Any]:
    """Parse an MDL string back into a dictionary for reconstruction."""
    data = yaml.safe_load(mdl_string)
    if not isinstance(data, dict):
        raise ValueError("MDL string must decode into a mapping")
    decoded: dict[str, Any] = data
    logger.info(
        "MDL: Decoded mind of '%s' (v%s)",
        decoded.get("agent_name", "Unknown"),
        decoded.get("version"),
    )
    return decoded

Orchestrator

sc_neurocore.core.orchestrator

TensorStream-aware orchestrator for sequencing registered processing modules.

CognitiveOrchestrator dataclass

Central orchestrator that sequences registered modules into a pipeline.

Connects disparate processing modules and routes a :class:TensorStream through them in execution order.

Source code in src/sc_neurocore/core/orchestrator.py
Python
23
24
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass
class CognitiveOrchestrator:
    """Central orchestrator that sequences registered modules into a pipeline.

    Connects disparate processing modules and routes a :class:`TensorStream`
    through them in execution order.
    """

    modules: Dict[str, Any] = field(default_factory=dict)
    active_goals: List[str] = field(default_factory=list)
    attention_focus: Optional[str] = None

    def register_module(self, name: str, module_obj: Any) -> None:
        """Register a named module object for later pipeline execution."""
        self.modules[name] = module_obj

    def set_attention(self, module_name: str) -> None:
        """Focus orchestrator resources on a specific module."""
        if module_name in self.modules:
            self.attention_focus = module_name
            logger.info("Orchestrator: Attention focused on '%s'.", module_name)

    def execute_pipeline(self, pipeline: List[str], initial_input: TensorStream) -> TensorStream:
        """Execute a sequence of modules, handling TensorStream conversions.

        Parameters
        ----------
        pipeline : list of str
            Ordered module names to execute; unknown names are skipped.
        initial_input : TensorStream
            Input stream fed to the first module in the pipeline.

        Returns
        -------
        TensorStream
            The stream produced by the final executed module.
        """
        current_stream = initial_input

        for module_name in pipeline:
            if module_name not in self.modules:
                logger.warning("Module %s not found.", module_name)
                continue

            module = self.modules[module_name]

            # Smart dispatch based on module type/method
            if hasattr(module, "forward"):
                # Many layers use 'forward'
                # Check what input it expects (rough heuristic)
                if "Quantum" in module.__class__.__name__:
                    input_data = current_stream.to_bitstream()
                else:
                    input_data = current_stream.to_prob()

                output_data = module.forward(input_data)

                # Wrap output back to stream
                if isinstance(output_data, np.ndarray):
                    if np.iscomplexobj(output_data):
                        current_stream = TensorStream(output_data, "quantum")
                    elif output_data.dtype == np.uint8:
                        current_stream = TensorStream(output_data, "bitstream")
                    else:
                        current_stream = TensorStream(output_data, "prob")

            elif hasattr(module, "step"):
                # Simple neurons or CPGs
                # Process scalar or vector step
                val = current_stream.to_prob()
                if isinstance(val, np.ndarray) and val.ndim > 0:
                    res = np.array([module.step(v) for v in val.flatten()])
                else:
                    res = module.step(val)
                current_stream = TensorStream.from_prob(res)

        return current_stream

register_module(name, module_obj)

Register a named module object for later pipeline execution.

Source code in src/sc_neurocore/core/orchestrator.py
Python
35
36
37
def register_module(self, name: str, module_obj: Any) -> None:
    """Register a named module object for later pipeline execution."""
    self.modules[name] = module_obj

set_attention(module_name)

Focus orchestrator resources on a specific module.

Source code in src/sc_neurocore/core/orchestrator.py
Python
39
40
41
42
43
def set_attention(self, module_name: str) -> None:
    """Focus orchestrator resources on a specific module."""
    if module_name in self.modules:
        self.attention_focus = module_name
        logger.info("Orchestrator: Attention focused on '%s'.", module_name)

execute_pipeline(pipeline, initial_input)

Execute a sequence of modules, handling TensorStream conversions.

Parameters

pipeline : list of str Ordered module names to execute; unknown names are skipped. initial_input : TensorStream Input stream fed to the first module in the pipeline.

Returns

TensorStream The stream produced by the final executed module.

Source code in src/sc_neurocore/core/orchestrator.py
Python
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def execute_pipeline(self, pipeline: List[str], initial_input: TensorStream) -> TensorStream:
    """Execute a sequence of modules, handling TensorStream conversions.

    Parameters
    ----------
    pipeline : list of str
        Ordered module names to execute; unknown names are skipped.
    initial_input : TensorStream
        Input stream fed to the first module in the pipeline.

    Returns
    -------
    TensorStream
        The stream produced by the final executed module.
    """
    current_stream = initial_input

    for module_name in pipeline:
        if module_name not in self.modules:
            logger.warning("Module %s not found.", module_name)
            continue

        module = self.modules[module_name]

        # Smart dispatch based on module type/method
        if hasattr(module, "forward"):
            # Many layers use 'forward'
            # Check what input it expects (rough heuristic)
            if "Quantum" in module.__class__.__name__:
                input_data = current_stream.to_bitstream()
            else:
                input_data = current_stream.to_prob()

            output_data = module.forward(input_data)

            # Wrap output back to stream
            if isinstance(output_data, np.ndarray):
                if np.iscomplexobj(output_data):
                    current_stream = TensorStream(output_data, "quantum")
                elif output_data.dtype == np.uint8:
                    current_stream = TensorStream(output_data, "bitstream")
                else:
                    current_stream = TensorStream(output_data, "prob")

        elif hasattr(module, "step"):
            # Simple neurons or CPGs
            # Process scalar or vector step
            val = current_stream.to_prob()
            if isinstance(val, np.ndarray) and val.ndim > 0:
                res = np.array([module.step(v) for v in val.flatten()])
            else:
                res = module.step(val)
            current_stream = TensorStream.from_prob(res)

    return current_stream

Tensor Stream

sc_neurocore.core.tensor_stream

Tensor container with conversions between probability, bitstream, and quantum domains.

TensorStream dataclass

Unified tensor container for sc-neurocore.

Handles automatic conversion between the probability, bitstream, and quantum domains.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
19
20
21
22
23
24
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
@dataclass
class TensorStream:
    """Unified tensor container for sc-neurocore.

    Handles automatic conversion between the probability, bitstream, and
    quantum domains.
    """

    data: np.ndarray[Any, Any]
    domain: str  # 'prob', 'bitstream', 'quantum', 'spike'

    @classmethod
    def from_prob(cls, probs: np.ndarray[Any, Any]) -> TensorStream:
        """Create a tensor stream whose data is already in probability form."""
        return cls(data=probs, domain="prob")

    def to_bitstream(self, length: int = 1024) -> np.ndarray[Any, Any]:
        """Convert probability-domain data into Bernoulli bitstreams."""
        if self.domain == "bitstream":
            return self.data
        if self.domain == "prob":
            # Vectorized Bernoulli
            rands = np.random.random((*self.data.shape, length))
            return (rands < self.data[..., None]).astype(np.uint8)
        raise ValueError(f"Cannot convert {self.domain} to bitstream directly.")

    def to_prob(self) -> np.ndarray[Any, Any]:
        """Convert supported domains into probability-domain tensors."""
        if self.domain == "prob":
            return self.data
        if self.domain == "bitstream":
            # Mean along the last axis (time)
            mean_prob: np.ndarray[Any, Any] = np.mean(self.data, axis=-1)
            return mean_prob
        if self.domain == "quantum":
            # Born Rule: p = |beta|^2
            born_prob: np.ndarray[Any, Any] = np.abs(self.data[..., 1]) ** 2
            return born_prob
        return self.data  # Fallback

    def to_quantum(self) -> np.ndarray[Any, Any]:
        """Convert probability-domain data into two-amplitude quantum encoding."""
        if self.domain == "quantum":
            return self.data
        p = np.clip(self.to_prob(), 0.0, 1.0)
        # Amplitude encoding: |psi> = sqrt(1-p)|0> + sqrt(p)|1>
        # Measurement P(|1>) = |beta|^2 = p — preserves SC probability exactly.
        # Matches CategoryTheoryBridge.stochastic_to_quantum().
        alpha = np.sqrt(1.0 - p)
        beta = np.sqrt(p)
        return np.stack([alpha, beta], axis=-1).astype(complex)

from_prob(probs) classmethod

Create a tensor stream whose data is already in probability form.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
30
31
32
33
@classmethod
def from_prob(cls, probs: np.ndarray[Any, Any]) -> TensorStream:
    """Create a tensor stream whose data is already in probability form."""
    return cls(data=probs, domain="prob")

to_bitstream(length=1024)

Convert probability-domain data into Bernoulli bitstreams.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
35
36
37
38
39
40
41
42
43
def to_bitstream(self, length: int = 1024) -> np.ndarray[Any, Any]:
    """Convert probability-domain data into Bernoulli bitstreams."""
    if self.domain == "bitstream":
        return self.data
    if self.domain == "prob":
        # Vectorized Bernoulli
        rands = np.random.random((*self.data.shape, length))
        return (rands < self.data[..., None]).astype(np.uint8)
    raise ValueError(f"Cannot convert {self.domain} to bitstream directly.")

to_prob()

Convert supported domains into probability-domain tensors.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
45
46
47
48
49
50
51
52
53
54
55
56
57
def to_prob(self) -> np.ndarray[Any, Any]:
    """Convert supported domains into probability-domain tensors."""
    if self.domain == "prob":
        return self.data
    if self.domain == "bitstream":
        # Mean along the last axis (time)
        mean_prob: np.ndarray[Any, Any] = np.mean(self.data, axis=-1)
        return mean_prob
    if self.domain == "quantum":
        # Born Rule: p = |beta|^2
        born_prob: np.ndarray[Any, Any] = np.abs(self.data[..., 1]) ** 2
        return born_prob
    return self.data  # Fallback

to_quantum()

Convert probability-domain data into two-amplitude quantum encoding.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
59
60
61
62
63
64
65
66
67
68
69
def to_quantum(self) -> np.ndarray[Any, Any]:
    """Convert probability-domain data into two-amplitude quantum encoding."""
    if self.domain == "quantum":
        return self.data
    p = np.clip(self.to_prob(), 0.0, 1.0)
    # Amplitude encoding: |psi> = sqrt(1-p)|0> + sqrt(p)|1>
    # Measurement P(|1>) = |beta|^2 = p — preserves SC probability exactly.
    # Matches CategoryTheoryBridge.stochastic_to_quantum().
    alpha = np.sqrt(1.0 - p)
    beta = np.sqrt(p)
    return np.stack([alpha, beta], axis=-1).astype(complex)