Per-instrument stream registry#

Overview#

Each instrument carries a single source of truth for streaming data declarations: one record per group that the ESS filewriter would write into a NeXus file. NeXus-rich instruments auto-generate the bulk of the registry from a geometry artifact, with code-side overrides for the few names that need to differ.

Two routing-layer consumers shaped the design:

  • Dynamic detector transforms drive entries in a depends_on chain from live f144 streams.

  • Chopper workflows drive per-chopper geometry assembly from live f144 setpoint streams, including synthesised streams that are emitted in-process rather than read from Kafka.

Both need a stream → Sciline-key routing layer with per-consumer scoping.

Shape#

Stream and F144Stream records#

Stream is a frozen dataclass for any streaming group; F144Stream subclasses it for the f144 writer module and adds units. The base carries the NeXus identity (nexus_path, nx_class) and the wire identity (topic, source) — the facts the filewriter collocates as group attrs at a specific HDF5 path.

@dataclass(frozen=True, slots=True, kw_only=True)
class Stream:
    writer_module: str
    nexus_path: str | None = None
    topic: str | None = None
    source: str | None = None
    nx_class: str = ''


@dataclass(frozen=True, slots=True, kw_only=True)
class F144Stream(Stream):
    units: str | None = None
    writer_module: str = 'f144'
    nx_class: str = 'NXlog'

Instrument.streams: dict[str, Stream] holds streams of all writer modules, keyed by the instrument-facing stream name. The name is the routing handle for every non-kafka-boundary consumer; topic / source are read only by the Kafka boundary (streams.py per instrument, message_adapter, kafka source). Per-kind views are filters (instrument.f144_streams returns dict[str, F144Stream]).

nexus_path is on the base because consumers that splice live values into a NeXus structure — transform-chain patching, chopper DataGroup assembly — dispatch on it. It is None only for synthesised streams.

Synthesised streams#

A synthesised stream (no NeXus row, emitted in-process by a synthesiser) is Stream(topic=None, source=None, nexus_path=None, ...). The predicate “synthesised” is stream.topic is None. __post_init__ enforces that topic and source are both None or both set.

Consumers are oblivious to the distinction: workflow harnesses, timeseries UI, and binding resolution all index into streams by name. The only place that filters by topic is None is the Kafka-boundary StreamLUT builder, which has no entry to emit for a stream that never arrives over Kafka.

LogContextBinding#

The runtime overlay — wiring f144 streams into Sciline workflows — is a separate list referencing streams by name:

@dataclass(frozen=True, slots=True, kw_only=True)
class LogContextBinding:
    stream_name: str            # references Instrument.streams[stream_name]
    workflow_key: Any           # Sciline key type (e.g. InstrumentAngle[SampleRun])
    dependent_sources: frozenset[str]  # spec source_names that pick up this binding

Instrument.log_context_bindings: list[LogContextBinding]. A binding declares only “stream X feeds Sciline key K, scoped to dependents D”; it does not mirror units or nexus_path, which live on the stream record it references.

Bindings live separately from the record because dependent_sources and workflow_key belong to a binding, not to a stream: a single stream can participate in multiple bindings with different scoping (e.g., feeding two Sciline workflows triggered by different sources). A field on the stream record would force one global answer.

Factories derive StreamProcessorWorkflow.context_keys by calling instrument.get_context_keys(source_name), which filters the binding list by dependent_sources. The aux-source routing mechanism iterates log_context_bindings once and filters by dependent_sources & spec.source_names.

The same binding type covers both known consumers:

  • Dynamic-transforms: the detector spec picks up the live readback as an aux input, keyed by a TransformValueLog subclass, and patches it into the transform chain.

  • Chopper-workflow: a chopper workflow spec is triggered by a synthetic chopper_cascade_in_phase stream; every chopper f144 stream declares itself as an aux input to that trigger and feeds a typed Sciline key.

Dispatch between the two is driven by the workflow_key type and the Sciline graph downstream, not by separate binding kinds.

Display info: source_metadata, not Stream#

title and description live in Instrument.source_metadata: dict[str, SourceMetadata], the side-dict that already holds display info for detectors and monitors. Stream is data plumbing; display labels have a different audience and lifecycle, and merging them onto the record would pull human-readable strings into every record-equality test, override path, and parser fixture.

Codegen seam#

For instruments with O(100) NXlogs the registry is bootstrapped offline by the CLI generator python -m ess.livedata.nexus_helpers <geometry.nxs> --generate. The generator emits a checked-in streams_parsed.py containing a path-keyed dict literal:

# config/instruments/<inst>/streams_parsed.py  -- auto-generated
PARSED_STREAMS: dict[str, F144Stream] = {
    '/entry/instrument/sample/temperature': F144Stream(
        nexus_path='/entry/instrument/sample/temperature',
        source='SAMPLE-Tmp:RBV', topic='<inst>_sample', units='K',
    ),
    ...
}

The hand-edited specs.py composes the final name-keyed registry via name_streams:

from .streams_parsed import PARSED_STREAMS

streams = name_streams(
    PARSED_STREAMS,
    rename={
        '/entry/instrument/detector_carriage/value': 'detector_carriage_value',
    },
)

name_streams auto-suggests an instrument-facing name from the tail of each NeXus path (generic container groups like entry / instrument / sample are filtered out) and accepts a rename= map keyed by nexus_path. It raises if a rename key matches no parsed entry, or if the resulting names collide.

Codegen rather than runtime parsing: backend and dashboard both import the per-instrument config module at startup, and the dashboard has no file-access path. Parsing at runtime would force a Pooch download with cold-start dependencies on cache state and network. Codegen keeps both processes file-free at startup and makes the parser output reviewable as code. Stream is a frozen dataclass that round-trips through any serialisation layer, so moving to dynamic distribution later is a local change to who populates Instrument.streams.

For instruments with few streams (today: dummy), the registry is a hand-written dict[str, F144Stream] literal and the codegen path is not used.

Construction-time invariants#

Misconfiguration in the registry, bindings, or overrides crashes the process at startup, not at pytest time:

  • Instrument.__post_init__ raises if any log_context_bindings[i].stream_name is missing from Instrument.streams.

  • name_streams raises if a rename key matches no parsed entry, or if names collide.

  • End of Instrument.load_factories() raises if any log_context_bindings[i].dependent_sources element does not match the source_names of some registered spec. This runs after factory loading because spec registration populates source_names.

The dashboard and backend both import Instrument at startup, so any typo in specs.py (wrong stream_name, missing rename target, unknown dependent source) crashes the process before traffic starts.

Consumer-specific reachability checks (e.g., “every binding’s nexus_path lies on a depends_on chain”) are the consuming module’s responsibility, not the registry’s. The registry validates registry invariants only.

Open questions#

  1. Generalising overlay beyond f144. When ev44/tdct streams gain per-stream runtime config (rate hints, decoder choices), each gets its own Stream subclass. A consumer that needs a binding shape other than the f144 triple (stream_name, workflow_key, dependent_sources) gets its own binding type — LogContextBinding unifies f144 Sciline-key consumers, not all writer modules.

  2. Codegen drift protection. streams_parsed.py is regenerated by hand; staleness against the latest geometry file is not caught automatically. An earlier drift test was dropped because it skipped in CI (the geometry files are not in Pooch) and so gave false confidence without catching the realistic failure mode. A real check needs a stable geometry-file input pinned in Pooch first.