From 6940614f6de7b0184de88e6fa8c6f7efd8fc7789 Mon Sep 17 00:00:00 2001 From: x10sa Date: Fri, 22 May 2026 16:23:46 +0200 Subject: [PATCH] Automatic backup triggered by new deployment --- pxii_bec/macros/check2.py | 394 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 394 insertions(+) create mode 100644 pxii_bec/macros/check2.py diff --git a/pxii_bec/macros/check2.py b/pxii_bec/macros/check2.py new file mode 100644 index 0000000..e2b4292 --- /dev/null +++ b/pxii_bec/macros/check2.py @@ -0,0 +1,394 @@ +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable + +from bec_lib.device import Signal, Positioner + + +# ------------------------------------------------------------------- +# Status Enum +# ------------------------------------------------------------------- + + +class Status(Enum): + OK = 0 + WARNING = 1 + ERROR = 2 + UNKNOWN = 3 + + @property + def color(self): + return { + Status.OK: "green", + Status.WARNING: "orange", + Status.ERROR: "red", + Status.UNKNOWN: "gray", + }[self] + + +# ------------------------------------------------------------------- +# Health Result Object +# ------------------------------------------------------------------- + + +@dataclass +class HealthCheckResult: + + name: str + status: Status + + value: Any = None + + message: str = "" + + category: str = "general" + + def __str__(self): + + if self.status == Status.OK: + return f"[{self.status.name}] {self.name}" + + return ( + f"[{self.status.name}] " + f"{self.name}: {self.message}" + ) + + +# ------------------------------------------------------------------- +# Configuration +# ------------------------------------------------------------------- + +@dataclass +class BeamlineHealthConfig: + + signal_rules: dict[str, Callable] = field( + default_factory=lambda: { + "cam": lambda x: x != 0, + "bpm": lambda x: x != 0, + } + ) + + motor_tolerances: dict[str, float] = field( + default_factory=lambda: { + # examples + # "mono_theta": 0.001, + # "detector_z": 0.1, + } + ) + + default_motor_tolerance: float = 0.02 + + +# ------------------------------------------------------------------- +# Device Collection +# ------------------------------------------------------------------- + + +def get_devices(): + return list(dev.items()) + + +# ------------------------------------------------------------------- +# Signal Checks +# ------------------------------------------------------------------- + + +def check_signals(devices, config: BeamlineHealthConfig): + + results = [] + + signal_devices = [ + (name, obj) + for name, obj in devices + if isinstance(obj, Signal) + ] + + for name, obj in signal_devices: + + try: + data = obj.read() + actual = data[name]["value"] + + except Exception as e: + + results.append( + HealthCheckResult( + name=name, + status=Status.UNKNOWN, + message=f"Failed to read signal: {e}", + category="signals", + ) + ) + + continue + + matched = False + + for keyword, rule in config.signal_rules.items(): + + if keyword in name: + + matched = True + + try: + passed = rule(actual) + + except Exception as e: + + results.append( + HealthCheckResult( + name=name, + status=Status.UNKNOWN, + value=actual, + message=f"Rule evaluation failed: {e}", + category="signals", + ) + ) + + break + + if passed: + + results.append( + HealthCheckResult( + name=name, + status=Status.OK, + value=actual, + category="signals", + ) + ) + + else: + + results.append( + HealthCheckResult( + name=name, + status=Status.ERROR, + value=actual, + message=f"Signal value {actual} failed validation", + category="signals", + ) + ) + + break + + if not matched: + continue + + return results + + +# ------------------------------------------------------------------- +# Motor Checks +# ------------------------------------------------------------------- + + +def check_motors(devices, config: BeamlineHealthConfig): + + results = [] + + motor_devices = [ + (name, obj) + for name, obj in devices + if isinstance(obj, Positioner) + ] + + for name, obj in motor_devices: + + try: + + data = obj.read() + + actual = data[name]["value"] + + error_code = obj.motor_status.get() + + move_state = obj.motor_is_moving.get() + + except Exception as e: + + results.append( + HealthCheckResult( + name=name, + status=Status.UNKNOWN, + message=f"Failed to read motor: {e}", + category="motors", + ) + ) + + continue + + # ----------------------------------------------------------- + # Error state + # ----------------------------------------------------------- + + if error_code != 0: + + results.append( + HealthCheckResult( + name=name, + status=Status.ERROR, + value=error_code, + message=f"motor error code: {error_code}", + category="motors", + ) + ) + + continue + + # ----------------------------------------------------------- + # Moving state + # ----------------------------------------------------------- + + if move_state != 0: + + results.append( + HealthCheckResult( + name=name, + status=Status.WARNING, + value=move_state, + message="motor is currently moving", + category="motors", + ) + ) + + continue + + # ----------------------------------------------------------- + # Setpoint comparison + # ----------------------------------------------------------- + + sp_key = f"{name}_user_setpoint" + + if sp_key in data: + + setpoint = data[sp_key]["value"] + + diff = abs(actual - setpoint) + + tolerance = config.motor_tolerances.get( + name, + config.default_motor_tolerance, + ) + + if diff > tolerance: + + results.append( + HealthCheckResult( + name=name, + status=Status.WARNING, + value=diff, + message=( + f"Setpoint {setpoint:.4g} differs " + f"from readback {actual:.4g} " + f"by {diff:.4g}" + ), + category="motors", + ) + ) + + else: + + results.append( + HealthCheckResult( + name=name, + status=Status.OK, + value=actual, + category="motors", + ) + ) + + else: + + results.append( + HealthCheckResult( + name=name, + status=Status.UNKNOWN, + message="No setpoint available", + category="motors", + ) + ) + + return results + + +# ------------------------------------------------------------------- +# Main Check Entry Point +# ------------------------------------------------------------------- + + +def check2(config: BeamlineHealthConfig | None = None): + if config is None: + config = BeamlineHealthConfig() + + devices = get_devices() + + results = [] + + results.extend(check_signals(devices, config)) + + results.extend(check_motors(devices, config)) + + # --------------------------------------------------------------- + # Sort by severity + # --------------------------------------------------------------- + + results.sort( + key=lambda r: r.status.value, + ) + + return results + + +# ------------------------------------------------------------------- +# Summary Printer +# ------------------------------------------------------------------- + + +def print_summary(results): + + n_ok = sum(r.status == Status.OK for r in results) + + n_warn = sum(r.status == Status.WARNING for r in results) + + n_err = sum(r.status == Status.ERROR for r in results) + + n_unknown = sum(r.status == Status.UNKNOWN for r in results) + + print("\n==============================") + print(" Beamline Health Summary") + print("==============================") + print(f"OK : {n_ok}") + print(f"WARNING : {n_warn}") + print(f"ERROR : {n_err}") + print(f"UNKNOWN : {n_unknown}") + print("==============================\n") + + +# ------------------------------------------------------------------- +# CLI Entry Point +# ------------------------------------------------------------------- + + +def run_check(show_all=False): + + results = check2() + problem_results = [ + r for r in results + if r.status != Status.OK + ] + + print_summary(results) + + if not show_all: + + for result in problem_results: + print(result) + else: + for result in results: + print(result) + + +# ------------------------------------------------------------------- +# Script Execution +# -------------------------------------------------------------------