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

MindDescriptionLanguage

Parser for Mind Description Language (MDL). A universal, substrate-independent format for archiving consciousness.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
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
class MindDescriptionLanguage:
    """
    Parser for Mind Description Language (MDL).
    A universal, substrate-independent format for archiving consciousness.
    """

    @staticmethod
    def encode(orchestrator: Any, agent_name: str) -> str:
        """
        Exports the Orchestrator state to YAML MDL.
        """
        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 yaml.dump(asdict(mdl), sort_keys=False)

    @staticmethod
    def decode(mdl_string: str) -> Dict[str, Any]:
        """
        Parses MDL back to a dictionary (for reconstruction).
        """
        data = yaml.safe_load(mdl_string)
        logger.info(
            "MDL: Decoded mind of '%s' (v%s)",
            data.get("agent_name", "Unknown"),
            data.get("version"),
        )
        return data

encode(orchestrator, agent_name) staticmethod

Exports the Orchestrator state to YAML MDL.

Source code in src/sc_neurocore/core/mdl_parser.py
Python
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@staticmethod
def encode(orchestrator: Any, agent_name: str) -> str:
    """
    Exports the Orchestrator state to YAML MDL.
    """
    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 yaml.dump(asdict(mdl), sort_keys=False)

decode(mdl_string) staticmethod

Parses MDL back to a dictionary (for reconstruction).

Source code in src/sc_neurocore/core/mdl_parser.py
Python
55
56
57
58
59
60
61
62
63
64
65
66
@staticmethod
def decode(mdl_string: str) -> Dict[str, Any]:
    """
    Parses MDL back to a dictionary (for reconstruction).
    """
    data = yaml.safe_load(mdl_string)
    logger.info(
        "MDL: Decoded mind of '%s' (v%s)",
        data.get("agent_name", "Unknown"),
        data.get("version"),
    )
    return data

Orchestrator

sc_neurocore.core.orchestrator

CognitiveOrchestrator dataclass

Central Orchestrator for sc-neurocore Agents. Connects disparate modules into a functional pipeline.

Source code in src/sc_neurocore/core/orchestrator.py
Python
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class CognitiveOrchestrator:
    """
    Central Orchestrator for sc-neurocore Agents.
    Connects disparate modules into a functional pipeline.
    """

    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:
        self.modules[name] = module_obj

    def set_attention(self, module_name: str) -> None:
        """Focuses 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:
        """
        Executes a sequence of modules.
        Automatically handles TensorStream conversions.
        """
        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

set_attention(module_name)

Focuses resources on a specific module.

Source code in src/sc_neurocore/core/orchestrator.py
Python
35
36
37
38
39
def set_attention(self, module_name: str) -> None:
    """Focuses 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)

Executes a sequence of modules. Automatically handles TensorStream conversions.

Source code in src/sc_neurocore/core/orchestrator.py
Python
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
def execute_pipeline(self, pipeline: List[str], initial_input: TensorStream) -> TensorStream:
    """
    Executes a sequence of modules.
    Automatically handles TensorStream conversions.
    """
    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

TensorStream dataclass

Unified Data Structure for sc-neurocore. Handles automatic conversion between domains.

Source code in src/sc_neurocore/core/tensor_stream.py
Python
16
17
18
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
@dataclass
class TensorStream:
    """
    Unified Data Structure for sc-neurocore.
    Handles automatic conversion between domains.
    """

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

    @classmethod
    def from_prob(cls, probs: np.ndarray[Any, Any]) -> TensorStream:
        return cls(data=probs, domain="prob")

    def to_bitstream(self, length: int = 1024) -> np.ndarray[Any, Any]:
        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]:
        if self.domain == "prob":
            return self.data
        if self.domain == "bitstream":
            # Mean along the last axis (time)
            return np.mean(self.data, axis=-1)
        if self.domain == "quantum":
            # Born Rule: p = |beta|^2
            return np.abs(self.data[..., 1]) ** 2
        return self.data  # Fallback

    def to_quantum(self) -> np.ndarray[Any, Any]:
        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)