This commit is contained in:
x10sa
@@ -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
|
||||
# -------------------------------------------------------------------
|
||||
Reference in New Issue
Block a user