How a neutron-data message is identified at each layer of the reduction stack, and the points where that identifier is translated from one naming scheme to the next.
A reduction workflow is a sciline graph: its inputs are addressed by Python types (e.g. NeXusData[Incident, SampleRun]). Data, however, enters the system from Kafka identified by a (topic, source_name) pair, and a user configuring a workflow thinks in terms of a role — “the incident monitor used for normalization” — which may be filled by any of several physical streams.
Bridging those views means a message’s identifier passes through four naming schemes on its way to a provider. This page traces the layers of encapsulation the data flows through, shows which scheme each layer speaks, and pinpoints the three places a translation happens.
(topic, source_name) — raw Kafka identity
role — logical input a workflow consumes
canonical stream_name — the NeXus name from the instrument config
sciline.Key — the graph’s type-addressed input
Each box wraps the one below it. A message arrives at the top as raw Kafka bytes and descends until its value reaches a sciline provider. The colour of each identifier marks the scheme it is in.
get_stream_id(topic, source_name) → StreamId(kind, name) via the StreamLUT lookup table.StreamId; emit WorkflowData.data : { StreamId(name) : value }._filter_data_for_job splits each batch by stream.name:stream.name ∈ job.source_names → primary_data[stream_name]stream.name ∈ job.input_stream_names → aux_data[stream_name] (auxiliary and context payloads alike; demultiplexed at translation 3)job.missing_context(available)) holds the job until each required context stream has a value.input_streams / gating_streams : set[stream_name] — plain name sets, no mapping.add() forwards {**primary_data, **aux_data} unchanged.dynamic_keys : { stream_name → sciline.Key } — fed to accumulate()context_keys : { stream_name → sciline.Key } — fed to set_context()accumulate; which dict the name is in decides accumulate vs. set_context) data pathaccumulate() / set_context() / finalize() — addressed by sciline.Key.finalize() → { output_name : DataArray } → JobResult → da00 message → KafkaA workflow consumes three kinds of input. All three arrive on the data path keyed by stream_name and all three end up inside the wrapped StreamProcessor, but they differ in how their stream binding is chosen, how they are fed into the sciline graph, and whether they hold up the job.
| Input | What it is | Stream binding chosen by | Fed into the graph via | Gates? |
|---|---|---|---|---|
| Primary | The detector data being reduced | The job’s identity (source_name) |
accumulate() — summed over time |
no |
| Auxiliary (dynamic) |
Extra measured data, e.g. monitor counts for normalization | A user-selected role (translation 2) | accumulate() — summed over time |
no |
| Context | Slowly-varying parameters of the reduction: motion / geometry positions, sample temperature | A framework-routed ContextBinding (stream-named) |
set_context() — latest value, persistent |
yes |
Auxiliary vs. context. Both are non-primary, but they answer different questions. An auxiliary input is more data to reduce: it is accumulated, so each batch adds to a running total and an empty batch is harmless. A context input is a parameter of the reduction: it is injected with set_context() and is stateful — a value set in one batch persists into every later batch until overwritten, and a batch carrying no new value leaves the parameter unchanged. Because a run with an uninitialized parameter would crash or silently misattribute its output, the JobManager gates the job until each context stream has produced a value (ADR 0002); auxiliary inputs never gate.
ROI is a deliberate hybrid. It is declared as an AuxSources — a user-selectable input, so its stream binding is chosen at translation 2 like other auxiliary inputs — but the factory feeds it through set_context() rather than accumulate(), and it does not gate. “No ROI selected” is a safe default (an empty request), so there is no readiness to wait on (ADR 0002). It is not a ContextBinding.
The primary is the degenerate case. It is a logical slot too — named by its sciline key, NeXusData[NXdetector, Run] — but its stream binding is the job’s identity (source_name) rather than a user-selectable role, so translation 2 is the identity function and the factory keys it directly, without aux_source_names. Context inputs likewise carry no role: their wire name equals their stream name. Translation 2 therefore touches only auxiliary inputs.
The role scheme is the one a workflow author and a user reason in for auxiliary inputs. It is resolved to a canonical stream name once, in the instrument factory that constructs the workflow — before any data flows — so that the workflow object itself is built entirely in the stream-name scheme.
aux_source_names, the role → stream map that JobFactory computes via AuxSources.render (defaults overlaid with the user’s selection).dynamic_keys by canonical stream name directly, resolving each role inline:{ source_name: NeXusData[NXdetector, …], aux_source_names['incident_monitor']: NeXusData[Incident, …] }KeyError at job creation, visible in factory code. The role scheme never enters the workflow object.| # | Boundary | From → To | Performed by | Path |
|---|---|---|---|---|
| 1 | Kafka ingress | (topic, source_name) → stream_name | MessageAdapter.get_stream_id + StreamLUT |
data |
| 2 | Workflow construction | role → stream_name (building dynamic_keys) |
Instrument factory, inline aux_source_names[role] |
build-time |
| 3 | Workflow runtime | stream_name → sciline.Key | StreamProcessorWorkflow.accumulate (dict lookup) |
data |
Only one scheme is on the data path. Translation 2 happens at construction; once messages flow, the entire stack from WorkflowData down to the StreamProcessorWorkflow dict lookup speaks a single language — stream_name. Only the innermost shell (StreamProcessor / sciline) speaks sciline keys.
Two independent sources must agree on the canonical name. The StreamLUT drives translation 1 on ingress; the render / aux_source_names map drives translation 2 in the factory. Nothing cross-checks the two — a stream that is present in both but spelled differently routes silently to nowhere, surfacing only as an input that never accumulates.
The role scheme is confined to the config layer. It lives in AuxSources and the brief aux_source_names[role] lookups in factories, and is fully consumed before the first message moves. Neither the Job nor the runtime path is aware of roles.