Gateway Sessions

Web gateway instrument sessions and streaming poll loop

Instrument session layer: one worker thread per instrument.

All SCPI I/O for a session happens on its single worker thread (FIFO job queue), so compound operations are atomic and the non-thread-safe connection is never shared across threads (AUDIT.md C2). This module is FastAPI-free: the API layer adapts the returned concurrent.futures.Future to asyncio.

SessionError

Bases: RuntimeError

Session is not in a state that can accept jobs (maps to HTTP 409).

InstrumentSession

InstrumentSession(label: str, scope: Oscilloscope, mock: bool, address: Optional[str], poll_interval: float)

Source code in scpi_control/server/sessions.py

def __init__(self, label: str, scope: Oscilloscope, mock: bool, address: Optional[str], poll_interval: float):
    self.id = uuid.uuid4().hex[:8]
    self.label = label
    self.mock = mock
    self.address = address
    self.state = "connecting"
    self.idn = ""
    self.model = ""
    self.dialect = ""
    self.num_channels = 0
    self.error_detail: Optional[str] = None
    self._scope = scope
    self._poll_interval = poll_interval
    self._queue: "queue.Queue" = queue.Queue()
    self._closed = threading.Event()
    self._thread = threading.Thread(target=self._worker, name="scpi-session-{0}".format(self.id), daemon=True)
    self._subscribers: List[Callable[[Dict[str, Any]], None]] = []
    self._subscribers_lock = threading.Lock()
    self.measurements: List[Tuple[int, str]] = []
    self._poll_count = 0
    # Server-owned analysis config. Request coroutines swap these whole
    # dicts atomically (never mutate in place); the worker thread only reads.
    self.spectrum_config: Dict[str, Any] = {"enabled": False, "channel": 1, "window": "hanning", "db": True}
    self.filters: Dict[int, Dict[str, Any]] = {n: {"source": 1, "kind": "lowpass", "cutoff_low": None, "cutoff_high": None, "order": 5, "enabled": False} for n in (1, 2)}
    self.active_reference: Optional[Dict[str, Any]] = None  # {"name", "channel", "data": {"time","voltage",...}}
    self._shown: set = set()  # trace keys (M1/M2/F1/F2/SPEC) live on subscribers' canvases; worker-thread-only
    self.recorder = TrendRecorder()  # internally locked: worker appends, request threads control/read
    self.owner = ""
    self.owner_last_active = time.monotonic()
    # Live stream watchers, keyed by identity (not by "is this the
    # owner"): a Counter rather than a set because the same identity may
    # open two tabs and close one, so a plain membership test would drop
    # the identity on the first close while the second tab is still live.
    self._watchers: "Counter[str]" = Counter()

touch

touch() -> None

Record owner activity; feeds the abandoned-session claim rule.

Source code in scpi_control/server/sessions.py

def touch(self) -> None:
    """Record owner activity; feeds the abandoned-session claim rule."""
    self.owner_last_active = time.monotonic()

owner_watching

owner_watching() -> bool

True while at least one live stream opened by the current owner is connected.

Evaluated at check time against the current owner and the live per-identity watcher counts (see mark_owner_watching) -- never a snapshot taken when some stream connected. Ownership can change mid-stream via claim() or an explicit handoff, so "is the owner watching" must always be asked fresh: a snapshot fails open (a non-owner who claims the session while already watching would not be protected) and also sticks stale (a former owner's lingering socket would keep blocking claims after handoff).

A watching owner is never idle even though only writes touch() owner_last_active -- the claim rule refuses outright while this is true, regardless of the idle threshold.

Source code in scpi_control/server/sessions.py

def owner_watching(self) -> bool:
    """True while at least one live stream opened by the current owner is connected.

    Evaluated at check time against the *current* owner and the live
    per-identity watcher counts (see mark_owner_watching) -- never a
    snapshot taken when some stream connected. Ownership can change
    mid-stream via claim() or an explicit handoff, so "is the owner
    watching" must always be asked fresh: a snapshot fails open (a
    non-owner who claims the session while already watching would not
    be protected) and also sticks stale (a former owner's lingering
    socket would keep blocking claims after handoff).

    A watching owner is never idle even though only writes touch()
    owner_last_active -- the claim rule refuses outright while this is
    true, regardless of the idle threshold.
    """
    return bool(self.owner) and self._watchers[self.owner] > 0

mark_owner_watching

mark_owner_watching(identity: str) -> Callable[[], None]

Register that identity opened the live stream; returns an unmark callback.

Tracks the watching identity itself, not whether it happened to be the owner at connect time -- owner_watching() does that comparison later, at check time, against whoever owns the session then. The returned callback decrements exactly once no matter how many times it is called, so it is safe to invoke unconditionally from a finally block on any disconnect path (clean close, error, or cancellation).

Source code in scpi_control/server/sessions.py

def mark_owner_watching(self, identity: str) -> Callable[[], None]:
    """Register that ``identity`` opened the live stream; returns an unmark callback.

    Tracks the watching identity itself, not whether it happened to be
    the owner at connect time -- owner_watching() does that comparison
    later, at check time, against whoever owns the session *then*. The
    returned callback decrements exactly once no matter how many times
    it is called, so it is safe to invoke unconditionally from a
    ``finally`` block on any disconnect path (clean close, error, or
    cancellation).
    """
    self._watchers[identity] += 1
    released = False

    def unmark() -> None:
        nonlocal released
        if not released:
            released = True
            self._watchers[identity] -= 1
            if self._watchers[identity] <= 0:
                del self._watchers[identity]

    return unmark

SessionManager

SessionManager(allowed_ports: Optional[frozenset] = None, max_sessions: int = 8)

Registry of live sessions. create() connects before registering.

Source code in scpi_control/server/sessions.py

def __init__(self, allowed_ports: Optional[frozenset] = None, max_sessions: int = 8) -> None:
    if max_sessions < 1:
        # A cap below 1 doesn't disable the gateway -- it silently makes
        # every create() a 409 "session limit reached (0)", which reads
        # like a bug rather than a configuration mistake. Reject eagerly
        # here (the library boundary) rather than relying solely on the
        # CLI to catch it, since SessionManager is also constructed
        # directly by embedders and tests.
        raise ValueError("max_sessions must be at least 1 (got {0})".format(max_sessions))
    self._sessions: Dict[str, InstrumentSession] = {}
    self._lock = threading.Lock()
    self.allowed_ports = allowed_ports
    self.max_sessions = max_sessions
    # Slots claimed by an in-flight create() that hasn't registered yet.
    # Counted alongside len(self._sessions) under the same lock as the
    # cap check so the check and the reservation are one atomic step --
    # see create() for why that matters.
    self._pending = 0