diff --git a/.github/workflows/build-container.yml b/.github/workflows/build-container.yml
index d43f075..0b451f0 100644
--- a/.github/workflows/build-container.yml
+++ b/.github/workflows/build-container.yml
@@ -4,7 +4,7 @@ on:
push:
branches:
- main
- - pool
+ - bmad
tags:
- "v*"
workflow_dispatch:
diff --git a/deploy/kubernetes/kustomization.yaml b/deploy/kubernetes/kustomization.yaml
index 2ae903f..8270a32 100644
--- a/deploy/kubernetes/kustomization.yaml
+++ b/deploy/kubernetes/kustomization.yaml
@@ -13,9 +13,9 @@ resources:
- ingress.yaml
- quad-scan-ingress.yaml
-# images:
-# - name: ghcr.io/slaclab/lume-visualizations
-# newTag: pool
+images:
+ - name: ghcr.io/slaclab/lume-visualizations
+ newTag: bmad
configMapGenerator:
- name: lume-live-monitor-ui
diff --git a/deploy/kubernetes/live-monitor-ui/live_stream_monitor.py b/deploy/kubernetes/live-monitor-ui/live_stream_monitor.py
index 45f592f..b70af5c 100644
--- a/deploy/kubernetes/live-monitor-ui/live_stream_monitor.py
+++ b/deploy/kubernetes/live-monitor-ui/live_stream_monitor.py
@@ -41,7 +41,7 @@ def imports():
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
- from lume_visualizations.beam_monitor import StagedModelImageSource
+ from lume_visualizations.beam_monitor import ModelImageSource, MODEL_INFO, MODELS
from lume_visualizations.config import (
EPICS_INPUT_PVS,
MANUAL_INPUT_PVS,
@@ -66,19 +66,31 @@ def imports():
SCREEN_KEYS,
BeamDashboard,
EpicsInputProvider,
- StagedModelImageSource,
+ MODEL_INFO,
+ MODELS,
+ ModelImageSource,
VisibilitySettings,
)
+@app.cell
+def model_selector(MODELS, mo):
+ model_dropdown = mo.ui.dropdown(
+ options=list(MODELS.keys()),
+ value="cu_hxr_staged",
+ label="Model",
+ )
+ return (model_dropdown,)
+
+
@app.cell
def header(mo):
mo.md("# LUME Live Stream Monitor")
@app.cell
-def source_setup(EpicsInputProvider, EPICS_INPUT_PVS, FAKE_INPUT_SPECS, StagedModelImageSource):
- source = StagedModelImageSource.create_default()
+def source_setup(EpicsInputProvider, EPICS_INPUT_PVS, FAKE_INPUT_SPECS, ModelImageSource, model_dropdown):
+ source = ModelImageSource(model_name=model_dropdown.value, reset_values={})
provider = EpicsInputProvider()
# Keep fake and real EPICS deployments on the same explicit PV contract.
model_input_names = list(EPICS_INPUT_PVS)
@@ -103,7 +115,7 @@ def interactive_dashboard_setup(BeamDashboard):
@app.cell
-def live_controls(mo, SCREEN_KEYS):
+def live_controls(MODEL_INFO, mo, SCREEN_KEYS, model_dropdown):
live_screen_dropdown = mo.ui.dropdown(
options=SCREEN_KEYS, value="OTR4", label="Screen"
)
@@ -127,14 +139,28 @@ def live_controls(mo, SCREEN_KEYS):
live_show_emit_y = mo.ui.checkbox(value=True, label="eps_n,y")
live_show_twiss_a_beta = mo.ui.checkbox(value=True, label="x.beta")
live_show_twiss_b_beta = mo.ui.checkbox(value=True, label="y.beta")
+ _rows = []
+ for _k, _v in MODEL_INFO.items():
+ _rows.append("" + _k + ": " + _v["description"])
+ _model_info = mo.Html(
+ ""
+ "ⓘ info
"
+ ""
+ + "
".join(_rows)
+ + "
"
+ )
live_controls_ui = mo.vstack(
[
mo.hstack(
- [
- live_screen_dropdown,
- live_poll_period_slider,
- live_image_scale_mode,
- ],
+ [model_dropdown, _model_info],
+ gap="0.5rem",
+ justify="start",
+ align="center",
+ ),
+ mo.hstack(
+ [live_screen_dropdown, live_poll_period_slider, live_image_scale_mode],
gap="1.0rem",
justify="start",
),
@@ -171,7 +197,7 @@ def live_controls(mo, SCREEN_KEYS):
@app.cell
-def interactive_controls(mo, SCREEN_KEYS):
+def interactive_controls(MODEL_INFO, mo, SCREEN_KEYS, model_dropdown, apply_machine_btn):
interactive_screen_dropdown = mo.ui.dropdown(
options=SCREEN_KEYS, value="OTR4", label="Screen"
)
@@ -187,22 +213,45 @@ def interactive_controls(mo, SCREEN_KEYS):
interactive_show_emit_y = mo.ui.checkbox(value=True, label="eps_n,y")
interactive_show_twiss_a_beta = mo.ui.checkbox(value=True, label="x.beta")
interactive_show_twiss_b_beta = mo.ui.checkbox(value=True, label="y.beta")
- # All display controls in a single compact row above the dashboard
- interactive_controls_ui = mo.hstack(
+ _rows = []
+ for _k, _v in MODEL_INFO.items():
+ _rows.append("" + _k + ": " + _v["description"])
+ _model_info = mo.Html(
+ ""
+ "ⓘ info
"
+ ""
+ + "
".join(_rows)
+ + "
"
+ )
+ interactive_controls_ui = mo.vstack(
[
- interactive_screen_dropdown,
- interactive_image_scale_mode,
- mo.md("**Show:**"),
- interactive_show_sigma_x,
- interactive_show_sigma_y,
- interactive_show_sigma_z,
- interactive_show_emit_x,
- interactive_show_emit_y,
- interactive_show_twiss_a_beta,
- interactive_show_twiss_b_beta,
+ mo.hstack(
+ [model_dropdown, _model_info],
+ gap="0.5rem",
+ justify="start",
+ align="center",
+ ),
+ mo.hstack(
+ [
+ interactive_screen_dropdown,
+ interactive_image_scale_mode,
+ mo.md("**Show:**"),
+ interactive_show_sigma_x,
+ interactive_show_sigma_y,
+ interactive_show_sigma_z,
+ interactive_show_emit_x,
+ interactive_show_emit_y,
+ interactive_show_twiss_a_beta,
+ interactive_show_twiss_b_beta,
+ apply_machine_btn
+ ],
+ gap="1.0",
+ justify="start",
+ ),
],
- gap="1.0",
- justify="start",
+ gap="0.8rem",
)
return (
interactive_controls_ui,
@@ -226,6 +275,7 @@ def interactive_slider_controls(
mo,
set_interactive_eval_trigger,
slider_display_values,
+ source
):
slider_labels = {
"SOLN:IN20:121:BCTRL": "SOLN:IN20:121:BCTRL",
@@ -239,6 +289,10 @@ def interactive_slider_controls(
"QUAD:IN20:441:BCTRL": "QUAD:IN20:441:BCTRL",
"QUAD:IN20:511:BCTRL": "QUAD:IN20:511:BCTRL",
"QUAD:IN20:525:BCTRL": "QUAD:IN20:525:BCTRL",
+ "QUAD:IN20:631:BCTRL": "QUAD:IN20:631:BCTRL",
+ "QUAD:IN20:651:BCTRL": "QUAD:IN20:651:BCTRL",
+ "XCOR:IN20:641:BCTRL": "XCOR:IN20:641:BCTRL",
+ "YCOR:IN20:642:BCTRL": "YCOR:IN20:642:BCTRL",
}
slider_specs = {spec.pv_name: spec for spec in FAKE_INPUT_SPECS}
# Resolve display overrides once (set by the "apply machine values" button).
@@ -248,12 +302,18 @@ def interactive_slider_controls(
slider_rows = []
current_row = []
for index, pv_name in enumerate(MANUAL_INPUT_PVS):
+ if pv_name not in source._writable_variable_names:
+ continue
spec = slider_specs[pv_name]
_range = float(spec.maximum) - float(spec.minimum)
- _step = max(round(_range / 1000, 3), 0.001)
+ if _range <= 0:
+ _start, _stop, _step = -0.1, 0.1, 0.001
+ else:
+ _start, _stop = float(spec.minimum), float(spec.maximum)
+ _step = max(round(_range / 1000, 3), 0.001)
slider = mo.ui.slider(
- start=float(spec.minimum),
- stop=float(spec.maximum),
+ start=_start,
+ stop=_stop,
step=_step,
value=round(float(_display_vals.get(pv_name, initial_inputs[pv_name])), 3),
label=slider_labels.get(pv_name, pv_name),
@@ -377,7 +437,6 @@ def interactive_visibility_sync(
@app.cell
def layout(
active_tab,
- apply_machine_btn,
interactive_controls_ui,
interactive_dashboard_widget,
interactive_slider_controls_ui,
@@ -396,11 +455,7 @@ def layout(
)
interactive_content = mo.vstack(
[
- mo.hstack(
- [interactive_controls_ui, apply_machine_btn],
- widths=[0.55, 0.45],
- gap="1rem",
- ),
+ interactive_controls_ui,
interactive_dashboard_widget,
interactive_slider_controls_ui,
interactive_status,
@@ -523,7 +578,7 @@ def interactive_eval(
manual_values = {
name: float(interactive_sliders[name].value)
- for name in MANUAL_INPUT_PVS
+ for name in MANUAL_INPUT_PVS if name in source._writable_variable_names
}
frame = source.snapshot(
screen,
diff --git a/lume_visualizations/beam_monitor.py b/lume_visualizations/beam_monitor.py
index 0961ddd..49370f5 100644
--- a/lume_visualizations/beam_monitor.py
+++ b/lume_visualizations/beam_monitor.py
@@ -17,6 +17,7 @@
EPICS_INPUT_PVS,
MODEL_INPUT_NAMES,
SCREEN_CONFIGS,
+ EXCLUDED_EPICS_PVS,
resolve_lcls_lattice_path,
)
@@ -36,54 +37,19 @@ def _create_cu_hxr_staged_model(start_element="OTR2", end_element="OTR4"):
return get_cu_hxr_staged_model(start_element=start_element, end_element=end_element, track_beam=True)
-def _create_safe_cu_hxr_staged_model(lattice_path: str, start_element="OTR2", end_element="OTR4"):
- """Not currently used"""
- from pytao import Tao
- from lume_bmad.model import LUMEBmadModel
- from virtual_accelerator.bmad.cu_transformer import CUBmadTransformer
- from virtual_accelerator.bmad.variables import (
- get_cu_hxr_screen_variables,
- get_variables,
- )
- from virtual_accelerator.models import cu_hxr as va_cu_hxr
- from virtual_accelerator.models.staged_model import StagedModel
- from virtual_accelerator.surrogates.injector_surrogate import InjectorSurrogate
- from virtual_accelerator.utils.variables import (
- get_epics_to_name_or_overlay_mapping,
- split_control_and_observable,
- )
-
- init_file = Path(lattice_path) / "bmad" / "models" / "cu_hxr" / "tao.init"
-
- with _tao_model_workdir(lattice_path):
- tao = Tao(f"-init {init_file} -noplot -slice_lattice {start_element}:{end_element}")
-
- control_name_to_element_name = get_epics_to_name_or_overlay_mapping()
- variables = get_variables(tao)
- control_variables, observable_variables = split_control_and_observable(variables)
-
- screens = ["OTR3", "OTR4", "OTR11", "OTR12", "OTR21"]
- control_variables, screen_attributes, used_screens = get_cu_hxr_screen_variables(
- tao, control_variables, screens
- )
-
- transformer = CUBmadTransformer(
- control_name_to_bmad=control_name_to_element_name,
- screen_attributes=screen_attributes,
- )
- bmad_model = LUMEBmadModel(
- tao=tao,
- control_variables=control_variables,
- output_variables=observable_variables,
- transformer=transformer,
- dump_locations=used_screens,
- )
-
- beam_path = (Path(va_cu_hxr.__file__).parent / ".." / "bmad" / "bmad_set_beam2000_pg").resolve()
- bmad_model.tao.cmd(f"set beam_init position_file = {beam_path}")
- bmad_model.set({"track_type": 1})
-
- return StagedModel([InjectorSurrogate(), bmad_model])
+def _create_cu_hxr_bmad_model(start_element="OTR2", end_element="OTR4"):
+ from virtual_accelerator.models.cu_hxr import get_cu_hxr_bmad_model
+ return get_cu_hxr_bmad_model(start_element=start_element, end_element=end_element, track_beam=True)
+
+MODELS = {
+ "cu_hxr_staged": _create_cu_hxr_staged_model,
+ "cu_hxr_bmad": _create_cu_hxr_bmad_model,
+}
+
+MODEL_INFO = {
+ "cu_hxr_staged": {"description": "Staged LUMEModel chaining the LCLS Cu Injector ML model (predicting at OTR2)with a Bmad beamline simulation tracking from OTR2 to OTR4."},
+ "cu_hxr_bmad": {"description": "LUMEModel of Bmad linac simulation tracking from OTR2 to OTR4."},
+}
# ---------------------------------------------------------------------------
@@ -121,21 +87,22 @@ class BeamFrame:
# ---------------------------------------------------------------------------
-class StagedModelImageSource:
+class ModelImageSource:
"""Read beam images, beam phase space, and scalars from a staged model."""
thread_safe = False
def __init__(
self,
- model,
+ model_name: str,
max_scatter_points: int = 3000,
reset_values: Optional[dict[str, object]] = None,
twiss_s_pv: str = "s",
twiss_a_beta_pv: str = "x.beta",
twiss_b_beta_pv: str = "y.beta",
):
- self.model = model
+ self.model_name = model_name
+ self.model = MODELS.get(model_name)()
self.max_scatter_points = max_scatter_points
self.reset_values = reset_values or {}
self.twiss_s_pv = twiss_s_pv
@@ -145,34 +112,22 @@ def __init__(
name
for name, variable in self.model.supported_variables.items()
if not getattr(variable, "read_only", False)
+ and name not in EXCLUDED_EPICS_PVS
}
+ self.reset_values = reset_values or {}
+ self.lattice_path = resolve_lcls_lattice_path()
+ os.environ["LCLS_LATTICE"] = self.lattice_path
@classmethod
def create_default(cls):
lattice_path = resolve_lcls_lattice_path()
os.environ["LCLS_LATTICE"] = lattice_path
- model = _create_cu_hxr_staged_model()
- #model = _create_safe_cu_hxr_staged_model(lattice_path)
- return cls(model=model, reset_values={})
+ return cls(model_name="cu_hxr_staged", reset_values={})
def reset(self) -> None:
if self.reset_values:
self.model.set(self.reset_values)
- def get_model_input_names(self) -> list[str]:
- torch_model = self.model.lume_model_instances[0].model.torch_model
- return list(torch_model.input_names)
-
- def get_current_inputs(self, input_names: Optional[list[str]] = None) -> dict[str, float]:
- names = input_names or self.get_model_input_names()
- values = self.model.get(names)
- return {name: float(values[name]) for name in names}
-
- def get_writable_model_input_names(self) -> list[str]:
- return [
- name for name in self.get_model_input_names() if name in self._writable_variable_names
- ]
-
def _filter_writable_updates(
self, control_updates: Mapping[str, float]
) -> dict[str, float]:
@@ -205,7 +160,7 @@ def snapshot(
]
if screen.image_pv:
pvs.insert(0, screen.image_pv)
- if screen.scalar_mode == "pvs":
+ if screen.scalar_mode == "pvs" and self.model_name == "cu_hxr_staged":
pvs.extend(
[
screen.xrms_pv,
@@ -251,7 +206,7 @@ def snapshot(
)
def _extract_scalars(self, screen, result: Mapping[str, object], beam) -> tuple[float, float, float, float, float]:
- if screen.scalar_mode == "pvs":
+ if screen.scalar_mode == "pvs" and self.model_name == "cu_hxr_staged":
return (
float(result[screen.xrms_pv]),
float(result[screen.yrms_pv]),
@@ -263,6 +218,7 @@ def _extract_scalars(self, screen, result: Mapping[str, object], beam) -> tuple[
if beam is None:
return (0.0, 0.0, 0.0, 0.0, 0.0)
+ # for bmad-only model, OTR2 scalars are from "input_beam" particle distribution
return (
float(beam["sigma_x"]) * 1e6,
float(beam["sigma_y"]) * 1e6,
@@ -281,263 +237,4 @@ def _extract_scatter(self, beam) -> tuple[Optional[np.ndarray], Optional[np.ndar
x = x[indices]
px = px[indices]
return (x * 1e6, px)
-
-
-# ---------------------------------------------------------------------------
-# Concrete implementation: virtual-accelerator StagedModel
-# ---------------------------------------------------------------------------
-
-
-def default_manual_input_values() -> dict[str, float]:
- return {name: 0.0 for name in MODEL_INPUT_NAMES}
-
-
-def _is_virtualapple_emulated_x86() -> bool:
- if platform.machine().lower() != "x86_64":
- return False
-
- cpuinfo_path = Path("/proc/cpuinfo")
- if not cpuinfo_path.exists():
- return False
-
- try:
- cpuinfo = cpuinfo_path.read_text(encoding="utf-8", errors="ignore")
- except OSError:
- return False
-
- return "VirtualApple" in cpuinfo
-
-
-class SyntheticLiveImageSource:
- """Container-safe fallback source for amd64-on-Apple emulation."""
-
- thread_safe = True
-
- def __init__(
- self,
- image_shape: tuple[int, int] = (256, 256),
- max_scatter_points: int = 3000,
- ):
- from lume_visualizations.fake_epics_ioc import FAKE_INPUT_SPECS
-
- defaults = {spec.pv_name: float(spec.default) for spec in FAKE_INPUT_SPECS}
- self.default_inputs = {
- name: float(defaults.get(name, 0.0))
- for name in EPICS_INPUT_PVS
- }
- self.current_inputs = dict(self.default_inputs)
- self.image_shape = image_shape
- self.max_scatter_points = max_scatter_points
-
- def reset(self) -> None:
- self.current_inputs = dict(self.default_inputs)
-
- def get_model_input_names(self) -> list[str]:
- return list(EPICS_INPUT_PVS)
-
- def get_current_inputs(self, input_names: Optional[list[str]] = None) -> dict[str, float]:
- names = input_names or self.get_model_input_names()
- return {name: float(self.current_inputs[name]) for name in names}
-
- def get_writable_model_input_names(self) -> list[str]:
- return list(EPICS_INPUT_PVS)
-
- def snapshot(
- self,
- screen_key: str,
- control_updates: Optional[Mapping[str, float]] = None,
- x_axis_value: float | datetime = 0.0,
- frame_index: int = 0,
- image_caption: str = "",
- title_suffix: str = "",
- ) -> BeamFrame:
- if control_updates:
- for key, value in control_updates.items():
- if key in self.current_inputs:
- self.current_inputs[key] = float(value)
-
- screen = SCREEN_CONFIGS[screen_key]
- screen_scale = {"OTR2": 0.92, "OTR3": 1.0, "OTR4": 1.08}.get(screen_key, 1.0)
- focus_term = (
- 0.25 * self.current_inputs["QUAD:IN20:361:BCTRL"]
- - 0.18 * self.current_inputs["QUAD:IN20:371:BCTRL"]
- + 0.12 * self.current_inputs["QUAD:IN20:425:BCTRL"]
- - 0.09 * self.current_inputs["QUAD:IN20:441:BCTRL"]
- )
- phase_term = 0.05 * (
- self.current_inputs["ACCL:IN20:300:L0A_PDES"]
- + self.current_inputs["ACCL:IN20:400:L0B_PDES"]
- )
- solenoid_term = 40.0 * self.current_inputs["SOLN:IN20:121:BCTRL"]
- xrms_base = self.current_inputs["CAMR:IN20:186:XRMS"]
- yrms_base = self.current_inputs["CAMR:IN20:186:YRMS"]
-
- xrms_um = max(60.0, screen_scale * xrms_base + solenoid_term + 22.0 * np.sin(focus_term))
- yrms_um = max(60.0, screen_scale * yrms_base - 0.6 * solenoid_term + 18.0 * np.cos(focus_term))
- sigma_z_um = max(
- 200.0,
- 220.0
- + 120.0 * self.current_inputs["FBCK:BCI0:1:CHRG_S"]
- + 0.04 * (xrms_base + yrms_base)
- + 18.0 * np.cos(phase_term),
- )
- norm_emit_x_um_rad = max(0.08, 0.42 + 0.03 * np.sin(focus_term + 0.4 * phase_term))
- norm_emit_y_um_rad = max(0.08, 0.39 + 0.03 * np.cos(focus_term - 0.3 * phase_term))
-
- rng = np.random.default_rng(seed=2719 + frame_index)
- beam_x_um = rng.normal(0.0, xrms_um, size=self.max_scatter_points)
- beam_px_evc = rng.normal(0.0, max(0.02, yrms_um / 1800.0), size=self.max_scatter_points)
-
- twiss_s = np.linspace(0.0, 40.0, 200)
- twiss_a_beta = 5.5 + 1.3 * np.sin(twiss_s / 6.0 + 0.1 * focus_term)
- twiss_b_beta = 7.8 + 1.8 * np.cos(twiss_s / 7.5 - 0.08 * focus_term)
-
- image = None
- if screen.image_pv:
- nrow, ncol = self.image_shape
- cy, cx = nrow / 2, ncol / 2
- sigma_x_px = max(4.0, xrms_um / 18.0)
- sigma_y_px = max(4.0, yrms_um / 18.0)
- y_idx, x_idx = np.mgrid[0:nrow, 0:ncol]
- image = np.exp(
- -0.5 * ((x_idx - cx) / sigma_x_px) ** 2
- - 0.5 * ((y_idx - cy) / sigma_y_px) ** 2
- )
- image += rng.normal(0.0, 0.02, size=image.shape)
- image = np.clip(image, 0.0, None)
-
- return BeamFrame(
- screen_key=screen.key,
- screen_label=screen.label,
- x_axis_value=x_axis_value,
- xrms_um=float(xrms_um),
- yrms_um=float(yrms_um),
- sigma_z_um=float(sigma_z_um),
- norm_emit_x_um_rad=float(norm_emit_x_um_rad),
- norm_emit_y_um_rad=float(norm_emit_y_um_rad),
- image=image,
- image_message=screen.image_message if image is None else "",
- image_caption=image_caption,
- beam_x_um=beam_x_um,
- beam_px_evc=beam_px_evc,
- twiss_s=twiss_s,
- twiss_a_beta=twiss_a_beta,
- twiss_b_beta=twiss_b_beta,
- title_suffix=title_suffix,
- frame_index=frame_index,
- timestamp=time.time(),
- )
-
-
-# ---------------------------------------------------------------------------
-# Concrete implementation: synthetic / mock source (no hardware required)
-# ---------------------------------------------------------------------------
-
-
-class MockImageSource:
- """
- Generates synthetic beam images and scalars for development / CI.
-
- The beam spot is a 2-D Gaussian whose sigma_x grows with |scan_value|,
- mimicking a real quadrupole scan.
-
- Parameters
- ----------
- image_shape:
- (nRow, nCol) of the fake OTR image.
- n_particles:
- Number of particles in the phase-space scatter.
- noise_level:
- Gaussian noise amplitude added to the image.
- sleep_s:
- Artificial latency per step to simulate computation time.
- """
-
- thread_safe = True
-
- def __init__(
- self,
- image_shape: tuple[int, int] = (256, 256),
- n_particles: int = 3000,
- noise_level: float = 0.02,
- sleep_s: float = 0.2,
- ):
- self.image_shape = image_shape
- self.n_particles = n_particles
- self.noise_level = noise_level
- self.sleep_s = sleep_s
-
- def snapshot(
- self,
- screen_key: str,
- x_axis_value: float | datetime,
- frame_index: int,
- image_caption: str,
- title_suffix: str,
- ) -> BeamFrame:
- if self.sleep_s > 0:
- time.sleep(self.sleep_s)
-
- nrow, ncol = self.image_shape
- cy, cx = nrow / 2, ncol / 2
-
- # Spot size varies with quad strength
- position_value = frame_index if isinstance(x_axis_value, datetime) else float(x_axis_value)
- sigma_x_px = max(4.0, 20.0 + 2.5 * position_value)
- sigma_y_px = max(4.0, 20.0 - 1.5 * position_value)
-
- y_idx, x_idx = np.mgrid[0:nrow, 0:ncol]
- image = np.exp(
- -0.5 * ((x_idx - cx) / sigma_x_px) ** 2
- - 0.5 * ((y_idx - cy) / sigma_y_px) ** 2
- )
- image += np.random.normal(0, self.noise_level, image.shape)
- image = np.clip(image, 0, None)
-
- # Scalar diagnostics
- res_um = 12.0 # µm/px (approximate)
- xrms = sigma_x_px * res_um
- yrms = sigma_y_px * res_um
- norm_emit_x = max(1e-7, 5e-7 + position_value * 2e-8)
- norm_emit_y = max(1e-7, 5e-7 - position_value * 1.5e-8)
- sigma_z = 4.6e-4
-
- # Phase-space scatter
- bx = np.random.normal(0, xrms, self.n_particles)
- bpx = np.random.normal(0, yrms, self.n_particles)
-
- # Simple synthetic Twiss functions along s
- twiss_s = np.linspace(0.0, 40.0, 200)
- twiss_a_beta = 6.0 + 1.5 * np.sin(twiss_s / 6.0 + 0.15 * position_value)
- twiss_b_beta = 8.0 + 2.0 * np.cos(twiss_s / 7.5 - 0.12 * position_value)
-
- return BeamFrame(
- screen_key=screen_key,
- screen_label=screen_key,
- x_axis_value=x_axis_value,
- xrms_um=xrms,
- yrms_um=yrms,
- sigma_z_um=sigma_z * 1e6,
- norm_emit_x_um_rad=norm_emit_x * 1e6,
- norm_emit_y_um_rad=norm_emit_y * 1e6,
- image=image,
- image_caption=image_caption,
- beam_x_um=bx,
- beam_px_evc=bpx,
- twiss_s=twiss_s,
- twiss_a_beta=twiss_a_beta,
- twiss_b_beta=twiss_b_beta,
- title_suffix=title_suffix,
- frame_index=frame_index,
- timestamp=time.time(),
- )
-
- def get_frame(self, scan_pv: str, scan_value: float, step_index: int) -> BeamFrame:
- return self.snapshot(
- screen_key="OTR4",
- x_axis_value=scan_value,
- frame_index=step_index,
- image_caption=f"{scan_pv} = {scan_value:.2f} kG",
- title_suffix=f"Mock step {step_index + 1}",
- )
-
+
\ No newline at end of file
diff --git a/lume_visualizations/config.py b/lume_visualizations/config.py
index fde20c5..830d93f 100644
--- a/lume_visualizations/config.py
+++ b/lume_visualizations/config.py
@@ -27,12 +27,12 @@
"QUAD:IN20:441:BCTRL",
"QUAD:IN20:511:BCTRL",
"QUAD:IN20:525:BCTRL",
+ "QUAD:IN20:631:BCTRL",
+ "QUAD:IN20:651:BCTRL",
+ "XCOR:IN20:641:BCTRL",
+ "YCOR:IN20:642:BCTRL",
]
-# EXTRA_MACHINE_INPUTS = [
-# "CAMR:IN20:186:XRMS",
-# "CAMR:IN20:186:YRMS"
-# ]
EXTRA_MACHINE_INPUTS = []
EXCLUDED_EPICS_PVS = (
@@ -43,6 +43,8 @@
"FBCK:BCI0:1:CHRG_S",
"ACCL:IN20:300:L0A_ADES",
"ACCL:IN20:400:L0B_ADES",
+ "input_beam",
+ "track_type"
)
EPICS_INPUT_PVS = [
@@ -63,6 +65,10 @@
"QUAD:IN20:441:BCTRL",
"QUAD:IN20:511:BCTRL",
"QUAD:IN20:525:BCTRL",
+ "QUAD:IN20:631:BCTRL",
+ "QUAD:IN20:651:BCTRL",
+ "XCOR:IN20:641:BCTRL",
+ "YCOR:IN20:642:BCTRL",
]
diff --git a/lume_visualizations/fake_epics_ioc.py b/lume_visualizations/fake_epics_ioc.py
index b561c18..1df9f54 100644
--- a/lume_visualizations/fake_epics_ioc.py
+++ b/lume_visualizations/fake_epics_ioc.py
@@ -25,7 +25,7 @@ class FakePVSpec:
period_s: float
phase_offset_rad: float = 0.0
-
+SCALE = 0.25
FAKE_INPUT_SPECS = [
#FakePVSpec("camr_in20_186_r_dist", "CAMR:IN20:186:R_DIST", 423.867825, 210.21247820852545, 499.9996083265339, 23.0, 0.0),
#FakePVSpec("camr_in20_186_xrms", "CAMR:IN20:186:XRMS", 300, 200, 400, 23.0, 0.0),
@@ -41,6 +41,10 @@ class FakePVSpec:
FakePVSpec("quad_in20_441_bctrl", "QUAD:IN20:441:BCTRL", -0.17938799998564897, -1.0782202690353522, 7.559878303179915, 22.0, 3.6),
FakePVSpec("quad_in20_511_bctrl", "QUAD:IN20:511:BCTRL", 2.852171999771826, -1.0792451325247663, 7.5582919025608595, 39.0, 4.0),
FakePVSpec("quad_in20_525_bctrl", "QUAD:IN20:525:BCTRL", -3.218399988942528, -7.557932980106783, -1.0800286565992732, 15.0, 4.4),
+ FakePVSpec("quad_in20_631_bctrl", "QUAD:IN20:631:BCTRL", 7.335358881640616, 7.335358881640616*(1-SCALE), 7.335358881640616*(1+SCALE), 1.0, 4.7),
+ FakePVSpec("quad_in20_651_bctrl", "QUAD:IN20:651:BCTRL", -5.821093449409309, -5.821093449409309*(1+SCALE), -5.821093449409309*(1-SCALE), 29.0, 5.0),
+ FakePVSpec("xcor_in20_641_bctrl", "XCOR:IN20:641:BCTRL", 0.0, 0.0, 0.0, 19.0, 5.3),
+ FakePVSpec("ycor_in20_642_bctrl", "YCOR:IN20:642:BCTRL", 0.0, 0.0, 0.0, 19.0, 5.6),
]
@@ -82,7 +86,10 @@ class FakeLumeInputIOC(PVGroup):
quad_in20_441_bctrl = pvproperty(name="QUAD:IN20:441:BCTRL", value=-0.17938799998564897)
quad_in20_511_bctrl = pvproperty(name="QUAD:IN20:511:BCTRL", value=2.852171999771826)
quad_in20_525_bctrl = pvproperty(name="QUAD:IN20:525:BCTRL", value=-3.218399988942528)
-
+ quad_in20_631_bctrl = pvproperty(name="QUAD:IN20:631:BCTRL", value=-0.5)
+ quad_in20_651_bctrl = pvproperty(name="QUAD:IN20:651:BCTRL", value=0.5)
+ xcor_in20_641_bctrl = pvproperty(name="XCOR:IN20:641:BCTRL", value=0.0)
+ ycor_in20_642_bctrl = pvproperty(name="YCOR:IN20:642:BCTRL", value=0.0)
def __init__(self, *args, update_period: float = 0.5, noise_scale: float = 0.03, **kwargs):
super().__init__(*args, **kwargs)
self.update_period = update_period
diff --git a/lume_visualizations/live_stream_monitor.py b/lume_visualizations/live_stream_monitor.py
index 45f592f..b70af5c 100644
--- a/lume_visualizations/live_stream_monitor.py
+++ b/lume_visualizations/live_stream_monitor.py
@@ -41,7 +41,7 @@ def imports():
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
- from lume_visualizations.beam_monitor import StagedModelImageSource
+ from lume_visualizations.beam_monitor import ModelImageSource, MODEL_INFO, MODELS
from lume_visualizations.config import (
EPICS_INPUT_PVS,
MANUAL_INPUT_PVS,
@@ -66,19 +66,31 @@ def imports():
SCREEN_KEYS,
BeamDashboard,
EpicsInputProvider,
- StagedModelImageSource,
+ MODEL_INFO,
+ MODELS,
+ ModelImageSource,
VisibilitySettings,
)
+@app.cell
+def model_selector(MODELS, mo):
+ model_dropdown = mo.ui.dropdown(
+ options=list(MODELS.keys()),
+ value="cu_hxr_staged",
+ label="Model",
+ )
+ return (model_dropdown,)
+
+
@app.cell
def header(mo):
mo.md("# LUME Live Stream Monitor")
@app.cell
-def source_setup(EpicsInputProvider, EPICS_INPUT_PVS, FAKE_INPUT_SPECS, StagedModelImageSource):
- source = StagedModelImageSource.create_default()
+def source_setup(EpicsInputProvider, EPICS_INPUT_PVS, FAKE_INPUT_SPECS, ModelImageSource, model_dropdown):
+ source = ModelImageSource(model_name=model_dropdown.value, reset_values={})
provider = EpicsInputProvider()
# Keep fake and real EPICS deployments on the same explicit PV contract.
model_input_names = list(EPICS_INPUT_PVS)
@@ -103,7 +115,7 @@ def interactive_dashboard_setup(BeamDashboard):
@app.cell
-def live_controls(mo, SCREEN_KEYS):
+def live_controls(MODEL_INFO, mo, SCREEN_KEYS, model_dropdown):
live_screen_dropdown = mo.ui.dropdown(
options=SCREEN_KEYS, value="OTR4", label="Screen"
)
@@ -127,14 +139,28 @@ def live_controls(mo, SCREEN_KEYS):
live_show_emit_y = mo.ui.checkbox(value=True, label="eps_n,y")
live_show_twiss_a_beta = mo.ui.checkbox(value=True, label="x.beta")
live_show_twiss_b_beta = mo.ui.checkbox(value=True, label="y.beta")
+ _rows = []
+ for _k, _v in MODEL_INFO.items():
+ _rows.append("" + _k + ": " + _v["description"])
+ _model_info = mo.Html(
+ ""
+ "ⓘ info
"
+ ""
+ + "
".join(_rows)
+ + "
"
+ )
live_controls_ui = mo.vstack(
[
mo.hstack(
- [
- live_screen_dropdown,
- live_poll_period_slider,
- live_image_scale_mode,
- ],
+ [model_dropdown, _model_info],
+ gap="0.5rem",
+ justify="start",
+ align="center",
+ ),
+ mo.hstack(
+ [live_screen_dropdown, live_poll_period_slider, live_image_scale_mode],
gap="1.0rem",
justify="start",
),
@@ -171,7 +197,7 @@ def live_controls(mo, SCREEN_KEYS):
@app.cell
-def interactive_controls(mo, SCREEN_KEYS):
+def interactive_controls(MODEL_INFO, mo, SCREEN_KEYS, model_dropdown, apply_machine_btn):
interactive_screen_dropdown = mo.ui.dropdown(
options=SCREEN_KEYS, value="OTR4", label="Screen"
)
@@ -187,22 +213,45 @@ def interactive_controls(mo, SCREEN_KEYS):
interactive_show_emit_y = mo.ui.checkbox(value=True, label="eps_n,y")
interactive_show_twiss_a_beta = mo.ui.checkbox(value=True, label="x.beta")
interactive_show_twiss_b_beta = mo.ui.checkbox(value=True, label="y.beta")
- # All display controls in a single compact row above the dashboard
- interactive_controls_ui = mo.hstack(
+ _rows = []
+ for _k, _v in MODEL_INFO.items():
+ _rows.append("" + _k + ": " + _v["description"])
+ _model_info = mo.Html(
+ ""
+ "ⓘ info
"
+ ""
+ + "
".join(_rows)
+ + "
"
+ )
+ interactive_controls_ui = mo.vstack(
[
- interactive_screen_dropdown,
- interactive_image_scale_mode,
- mo.md("**Show:**"),
- interactive_show_sigma_x,
- interactive_show_sigma_y,
- interactive_show_sigma_z,
- interactive_show_emit_x,
- interactive_show_emit_y,
- interactive_show_twiss_a_beta,
- interactive_show_twiss_b_beta,
+ mo.hstack(
+ [model_dropdown, _model_info],
+ gap="0.5rem",
+ justify="start",
+ align="center",
+ ),
+ mo.hstack(
+ [
+ interactive_screen_dropdown,
+ interactive_image_scale_mode,
+ mo.md("**Show:**"),
+ interactive_show_sigma_x,
+ interactive_show_sigma_y,
+ interactive_show_sigma_z,
+ interactive_show_emit_x,
+ interactive_show_emit_y,
+ interactive_show_twiss_a_beta,
+ interactive_show_twiss_b_beta,
+ apply_machine_btn
+ ],
+ gap="1.0",
+ justify="start",
+ ),
],
- gap="1.0",
- justify="start",
+ gap="0.8rem",
)
return (
interactive_controls_ui,
@@ -226,6 +275,7 @@ def interactive_slider_controls(
mo,
set_interactive_eval_trigger,
slider_display_values,
+ source
):
slider_labels = {
"SOLN:IN20:121:BCTRL": "SOLN:IN20:121:BCTRL",
@@ -239,6 +289,10 @@ def interactive_slider_controls(
"QUAD:IN20:441:BCTRL": "QUAD:IN20:441:BCTRL",
"QUAD:IN20:511:BCTRL": "QUAD:IN20:511:BCTRL",
"QUAD:IN20:525:BCTRL": "QUAD:IN20:525:BCTRL",
+ "QUAD:IN20:631:BCTRL": "QUAD:IN20:631:BCTRL",
+ "QUAD:IN20:651:BCTRL": "QUAD:IN20:651:BCTRL",
+ "XCOR:IN20:641:BCTRL": "XCOR:IN20:641:BCTRL",
+ "YCOR:IN20:642:BCTRL": "YCOR:IN20:642:BCTRL",
}
slider_specs = {spec.pv_name: spec for spec in FAKE_INPUT_SPECS}
# Resolve display overrides once (set by the "apply machine values" button).
@@ -248,12 +302,18 @@ def interactive_slider_controls(
slider_rows = []
current_row = []
for index, pv_name in enumerate(MANUAL_INPUT_PVS):
+ if pv_name not in source._writable_variable_names:
+ continue
spec = slider_specs[pv_name]
_range = float(spec.maximum) - float(spec.minimum)
- _step = max(round(_range / 1000, 3), 0.001)
+ if _range <= 0:
+ _start, _stop, _step = -0.1, 0.1, 0.001
+ else:
+ _start, _stop = float(spec.minimum), float(spec.maximum)
+ _step = max(round(_range / 1000, 3), 0.001)
slider = mo.ui.slider(
- start=float(spec.minimum),
- stop=float(spec.maximum),
+ start=_start,
+ stop=_stop,
step=_step,
value=round(float(_display_vals.get(pv_name, initial_inputs[pv_name])), 3),
label=slider_labels.get(pv_name, pv_name),
@@ -377,7 +437,6 @@ def interactive_visibility_sync(
@app.cell
def layout(
active_tab,
- apply_machine_btn,
interactive_controls_ui,
interactive_dashboard_widget,
interactive_slider_controls_ui,
@@ -396,11 +455,7 @@ def layout(
)
interactive_content = mo.vstack(
[
- mo.hstack(
- [interactive_controls_ui, apply_machine_btn],
- widths=[0.55, 0.45],
- gap="1rem",
- ),
+ interactive_controls_ui,
interactive_dashboard_widget,
interactive_slider_controls_ui,
interactive_status,
@@ -523,7 +578,7 @@ def interactive_eval(
manual_values = {
name: float(interactive_sliders[name].value)
- for name in MANUAL_INPUT_PVS
+ for name in MANUAL_INPUT_PVS if name in source._writable_variable_names
}
frame = source.snapshot(
screen,