Skip to content

Add log stream name mapping support #589

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SimonHeybrock merged 4 commits into from
Dec 15, 2025
Merged

Add log stream name mapping support #589

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ef6cbe7
Add log stream name mapping support to StreamMapping
SimonHeybrock Dec 8, 2025
bbd7a5c
Extend nexus_helpers with units extraction and code generation
SimonHeybrock Dec 8, 2025
d007bd4
Address PR review comments for nexus_helpers
SimonHeybrock Dec 10, 2025
158f421
Do not strip suffixes from suggested internal names
SimonHeybrock Dec 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions src/ess/livedata/config/instruments/_ess.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,12 @@ def _make_dev_area_detectors(
}


def _make_dev_logs(*, instrument: str, log_names: list[str]) -> StreamLUT:
"""Create log stream mapping for dev mode where source_name equals internal name."""
topic = f'{instrument}_motion'
return {InputStreamKey(topic=topic, source_name=name): name for name in log_names}


def _make_dev_beam_monitors(
instrument: str, monitor_names: list[str] | None = None
) -> StreamLUT:
Expand Down Expand Up @@ -80,22 +86,26 @@ def make_dev_stream_mapping(
detector_names: list[str],
area_detector_names: list[str] | None = None,
monitor_names: list[str] | None = None,
log_names: list[str] | None = None,
) -> StreamMapping:
motion_topic = f'{instrument}_motion'
log_topics = {motion_topic}
area_detectors = (
_make_dev_area_detectors(
instrument=instrument, area_detectors=area_detector_names
)
if area_detector_names
else {}
)
logs = (
_make_dev_logs(instrument=instrument, log_names=log_names)
if log_names
else None
)
return StreamMapping(
instrument=instrument,
detectors=_make_dev_detectors(instrument=instrument, detectors=detector_names),
monitors=_make_dev_beam_monitors(instrument, monitor_names=monitor_names),
area_detectors=area_detectors,
log_topics=log_topics,
logs=logs,
**_make_livedata_topics(instrument),
)

Expand All @@ -106,6 +116,5 @@ def make_common_stream_mapping_inputs(
return {
'instrument': instrument,
'monitors': _make_cbm_monitors(instrument, monitor_names=monitor_names),
'log_topics': None,
**_make_livedata_topics(instrument),
}
134 changes: 130 additions & 4 deletions src/ess/livedata/config/instruments/bifrost/specs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,11 +231,137 @@ class QMapOutputs(WorkflowOutputsBase):
'113_psd1',
]

# Some example motions used for testing, probably not reflecting reality
# Combined f144 log stream configuration.
# Maps internal name -> {source: Kafka source name, units: unit string, topic: topic}
# Generated using: python -m ess.livedata.nexus_helpers <file> --generate --topic <t>
f144_log_streams: dict[str, dict[str, str]] = {
Copy link
Member

@nvaytet nvaytet Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is a mapping only applicable to Bifrost, or will other instruments follow later?

Copy link
Member Author

@SimonHeybrock SimonHeybrock Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, others will follow later, but every instrument will have its own unique mapping (or maybe with some overlap, for shared equipment).

# Motion streams (topic: bifrost_motion)
'attenuator_1': {
'source': 'BIFRO-AttChg:MC-Pne-01:ShtAuxBits07',
'units': 'dimensionless',
'topic': 'bifrost_motion',
},
'attenuator_2': {
'source': 'BIFRO-AttChg:MC-Pne-02:ShtAuxBits07',
'units': 'dimensionless',
'topic': 'bifrost_motion',
},
'attenuator_3': {
'source': 'BIFRO-AttChg:MC-Pne-03:ShtAuxBits07',
'units': 'dimensionless',
'topic': 'bifrost_motion',
},
'detector_rotation': {
'source': 'BIFRO-DtCar:MC-RotZ-01:Mtr.RBV',
'units': 'deg',
'topic': 'bifrost_motion',
},
'get_lost_tube': {
'source': 'BIFRO-InBm:MC-Pne-01:ShtAuxBits07',
'units': 'dimensionless',
'topic': 'bifrost_motion',
},
'goniometer_x': {
'source': 'BIFRO-SpGon:MC-RotX-01:Mtr.RBV',
'units': 'deg',
'topic': 'bifrost_motion',
},
'goniometer_y': {
'source': 'BIFRO-SpGon:MC-RotY-01:Mtr.RBV',
'units': 'deg',
'topic': 'bifrost_motion',
},
'sample_rotation': {
'source': 'BIFRO-SpRot:MC-RotZ-01:Mtr.RBV',
'units': 'deg',
'topic': 'bifrost_motion',
},
'slit_bottom': {
'source': 'BIFRO-SpSl1:MC-SlZm-01:PzMtr.RBV',
'units': 'mm',
'topic': 'bifrost_motion',
},
'slit_left': {
'source': 'BIFRO-SpSl1:MC-SlYp-01:PzMtr.RBV',
'units': 'mm',
'topic': 'bifrost_motion',
},
'slit_right': {
'source': 'BIFRO-SpSl1:MC-SlYm-01:PzMtr.RBV',
'units': 'mm',
'topic': 'bifrost_motion',
},
'slit_top': {
'source': 'BIFRO-SpSl1:MC-SlZp-01:PzMtr.RBV',
'units': 'mm',
'topic': 'bifrost_motion',
},
'slit_position': {
'source': 'BIFRO-SpSl1:MC-LinX-01:PzMtr-PosReadback',
'units': 'mm',
'topic': 'bifrost_motion',
},
# Sample environment streams (topic: bifrost_sample_env)
'heater_1': {
'source': 'YMIR-SEE:SE-LS336-004:HTR1',
Copy link
Member

@nvaytet nvaytet Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Strange to see Ymir stuff in here? (or maybe not?)

Copy link
Member Author

@SimonHeybrock SimonHeybrock Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comes from the latest coda file.

'units': 'W',
'topic': 'bifrost_sample_env',
},
'heater_2': {
'source': 'YMIR-SEE:SE-LS336-004:HTR2',
'units': 'W',
'topic': 'bifrost_sample_env',
},
'temperature_0': {
'source': 'YMIR-SEE:SE-LS336-004:KRDG0',
'units': 'K',
'topic': 'bifrost_sample_env',
},
'temperature_1': {
'source': 'YMIR-SEE:SE-LS336-004:KRDG1',
'units': 'K',
'topic': 'bifrost_sample_env',
},
'temperature_2': {
'source': 'YMIR-SEE:SE-LS336-004:KRDG2',
'units': 'K',
'topic': 'bifrost_sample_env',
},
'temperature_3': {
'source': 'YMIR-SEE:SE-LS336-004:KRDG3',
'units': 'K',
'topic': 'bifrost_sample_env',
},
'temperature_setpoint': {
'source': 'YMIR-SEE:SE-LS336-004:SETP_S1',
'units': 'K',
'topic': 'bifrost_sample_env',
},
'sensor_0': {
'source': 'YMIR-SEE:SE-LS336-004:SRDG0',
'units': 'V',
'topic': 'bifrost_sample_env',
},
'sensor_1': {
'source': 'YMIR-SEE:SE-LS336-004:SRDG1',
'units': 'V',
'topic': 'bifrost_sample_env',
},
'sensor_2': {
'source': 'YMIR-SEE:SE-LS336-004:SRDG2',
'units': 'V',
'topic': 'bifrost_sample_env',
},
'sensor_3': {
'source': 'YMIR-SEE:SE-LS336-004:SRDG3',
'units': 'V',
'topic': 'bifrost_sample_env',
},
}

# Derived from f144_log_streams for use by the Instrument
f144_attribute_registry = {
'detector_rotation': {'units': 'deg'},
'sample_rotation': {'units': 'deg'},
'sample_temperature': {'units': 'K'},
name: {'units': info['units']} for name, info in f144_log_streams.items()
}

# Create instrument
Expand Down
17 changes: 16 additions & 1 deletion src/ess/livedata/config/instruments/bifrost/streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from ess.livedata.kafka import InputStreamKey, StreamLUT, StreamMapping

from .._ess import make_common_stream_mapping_inputs, make_dev_stream_mapping
from .specs import monitors
from .specs import f144_log_streams, monitors


def _bifrost_generator() -> Generator[tuple[str, tuple[int, int]]]:
Expand Down Expand Up @@ -61,16 +61,31 @@ def _make_bifrost_detectors() -> StreamLUT:
}


def _make_bifrost_logs() -> StreamLUT:
"""
Bifrost log data mapping.

Derives StreamLUT from f144_log_streams, mapping Kafka source names
(EPICS PV names) to ESSlivedata-internal stream names.
"""
return {
InputStreamKey(topic=info['topic'], source_name=info['source']): internal_name
for internal_name, info in f144_log_streams.items()
}


stream_mapping = {
StreamingEnv.DEV: make_dev_stream_mapping(
'bifrost',
detector_names=list(detector_fakes),
monitor_names=monitors,
log_names=list(f144_log_streams.keys()),
),
StreamingEnv.PROD: StreamMapping(
**make_common_stream_mapping_inputs(
instrument='bifrost', monitor_names=monitors
),
detectors=_make_bifrost_detectors(),
logs=_make_bifrost_logs(),
),
}
2 changes: 2 additions & 0 deletions src/ess/livedata/config/instruments/dummy/streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from ess.livedata.kafka import InputStreamKey, StreamLUT, StreamMapping

from .._ess import make_common_stream_mapping_inputs, make_dev_stream_mapping
from .specs import instrument

detector_fakes = {'panel_0': (1, 128**2)}

Expand Down Expand Up @@ -34,6 +35,7 @@ def _make_dummy_area_detectors() -> StreamLUT:
'dummy',
detector_names=list(detector_fakes),
area_detector_names=list(area_detector_fakes),
log_names=list(instrument.f144_attribute_registry.keys()),
),
StreamingEnv.PROD: StreamMapping(
**make_common_stream_mapping_inputs(instrument='dummy'),
Expand Down
12 changes: 6 additions & 6 deletions src/ess/livedata/kafka/stream_mapping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ def __init__(
detectors: StreamLUT,
monitors: StreamLUT,
area_detectors: StreamLUT | None = None,
log_topics: set[KafkaTopic] | None = None,
logs: StreamLUT | None = None,
livedata_commands_topic: str,
livedata_data_topic: str,
livedata_responses_topic: str,
Expand All @@ -43,9 +43,7 @@ def __init__(
self._detectors = detectors
self._monitors = monitors
self._area_detectors = area_detectors or {}
# Currently we simply reuse the source_name as the stream name
self._logs = None
self._log_topics = log_topics or set()
self._logs = logs
self._livedata_commands_topic = livedata_commands_topic
self._livedata_data_topic = livedata_data_topic
self._livedata_responses_topic = livedata_responses_topic
Expand Down Expand Up @@ -94,8 +92,10 @@ def monitor_topics(self) -> set[KafkaTopic]:

@property
def log_topics(self) -> set[KafkaTopic]:
"""Returns the list of log topics."""
return self._log_topics
"""Returns the set of log topics."""
if self._logs is None:
return set()
return {stream.topic for stream in self._logs.keys()}

@property
def detectors(self) -> StreamLUT:
Expand Down
Loading
Loading