The SynapseDevice gRPC service defines the control plane for a Synapse device. Clients use these RPCs to inspect and operate a device. Data produced by a running signal chain is exposed separately through Taps.

See the SynapseDevice service definition for the canonical request and response types.

Info

Info returns the current identity and state of a Synapse device.

rpc Info(google.protobuf.Empty) returns (DeviceInfo)

The headstage implementation returns kOk for a healthy device. If the device is in its error state, the response reports kUndefinedError and includes the message Device in error state.

import synapse as syn

device = syn.Device("192.168.42.1")
info = device.info()

if info is None:
    raise RuntimeError("Unable to read device information")

print(info.name)
print(info.status.state)
print(info.peripherals)

Configure

Configure validates and installs a signal chain on the device.

rpc Configure(DeviceConfiguration) returns (Status)

A DeviceConfiguration contains:

• nodes: the configuration for every node in the signal chain
• connections: directed connections from each source node ID to a destination node ID

See JSON Configuration for instructions and complete examples.

The device validates node support, peripheral IDs, node parameters, connection compatibility, and available resources before accepting the configuration. You can inspect the returned Status.code and Status.message to determine whether configuration succeeded.

import synapse as syn

channels = [
    syn.Channel(id=channel_id, electrode_id=channel_id, reference_id=520)
    for channel_id in range(32)
]
broadband = syn.BroadbandSource(
    peripheral_id=1000,
    bit_width=12,
    sample_rate_hz=10000,
    gain=1.0,
    signal=syn.SignalConfig(
        electrode=syn.ElectrodeConfig(
            channels=channels,
            low_cutoff_hz=57.0,
            high_cutoff_hz=4365.0,
        )
    ),
)

config = syn.Config()
config.add_node(broadband)

device = syn.Device("192.168.42.1")
status = device.configure_with_status(config)
if status is None or status.code != 0:
    raise RuntimeError(status.message if status else "Configure RPC failed")

Start

Start starts the signal chain previously installed by Configure. See Device Lifecycle for the complete state model.

rpc Start(google.protobuf.Empty) returns (Status)

The headstage implementation returns:

• kOk when the device starts.
• kInvalidConfiguration when the device has not been configured.
• kFailedPrecondition when the device is already running or a configured peripheral cannot be found.
• kInternalError when the device cannot start.

import synapse as syn

device = syn.Device("192.168.42.1")
status = device.start_with_status()

if status is None or status.code != 0:
    raise RuntimeError(status.message if status else "Start RPC failed")

Stop

Stop stops the running signal chain. See Device Lifecycle for the complete state model.

rpc Stop(google.protobuf.Empty) returns (Status)

The headstage implementation returns kOk when the device stops, kFailedPrecondition when the device is not running, and kInternalError when the device cannot stop.

import synapse as syn

device = syn.Device("192.168.42.1")
status = device.stop_with_status()

if status is None or status.code != 0:
    raise RuntimeError(status.message if status else "Stop RPC failed")

Query

Query runs a diagnostic or metadata operation and returns one response.

rpc Query(QueryRequest) returns (QueryResponse)

Only one Query or StreamQuery can execute at a time.

The headstage implementation handles:

• impedance_query: measure the magnitude and phase for selected electrodes on a peripheral.
• self_test_query: run the self-test exposed by a peripheral.
• list_taps_query: list the Taps currently exposed by the configured signal chain.
• get_settings_query: return current device settings and the device-defined settings schema.

The response contains a Status and the response message corresponding to the selected query. A concurrent Query or StreamQuery returns gRPC RESOURCE_EXHAUSTED.

import synapse as syn
from synapse.api.query_pb2 import QueryRequest

device = syn.Device("192.168.42.1")
request = QueryRequest(
    query_type=QueryRequest.kListTaps,
    list_taps_query={},
)
response = device.query(request)

if response is None or response.status.code != 0:
    raise RuntimeError(response.status.message if response else "Query RPC failed")

for tap in response.list_taps_response.taps:
    print(tap)

Read device settings

Settings are read through Query using get_settings_query.

The resulting GetSettingsResponse contains the current DeviceSettings.values and a schema describing every accepted key, value type, default, description, and allowed value.

import synapse as syn
from synapse.client import settings

device = syn.Device("192.168.42.1")

print(settings.get_available_settings(device))
print(settings.get_all_settings(device))

StreamQuery

StreamQuery runs a diagnostic operation and streams results as they become available.

rpc StreamQuery(StreamQueryRequest) returns (stream StreamQueryResponse)

Only one Query or StreamQuery can execute at a time.

The headstage implementation supports streaming impedance and self-test queries. Each response includes a status code, message, timestamp, and the impedance or self-test result. Unsupported query types return gRPC INVALID_ARGUMENT.

import synapse as syn
from synapse.api.query_pb2 import QueryRequest, StreamQueryRequest

device = syn.Device("192.168.42.1")
request = StreamQueryRequest(
    request=QueryRequest(
        query_type=QueryRequest.kImpedance,
        impedance_query={
            "peripheral_id": 100,
            "electrode_ids": [0, 1, 2, 3],
        },
    )
)

for response in device.stream_query(request):
    if response.code != 0:
        raise RuntimeError(response.message)
    for measurement in response.impedance.measurements:
        print(measurement.electrode_id, measurement.magnitude, measurement.phase)

DeployApp

DeployApp uploads and installs a packaged Synapse App.

rpc DeployApp(stream AppPackageChunk) returns (stream AppDeployResponse)

The client first sends PackageMetadata, including the package name, version, size, and SHA-256 checksum, followed by chunks of package data. The device streams informational and status responses while it validates and installs the package. A failed installation returns a gRPC INTERNAL error.

The Python client's deployment helper calculates the package metadata, streams the package, and processes the deployment responses:

from synapse.cli.deploy import deploy_package

deployed = deploy_package(
    "192.168.42.1",
    "/path/to/synapse-example-app_0.1.0_arm64.deb",
)
if not deployed:
    raise RuntimeError("DeployApp RPC failed")

ListApps

ListApps returns the Synapse Apps installed on the device.

rpc ListApps(ListAppsRequest) returns (ListAppsResponse)

Each AppInfo in the response contains the installed app's name and version. ListAppsRequest has no fields.

import synapse as syn

device = syn.Device("192.168.42.1")
response = device.list_apps()

if response is None:
    raise RuntimeError("ListApps RPC failed")

for app in response.apps:
    print(app.name, app.version)

ListFiles

ListFiles returns the files exposed by the device. See Filesystem for the device filesystem mental model.

rpc ListFiles(google.protobuf.Empty) returns (ListFilesResponse)

Each file entry contains its name, size, creation and modification timestamps, and type.

The Python client exposes this RPC through the generated stub:

import synapse as syn
from google.protobuf.empty_pb2 import Empty

device = syn.Device("192.168.42.1")
response = device.rpc.ListFiles(Empty())

for file in response.files:
    print(file.name, file.size, file.type)

WriteFile

WriteFile writes the supplied bytes to a named file on the device. See Filesystem for the device filesystem mental model, reserved folders, and SciFi 2 USB access.

rpc WriteFile(WriteFileRequest) returns (WriteFileResponse)

The request contains name and data. The response returns the file name and number of bytes written.

import synapse as syn
from synapse.api.files_pb2 import WriteFileRequest

device = syn.Device("192.168.42.1")
response = device.rpc.WriteFile(
    WriteFileRequest(name="/data/example.bin", data=b"example data")
)
print(response.name, response.bytes_written)

ReadFile

ReadFile downloads a named file as a server stream. See Filesystem for the device filesystem mental model, reserved folders, and SciFi 2 USB access.

rpc ReadFile(ReadFileRequest) returns (stream ReadFileResponse)

Each response contains a chunk of file data, its starting offset, and the total file length. Use the offset and total length to assemble and verify the downloaded file.

import synapse as syn
from synapse.api.files_pb2 import ReadFileRequest

device = syn.Device("192.168.42.1")

with open("example.bin", "wb") as output:
    for response in device.rpc.ReadFile(
        ReadFileRequest(name="/data/example.bin")
    ):
        output.seek(response.start_offset)
        output.write(response.data)

DeleteFile

DeleteFile deletes a named file from the device. See Filesystem for the device filesystem mental model, reserved folders, and SciFi 2 USB access.

rpc DeleteFile(DeleteFileRequest) returns (DeleteFileResponse)

The response identifies the file and reports the resulting StatusCode.

import synapse as syn
from synapse.api.files_pb2 import DeleteFileRequest
from synapse.api.status_pb2 import StatusCode

device = syn.Device("192.168.42.1")
response = device.rpc.DeleteFile(
    DeleteFileRequest(name="/data/example.bin")
)

if response.status_code != StatusCode.kOk:
    raise RuntimeError(f"DeleteFile failed: {response.status_code}")

GetLogs

GetLogs returns retained log entries matching a time range and minimum severity.

rpc GetLogs(LogQueryRequest) returns (LogQueryResponse)

Set either:

• since_ms to request entries from a recent interval, or
• start_time_ns and end_time_ns to request an explicit time range.

When since_ms is set, the headstage ignores the explicit start and end times. If min_level is unspecified, the headstage defaults to LOG_LEVEL_INFO. Each returned entry includes its timestamp, severity, source, and message.

import synapse as syn

device = syn.Device("192.168.42.1")
response = device.get_logs(log_level="WARNING", since_ms=60_000)

if response is None:
    raise RuntimeError("GetLogs RPC failed")

for entry in response.entries:
    print(entry.timestamp_ns, entry.level, entry.source, entry.message)

TailLogs

TailLogs streams new log entries at or above a minimum severity.

rpc TailLogs(TailLogsRequest) returns (stream LogEntry)

The stream remains active until the client cancels it or the server's stream duration limit is reached. The headstage limits the number of simultaneous log subscribers and returns gRPC RESOURCE_EXHAUSTED when no subscription slot is available.

import synapse as syn

device = syn.Device("192.168.42.1")

for entry in device.tail_logs(log_level="INFO"):
    print(entry.timestamp_ns, entry.level, entry.source, entry.message)

UpdateDeviceSettings

UpdateDeviceSettings validates, persists, and applies one or more settings. See the Settings concept for schema behavior. Device-specific settings are listed in the Product Details for SciFi and SciFi 2.

rpc UpdateDeviceSettings(UpdateDeviceSettingsRequest)
    returns (UpdateDeviceSettingsResponse)

Only keys present in settings.values are updated. The response includes a Status and the resulting settings. The accepted keys are device-defined; clients should read the schema before constructing an update.

Current SciFi settings include the user-facing device name, timestamp source, FPGA clock frequency, and per-network WiFi MAC address policy.

import synapse as syn
from synapse.client import settings

device = syn.Device("192.168.42.1")
device.stop()

updated_name = settings.set_setting(device, "name", "recording-room-a")
print(updated_name)