127 lines
3.4 KiB
Python
127 lines
3.4 KiB
Python
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from aarecommon.models.beamline import MXBeamline
|
|
|
|
|
|
class BeamlineYAMLConfig:
|
|
def __init__(self):
|
|
self.beamline: MXBeamline = mx_beamline()
|
|
self.config = self._load_config()
|
|
|
|
def _load_config(self) -> dict[str, Any]:
|
|
config_dir = Path(__file__).parent / "beamline_configs"
|
|
yaml_file = config_dir / f"{self.beamline.value.lower()}.yaml"
|
|
|
|
if not yaml_file.exists():
|
|
raise FileNotFoundError(f"Config file not found: {yaml_file}")
|
|
|
|
with open(yaml_file, "r") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
return self.config.get(key, default)
|
|
|
|
|
|
def mx_beamline() -> MXBeamline:
|
|
name = os.getenv("BEAMLINE")
|
|
if name is None:
|
|
raise ValueError("Please set the BEAMLINE environment variable to run AareDAQ")
|
|
name = name.strip().upper()
|
|
if name in MXBeamline.__members__:
|
|
return MXBeamline[name]
|
|
raise ValueError(
|
|
f"{name} is not a valid value for BEAMLINE. Please set one of {MXBeamline.__members__}"
|
|
)
|
|
|
|
|
|
def get_jfjoch_url(bl: MXBeamline) -> str:
|
|
"""Centralized URL resolution for JFJoch services."""
|
|
match bl:
|
|
case MXBeamline.X10SA:
|
|
return cfg_get("daq.hardware.jfjoch_url", "http://sls-gpu-002:8080")
|
|
case MXBeamline.X06DA:
|
|
return cfg_get("daq.hardware.jfjoch_url", "http://sls-gpu-001:8080")
|
|
case MXBeamline.SIMULATED:
|
|
return cfg_get("daq.hardware.jfjoch_url", "http://localhost:8080")
|
|
case MXBeamline.X06SA:
|
|
raise NotImplementedError("X06SA beamline not supported yet")
|
|
case _:
|
|
raise ValueError(f"unknown beamline {bl}")
|
|
|
|
|
|
def get_beamline_config() -> BeamlineYAMLConfig:
|
|
beamline_config = BeamlineYAMLConfig()
|
|
return beamline_config
|
|
|
|
|
|
def cfg_get(path: str, default: Any = None) -> Any:
|
|
"""
|
|
Hierarchical getter for YAML entries using dotted paths, e.g.:
|
|
cfg_get("endpoints.smargon_base")
|
|
cfg_get("epics.pv_prefix")
|
|
"""
|
|
beamline_config = get_beamline_config()
|
|
node = beamline_config.config
|
|
for part in path.split("."):
|
|
if not isinstance(node, dict) or part not in node:
|
|
return default
|
|
node = node[part]
|
|
return node
|
|
|
|
|
|
def jfjoch_url() -> str | None:
|
|
return cfg_get("shared.jfjoch.jfjoch_url")
|
|
|
|
|
|
def smargon_url() -> str | None:
|
|
return cfg_get("daq.hardware.smargon_url")
|
|
|
|
|
|
def aerotech_url() -> str | None:
|
|
return cfg_get("daq.hardware.aerotech_url")
|
|
|
|
|
|
def tell_url() -> str | None:
|
|
return cfg_get("daq.hardware.tell_url")
|
|
|
|
|
|
def bec_host() -> str | None:
|
|
return cfg_get("daq.hardware.bec_url")
|
|
|
|
|
|
def redis_host() -> str | None:
|
|
return cfg_get("daq.hardware.redis_url")
|
|
|
|
|
|
def gui_sample_camera_zmq() -> str | None:
|
|
return cfg_get("gui.cameras.sample_camera_zmq_url")
|
|
|
|
|
|
def gui_prediction_zmq() -> str | None:
|
|
return cfg_get("gui.cameras.prediction_zmq_url")
|
|
|
|
|
|
def gui_beamline_camera_addr() -> str | None:
|
|
return cfg_get("gui.cameras.beamline_camera_url")
|
|
|
|
|
|
def gui_gonio_camera_addr() -> str | None:
|
|
return cfg_get("gui.cameras.gonio_camera_url")
|
|
|
|
|
|
def gui_gonio_camera_id() -> int | None:
|
|
return cfg_get("gui.cameras.gonio_camera_id")
|
|
|
|
|
|
def daq_base_url() -> str | None:
|
|
return cfg_get("gui.daq.daq_url")
|
|
|
|
|
|
# Global instance for easy import
|
|
if __name__ == "__main__":
|
|
beamline_config = BeamlineYAMLConfig()
|