| """Catalyst Cloud — Interactive Neuromorphic Simulator Demo. |
| |
| Runs spiking neural network simulations via the Catalyst Cloud API |
| and displays spike raster plots. |
| """ |
|
|
| import gradio as gr |
| import requests |
| import numpy as np |
|
|
| API_URL = "https://api.catalyst-neuromorphic.com" |
|
|
|
|
| def signup(email: str) -> str: |
| """Sign up for a free API key.""" |
| if not email or "@" not in email: |
| return "Please enter a valid email address." |
| try: |
| resp = requests.post(f"{API_URL}/v1/signup", json={"email": email, "tier": "free"}, timeout=15) |
| if resp.status_code == 200: |
| data = resp.json() |
| return f"Your API key: {data['api_key']}\n\nSave this — it's shown only once.\nFree tier: {data['limits']['max_neurons']} neurons, {data['limits']['max_timesteps']} timesteps, {data['limits']['max_jobs_per_day']} jobs/day." |
| else: |
| return f"Error: {resp.json().get('detail', resp.text)}" |
| except Exception as e: |
| return f"Connection error: {e}" |
|
|
|
|
| def run_simulation(api_key: str, input_size: int, hidden_size: int, output_size: int, |
| topology: str, weight: int, sparsity: float, timesteps: int, |
| input_current: int): |
| """Run a simulation and return results + raster plot.""" |
| if not api_key or not api_key.startswith("cn_live_"): |
| return "Enter a valid API key (starts with cn_live_)", None |
|
|
| headers = {"X-API-Key": api_key, "Content-Type": "application/json"} |
|
|
| |
| populations = [{"label": "input", "size": input_size, "params": {"threshold": 1000}}] |
| connections = [] |
|
|
| if hidden_size > 0: |
| populations.append({"label": "hidden", "size": hidden_size, "params": {"threshold": 1000}}) |
| connections.append({ |
| "source": "input", "target": "hidden", |
| "topology": topology, "weight": weight, "p": sparsity, |
| }) |
| if output_size > 0: |
| populations.append({"label": "output", "size": output_size, "params": {"threshold": 1000}}) |
| connections.append({ |
| "source": "hidden", "target": "output", |
| "topology": topology, "weight": weight, "p": sparsity, |
| }) |
| elif output_size > 0: |
| populations.append({"label": "output", "size": output_size, "params": {"threshold": 1000}}) |
| connections.append({ |
| "source": "input", "target": "output", |
| "topology": topology, "weight": weight, "p": sparsity, |
| }) |
|
|
| total = sum(p["size"] for p in populations) |
| if total > 1024: |
| return f"Total neurons ({total}) exceeds free tier limit (1024). Reduce sizes.", None |
|
|
| try: |
| |
| resp = requests.post(f"{API_URL}/v1/networks", headers=headers, |
| json={"populations": populations, "connections": connections}, timeout=15) |
| if resp.status_code != 200: |
| return f"Network error: {resp.json().get('detail', resp.text)}", None |
|
|
| network_id = resp.json()["network_id"] |
|
|
| |
| resp = requests.post(f"{API_URL}/v1/jobs", headers=headers, json={ |
| "network_id": network_id, |
| "timesteps": timesteps, |
| "stimuli": [{"population": "input", "current": input_current}], |
| }, timeout=15) |
| if resp.status_code != 200: |
| return f"Job error: {resp.json().get('detail', resp.text)}", None |
|
|
| job_id = resp.json()["job_id"] |
|
|
| |
| import time |
| for _ in range(60): |
| time.sleep(0.5) |
| resp = requests.get(f"{API_URL}/v1/jobs/{job_id}", headers=headers, timeout=15) |
| job = resp.json() |
| if job["status"] == "completed": |
| break |
| if job["status"] == "failed": |
| return f"Simulation failed: {job.get('error_message', 'Unknown error')}", None |
| else: |
| return "Timeout waiting for simulation to complete.", None |
|
|
| result = job["result"] |
|
|
| |
| resp = requests.get(f"{API_URL}/v1/jobs/{job_id}/spikes", headers=headers, timeout=15) |
| spikes = resp.json()["spike_trains"] |
|
|
| |
| import matplotlib |
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
|
|
| fig, ax = plt.subplots(figsize=(10, 5)) |
| fig.patch.set_facecolor("#0a0a0a") |
| ax.set_facecolor("#0a0a0a") |
|
|
| colors = {"input": "#4A9EFF", "hidden": "#FF6B6B", "output": "#50C878"} |
| neuron_offset = 0 |
| yticks = [] |
| yticklabels = [] |
|
|
| for pop_label in [p["label"] for p in populations]: |
| pop_size = next(p["size"] for p in populations if p["label"] == pop_label) |
| pop_spikes = spikes.get(pop_label, {}) |
| color = colors.get(pop_label, "#FFFFFF") |
|
|
| for neuron_str, times in pop_spikes.items(): |
| neuron_idx = int(neuron_str) |
| y = neuron_offset + neuron_idx |
| ax.scatter(times, [y] * len(times), s=1, c=color, marker="|", linewidths=0.5) |
|
|
| mid = neuron_offset + pop_size // 2 |
| yticks.append(mid) |
| yticklabels.append(f"{pop_label}\n({pop_size})") |
| neuron_offset += pop_size |
|
|
| ax.set_xlabel("Timestep", color="white", fontsize=11) |
| ax.set_ylabel("Neuron", color="white", fontsize=11) |
| ax.set_title("Spike Raster Plot", color="white", fontsize=13, fontweight="bold") |
| ax.set_yticks(yticks) |
| ax.set_yticklabels(yticklabels, fontsize=9) |
| ax.tick_params(colors="white") |
| ax.spines["bottom"].set_color("#333") |
| ax.spines["left"].set_color("#333") |
| ax.spines["top"].set_visible(False) |
| ax.spines["right"].set_visible(False) |
| ax.set_xlim(-1, timesteps + 1) |
| ax.set_ylim(-1, neuron_offset) |
|
|
| plt.tight_layout() |
|
|
| summary = ( |
| f"Total spikes: {result['total_spikes']}\n" |
| f"Timesteps: {result['timesteps']}\n" |
| f"Compute time: {job['compute_seconds']:.4f}s\n\n" |
| f"Firing rates:\n" + |
| "\n".join(f" {k}: {v:.4f}" for k, v in result["firing_rates"].items()) |
| ) |
|
|
| return summary, fig |
|
|
| except Exception as e: |
| return f"Error: {e}", None |
|
|
|
|
| |
|
|
| with gr.Blocks( |
| title="Catalyst Cloud — Neuromorphic Simulator", |
| theme=gr.themes.Base( |
| primary_hue="blue", |
| neutral_hue="slate", |
| ), |
| ) as demo: |
| gr.Markdown(""" |
| # Catalyst Cloud — Neuromorphic Simulator |
| Run hardware-accurate spiking neural network simulations in the cloud. |
| Full Loihi 2 parity. No hardware required. |
| """) |
|
|
| with gr.Tab("Get API Key"): |
| email_input = gr.Textbox(label="Email", placeholder="you@lab.edu") |
| signup_btn = gr.Button("Sign up (free)") |
| signup_output = gr.Textbox(label="Result", lines=4) |
| signup_btn.click(signup, inputs=[email_input], outputs=[signup_output]) |
|
|
| with gr.Tab("Simulate"): |
| api_key_input = gr.Textbox(label="API Key", placeholder="cn_live_...", type="password") |
|
|
| with gr.Row(): |
| input_size = gr.Slider(1, 500, value=50, step=1, label="Input neurons") |
| hidden_size = gr.Slider(0, 500, value=30, step=1, label="Hidden neurons (0 = skip)") |
| output_size = gr.Slider(0, 200, value=20, step=1, label="Output neurons (0 = skip)") |
|
|
| with gr.Row(): |
| topology = gr.Dropdown( |
| ["all_to_all", "one_to_one", "random_sparse"], |
| value="random_sparse", label="Topology", |
| ) |
| weight = gr.Slider(100, 2000, value=800, step=50, label="Synaptic weight") |
| sparsity = gr.Slider(0.01, 1.0, value=0.3, step=0.01, label="Connection probability") |
|
|
| with gr.Row(): |
| timesteps = gr.Slider(10, 1000, value=200, step=10, label="Timesteps") |
| input_current = gr.Slider(500, 10000, value=5000, step=500, label="Input current") |
|
|
| run_btn = gr.Button("Run simulation", variant="primary") |
|
|
| with gr.Row(): |
| result_text = gr.Textbox(label="Results", lines=8) |
| raster_plot = gr.Plot(label="Spike raster") |
|
|
| run_btn.click( |
| run_simulation, |
| inputs=[api_key_input, input_size, hidden_size, output_size, |
| topology, weight, sparsity, timesteps, input_current], |
| outputs=[result_text, raster_plot], |
| ) |
|
|
| gr.Markdown(""" |
| --- |
| [Website](https://catalyst-neuromorphic.com/cloud) | |
| [API Docs](https://catalyst-neuromorphic.com/cloud/docs) | |
| [Pricing](https://catalyst-neuromorphic.com/cloud/pricing) | |
| [pip install catalyst-cloud](https://pypi.org/project/catalyst-cloud/) |
| """) |
|
|
| demo.launch() |
|
|
