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 
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
class MindDescriptionLanguage:
    """
    Parser for Mind Description Language (MDL).
    A universal, substrate-independent format for archiving consciousness.
    """

    @staticmethod
    def encode(orchestrator, 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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@staticmethod
def encode(orchestrator, 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 
52
53
54
55
56
57
58
59
60
61
62
63
@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 
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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@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 
32
33
34
35
36
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 
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
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 
13
14
15
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
@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]) -> None:
        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)