Skip to content

Python API Reference

The Device class is Splice's programmatic, no-database API. It wraps the same NETCONF discovery, catalog-building, and polling core used by the CLI, but keeps all state in memory — no PostgreSQL, no config files required. This makes it ideal for scripting, notebooks, tests, and AI-agent workflows.

Construct a Device against real hardware (Device(host, username=..., password=...)) or a hardware-free simulator (Device("sim://openroadm")), then discover paths, read telemetry, push config, or start a gNMI server — all from Python.

Usage

from splice.api.device import Device

# Hardware-free: run the full pipeline against a simulated OpenROADM device.
with Device("sim://openroadm") as dev:
    dev.discover()            # NETCONF capabilities → YANG modules
    dev.fetch_yang()          # download module sources
    dev.build_catalog()       # build the in-memory path catalog

    for path in dev.find_paths("optical-power"):
        print(path, dev.get(path))

The context manager (with ...) opens the NETCONF session on entry and closes it on exit. See the full method reference below.

splice.api.device.Device

Programmatic Python API for a single NETCONF device.

Provides connect/discover, get, set, subscribe, and optional gRPC serving -- all without requiring a database or YAML config files.

Quick start (full pipeline)::

device = Device("10.0.0.1", username="admin", password="secret")
device.connect()
data = device.get("/interfaces/interface")
device.close()

Granular steps (skip expensive discovery)::

device = Device("10.0.0.1", username="admin", password="secret",
                yang_dir="/path/to/cached/yang")
device.connect_netconf()          # SSH only
device.build_catalog()            # reuse existing YANG files
data = device.get("/interfaces/interface")

Context-manager::

with Device("10.0.0.1", username="admin", password="secret") as dev:
    data = dev.get("/interfaces/interface")

connect(*, skip_discover=False, skip_fetch_yang=False, skip_catalog=False)

Connect to the NETCONF device and run the discovery pipeline.

By default runs all steps:

  1. Establish NETCONF session
  2. Discover YANG modules
  3. Fetch YANG schemas to disk
  4. Build in-memory catalog
  5. Initialize default polling jobs
  6. Optionally create Kafka publisher

Use keyword arguments to skip expensive steps when reusing cached data (e.g. a pre-populated yang_dir):

  • skip_discover=True -- skip module discovery (step 2). Requires either a previous discover() call or skip_fetch_yang=True as well.
  • skip_fetch_yang=True -- skip YANG schema fetch (step 3). Requires yang_dir to already contain .yang files.
  • skip_catalog=True -- skip catalog build (steps 4-5). You must call build_catalog() later before using get(), set(), or subscribe().

connect_netconf()

Establish the NETCONF SSH session only (step 1).

Creates NetconfClient (or SimulatedNetconfClient when in simulation mode) and NetconfClientPool. Does not run any discovery. After this call, low-level pool methods are available but get()/set()/subscribe() will not work until build_catalog() has been called.

discover()

Discover YANG modules from the device (step 2).

Returns the list of discovered module dicts and stores them internally for fetch_yang() to use.

Requires connect_netconf() to have been called first.

fetch_yang(throttle_sec=0.3)

Fetch YANG schemas from the device to yang_dir (step 3).

Skips files that already exist on disk, so this is safe to call repeatedly. Returns the number of schemas written/found.

Requires discover() to have been called first (or modules to be set via _discovered_modules).

build_catalog()

Build in-memory catalog from YANG files on disk (step 4).

Reads all .yang files from yang_dir, builds the catalog, alias map, type map, and list keys, then initializes default polling jobs.

This method does not require a NETCONF connection -- it only needs YANG files on disk. This makes it possible to:

  • Pre-populate yang_dir and skip discovery entirely
  • Call connect(skip_discover=True, skip_fetch_yang=True) followed by build_catalog() for instant startup

close()

Tear down all resources.

get(path)

Read data from the device via a direct NETCONF get.

Accepts canonical, alias, or keyed paths. Returns a list of update dicts: [{"path": ..., "value": ..., "timestamp_ns": ...}, ...]

set(path, value, operation='merge', datastore='auto', use_lock=False)

Write a configuration value via NETCONF edit-config.

Returns a dict with path and operation on success.

subscribe(*paths, interval=30, batch=True)

Subscribe to periodic polling updates for the given paths.

Returns a Subscription iterator. Each iteration yields the full snapshot of updates received in the latest poll cycle.

serve(port=9339, bind='0.0.0.0', heartbeat_sec=10, set_enabled=True, blocking=False)

Start a gRPC gNMI server backed by this device's cache.

If blocking=True the call blocks until the server is stopped.

find_paths(pattern)

Return catalog paths matching a glob pattern.

search_paths(query, top_k=10)

Search catalog paths using natural language semantic search.

Returns list of (path, score) tuples ranked by relevance. Ranking is hybrid: BM25 keyword match (always available) blended with embedding similarity when sentence-transformers or the openai package is installed.

Example::

results = device.search_paths("Q-factor optical quality")
for path, score in results:
    print(f"{score:.3f}  {path}")

doctor()

Run diagnostic checks on the device connection.

Returns dict with: - connected: bool - circuit_breaker: dict (state, failures, cooldown) - catalog_paths: int (number of discovered paths) - yang_modules: int - capabilities: list[str]

export_catalog(format='json')

Export the device catalog.

Parameters:

Name Type Description Default
format str

"json" returns JSON string, "dict" returns list of dicts

 'json'