Skip to content

OOS (Organoid Operating System)

The OOS layer manages the compilation and execution of problems on the OPU.

Process

pykoppu.oos.Process

Represents a computing process on the OPU.

Source code in pykoppu/oos/process.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
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
class Process:
    """
    Represents a computing process on the OPU.
    """

    def __init__(self, problem: Any, backend: str = "cpu", t: float = 1000.0):
        """
        Initialize a process.

        Args:
            problem: The problem instance to solve.
            backend (str): The backend driver to use. Defaults to "cpu".
            t (float): Total simulation duration in milliseconds. Defaults to 1000.0.
        """
        self.problem = problem
        self.backend = backend
        self.t = t
        self.compiler = BioCompiler()
        self.driver = connect(backend)

    def run(self, backend: Optional[str] = None) -> SimulationResult:
        """
        Run the process.

        Args:
            backend (Optional[str]): Override the backend driver for this run.

        Returns:
            SimulationResult: The result of the computation.
        """
        # 0. Handle Backend Override
        if backend is not None and backend != self.backend:
            self.driver.disconnect()
            self.backend = backend
            self.driver = connect(self.backend)
        # 1. Compile
        instructions = self.compiler.compile(self.problem, duration=self.t)

        # 2. Execute
        try:
            # Driver now returns (state, energy, spikes)
            raw_result = self.driver.execute(instructions)

            # Handle different return types for backward compatibility or different drivers
            if isinstance(raw_result, tuple) and len(raw_result) == 3:
                final_state, energy_trace, spike_data = raw_result
            elif isinstance(raw_result, dict) and 'state' in raw_result:
                # Legacy fallback
                final_state = raw_result['state']
                energy_trace = []
                spike_data = ([], [])
            else:
                # Fallback for unknown format
                final_state = np.array([])
                energy_trace = []
                spike_data = ([], [])

            # Apply problem-specific energy offset
            offset = getattr(self.problem, 'offset', 0.0)
            if len(energy_trace) > 0:
                energy_trace = np.array(energy_trace) + offset

                # Normalize energy to [0, 1]
                e_min = np.min(energy_trace)
                e_max = np.max(energy_trace)
                if e_max > e_min:
                    energy_trace = (energy_trace - e_min) / (e_max - e_min)
                else:
                    energy_trace = np.zeros_like(energy_trace)

        finally:
            self.driver.disconnect()

        # 3. Evaluate Metrics
        metrics = {}
        if hasattr(self.problem, 'evaluate'):
            metrics = self.problem.evaluate(final_state)

        # 4. Construct Result
        return SimulationResult(
            solution=final_state,
            energy_history=energy_trace,
            spikes=spike_data,
            metrics=metrics,
            metadata={"backend": self.backend}
        )

__init__(problem, backend='cpu', t=1000.0)

Initialize a process.

Parameters:

Name Type Description Default
problem Any

The problem instance to solve.

 required
backend str

The backend driver to use. Defaults to "cpu".

 'cpu'
t float

Total simulation duration in milliseconds. Defaults to 1000.0.

 1000.0
Source code in pykoppu/oos/process.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, problem: Any, backend: str = "cpu", t: float = 1000.0):
    """
    Initialize a process.

    Args:
        problem: The problem instance to solve.
        backend (str): The backend driver to use. Defaults to "cpu".
        t (float): Total simulation duration in milliseconds. Defaults to 1000.0.
    """
    self.problem = problem
    self.backend = backend
    self.t = t
    self.compiler = BioCompiler()
    self.driver = connect(backend)

run(backend=None)

Run the process.

Parameters:

Name Type Description Default
backend Optional[str]

Override the backend driver for this run.

 None

Returns:

Name Type Description
SimulationResult SimulationResult

The result of the computation.
Source code in pykoppu/oos/process.py 
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
def run(self, backend: Optional[str] = None) -> SimulationResult:
    """
    Run the process.

    Args:
        backend (Optional[str]): Override the backend driver for this run.

    Returns:
        SimulationResult: The result of the computation.
    """
    # 0. Handle Backend Override
    if backend is not None and backend != self.backend:
        self.driver.disconnect()
        self.backend = backend
        self.driver = connect(self.backend)
    # 1. Compile
    instructions = self.compiler.compile(self.problem, duration=self.t)

    # 2. Execute
    try:
        # Driver now returns (state, energy, spikes)
        raw_result = self.driver.execute(instructions)

        # Handle different return types for backward compatibility or different drivers
        if isinstance(raw_result, tuple) and len(raw_result) == 3:
            final_state, energy_trace, spike_data = raw_result
        elif isinstance(raw_result, dict) and 'state' in raw_result:
            # Legacy fallback
            final_state = raw_result['state']
            energy_trace = []
            spike_data = ([], [])
        else:
            # Fallback for unknown format
            final_state = np.array([])
            energy_trace = []
            spike_data = ([], [])

        # Apply problem-specific energy offset
        offset = getattr(self.problem, 'offset', 0.0)
        if len(energy_trace) > 0:
            energy_trace = np.array(energy_trace) + offset

            # Normalize energy to [0, 1]
            e_min = np.min(energy_trace)
            e_max = np.max(energy_trace)
            if e_max > e_min:
                energy_trace = (energy_trace - e_min) / (e_max - e_min)
            else:
                energy_trace = np.zeros_like(energy_trace)

    finally:
        self.driver.disconnect()

    # 3. Evaluate Metrics
    metrics = {}
    if hasattr(self.problem, 'evaluate'):
        metrics = self.problem.evaluate(final_state)

    # 4. Construct Result
    return SimulationResult(
        solution=final_state,
        energy_history=energy_trace,
        spikes=spike_data,
        metrics=metrics,
        metadata={"backend": self.backend}
    )

Result

pykoppu.oos.SimulationResult

Rich result object for KOPPU simulations.

Attributes:

Name Type Description
solution ndarray

The final state vector.
energy_history ndarray

The energy evolution over time.
spikes Tuple[ndarray, ndarray]

Tuple of (spike_times, neuron_indices).
metrics Dict[str, Any]

Evaluation metrics (validity, etc.).
metadata Dict[str, Any]

Simulation metadata.
Source code in pykoppu/oos/result.py 
 11
 12
 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
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class SimulationResult:
    """
    Rich result object for KOPPU simulations.

    Attributes:
        solution (np.ndarray): The final state vector.
        energy_history (np.ndarray): The energy evolution over time.
        spikes (Tuple[np.ndarray, np.ndarray]): Tuple of (spike_times, neuron_indices).
        metrics (Dict[str, Any]): Evaluation metrics (validity, etc.).
        metadata (Dict[str, Any]): Simulation metadata.
    """

    def __init__(
        self,
        solution: np.ndarray,
        energy_history: List[float],
        spikes: Tuple[np.ndarray, np.ndarray],
        metrics: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None
    ):
        self.solution = np.array(solution)
        self.energy_history = np.array(energy_history)
        self.spikes = spikes
        self.metrics = metrics or {}
        self.metadata = metadata or {}

    def plot(self):
        """
        Generate a 3-panel visualization of the simulation.

        1. Temporal Evolution (Raster Plot)
        2. Final State (Heatmap/Bar)
        3. System Energy (Time Series)
        """
        import matplotlib.pyplot as plt
        import seaborn as sns
        sns.set_theme(style="whitegrid")
        fig, axes = plt.subplots(3, 1, figsize=(10, 12), sharex=False)

        # 1. Raster Plot
        spike_times, neuron_indices = self.spikes
        if len(spike_times) > 0:
            axes[0].scatter(spike_times, neuron_indices, s=2, c='black', alpha=0.6)
            axes[0].set_ylabel("Neuron Index")
            axes[0].set_title("Temporal Evolution (Spike Raster)")
        else:
            axes[0].text(0.5, 0.5, "No Spikes Recorded", ha='center', va='center')

        # 2. Final State
        # Visualize as a bar chart or heatmap depending on size
        n = len(self.solution)
        if n <= 50:
            sns.barplot(x=list(range(n)), y=self.solution, hue=list(range(n)), ax=axes[1], palette="viridis", legend=False)
            axes[1].set_ylabel("Excitability Probability")
            axes[1].set_xlabel("Neuron Index")
            axes[1].set_title("Final State")
        else:
            sns.heatmap(self.solution.reshape(1, -1), ax=axes[1], cmap="viridis", cbar=True)
            axes[1].set_title("Final State (Heatmap)")
            axes[1].set_yticks([])

        # 3. System Energy
        if len(self.energy_history) > 0:
            # Plot Energy Trace
            axes[2].plot(self.energy_history, color='red', linewidth=1.5, label='Energy Trace')

            # Calculate Statistics
            e_max = np.max(self.energy_history)
            e_min = np.min(self.energy_history)
            e_mean = np.mean(self.energy_history)
            e_final = self.energy_history[-1]

            # Linear Regression
            x = np.arange(len(self.energy_history))
            y = self.energy_history
            slope, intercept = np.polyfit(x, y, 1)
            trend_line = slope * x + intercept

            # Plot Trend Line
            axes[2].plot(x, trend_line, color='blue', linestyle='--', linewidth=1.5, label=f'Trend (slope={slope:.2e})')

            # Display Stats
            stats_text = (
                f"Max: {e_max:.4f}\n"
                f"Min: {e_min:.4f}\n"
                f"Mean: {e_mean:.4f}\n"
                f"Final: {e_final:.4f}"
            )

            # Legend in upper right
            axes[2].legend(loc='upper right', bbox_to_anchor=(1.0, 1.0))

            # Place text box below the legend (approx y=0.78)
            axes[2].text(1.0, 0.78, stats_text, transform=axes[2].transAxes, 
                         fontsize=10, verticalalignment='top', horizontalalignment='right',
                         bbox=dict(boxstyle='round', facecolor='white', alpha=0.8))

            axes[2].set_ylabel("Energy (H)")
            axes[2].set_xlabel("Time Step")
            axes[2].set_title("System Energy Evolution")
        else:
            axes[2].text(0.5, 0.5, "No Energy Data", ha='center', va='center')

        plt.tight_layout()
        plt.show()

    def __repr__(self):
        return f"SimulationResult(metrics={self.metrics})"

plot()

Generate a 3-panel visualization of the simulation.

  1. Temporal Evolution (Raster Plot)
  2. Final State (Heatmap/Bar)
  3. System Energy (Time Series)
Source code in pykoppu/oos/result.py 
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def plot(self):
    """
    Generate a 3-panel visualization of the simulation.

    1. Temporal Evolution (Raster Plot)
    2. Final State (Heatmap/Bar)
    3. System Energy (Time Series)
    """
    import matplotlib.pyplot as plt
    import seaborn as sns
    sns.set_theme(style="whitegrid")
    fig, axes = plt.subplots(3, 1, figsize=(10, 12), sharex=False)

    # 1. Raster Plot
    spike_times, neuron_indices = self.spikes
    if len(spike_times) > 0:
        axes[0].scatter(spike_times, neuron_indices, s=2, c='black', alpha=0.6)
        axes[0].set_ylabel("Neuron Index")
        axes[0].set_title("Temporal Evolution (Spike Raster)")
    else:
        axes[0].text(0.5, 0.5, "No Spikes Recorded", ha='center', va='center')

    # 2. Final State
    # Visualize as a bar chart or heatmap depending on size
    n = len(self.solution)
    if n <= 50:
        sns.barplot(x=list(range(n)), y=self.solution, hue=list(range(n)), ax=axes[1], palette="viridis", legend=False)
        axes[1].set_ylabel("Excitability Probability")
        axes[1].set_xlabel("Neuron Index")
        axes[1].set_title("Final State")
    else:
        sns.heatmap(self.solution.reshape(1, -1), ax=axes[1], cmap="viridis", cbar=True)
        axes[1].set_title("Final State (Heatmap)")
        axes[1].set_yticks([])

    # 3. System Energy
    if len(self.energy_history) > 0:
        # Plot Energy Trace
        axes[2].plot(self.energy_history, color='red', linewidth=1.5, label='Energy Trace')

        # Calculate Statistics
        e_max = np.max(self.energy_history)
        e_min = np.min(self.energy_history)
        e_mean = np.mean(self.energy_history)
        e_final = self.energy_history[-1]

        # Linear Regression
        x = np.arange(len(self.energy_history))
        y = self.energy_history
        slope, intercept = np.polyfit(x, y, 1)
        trend_line = slope * x + intercept

        # Plot Trend Line
        axes[2].plot(x, trend_line, color='blue', linestyle='--', linewidth=1.5, label=f'Trend (slope={slope:.2e})')

        # Display Stats
        stats_text = (
            f"Max: {e_max:.4f}\n"
            f"Min: {e_min:.4f}\n"
            f"Mean: {e_mean:.4f}\n"
            f"Final: {e_final:.4f}"
        )

        # Legend in upper right
        axes[2].legend(loc='upper right', bbox_to_anchor=(1.0, 1.0))

        # Place text box below the legend (approx y=0.78)
        axes[2].text(1.0, 0.78, stats_text, transform=axes[2].transAxes, 
                     fontsize=10, verticalalignment='top', horizontalalignment='right',
                     bbox=dict(boxstyle='round', facecolor='white', alpha=0.8))

        axes[2].set_ylabel("Energy (H)")
        axes[2].set_xlabel("Time Step")
        axes[2].set_title("System Energy Evolution")
    else:
        axes[2].text(0.5, 0.5, "No Energy Data", ha='center', va='center')

    plt.tight_layout()
    plt.show()