Building an SCPI Capability Registry with Runtime Discovery

Two units of the same oscilloscope model can differ by an installed bandwidth license, a memory-depth option, or a firmware revision that adds a subsystem — so control code that hard-codes “this model has segmented memory” is wrong on the bench next to it. This guide, part of SCPI Command-Set Standardization, builds a capability registry that probes each instrument at connect time via *IDN?, *OPT?, and targeted feature queries, then answers require("segmented_memory") from a cache instead of a per-model guess.

Discovery Scope and the Per-Unit Assumption Trap

The problem this solves is narrow and distinct from mapping vendor verbs onto a canonical grammar. The sibling guide on mapping vendor commands to a canonical SCPI layer answers “how do I phrase this command for this box”; capability discovery answers the prior question, “is this box able to do the thing at all”. A canonical layer that emits TRIGger:SEGMented ON to an instrument whose segmented-acquisition option is not licensed produces a -241,"Hardware missing" error, not a syntax error, and no amount of command normalization prevents it. The only reliable answer is to ask the hardware.

The assumptions are tight. Every instrument speaks SCPI over a message-based session allocated through a configured VISA Resource Manager, answers *IDN?, and populates a SCPI error queue you can read with :SYSTem:ERRor?. Discovery runs once per connection, before any acquisition, and its output is a typed, immutable record keyed by (model, firmware, options). Application code never branches on a model string directly; it calls require(cap) and receives either a green light or a precise, actionable error. Instruments that predate *OPT? or answer it with junk are handled by feature probes — issuing a candidate query and watching the error queue rather than trusting a capability table.

Capability as a Probed Set, Not a Static Table

A capability set is the union of three probe sources, computed once and then cached:

 $C = probe (*IDN?, *OPT?, {f_{1}, f_{2}, \dots, f_{n}}), require (c) = {ok raise c \in C c \in / C$

*IDN? yields the (manufacturer, model, serial, firmware) tuple; *OPT? yields a comma-delimited option list; and each feature probe f_i is a boolean function that issues a candidate command and inspects the error queue. The discovery cost is n + 2 round trips paid once. Every subsequent require(c) is a set-membership test — O(1) after caching — so gating a hot acquisition loop on capabilities adds no per-command I/O. This asymmetry is the whole point: pay a bounded discovery cost at connect, then never touch the wire to answer a capability question again.

Determinism matters as much as speed. A feature probe must be side-effect free — querying a limit, reading a mode, or issuing a command the instrument rejects cleanly, never one that arms a trigger or changes an output. The safe pattern is probe-then-restore: read the current value, attempt the feature query, classify the error-queue delta, and confirm no state changed. Where SYSTem:CAPability? is implemented (LXI-class instruments expose it as an IEEE 488.2 capability string), it short-circuits several probes with one query, but it is advisory: many instruments return only interface-level capabilities there, not option-gated features, so it supplements *OPT? rather than replacing it.

Three probes run once per connection; their union is frozen into a keyed record that answers every later require() from cache.

A Discovering CapabilityRegistry with a require() Guard

The implementation probes on discover(), folds the results into a frozen Capabilities record, caches it under the (model, firmware, options) key, and exposes require() as the single gate application code calls. Feature probes are declared as data, so adding a new capability is a one-line entry, not a code change. Error-queue classification reuses the categorization discipline detailed in Error Code Categorization: an execution error on a probe means the feature is absent, whereas an I/O timeout means the probe is inconclusive and must not be cached as a negative.

from __future__ import annotations

import logging
import re
from dataclasses import dataclass, field
from typing import Callable, Optional

import pyvisa

logger = logging.getLogger(__name__)

# "-241,\"Hardware missing\"" -> the numeric code is what we classify on.
_ERR_RE = re.compile(r"^\s*(-?\d+)\s*,")
# Codes that mean "the instrument understood you but cannot do it" => feature absent.
_ABSENT_CODES = frozenset({-241, -221, -224, -200, -100, -113})


class CapabilityError(RuntimeError):
    """Raised by require() when a demanded capability is not present."""


@dataclass(frozen=True)
class Capabilities:
    """Immutable, per-unit capability record. Keyed by (model, firmware, options)."""

    manufacturer: str
    model: str
    firmware: str
    serial: str
    options: frozenset[str]
    features: frozenset[str]

    @property
    def key(self) -> tuple[str, str, frozenset[str]]:
        return (self.model, self.firmware, self.options)

    def has(self, cap: str) -> bool:
        return cap in self.features or cap in self.options


@dataclass(frozen=True)
class FeatureProbe:
    """A side-effect-free query whose success proves a capability exists."""

    name: str
    query: str                      # a read-only query, never a state change
    option_hint: Optional[str] = None   # if this *OPT? token is present, skip probing


class CapabilityRegistry:
    """Discovers and caches what each instrument can actually do."""

    def __init__(self, probes: list[FeatureProbe]) -> None:
        self._probes = probes
        self._cache: dict[str, Capabilities] = {}

    def discover(self, session: pyvisa.resources.MessageBasedResource) -> Capabilities:
        """Probe *IDN?/*OPT?/features once and cache a typed record.

        Re-probes and replaces the cache entry whenever the (model, firmware,
        options) key changes, so a firmware flash or added license invalidates
        a stale record on the next connect.
        """
        idn = self._query(session, "*IDN?")
        manufacturer, model, serial, firmware = self._parse_idn(idn)
        options = self._parse_options(self._query(session, "*OPT?"))

        cache_key = f"{model}|{firmware}|{','.join(sorted(options))}"
        cached = self._cache.get(cache_key)
        if cached is not None:
            logger.debug("Capability cache hit for %s", cache_key)
            return cached

        features = self._probe_features(session, options)
        caps = Capabilities(
            manufacturer=manufacturer,
            model=model,
            firmware=firmware,
            serial=serial,
            options=frozenset(options),
            features=frozenset(features),
        )
        self._cache[cache_key] = caps
        logger.info("Discovered %s fw=%s opts=%s features=%s",
                    model, firmware, sorted(options), sorted(features))
        return caps

    def invalidate(self, session: pyvisa.resources.MessageBasedResource) -> None:
        """Drop the cached record for the unit on this session (call on reconnect)."""
        model, fw = self._parse_idn(self._query(session, "*IDN?"))[1::2]
        opts = ",".join(sorted(self._parse_options(self._query(session, "*OPT?"))))
        self._cache.pop(f"{model}|{fw}|{opts}", None)

    @staticmethod
    def require(caps: Capabilities, cap: str) -> None:
        """Guard: raise a precise error if the instrument lacks `cap`."""
        if not caps.has(cap):
            raise CapabilityError(
                f"{caps.model} (fw {caps.firmware}) lacks capability {cap!r}; "
                f"present options={sorted(caps.options)}, "
                f"features={sorted(caps.features)}"
            )

    def _probe_features(
        self, session: pyvisa.resources.MessageBasedResource, options: set[str]
    ) -> set[str]:
        present: set[str] = set()
        self._drain_errors(session)  # start from a clean queue
        for probe in self._probes:
            if probe.option_hint and probe.option_hint in options:
                present.add(probe.name)         # licensed => trust *OPT?, skip round trip
                continue
            try:
                session.query(probe.query)
            except pyvisa.errors.VisaIOError:
                logger.warning("Probe %s inconclusive (I/O); not caching", probe.name)
                continue                          # timeout != absent; leave undetermined
            if self._probe_ok(session):
                present.add(probe.name)
        return present

    def _probe_ok(self, session: pyvisa.resources.MessageBasedResource) -> bool:
        """True only if the probe left no 'feature absent' code in the queue."""
        absent = False
        for code, _msg in self._drain_errors(session):
            if code in _ABSENT_CODES:
                absent = True
        return not absent

    def _drain_errors(
        self, session: pyvisa.resources.MessageBasedResource
    ) -> list[tuple[int, str]]:
        drained: list[tuple[int, str]] = []
        while True:
            raw = self._query(session, ":SYSTem:ERRor?")
            match = _ERR_RE.match(raw)
            code = int(match.group(1)) if match else 0
            if code == 0:
                break
            drained.append((code, raw))
        return drained

    @staticmethod
    def _query(session: pyvisa.resources.MessageBasedResource, cmd: str) -> str:
        return session.query(cmd).strip()

    @staticmethod
    def _parse_idn(idn: str) -> tuple[str, str, str, str]:
        parts = [p.strip() for p in idn.split(",")]
        parts += [""] * (4 - len(parts))
        return parts[0], parts[1], parts[2], parts[3]

    @staticmethod
    def _parse_options(opt: str) -> set[str]:
        # "0" and empty mean no options on most vendors.
        tokens = {t.strip() for t in opt.split(",") if t.strip() and t.strip() != "0"}
        return tokens

The registry deliberately sits above the transport and below orchestration, the same seam that Protocol Abstraction Layers occupy: a legacy instrument wrapped to look like SCPI is discovered through the identical discover() path, and its wrapper simply answers *OPT? with a synthesized option string.

Validating Discovery Against Divergent *OPT? Strings

The verification target is that identical model firmware but different licensing produces different, correct gates. Drive it with fakes that answer distinct *OPT? strings and assert the feature membership flips:

import pytest


class FakeSession:
    """Minimal message-based stand-in that scripts *IDN?/*OPT?/error replies."""

    def __init__(self, idn: str, opt: str, absent_queries: set[str]) -> None:
        self._idn, self._opt = idn, opt
        self._absent = absent_queries
        self._queue: list[str] = []

    def query(self, cmd: str) -> str:
        if cmd == "*IDN?":
            return self._idn
        if cmd == "*OPT?":
            return self._opt
        if cmd == ":SYSTem:ERRor?":
            return self._queue.pop(0) if self._queue else '0,"No error"'
        if cmd in self._absent:
            self._queue.append('-241,"Hardware missing"')
        return "0"


def _registry() -> CapabilityRegistry:
    return CapabilityRegistry([
        FeatureProbe("segmented_memory", ":ACQuire:SEGMented:COUNt?", option_hint="SGM"),
        FeatureProbe("bandwidth_1ghz", ":CHANnel1:BANDwidth? MAX"),
    ])


def test_licensed_unit_gains_feature() -> None:
    reg = _registry()
    caps = reg.discover(FakeSession("KEYSIGHT,DSOX,US1,7.30", "SGM", set()))
    reg.require(caps, "segmented_memory")          # licensed => passes


def test_unlicensed_unit_is_gated() -> None:
    reg = _registry()
    sess = FakeSession("KEYSIGHT,DSOX,US2,7.30", "0", {":ACQuire:SEGMented:COUNt?"})
    caps = reg.discover(sess)
    with pytest.raises(CapabilityError):
        reg.require(caps, "segmented_memory")      # same model, no option => raises

On live hardware, confirm three signals: discover() logs an *OPT? set matching the license label on the chassis; a probe for an absent feature leaves the error queue at depth zero after discovery (the drain worked); and a second discover() on the same session returns instantly with a cache-hit log line and issues no probe traffic, observable on a bus analyzer. Where discovery writes provenance into captured datasets, route the record through Metadata Injection Workflows so every measurement carries the firmware and option set it was taken under.

Failure Modes Specific to Capability Probing

The probe command itself is unsupported. On an older instrument, :ACQuire:SEGMented:COUNt? is not merely unlicensed — it is an undefined header, yielding -113 and error-queue noise that a naive drain mistakes for an unrelated fault later. Diagnosis: classify -113 as “feature absent” (it is in _ABSENT_CODES), and always drain the queue before the next probe so one probe’s rejection never contaminates the next.
*OPT? lists a license the hardware cannot honor. A software option is provisioned but the physical module was pulled, so *OPT? reports it present while the feature errors on use. Diagnosis: treat option_hint as a fast path only for classes where license implies hardware; for module-gated features, force the live probe regardless of *OPT? and let the error queue be authoritative.
Stale cache after a firmware or option change. A unit is re-flashed or re-licensed between runs, but the process cached the old record under the old key. Because the key includes firmware and the sorted option set, the new *IDN?/*OPT? produces a different key and misses the stale entry; call invalidate() on reconnect for belt-and-suspenders eviction of the prior key.
Probing a busy instrument mid-acquisition. Issuing *OPT? while a sweep runs can block past the session timeout or interleave with acquisition reads, corrupting both. Diagnosis: gate discovery to the connect phase before any :INITiate, and if a reconnect lands mid-run, defer discovery behind *OPC? so the in-flight operation completes before any probe touches the wire.

Capability discovery turns “which model is this” — a fragile proxy — into “what can this exact unit do”, asked once and cached. When require() is the only capability check in the codebase, an unlicensed option fails at connect with a legible message instead of a cryptic -241 three hours into an unattended run.

SCPI Command-Set Standardization — the parent area this discovery step feeds.
Mapping Vendor Commands to a Canonical SCPI Layer — phrase the command once discovery confirms the box can run it.
Categorizing SCPI Error Codes for Automated Recovery — the classification a feature probe depends on.
Structuring a PyVISA Resource Manager for Multi-Vendor Labs — allocate the sessions discovery probes over.
Metadata Injection Workflows — stamp captured data with the discovered firmware and option set.

References

IVI Foundation SCPI specifications — *IDN?, *OPT?, and common-command semantics.
PyVISA documentation — message-based query and timeout behaviour.

← Back to SCPI Command-Set Standardization