neuroplatformv2 API Guide for Python Notebooks

neuroplatformv2 API Guide for Python Notebooks#

This guide is for people who want to use the FinalSpark NeuroPlatform SDK directly from Python, especially in a Jupyter notebook, or expose a controlled subset of the SDK to an LLM agent.

The SDK controls real laboratory hardware. Start with read-only database calls. Add triggers, stimulation, pumps, cameras, or other hardware controls only after validating the code with the laboratory operator.

In this repository, “API” primarily means the Python SDK interface. The controllers may communicate with internal HTTP, gRPC, MongoDB, or InfluxDB services, but users should call the Python controllers instead of depending on those internal protocols directly.

1) Choose the Right Interface#

Use case

Recommended interface

Calling style

Jupyter notebook

Classes in neuroplatformv2.core.*

Schema object, usually with top-level await

Standalone Python script

Classes in neuroplatformv2.core.*

asyncio.run(main())

LangChain/LangGraph agent

Classes in neuroplatformv2.tools.*, or a small custom tool

await tool.ainvoke({...})

Do not pass Pydantic request objects to an agent. Let the tool schema validate plain dictionaries generated by the model.

2) Requirements and Installation#

  • Python: >=3.11,<3.13

  • Network access to the FinalSpark laboratory services

  • Valid connection settings and authorization for the requested hardware

From a local clone:

python -m pip install -e .

For development, including Jupyter and plotting dependencies:

poetry install
poetry run python -m ipykernel install --user --name neuroplatformv2

Then select the neuroplatformv2 kernel in Jupyter.

3) Configure the Environment Before Importing#

The following settings are required when the SDK is imported:

export DB_PORT=8086
export INTAN_SOFTWARE_IP="..."
export TRIGGER_IP="..."

Frequently used optional settings include:

  • DB_IP, DB_TIMEOUT, DB_STAT_PORT

  • INTAN_SOFTWARE_PORT, INTAN_SOCK_TIMEOUT

  • TRIGGER_IP_PORT

  • TRIGGER_UV_IP, TRIGGER_UV_PORT, TRIGGER_450_PORT

  • CAMERA_IP, CAMERA_PORT

  • PUMP_1_IP, PUMP_2_IP, PUMP_3_IP, PUMP_PORT

  • MONGODB_URI_NEUROPLATFORM

  • TOTAL_ELECTRODES (default: 32)

Set TOTAL_ELECTRODES=64 for 64-channel hardware. It must be set before the Python process imports the SDK.

4) Notebook Rules#

Jupyter supports top-level await. Use it directly:

from neuroplatformv2.core.intan import IntanController
from neuroplatformv2.utils.schemas import CountDurationRequest

counts = await IntanController.count_spike(CountDurationRequest(duration=1000))
print(counts)

In a notebook:

  1. Use top-level await for async methods.

  2. Do not call asyncio.run(...); a notebook already has an event loop.

  3. Use timezone-aware UTC datetimes for database queries.

  4. Import controllers from their explicit core modules.

  5. Build controller inputs with classes from neuroplatformv2.utils.schemas.

5) A Read-Only Notebook Quick Start#

This example fetches recent spike events and produces a compact summary:

from datetime import datetime, timedelta, timezone

import pandas as pd

from neuroplatformv2.core.database import DatabaseController
from neuroplatformv2.utils.schemas import SpikeEventQuery

FSNAME = "fs511"  # Replace with your dataset identifier
stop = datetime.now(timezone.utc)
start = stop - timedelta(minutes=10)

spikes = await DatabaseController.get_spike_event(
    SpikeEventQuery(start=start, stop=stop, fsname=FSNAME)
)

if spikes is None or spikes.empty:
    print("No spike events returned.")
else:
    required = {"Time", "channel"}
    missing = required - set(spikes.columns)
    if missing:
        raise RuntimeError(
            f"Missing expected columns: {missing}; got {list(spikes.columns)}"
        )

    spikes = spikes.copy()
    spikes["Time"] = pd.to_datetime(spikes["Time"], utc=True, errors="coerce")
    spikes["channel"] = pd.to_numeric(spikes["channel"], errors="coerce")
    spikes = spikes.dropna(subset=["Time", "channel"])
    spikes["channel"] = spikes["channel"].astype(int)

    print(
        {
            "spikes": len(spikes),
            "channels": spikes["channel"].nunique(),
            "first": spikes["Time"].min(),
            "last": spikes["Time"].max(),
        }
    )
    display(spikes.head())

For intervals longer than 10 minutes, query in 10-minute chunks:

async def get_spike_events_chunked(fsname, start, stop, chunk_minutes=10):
    if start.tzinfo is None or stop.tzinfo is None:
        raise ValueError("start and stop must be timezone-aware")
    if start >= stop:
        raise ValueError("start must be earlier than stop")

    frames = []
    cursor = start
    while cursor < stop:
        chunk_stop = min(cursor + timedelta(minutes=chunk_minutes), stop)
        part = await DatabaseController.get_spike_event(
            SpikeEventQuery(start=cursor, stop=chunk_stop, fsname=fsname)
        )
        if part is not None and not part.empty:
            frames.append(part)
        cursor = chunk_stop

    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()


stop = datetime.now(timezone.utc)
start = stop - timedelta(hours=1)
spikes = await get_spike_events_chunked("fs511", start, stop)
print("rows:", len(spikes))

6) Common Controller Calls#

Database (read-only)#

from datetime import datetime, timedelta, timezone

from neuroplatformv2.core.database import DatabaseController
from neuroplatformv2.utils.schemas import (
    FlowrateQuery,
    LabQuery,
    NoiseQuery,
    SpikeCountQuery,
    SpikeEventQuery,
    TriggersQuery,
)

stop = datetime.now(timezone.utc)
start = stop - timedelta(minutes=10)

events = await DatabaseController.get_spike_event(
    SpikeEventQuery(start=start, stop=stop, fsname="fs511")
)
counts = await DatabaseController.get_spike_count(
    SpikeCountQuery(start=start, stop=stop, fsname="fs511")
)
triggers = await DatabaseController.get_all_triggers(
    TriggersQuery(start=start, stop=stop)
)
noise = await DatabaseController.get_noise_std(NoiseQuery(start=start, stop=stop))
temperature = await DatabaseController.temperature(
    LabQuery(start=start, stop=stop, incubator=1)
)
flowrate = await DatabaseController.flowrate(
    FlowrateQuery(start=start, stop=stop, port=1)
)

Intan spike count and triggers#

A digital trigger pattern must contain exactly 16 integers. Prefer async methods in async workflows. UV and 450 nm trigger duration t is in milliseconds and must be in 1..1000.

from neuroplatformv2.core.intan import IntanController
from neuroplatformv2.core.trigger import TriggerController
from neuroplatformv2.core.trigger450 import Trigger450Controller
from neuroplatformv2.core.triggeruv import TriggerUVController
from neuroplatformv2.utils.schemas import CountDurationRequest, TriggerUVSendRequest

counts = await IntanController.count_spike(CountDurationRequest(duration=1000))
digital_result = await TriggerController.trigger_a_sender([1] + [0] * 15)
uv_result = await TriggerUVController.trigger_uv_send(TriggerUVSendRequest(t=100))
blue_result = await Trigger450Controller.trigger_450_send(TriggerUVSendRequest(t=100))

7) Electrical Stimulation Safety and Example#

Electrical stimulation changes real hardware state. Validate the electrode mapping, charge, amplitude, duration, polarity, and booking before running it.

upload_stimparam(...) temporarily stops system recording. After every upload, wait at least five seconds before triggering, starting stimulation, or reading data so recording is active again.

from time import sleep

from neuroplatformv2.core.intan import IntanController
from neuroplatformv2.core.trigger import TriggerController
from neuroplatformv2.utils.enumerations import StimPolarity, StimShape
from neuroplatformv2.utils.schemas import StimParam

electrodes = [0]
params = [
    StimParam(
        index=0,
        enable=True,
        trigger_key=0,
        stim_shape=StimShape.Biphasic,
        polarity=StimPolarity.NegativeFirst,
        phase_duration1=100.0,
        phase_duration2=100.0,
        phase_amplitude1=5.0,
        phase_amplitude2=5.0,
        nb_pulse=0,
    )
]

await IntanController.send_stimparam(params)
await IntanController.upload_stimparam(electrodes)
sleep(5)  # Mandatory: allow recording to resume.

result = await TriggerController.trigger_a_sender([1] + [0] * 15)
print(result)

# Disable stimulation when finished or during error recovery.
print(await IntanController.disable_all_stim())

8) Standalone Python Script#

Top-level await is notebook-specific. In a .py file, use one event loop:

import asyncio

from neuroplatformv2.core.intan import IntanController
from neuroplatformv2.utils.schemas import CountDurationRequest


async def main():
    counts = await IntanController.count_spike(
        CountDurationRequest(duration=1000)
    )
    print(counts)


if __name__ == "__main__":
    asyncio.run(main())

9) Adding Hardware Tools to an Agent#

Treat every mutating tool as privileged. A production agent should have:

  1. A small allowlist of tools required for one workflow.

  2. Explicit bounds in Pydantic schemas, in addition to prompt instructions.

  3. A human approval step before each trigger, stimulation, pump, light, or camera capture.

  4. A maximum number of trials, pulses, stimulated electrodes, and total runtime.

  5. Structured logs containing tool name, validated inputs, timestamps, outputs, and errors.

  6. A cleanup path that disables stimulation and safely stops controlled hardware.

  7. No arbitrary Python REPL, shell, filesystem, or credential access unless essential and separately sandboxed.

Do not rely on the system prompt as a safety boundary. Validate arguments in code and keep dangerous tools out of the agent’s tool list until approval has been granted.

A direct tool test should be completed before connecting any hardware-changing tool to a model.

10) Troubleshooting#

Import fails with missing environment variables#

Set DB_PORT, INTAN_SOFTWARE_IP, and TRIGGER_IP before importing the package, then restart the Python process or notebook kernel.

RuntimeError: asyncio.run() cannot be called from a running event loop#

In Jupyter, remove asyncio.run(...) and call the async function with top-level await.

Pydantic validation error#

Pass a schema instance to controller methods and a plain dictionary to LangChain tools. Check units and enum values. Validation failures should stop the hardware call; do not bypass them.

Database query returns no rows#

Check the dataset identifier (fsname), UTC interval, database access, and eventual ingestion delay. Do not replace missing real data with simulated data unless simulation was explicitly requested.

Expected DataFrame columns are missing#

Inspect df.columns.tolist() and fail clearly. For spike-event analysis, expected columns are normally Time, channel, and, when available, Max Amplitude.

Agent calls the wrong tool or invents results#

Reduce the tool list, improve each tool’s description, return small JSON-serializable results, require use of a data tool in the system message, and reject answers not grounded in a tool result.

11) Quick Reference#

  • Notebook async call: await Controller.method(Request(...))

  • Standalone async call: asyncio.run(main())

  • LangChain tool call: await tool.ainvoke({"field": value})

  • Database timestamps: timezone-aware UTC

  • Long database intervals: query in 10-minute chunks

  • Trigger pattern: exactly 16 integers

  • upload_stimparam: always wait at least 5 seconds before triggering or readback

  • Agent inputs: plain dictionaries validated by tool schemas

  • Agent outputs: compact JSON-safe values rather than full DataFrames

  • Hardware-changing agent tools: operator approval, strict bounds, logging, and cleanup