fixes mainly

This commit is contained in:
2026-09-06 16:34:29 +02:00
parent 2e6a97fef5
commit bb3c8a2b56
36 changed files with 2893 additions and 525 deletions
+74 -4
View File
@@ -193,6 +193,38 @@ class Daq(Assembly):
self._running_ids = count()
self._event_master = event_master
self._detectors_event_code = detectors_event_code
# Dedicated CA-monitor cache for the detectors' event-code
# frequency, read by `rate_multiplicator` on *every* scan step (via
# `retrieve()`, once per `stop()`). Reading it with a fresh
# get_current_value() each time is the exact `pulse_id` bug above,
# on a different PV: pyepics' PV.get() returns None on a transient
# CA hiccup instead of raising, and `int(100 / freq)` on that None
# killed a real scan mid-run (SIN-TIMAST-TMA:Evt-52-Freq-I, hundreds
# of channels transiently disconnected under CA congestion - see
# the ca_tuning "silent None" warning this raised). Monitor once
# here, same pattern as the pulse_id cache above, and read the cache
# in `rate_multiplicator` instead of gets on the hot path.
self._frequency_detector = None
self._frequency_monitor = None
self._frequency_latest = {"value": None}
if event_master is not None and detectors_event_code is not None:
try:
self._frequency_detector = event_master.__dict__[
f"code{detectors_event_code:03d}"
].frequency
mon = self._frequency_detector.set_current_value_callback(
func="latest"
)
mon.start() # one get to seed the cache, then monitor-only
self._frequency_monitor = mon
self._frequency_latest = mon.data
except Exception:
# A code with no live frequency (MasterEventCodeFix, used for
# the CTA sequencer's fixed-delay codes) has no PV to
# monitor at all - get_detector_code_frequency()'s own
# fallback below handles that by raising a clear error
# instead of a bare TypeError on `100 / None`.
pass
self.name = name
self.namespace = namespace
self.checker = checker
@@ -276,10 +308,38 @@ class Daq(Assembly):
@property
def rate_multiplicator(self):
freq = self._event_master.__dict__[
f"code{self._detectors_event_code:03d}"
].frequency.get_current_value()
return int(100 / freq)
return int(100 / self.get_detector_code_frequency())
def get_detector_code_frequency(self):
"""Frequency (Hz) of the detectors' event code -- never `None`.
Reads the CA-monitor cache set up in `__init__`, not a fresh get:
this is called once per scan step (`rate_multiplicator`, from
`retrieve()`), which is exactly the traffic pattern that made the
`pulse_id` bug (see `get_pulse_id`'s docstring) real rather than
theoretical. Falls back to one direct read only if the monitor was
never able to attach (e.g. the code was resolved before the
underlying component finished initializing), and raises rather than
ever returning `None` for the caller to divide by.
"""
value = self._frequency_latest.get("value")
if value is not None:
return value
if self._frequency_detector is not None:
try:
value = self._frequency_detector.get_current_value()
except Exception:
value = None
if value is not None:
return value
pvname = getattr(self._frequency_detector, "pvname", "<unknown PV>")
raise TimeoutError(
f"could not determine the frequency of detectors_event_code="
f"{self._detectors_event_code} ({pvname}): no monitored value "
"yet and a direct read also failed or returned None. If this "
"code has no live frequency (e.g. a fixed-delay CTA sequencer "
"code), detectors_event_code is pointed at the wrong one."
)
@property
def pgroup(self):
@@ -1606,6 +1666,16 @@ class Daq(Assembly):
def _create_runtable_metadata_append_status_to_runtable(
self, scan, append_status_info=True, **kwargs
):
# Lost when this was split out of the old combined elog+run_table
# callback (see eco.acquisition.counters_tmp, which still has the
# guard in the equivalent spot) - without it, append_status_info=False
# still built scan metadata and called run_table.append_run(), whose
# very first use in a session authenticates to Google Sheets
# (Run_Table2.__init__ -> Gsheet_API), a genuinely slow, easily
# mistaken-for-init_all pause that append_status_info=False is
# supposed to buy out of.
if not append_status_info:
return
print("run_table appending run")
runno = scan.daq_run_number.get_current_value()
+39 -31
View File
@@ -13,6 +13,7 @@ from pathlib import Path
import colorama
from eco.elements.protocols import Adjustable, is_adjustable, resolve_lazy
from eco.epics_utils import ca_tuning
from eco.utilities.datafiles import open_group_writable
from eco.utilities.utilities import (
NumpyEncoder,
@@ -357,38 +358,45 @@ class StepScan(Assembly):
pass
statstr += " ; Ctrs "
if not self.has_callbacks_step_counting():
acs = []
for ctr in self.counters:
acq = ctr.acquire(
scan=self, Npulses=self.pulses_per_step[0], **self.callbacks_kwargs
) # TODO make sure step-individual aquisition argument is possible.
acs.append(acq)
try:
if hasattr(ctr, "name"):
statstr += f"{ctr.name}, "
except:
pass
filenames = []
for ta in acs:
ta.wait()
if hasattr(ta, "file_names"):
filenames.extend(ta.file_names)
else:
acs = []
for ctr in self.counters:
ctr.start(scan=self, **self.callbacks_kwargs)
try:
if hasattr(ctr, "name"):
statstr += f"{ctr.name}, "
except:
pass
self.run_callbacks_step_counting()
# The acquisition window: from here until every counter has stopped,
# channel access must not be disturbed. `sensitive_period` keeps the
# adaptive monitor sweeper from reconfiguring subscriptions
# mid-acquisition, and gives reads the more patient retry budget - a
# read that comes back None here costs the run, while a few extra
# milliseconds cost nothing. See eco.epics_utils.ca_tuning.
with ca_tuning.sensitive_period("scan step acquisition"):
if not self.has_callbacks_step_counting():
acs = []
for ctr in self.counters:
acq = ctr.acquire(
scan=self, Npulses=self.pulses_per_step[0], **self.callbacks_kwargs
) # TODO make sure step-individual aquisition argument is possible.
acs.append(acq)
try:
if hasattr(ctr, "name"):
statstr += f"{ctr.name}, "
except:
pass
filenames = []
for ta in acs:
ta.wait()
if hasattr(ta, "file_names"):
filenames.extend(ta.file_names)
else:
acs = []
for ctr in self.counters:
ctr.start(scan=self, **self.callbacks_kwargs)
try:
if hasattr(ctr, "name"):
statstr += f"{ctr.name}, "
except:
pass
self.run_callbacks_step_counting()
filenames = []
for ctr in self.counters:
resp = ctr.stop(scan=self, **self.callbacks_kwargs)
filenames.extend(resp["files"])
filenames = []
for ctr in self.counters:
resp = ctr.stop(scan=self, **self.callbacks_kwargs)
filenames.extend(resp["files"])
statstr = statstr[:-2] + " done."
print(statstr, end="\n")
+25 -258
View File
@@ -92,196 +92,17 @@ namespace.append_obj(
from . import bernina_vacuum
from . import bernina_event_timing
# First real trial of the generalized beamline-view prototype (see
# eco.elements.beamline_view / Assembly.mark_beamline) on the live namespace,
# in parallel with the untouched eco.xoptics.beamline_assembly.Beamline draft
# (eco.xoptics.beamline_bernina.make_bernina_front_end/make_bernina_experiment_
# hutch) -- z_source/kind values below are taken straight from those modules
# so they agree. Every "fel"-tagged component below also carries an
# organisational subtype -- "front_end" (SARFE10, up to the end-of-front-end
# shutter), "optics" (SAROP21 Bernina optics hutch), or "hutch" (the
# experiment hutch itself, see further down) -- navigable by prefix via
# mark_beamline's path/subtype doc. Purely additive bookkeeping:
# mark_beamline() never touches EPICS or constructs anything, so nothing
# here changes unless namespace.beamline (or .beamline_view(...)) is
# actually used. Try e.g.:
# namespace.beamline.fel # every "fel" position, any subtype
# namespace.beamline.fel.front_end # just this subtype
# namespace.beamline.fel.optics
# namespace.beamline.fel.hutch
# namespace.beamline.vacuum # the vacuum system, same 3 subtypes,
# # unfolding into each section's real
# # valve/gauge/pump devices
# The whole "fel"/"front_end" group (pshut_und, slit_und, mon_und, pshut_fe,
# att_fe, prof_fe) is delegated to bernina_front_end.py, which imports
# `namespace` back and self-registers - only needs to come after
# `namespace = Namespace(...)` above; see that module's docstring for why
# that's not a circular import.
from . import bernina_front_end # noqa: F401
from . import bernina_optics_hutch
from . import bernina_beamline_hutch
from . import bernina_laser
from . import bernina_hutch_devices
from . import bernina_alarms
from . import bernina_daq
# Alarm-overview panels mirroring the caqtdm "Alarms overview" launcher entry
# (S_charts.json -> alarms_caqtdm -> alarms.ui) and its two Papamoll pump-laser
# "Expert" sub-panels. See eco/devices_general/alarms/README.md.
namespace.append_obj(
"BerninaAlarmsOverview",
lazy=True,
name="alarms",
module_name="eco.devices_general.alarms",
)
namespace.append_obj(
"PapamollAlarms",
"26l_dean_1um_35fs",
lazy=True,
name="papamoll_alarms_35fs",
module_name="eco.devices_general.alarms",
)
namespace.append_obj(
"PapamollAlarms",
"26h_orr_510nm_100fs",
lazy=True,
name="papamoll_alarms_100fs",
module_name="eco.devices_general.alarms",
)
namespace.append_obj(
"AdjustableFS",
# "/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/config_JFs.json",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/config_JFs.json",
module_name="eco.elements.adjustable",
lazy=True,
name="config_JFs",
)
### channelsfor daq ###
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_JF.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_JF",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_BS.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_BS",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_BSCAM.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_BSCAM",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_CA.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_CA",
)
# namespace.append_obj(
# "MpodModule",
# "SARES21-PS7071",
# [1, 2, 3, 4],
# ["ch1", "ch2", "ch3", "ch4"],
# module_string="LV_OMPV_1",
# name="power_LV_patch1",
# lazy=True,
# module_name="eco.devices_general.powersockets",
# )
# namespace.append_obj(
# "MpodModule",
# "SARES21-PS7071",
# [5, 6, 7, 8],
# ["ch1", "ch2", "ch3", "ch4"],
# module_string="LV_OMPV_1",
# name="power_LV_patch2",
# lazy=True,
# module_name="eco.devices_general.powersockets",
# )
# new MPOD implementation
from eco.loptics.bernina_laser import Stage_LXT_Delay
# namespace.append_obj(
# "NEW_MpodModule",
# "SARES20-MPD1",
# [0, 1, 2, 3],
# ["ch1", "ch2", "ch3", "ch4"],
# module_string='1',
# name="power_LV_patch1",
# lazy=True,
# module_name="eco.devices_general.powersockets",
# )
# namespace.append_obj(
# "NEW_MpodModule",
# "SARES21-MPD1",
# [4, 5, 6, 7],
# ["ch4", "ch5", "ch6", "ch7"],
# module_string='1',
# name="power_LV_patch2",
# lazy=True,
# module_name="eco.devices_general.powersockets",
# )
# namespace.append_obj(
# "CheckerCA",
# module_name="eco.acquisition.checkers",
# pvname="SLAAR21-LTIM01-EVR0:CALCI",
# thresholds=[0.2, 10],
# required_fraction=0.6,
# filepath_thresholds="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds.json",
# filepath_fraction="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds_fraction.json",
# lazy=True,
# name="checker_mon_opt_ioxos",
# )
namespace.append_obj(
"CheckerBS",
module_name="eco.acquisition.checkers",
bs_channel="SAROP21-PBPS133:INTENSITY",
thresholds=[0.2, 10],
required_fraction=0.6,
filepath_thresholds="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds.json",
filepath_fraction="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds_fraction.json",
lazy=True,
name="checker",
)
##### standard DAQ #######
##### run table stuff #######
# TODO: need to check if the value property actually works here for the pgroup in the run table to make is dynamic!
@@ -330,81 +151,6 @@ namespace.append_obj(
)
# Take run status from a long-running eco.status_server process instead of
# initializing and reading *this* session's namespace at every scan start
# (see eco/status_server/README.md). On by default, pointed at the
# beamline's server - Daq.use_status_server() re-checks /health before every
# scan and falls back to the old local behaviour (with a printed warning) if
# it is unreachable, still initializing, or older than
# status_server_max_age, so a session never silently depends on the server
# being up. Override the URL, or set to "" / "off" / "none" to force the old
# always-local behaviour, with e.g.
# ECO_STATUS_SERVER=off scripts/eco-dev -s bernina
# An env var rather than a key in the shared bernina config JSON on purpose:
# which host (if any) runs a status server is a per-session choice, and that
# file is read by every session at the beamline.
_ECO_STATUS_SERVER_DEFAULT = "http://saresb-cons-04:8091"
_status_server = os.environ.get("ECO_STATUS_SERVER", _ECO_STATUS_SERVER_DEFAULT)
if _status_server.strip().lower() in ("", "off", "none", "false", "0"):
_status_server = None
if _status_server:
print(f"daq: taking run status from status server {_status_server} "
"(set ECO_STATUS_SERVER=off to always use the local namespace)")
namespace.append_obj(
"Daq",
instrument="bernina",
status_server=_status_server,
pgroup=NamespaceComponent(namespace, "config_bernina.pgroup"),
channels_JF=NamespaceComponent(namespace, "channels_JF"),
channels_BS=NamespaceComponent(namespace, "channels_BS"),
channels_BSCAM=NamespaceComponent(namespace, "channels_BSCAM"),
channels_CA=NamespaceComponent(namespace, "channels_CA"),
config_JFs=NamespaceComponent(namespace, "config_JFs"),
# pulse_id_adj="SLAAR21-LTIM01-EVR0:RX-PULSEID",
pulse_id_adj="SARES20-CVME-01-EVR0:RX-PULSEID",
event_master=NamespaceComponent(namespace, "event_master"),
detectors_event_code=50,
rate_multiplicator="auto",
name="daq",
namespace=namespace,
checker=NamespaceComponent(namespace, "checker"),
run_table=NamespaceComponent(namespace, "run_table"),
pulse_picker=NamespaceComponent(namespace, "xp"),
elog=NamespaceComponent(namespace, "elog"),
module_name="eco.acquisition.daq_client",
lazy=True,
)
namespace.append_obj(
"Scans",
# data_base_dir="scan_data",
# scan_info_dir=f"/sf/bernina/data/{config_bernina.pgroup()}/res/scan_info",
default_counters=[daq],
# default_counters=[NamespaceComponent(namespace,"daq")],
# default_counters=NamespaceComponent(namespace,"daq"),
callbacks_start_scan=[],
callbacks_end_step=[],
callbacks_end_scan=[],
# elog=elog,
name="scans",
module_name="eco.acquisition.scan",
lazy=True,
)
namespace.append_obj(
"Scans",
# data_base_dir="scan_data",
# scan_info_dir=f"/sf/bernina/data/{config_bernina.pgroup()}/res/scan_info",
default_counters=[],
callbacks_start_scan=[],
callbacks_end_step=[],
callbacks_end_scan=[],
name="scans_test",
module_name="eco.acquisition.scan",
lazy=True,
)
#####################################################################################################
## more temporary devices will be outcoupled to temorary module.
@@ -986,8 +732,9 @@ namespace.mark_beamline(
# TODO pgroup non dynamic here!
try:
import sys
import shutil
from ..utilities import TimeoutPath
from ..utilities.datafiles import ensure_dir
from ..utilities.datafiles import ensure_dir, ensure_group_writable
if TimeoutPath(f"/sf/bernina/data/{config_bernina.pgroup()}/res/").exists():
pgroup_eco_path = TimeoutPath(
@@ -999,6 +746,26 @@ try:
ensure_dir(pgroup_eco_path)
sys.path.append(pgroup_eco_path.as_posix())
pgroup_exp_path = pgroup_eco_path.get_path() / "bernina_exp.py"
if not any(pgroup_eco_path.get_path().iterdir()):
# Freshly created (empty) pgroup eco folder: seed it with an
# editable copy of the template, so there is somewhere obvious to
# add pgroup-specific devices/components without touching the
# checkout. ensure_dir already made the folder group-writable
# (setgid + group rwx), so the copy just needs the same treatment
# -- ensure_group_writable, not a hand-rolled chmod (see
# eco.utilities.datafiles).
shutil.copyfile(
Path(__file__).parent / "bernina_exp_template.py", pgroup_exp_path
)
ensure_group_writable(pgroup_exp_path)
if pgroup_exp_path.exists():
# pgroup_eco_path is on sys.path (above), so this is a plain
# top-level module import, not a package-relative one -- the
# per-pgroup file lives outside the eco package entirely.
import bernina_exp
else:
print(
"Could not access experiment folder, could be due to more systematic file system failure!"
+26
View File
@@ -0,0 +1,26 @@
from eco.bernina.bernina import namespace
from eco.utilities.config import NamespaceComponent
# Alarm-overview panels mirroring the caqtdm "Alarms overview" launcher entry
# (S_charts.json -> alarms_caqtdm -> alarms.ui) and its two Papamoll pump-laser
# "Expert" sub-panels. See eco/devices_general/alarms/README.md.
namespace.append_obj(
"BerninaAlarmsOverview",
lazy=True,
name="alarms",
module_name="eco.devices_general.alarms",
)
namespace.append_obj(
"PapamollAlarms",
"26l_dean_1um_35fs",
lazy=True,
name="papamoll_alarms_35fs",
module_name="eco.devices_general.alarms",
)
namespace.append_obj(
"PapamollAlarms",
"26h_orr_510nm_100fs",
lazy=True,
name="papamoll_alarms_100fs",
module_name="eco.devices_general.alarms",
)
+154
View File
@@ -0,0 +1,154 @@
import os
from eco.bernina.bernina import namespace
from eco.utilities.config import NamespaceComponent
from eco import bernina
namespace.append_obj(
"AdjustableFS",
# "/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/config_JFs.json",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/config_JFs.json",
module_name="eco.elements.adjustable",
lazy=True,
name="config_JFs",
)
### channelsfor daq ###
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_JF.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_JF",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_BS.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_BS",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_BSCAM.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_BSCAM",
)
namespace.append_obj(
"AdjustableFS",
"/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/channels_CA.json",
module_name="eco.elements.adjustable",
lazy=True,
name="channels_CA",
)
namespace.append_obj(
"CheckerBS",
module_name="eco.acquisition.checkers",
bs_channel="SAROP21-PBPS133:INTENSITY",
thresholds=[0.2, 10],
required_fraction=0.6,
filepath_thresholds="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds.json",
filepath_fraction="/sf/bernina/code/gac-bernina/eco_cnf_bernina/configuration/default_checker_thresholds_fraction.json",
lazy=True,
name="checker",
)
# Take run status from a long-running eco.status_server process instead of
# initializing and reading *this* session's namespace at every scan start
# (see eco/status_server/README.md). On by default, pointed at the server
# below - Daq.use_status_server() re-checks /health before every scan and
# falls back to the old local behaviour (with a printed warning) if it is
# unreachable, still initializing, or older than status_server_max_age, so a
# session never silently depends on the server being up.
#
# ECO_STATUS_SERVER is a *shell* environment variable, read once here at
# import time - it has to be set in the environment the eco session is
# started from (before `eco-dev`/`eco` runs), not from inside an already
# running IPython session, where setting it does nothing:
# ECO_STATUS_SERVER=off scripts/eco-dev -s bernina # this session only
# export ECO_STATUS_SERVER=off # every session after
# "" / "off" / "none" / "false" / "0" (case-insensitively) force the old,
# always-local behaviour; anything else is used as the server URL instead of
# the default below.
#
# An env var rather than a key in the shared bernina config JSON on purpose:
# which host (if any) runs a status server is a per-session choice, and that
# file is read by every session at the beamline.
#
# The default below is the *personal-checkout* status server used to develop
# and test this feature (see eco/status_server/DESIGN.md) - it is not yet a
# server started from the shared gac-bernina checkout, so treat it as a dev
# deployment: fine to rely on day to day, but not yet an officially operated
# service, and it can be restarted/moved without the same notice a
# production one would get.
_ECO_STATUS_SERVER_DEFAULT = "http://saresb-cons-04:8091"
_status_server = os.environ.get("ECO_STATUS_SERVER", _ECO_STATUS_SERVER_DEFAULT)
_status_server_is_default = "ECO_STATUS_SERVER" not in os.environ
if _status_server.strip().lower() in ("", "off", "none", "false", "0"):
_status_server = None
if _status_server:
kind = "dev/personal-checkout" if _status_server_is_default else "configured"
print(f"daq: taking run status from the {kind} status server "
f"{_status_server} (set ECO_STATUS_SERVER=off in the shell before "
"starting this session to always use the local namespace instead)")
namespace.append_obj(
"Daq",
instrument="bernina",
status_server=_status_server,
pgroup=NamespaceComponent(namespace, "config_bernina.pgroup"),
channels_JF=NamespaceComponent(namespace, "channels_JF"),
channels_BS=NamespaceComponent(namespace, "channels_BS"),
channels_BSCAM=NamespaceComponent(namespace, "channels_BSCAM"),
channels_CA=NamespaceComponent(namespace, "channels_CA"),
config_JFs=NamespaceComponent(namespace, "config_JFs"),
# pulse_id_adj="SLAAR21-LTIM01-EVR0:RX-PULSEID",
pulse_id_adj="SARES20-CVME-01-EVR0:RX-PULSEID",
event_master=NamespaceComponent(namespace, "event_master"),
detectors_event_code=50,
rate_multiplicator="auto",
name="daq",
namespace=namespace,
checker=NamespaceComponent(namespace, "checker"),
run_table=NamespaceComponent(namespace, "run_table"),
pulse_picker=NamespaceComponent(namespace, "xp"),
elog=NamespaceComponent(namespace, "elog"),
module_name="eco.acquisition.daq_client",
lazy=True,
)
namespace.append_obj(
"Scans",
# data_base_dir="scan_data",
# scan_info_dir=f"/sf/bernina/data/{config_bernina.pgroup()}/res/scan_info",
# default_counters=[daq],
default_counters=[NamespaceComponent(namespace,"daq")],
# default_counters=NamespaceComponent(namespace,"daq"),
callbacks_start_scan=[],
callbacks_end_step=[],
callbacks_end_scan=[],
# elog=elog,
name="scans",
module_name="eco.acquisition.scan",
lazy=True,
)
# namespace.append_obj(
# "Scans",
# # data_base_dir="scan_data",
# # scan_info_dir=f"/sf/bernina/data/{config_bernina.pgroup()}/res/scan_info",
# default_counters=[],
# callbacks_start_scan=[],
# callbacks_end_step=[],
# callbacks_end_scan=[],
# name="scans_test",
# module_name="eco.acquisition.scan",
# lazy=True,
# )
+6
View File
@@ -71,6 +71,7 @@ namespace.append_obj(
n_output_front=16,
n_output_rear=0,
name="evr_camserver72",
has_evr_sequencer=False,
module_name="eco.timing.event_timing_new_new",
lazy=True,
)
@@ -81,6 +82,7 @@ namespace.append_obj(
n_pulsers=16,
n_output_front=16,
n_output_rear=0,
has_evr_sequencer=False,
name="evr_camserver73",
module_name="eco.timing.event_timing_new_new",
lazy=True,
@@ -92,6 +94,7 @@ namespace.append_obj(
n_pulsers=16,
n_output_front=16,
n_output_rear=0,
has_evr_sequencer=False,
name="evr_camserver74",
module_name="eco.timing.event_timing_new_new",
lazy=True,
@@ -103,6 +106,7 @@ namespace.append_obj(
n_pulsers=16,
n_output_front=16,
n_output_rear=0,
has_evr_sequencer=False,
name="evr_camserver83",
module_name="eco.timing.event_timing_new_new",
lazy=True,
@@ -114,6 +118,7 @@ namespace.append_obj(
n_pulsers=16,
n_output_front=16,
n_output_rear=0,
has_evr_sequencer=False,
name="evr_camserver84",
module_name="eco.timing.event_timing_new_new",
lazy=True,
@@ -125,6 +130,7 @@ namespace.append_obj(
n_pulsers=16,
n_output_front=16,
n_output_rear=0,
has_evr_sequencer=False,
name="evr_camserver85",
module_name="eco.timing.event_timing_new_new",
lazy=True,
+27
View File
@@ -0,0 +1,27 @@
# This is only a template for creating experiment-specific files -- it is
# copied into the current pgroup's res/eco/ folder (as bernina_exp.py) the
# first time that folder is empty, then imported and left alone; edit the
# per-pgroup copy, not this one.
from eco.bernina.bernina import namespace
from eco.utilities.config import NamespaceComponent
from eco.elements.assembly import Assembly
class MyExp(Assembly):
"""A simple experiment with a timing master and a few devices."""
def __init__(self, adj_from_bernina, det_from_bernina, name=None):
super().__init__( name=name)
self._append(adj_from_bernina,name="test_adjustable", is_display=True, is_setting=True)
self._append(det_from_bernina,name="pulse_id", is_display=True, is_setting=False)
namespace.append_obj(MyExp,
NamespaceComponent(namespace,"dummy_adjustable"), # replace with a real adjustable name
NamespaceComponent(namespace,"event_master.pulse_id"),
name="my_exp_object_1",
lazy=True, # lazy=True: only resolves the components above once "my_exp_object_1" is actually touched
is_display=True,
is_setting=False,
)
+4 -2
View File
@@ -483,7 +483,9 @@ components = [
]
try:
components.extend(config["components"])
print("Did append additional components!")
_additional_components = config["components"]
components.extend(_additional_components)
if _additional_components:
print(f"Did append {len(_additional_components)} additional components!")
except:
print("Could not append components from config.")
+15 -10
View File
@@ -4,12 +4,17 @@ from time import time, sleep
import numpy as np
from epics import PV
from eco.epics_utils import ca_tuning
from eco.epics_utils.ca_tuning import (
CA_CONNECTION_TIMEOUT,
CA_INIT_CONNECTION_TIMEOUT,
note_successful_read,
report_none_read,
)
# The CA-backed classes in this module read through the same chokepoint as
# eco.epics_utils: it is what applies the retry for a channel that has
# worked before, and the silent-None diagnostics. Importing the diagnostics
# without ever calling them (as this module did) left these reads with no
# None handling at all.
from eco.epics_utils.adjustable import _read_pv
from eco.acquisition.utilities import Acquisition
from eco.aliases import Alias
@@ -27,7 +32,7 @@ class DetectorBsData(Assembly):
self.status_collection.append(self)
self.bschannel = bschannel
if epics_pv_available & epics_pv_availabe == "same":
self._pv = PV(pvname)
self._pv = ca_tuning.make_pv(pvname)
self._append(
AdjustablePvString, self.pvname + ".EGU", name="unit", is_setting=False
)
@@ -35,7 +40,7 @@ class DetectorBsData(Assembly):
self.alias = Alias(self.name, channel=self.pvname, channeltype="BS")
def get_current_value(self):
return self._pv.get()
return _read_pv(self._pv, name=getattr(self, "name", None))
def __call__(self):
return self.get_current_value()
@@ -50,7 +55,7 @@ class DetectorPvEnum(Assembly):
def __init__(self, pvname, name=None):
super().__init__(name=name)
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self.name = name
self.alias = Alias(name, channel=self.pvname, channeltype="CA")
self._resolve_lock = threading.Lock()
@@ -102,7 +107,7 @@ class DetectorPvEnum(Assembly):
return self._pv_enum(value)
def get_current_value(self):
return self.validate(self._pv.get())
return self.validate(_read_pv(self._pv, name=getattr(self, "name", None)))
def __call__(self):
return self.get_current_value()
@@ -112,12 +117,12 @@ class DetectorPvString:
def __init__(self, pvname, name=None, elog=None):
self.name = name
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self._elog = elog
self.alias = Alias(name, channel=self.pvname, channeltype="CA")
def get_current_value(self):
return self._pv.get()
return _read_pv(self._pv, name=getattr(self, "name", None))
def set_target_value(self, value, hold=False):
changer = lambda value: self._pv.put(bytes(value, "utf8"), wait=True)
@@ -141,7 +146,7 @@ class DetectorPvDataStream(Assembly):
super().__init__(name=name)
self.Id = pvname
self.pvname = pvname
self._pv = PV(pvname)
self._pv = ca_tuning.make_pv(pvname)
self.alias = Alias(self.name, channel=self.pvname, channeltype="CA")
self._append(
AdjustablePvString, self.pvname + ".EGU", name="unit", is_setting=False
@@ -255,4 +260,4 @@ class DetectorPvDataStream(Assembly):
data = property(get_data)
def get_current_value(self):
return self._pv.get()
return _read_pv(self._pv, name=getattr(self, "name", None))
+54 -63
View File
@@ -18,47 +18,38 @@ Two questions this answers
(``Stream.digitize().categorize()``, and the more general
``Scan(parameters=[...])`` it's built on) still work, and can it drive
per-scan-step binning the way ``escape.Array``'s scan-index binning does
for post-hoc data? Yes to the first part -- verified directly against a
live stream (fixed value-range bins, ``t.digitize(bins).categorize(i)``,
produced a correct multi-bin histogram). For step-scan binning, ``Scan``
with a synthetic "current step index" parameter (rather than a real bs
channel) is the right primitive -- but see the bug below, found and
worked around here rather than in ``escape.stream`` itself.
for post-hoc data? Yes -- verified directly against a live stream (fixed
value-range bins, ``t.digitize(bins).categorize(i)``, produced a correct
multi-bin histogram). For step-scan binning, ``Scan`` with a synthetic
"current step index" parameter (rather than a real bs channel) is the
right primitive -- see below.
The ``Scan`` open-bin bug (found, not yet fixed upstream)
------------------------------------------------------------
The ``Scan`` open-bin bug (found here, fixed upstream in escape-fel 0.2.7)
----------------------------------------------------------------------------
``escape.stream.Scan(parameters=[...])`` with no ``values=`` starts empty
and grows one bin at a time, the first time a genuinely new parameter value
is seen (``Scan._append()``). Each ``Stream`` wrapping it gets its own
``DataManager``, whose per-bin deque list is sized to
``len(scan._values)`` **at the moment that Stream/DataManager is
constructed** -- not kept in sync afterwards.
That is fine for a single channel (it always discovers a bin via its own
call to ``scan._append()`` before ever indexing into it). It is **not**
fine the moment a *second* channel shares the same growing ``Scan``: if
channel A's own event stream reaches a new step index before channel B's
does, A's call to ``scan._append()`` grows the *shared* ``_values`` list,
but B's own (unrelated) ``_data`` list is not grown to match. B's next
event then indexes past the end of its own list. Confirmed directly: this
raises ``IndexError: list index out of range`` inside
``Stream._appendEventData``, which runs as an ``EventWorker`` callback
inside a blanket ``try/except`` (``EventWorker.eventLoop``) -- so it never
raises up to caller code, it just prints
``"EventWorker callback error: list index out of range"`` once per event,
forever, and **that channel silently stops recording data for the rest of
the scan.**
Fix used here: never leave the shared ``Scan`` open-ended when more than
one channel is attached to it. Every bin is pre-declared up front
(``values=[(0.0,), (1.0,), ...]``) so every channel's ``DataManager`` is
born the correct final size regardless of which channel's data happens to
arrive first for any given step -- confirmed directly to eliminate the
crash and produce identical, correct per-step sample counts across three
simultaneously-binned channels. The number of steps is read from the
``scan`` object StepScan already passes into ``Counter.acquire``/``start``
(``len(scan._values_todo) + len(scan._values_done)``), so nothing new has
to be threaded through ``scan.py`` for this either.
``DataManager``. Before escape-fel 0.2.7, a ``DataManager``'s per-bin deque
list was only grown when *that* ``DataManager``'s own call to
``scan._append()`` reported a new bin -- so if a *second* channel sharing
the same growing ``Scan`` reached a given step only after some other
channel had already registered it, the second channel's own list was never
grown to match, and its next append indexed past the end of its own list.
That raised ``IndexError`` inside an ``EventWorker`` callback, silently
swallowed by ``EventWorker.eventLoop``'s blanket ``try/except`` -- so it
never surfaced as an exception, just endless
``"EventWorker callback error: list index out of range"`` console spam,
while the affected channel silently stopped recording data for the rest of
the scan. Confirmed directly (multiple independently-built ``Stream``s
sharing one dynamically-growing ``Scan``, reaching each new bin at
different times), reported upstream, and fixed in escape-fel 0.2.7
(``DataManager.append`` now grows its own buffers to cover whatever index
it needs, rather than trusting the shared ``doappend`` flag). Verified the
fix directly against the installed 0.2.7: the same reproduction that used
to lose a channel's data now records correctly, with the shared ``Scan``
left genuinely open-ended -- no pre-declared bin count needed, which is why
``BsStreamCounter`` below no longer computes ``n_steps`` from the scan
object the way an earlier version of this file did as a workaround.
Bin assignment: step index, not pulse-id math
-----------------------------------------------
@@ -213,7 +204,7 @@ class BsStreamCounter:
self.last_pulse_ids = {}
self.step_pulse_ids = {}
self._build_bins(n_steps=1) # standalone (no-scan) mode: one bin
self._build_bins()
self.callbacks_start_scan = [self._on_scan_start]
self.callbacks_start_step = []
@@ -222,14 +213,14 @@ class BsStreamCounter:
self.callbacks_end_scan = [self._on_scan_end]
# -- (re)building the shared per-step bin structure ---------------------
def _build_bins(self, n_steps):
def _build_bins(self):
# Open-ended (values=None): bins are created on demand as new step
# indices appear. Safe to share across every channel in self._raw
# regardless of which one's data reaches a given step first -- see
# the module docstring (escape-fel 0.2.7, DataManager.append).
from escape import stream as escape_stream
n_steps = max(1, n_steps or 1)
self._scan = escape_stream.Scan(
parameters=[self._step_source],
values=[(float(k),) for k in range(n_steps)],
)
self._scan = escape_stream.Scan(parameters=[self._step_source])
self._step_source.value = 0.0
self._step_index = 0
self.step_pulse_ids = {}
@@ -251,30 +242,31 @@ class BsStreamCounter:
self._channels = {}
def _on_scan_start(self, scan=None, **kwargs):
n_steps = None
if scan is not None:
n_steps = len(getattr(scan, "_values_todo", [])) + len(
getattr(scan, "_values_done", [])
)
self._teardown_bins()
self._build_bins(n_steps)
self._build_bins()
def _on_scan_end(self, scan=None, **kwargs):
self._teardown_bins()
self._build_bins(n_steps=1)
self._build_bins()
def close(self):
"""Unsubscribe every channel."""
self._teardown_bins()
# -- reading the current step's bin -------------------------------------
# Bins are created on demand (open-ended Scan) the first time a matching
# event actually arrives for a given channel, so `idx` may not exist yet
# in a freshly-advanced-to step -- treat that as "no samples yet", not
# an error.
def _bin(self, name, step_index=None):
idx = self._step_index if step_index is None else step_index
return self._channels[name]._dataManager._data[idx]
data = self._channels[name]._dataManager._data
return data[idx] if idx < len(data) else []
def _eventids(self, name, step_index=None):
idx = self._step_index if step_index is None else step_index
return self._channels[name]._dataManager._eventIds[idx]
eventids = self._channels[name]._dataManager._eventIds
return eventids[idx] if idx < len(eventids) else []
def _reduce_step(self, step_index, n0=None):
# n0: per-channel index into that bin to start reducing from -- data
@@ -347,22 +339,21 @@ class BsStreamCounter:
return {"files": []}
def _advance_step(self, scan):
"""Point the shared bin key at *scan*'s current step (or the single
standalone bin if no scan) and return ``(step_index, n0)``, where
``n0`` is how many samples were already in that step's bin per
"""Point the shared bin key at *scan*'s current step (or a single
standalone bin, index 0, if no scan) and return ``(step_index, n0)``,
where ``n0`` is how many samples were already in that step's bin per
channel -- data from *before* this call, to be excluded when
reducing (see ``_reduce_step``). For a fresh per-step bin this is
always 0; for the standalone single-ever-bin case it is whatever a
previous acquire()/get_current_value() call already left there.
Bins are created on demand (open-ended ``Scan``, see
``_build_bins``), so no "step exceeds allocated bins" check is
needed here -- any non-negative step index is valid, it simply
hasn't collected any samples yet until the first matching event
arrives.
"""
# No scan -> standalone use, always the single bin built by
# _build_bins(n_steps=1); a real scan's step index is used verbatim.
step_index = scan.next_step if scan is not None else 0
if step_index >= len(self._scan._values):
raise IndexError(
f"{self.name}: step {step_index} exceeds the {len(self._scan._values)} "
"bins allocated at scan start -- was callbacks_start_scan run for this scan?"
)
self._step_index = step_index
self._step_source.value = float(step_index)
ew = next(iter(self._channels.values()))._source.eventWorker
+2 -1
View File
@@ -2,6 +2,7 @@ from ..elements.assembly import Assembly
from ..aliases import Alias
from eco import ecocnf
from epics.pv import PV
from ..epics_utils import ca_tuning
# try:
# from bsread.bsavail import pollStream
@@ -74,7 +75,7 @@ class DetectorBsStream:
else:
self.pvname = cachannel
if self.pvname:
self._pv = PV(self.pvname, auto_monitor=False)
self._pv = ca_tuning.make_pv(self.pvname)
self.alias = Alias(name, channel=bs_channel, channeltype="BS")
_ensure_bs_event_worker()
+6 -5
View File
@@ -2,9 +2,10 @@ from epics import PV
from eco.epics_utils.ca_tuning import (
CA_CONNECTION_TIMEOUT,
CA_INIT_CONNECTION_TIMEOUT,
note_successful_read,
report_none_read,
)
# see eco.bs.detector: the diagnostics were imported here and never called,
# so a None from these reads was completely silent and unretried.
from eco.epics_utils.adjustable import _read_pv
import os
import numpy as np
import time
@@ -47,9 +48,9 @@ class PvRecord:
def get_current_value(self, readback=True):
if readback:
currval = self._pvreadback.get()
if not readback:
currval = self._pv.get()
currval = _read_pv(self._pvreadback, name=getattr(self, "name", None))
else:
currval = _read_pv(self._pv, name=getattr(self, "name", None))
return currval
def get_moveDone(self):
+50 -20
View File
@@ -1,9 +1,14 @@
import logging
import threading
import time
from enum import IntEnum
import numpy as np
from epics import PV
# the module itself as well as the names: the retry budget is read live from
# `ca_tuning` (it changes with `sensitive_period`), which a copied
# `from ... import` would freeze at import time.
from eco.epics_utils import ca_tuning
from eco.epics_utils.ca_tuning import (
CA_CONNECTION_TIMEOUT,
CA_INIT_CONNECTION_TIMEOUT,
@@ -12,6 +17,8 @@ from eco.epics_utils.ca_tuning import (
report_none_read,
)
logger = logging.getLogger(__name__)
from eco.aliases import Alias
from eco.elements.adjustable import (
AdjustableMemory,
@@ -106,8 +113,31 @@ def _read_pv(pv, name=None):
pvname = getattr(pv, "pvname", None)
if not pv.connected and not has_ever_succeeded(pvname):
value = pv.get(timeout=CA_INIT_CONNECTION_TIMEOUT)
else:
value = pv.get()
if value is None:
report_none_read(pv, name=name)
else:
note_successful_read(pvname)
return value
value = pv.get()
if value is None:
# A channel that has worked before and momentarily has not is the
# one case worth trying again: the failure is a dropped virtual
# circuit or a get that lost a race, both of which clear in
# milliseconds. Retrying here, at the single chokepoint every
# instrumented read goes through, is what stops the next caller
# having to grow its own private cache the way `Daq.get_pulse_id`
# and the event-code frequency did - and it is bounded, so an
# absent channel (handled above) never pays for it.
for attempt in range(ca_tuning.read_retries()):
time.sleep(ca_tuning.CA_READ_RETRY_DELAY)
value = pv.get()
if value is not None:
logger.debug(
"%s (%s) returned None, then a value on retry %d",
name or pvname, pvname, attempt + 1,
)
break
if value is None:
report_none_read(pv, name=name)
else:
@@ -207,31 +237,31 @@ class AdjustableAtomicPv:
# Alias(an, channel=".".join([pvname, af]), channeltype="CA")
# )
self._pv = PV(self.pvname, connection_timeout=CA_CONNECTION_TIMEOUT, count=element_count, auto_monitor=False)
self._pv = ca_tuning.make_pv(self.pvname, connection_timeout=CA_CONNECTION_TIMEOUT, count=element_count)
self._currentChange = None
self.accuracy = accuracy
if pvreadbackname is None:
self._pvreadback = PV(
self.pvname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvreadback = ca_tuning.make_pv(
self.pvname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
pvreadbackname = self.pvname
self.pvname = self.pvname
else:
self._pvreadback = PV(
pvreadbackname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvreadback = ca_tuning.make_pv(
pvreadbackname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
self.pvname = pvreadbackname
if pvlowlimname:
self._pvlowlim = PV(
pvlowlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvlowlim = ca_tuning.make_pv(
pvlowlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
else:
self._pvlowlim = None
if pvhighlimname:
self._pvhighlim = PV(
pvhighlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvhighlim = ca_tuning.make_pv(
pvhighlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
else:
self._pvhighlim = None
@@ -323,24 +353,24 @@ class AdjustablePv:
self.unit = AdjustableMemory(unit, name="unit")
if pvreadbackname is None:
self._pvreadback = PV(self.Id, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False)
self._pvreadback = ca_tuning.make_pv(self.Id, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT)
pvreadbackname = self.Id
self.pvname = self.Id
else:
self._pvreadback = PV(
pvreadbackname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvreadback = ca_tuning.make_pv(
pvreadbackname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
self.pvname = pvreadbackname
if pvlowlimname:
self._pvlowlim = PV(
pvlowlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvlowlim = ca_tuning.make_pv(
pvlowlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
else:
self._pvlowlim = None
if pvhighlimname:
self._pvhighlim = PV(
pvhighlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False
self._pvhighlim = ca_tuning.make_pv(
pvhighlimname, count=element_count, connection_timeout=CA_CONNECTION_TIMEOUT
)
else:
self._pvhighlim = None
@@ -483,7 +513,7 @@ class AdjustablePvEnum:
def __init__(self, pvname, pvname_set=None, name=None):
self.Id = pvname
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT * 2, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT * 2)
self.name = name
self._pv_set = PV(pvname_set, connection_timeout=CA_CONNECTION_TIMEOUT * 2) if pvname_set else None
self.alias = Alias(name, channel=self.Id, channeltype="CA")
@@ -624,7 +654,7 @@ class AdjustablePvString:
def __init__(self, pvname, name=None, elog=None):
self.name = name
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self._elog = elog
self.alias = Alias(name, channel=self.pvname, channeltype="CA")
+360
View File
@@ -16,6 +16,32 @@ time on a different value and each time fixed only for that value:
The pattern is always the same: a read that has no tolerance for a moment of
unavailability, and a caller that treats the resulting ``None`` as data.
THE GENERAL FIX (supersedes the per-PV ones above)
--------------------------------------------------
Each of those was fixed by hand-rolling ``auto_monitor=True`` for one PV.
That is the whole answer, generalised - see ``AUTO_MONITOR_DEFAULT`` below
for why, straight out of pyepics' source: with a monitor, a read is a dict
lookup that has no failure path at all; without one, every read is a network
round trip with two. eco's old ``auto_monitor=False`` default was therefore
not merely unhelpful, it was *the cause*.
So there are now two general mechanisms here instead of a growing list of
per-PV caches:
* **monitor by default, demote what is fast** - ``make_pv`` applies the
policy, and a background sweeper measures actual update rates and drops
the monitor on anything above ``AUTO_MONITOR_MAX_RATE``, remembering it
across sessions. Measured on bernina: 7 074 channels monitored, 68
demoted, ~8 % of one core standing cost.
* **retry at the chokepoint** - ``eco.epics_utils.adjustable._read_pv``
retries a read that has worked before (``CA_READ_RETRIES``), so a
momentary failure is absorbed once, for every caller, instead of each one
discovering it separately. A channel that has never produced a value is
not retried, so absent PVs stay cheap.
* **declare sensitive stretches** - ``sensitive_period`` marks a window
(a scan step's acquisition, say) where subscriptions must not be
reconfigured and reads get a more patient budget.
TRIAL - CONNECTION TIMEOUT
--------------------------
``CA_CONNECTION_TIMEOUT`` was ``0.05`` everywhere (22 hardcoded literals,
@@ -43,9 +69,13 @@ previous behaviour in one line. See the CLAUDE.md section "CA connection
timeout (trial)".
"""
import atexit
import json
import logging
import os
import threading
import time
from pathlib import Path
logger = logging.getLogger(__name__)
@@ -53,6 +83,73 @@ logger = logging.getLogger(__name__)
# TRIAL (see module docstring): was 0.05 everywhere. Revert to 0.05 to undo.
CA_CONNECTION_TIMEOUT = 1.0
# --------------------------------------------------------------------------
# Monitor policy: monitor by default, demote only what is actually fast
#
# `auto_monitor=False` was the eco-wide default, to keep subscription traffic
# down. It is also, from pyepics' own source, the direct cause of the silent
# `None` reads above. `PV.get_with_metadata` starts with
#
# if not self.wait_for_connection(timeout=timeout):
# return None
# if ((not use_monitor) or (not self.auto_monitor) or ... ):
# metad = ca.get_with_metadata(...)
# if metad is None:
# return
#
# so with `auto_monitor=False` **every** read is a network round trip with
# two independent ways to come back `None`, while with `auto_monitor=True`
# and a cached value that whole block is skipped: the read is a dict lookup
# that cannot time out, and a momentary circuit drop costs nothing. Every
# per-PV fix so far (`Daq.get_pulse_id`, the event-code frequency cache) has
# been hand-rolling `auto_monitor=True` for one PV at a time.
#
# The traffic argument for `False` turns out to apply to very few channels.
# Measured on the real bernina namespace (a 3-minute recording of all 7 719
# monitorable status channels, see eco/status_server/DESIGN.md section 15.2):
#
# >= 50 Hz 48 channels 87 % of all updates
# 10-50 Hz 9 channels 3 %
# 1-10 Hz 182 channels 9 %
# < 1 Hz 7 480 channels 1 %
#
# i.e. 0.6 % of channels produce seven eighths of the load, and monitoring
# the other 99.4 % is close to free. So: monitor everything by default, and
# demote the handful that prove to be fast. `_MonitorRateTracker` below does
# that automatically, and remembers them across sessions so the next one
# never subscribes to them at all.
AUTO_MONITOR_DEFAULT = True
# A channel updating faster than this gets demoted to auto_monitor=False.
# 10 Hz sits in the empty gap in the distribution above (the 10-50 Hz band
# holds 9 channels of 7 719), so the threshold is not delicately placed.
AUTO_MONITOR_MAX_RATE = 10.0
# How often the sweeper looks at accumulated counts.
AUTO_MONITOR_SWEEP_INTERVAL = 5.0
# Where the learned fast-channel list is remembered. Per user rather than
# shared: it is a local performance hint, not beamline configuration, and a
# per-user file has none of the group-permission problems a shared one in
# /sf/... would bring (see eco.utilities.datafiles).
AUTO_MONITOR_STATE_FILE = Path(
os.environ.get("ECO_CA_FAST_CHANNELS")
or (Path.home() / ".eco" / "ca_fast_channels.json")
)
# How many times a read that has worked before may be retried before it is
# reported as a silent None. See `_read_pv` in eco.epics_utils.adjustable:
# a channel that demonstrably works and momentarily does not is the exact
# case worth one more try, and the retry is skipped entirely for a channel
# that has never produced anything.
CA_READ_RETRIES = 2
CA_READ_RETRY_DELAY = 0.05
# Extra patience during a period the caller has declared sensitive (a scan
# step's acquisition window, say), where a failed read is far more expensive
# than a few extra milliseconds. See `sensitive_period`.
CA_READ_RETRIES_SENSITIVE = 4
# Budget used by `_wait_for_initialisation()` only. Deliberately still the
# old value: that call is a best-effort "is it there yet" during namespace
# init, and lengthening it would multiply init time by the number of absent
@@ -188,3 +285,266 @@ def report_none_read(pv, name=None, kind="read"):
except Exception:
# diagnostics must never be able to break a read
pass
# --------------------------------------------------------------------------
# adaptive monitor policy
_fast_lock = threading.RLock()
_fast_channels = set() # pvnames known to update faster than the threshold
_fast_dirty = False # something changed since the last save
_update_counts = {} # pvname -> updates since the last sweep
_tracked = {} # pvname -> (pv, callback_index)
_sweeper = None
_sensitive_depth = 0 # >0 while a caller has declared a sensitive period
def _load_fast_channels():
try:
with open(AUTO_MONITOR_STATE_FILE) as f:
names = json.load(f)
if isinstance(names, list):
with _fast_lock:
_fast_channels.update(str(n) for n in names)
logger.debug(
"ca_tuning: %d known fast channel(s) loaded from %s",
len(_fast_channels), AUTO_MONITOR_STATE_FILE,
)
except FileNotFoundError:
pass
except Exception:
logger.debug("ca_tuning: could not read %s", AUTO_MONITOR_STATE_FILE,
exc_info=True)
def save_fast_channels():
"""Persist the learned fast-channel list.
Worth persisting because the learning itself costs something: a session
that has to rediscover the ~50 fast channels subscribes to them for a few
seconds first. Remembering them means the next session never opens those
subscriptions at all.
"""
global _fast_dirty
with _fast_lock:
if not _fast_dirty:
return
names = sorted(_fast_channels)
_fast_dirty = False
try:
AUTO_MONITOR_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
tmp = AUTO_MONITOR_STATE_FILE.with_suffix(".tmp")
with open(tmp, "w") as f:
json.dump(names, f, indent=1)
os.replace(tmp, AUTO_MONITOR_STATE_FILE)
except Exception:
logger.debug("ca_tuning: could not write %s", AUTO_MONITOR_STATE_FILE,
exc_info=True)
def is_known_fast(pvname):
with _fast_lock:
return pvname in _fast_channels
def clear_fast_channels():
"""Forget everything learned, so the next reads re-measure from scratch.
For when a channel's rate has genuinely changed (a detector reconfigured,
an event code retimed) and it is stuck demoted from a previous session.
"""
global _fast_dirty
with _fast_lock:
_fast_channels.clear()
_fast_dirty = True
save_fast_channels()
def _count_update(pvname):
# The hot path: one dict increment per CA update, on libca's callback
# thread. Deliberately not locked - a lost increment costs nothing to a
# rate heuristic, and taking a lock here would put every monitored
# channel in the namespace through one contended lock.
_update_counts[pvname] = _update_counts.get(pvname, 0) + 1
def _track(pv):
"""Attach the rate counter to `pv` and remember it for the sweeper."""
pvname = getattr(pv, "pvname", None)
if not pvname or pvname in _tracked:
return
try:
# with_ctrlvars=False: pyepics defaults it to True, which issues a
# blocking get_ctrlvars() per already-connected PV - for a whole
# namespace that is exactly the get-storm this policy exists to
# avoid (same reason eco.status_server.monitor_store passes it).
index = pv.add_callback(
lambda pvname=pvname, **kw: _count_update(pvname),
with_ctrlvars=False,
)
except Exception:
return
_tracked[pvname] = (pv, index)
def _demote(pvname, pv, index, rate):
"""Stop monitoring a channel that updates too fast to be worth it."""
try:
# Never demote a channel somebody is deliberately monitoring: a
# recording, a Monitor(), a CallbackEpics all register their own
# callback, and clearing the subscription under them would silently
# stop their data. Ours is the only one we may account for.
if len(getattr(pv, "callbacks", {})) > 1:
return False
pv.remove_callback(index)
pv.auto_monitor = False
except Exception:
logger.debug("ca_tuning: could not demote %s", pvname, exc_info=True)
return False
logger.info(
"ca_tuning: %s updates at ~%.0f Hz (> %.0f Hz), dropping its monitor "
"- reads of it go back to a direct CA get.",
pvname, rate, AUTO_MONITOR_MAX_RATE,
)
return True
def _sweep_once(interval):
global _fast_dirty
if _sensitive_depth > 0:
# Do not reconfigure subscriptions in the middle of an acquisition:
# the point of a sensitive period is that nothing about channel
# access changes under it. Counts keep accumulating; the next sweep
# after it ends sees them.
return
# snapshot then zero, rather than clearing in place, so an update
# landing mid-sweep is counted against the next window instead of lost
counts = dict(_update_counts)
for name in counts:
_update_counts[name] = 0
demoted = []
for pvname, count in counts.items():
if count / interval <= AUTO_MONITOR_MAX_RATE:
continue
entry = _tracked.get(pvname)
if entry is None:
continue
pv, index = entry
if _demote(pvname, pv, index, count / interval):
demoted.append(pvname)
_tracked.pop(pvname, None)
_update_counts.pop(pvname, None)
if demoted:
with _fast_lock:
_fast_channels.update(demoted)
_fast_dirty = True
save_fast_channels()
def _sweep_loop():
last = time.time()
while True:
time.sleep(AUTO_MONITOR_SWEEP_INTERVAL)
now = time.time()
interval, last = max(now - last, 1e-6), now
try:
_sweep_once(interval)
except Exception:
logger.debug("ca_tuning: sweep failed", exc_info=True)
def _ensure_sweeper():
global _sweeper
if _sweeper is not None:
return
with _fast_lock:
if _sweeper is not None:
return
_sweeper = threading.Thread(
target=_sweep_loop, name="ca_tuning_monitor_sweeper", daemon=True
)
_sweeper.start()
def make_pv(pvname, auto_monitor=None, **kwargs):
"""Build a `PV` under the adaptive monitor policy.
Use this instead of `PV(...)` for anything eco reads repeatedly. It
monitors by default (see AUTO_MONITOR_DEFAULT for why that is both
faster and the fix for the silent-None class of bug), except for
channels already known to be too fast, and it registers the PV with the
sweeper that finds the rest.
`auto_monitor` still wins if given explicitly, for the cases that
genuinely know better than the policy.
"""
from epics import PV
if auto_monitor is None:
auto_monitor = AUTO_MONITOR_DEFAULT and not is_known_fast(pvname)
pv = PV(pvname, auto_monitor=auto_monitor, **kwargs)
if auto_monitor:
_track(pv)
_ensure_sweeper()
return pv
class sensitive_period:
"""Declare a stretch of time where channel access must not be disturbed.
Two things change while one is active: the sweeper leaves subscriptions
alone (reconfiguring a monitor mid-acquisition is exactly the wrong
moment), and reads get the more patient retry budget
(`CA_READ_RETRIES_SENSITIVE`), because inside a scan step a failed read
costs a run and a few extra milliseconds cost nothing.
Reentrant and thread-safe by depth counting, so nesting - a step inside
a scan inside a queue - behaves.
with ca_tuning.sensitive_period("scan step"):
...
"""
def __init__(self, what=""):
self.what = what
def __enter__(self):
global _sensitive_depth
with _fast_lock:
_sensitive_depth += 1
return self
def __exit__(self, *exc):
global _sensitive_depth
with _fast_lock:
_sensitive_depth = max(0, _sensitive_depth - 1)
return False
def in_sensitive_period():
return _sensitive_depth > 0
def read_retries():
"""How many extra attempts a read that has worked before may make."""
return CA_READ_RETRIES_SENSITIVE if _sensitive_depth > 0 else CA_READ_RETRIES
def monitor_report():
"""What the policy currently believes, for looking at from a session."""
with _fast_lock:
fast = sorted(_fast_channels)
return {
"auto_monitor_default": AUTO_MONITOR_DEFAULT,
"max_rate_hz": AUTO_MONITOR_MAX_RATE,
"monitored": len(_tracked),
"known_fast": fast,
"n_known_fast": len(fast),
"state_file": str(AUTO_MONITOR_STATE_FILE),
"in_sensitive_period": in_sensitive_period(),
}
_load_fast_channels()
atexit.register(save_fast_channels)
+5 -4
View File
@@ -5,6 +5,7 @@ from time import time, sleep
import numpy as np
from epics import PV
from eco.epics_utils.adjustable import _read_pv
from eco.epics_utils import ca_tuning
from eco.epics_utils.ca_tuning import (
CA_CONNECTION_TIMEOUT,
CA_INIT_CONNECTION_TIMEOUT,
@@ -50,7 +51,7 @@ class DetectorPvData(Assembly):
self._append(AdjustablePv, pvname, name="readback", is_setting=False)
# self.status_collection.append(self)
else:
self._pv = PV(pvname, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname)
self.alias = Alias(self.name, channel=self.pvname, channeltype="CA")
self.status_collection.append(self)
self.status_collection.append(self, selection="settings", recursive=False)
@@ -116,7 +117,7 @@ class DetectorPvEnum(Assembly):
def __init__(self, pvname, name=None):
super().__init__(name=name)
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self.name = name
self.alias = Alias(name, channel=self.pvname, channeltype="CA")
self._resolve_lock = threading.Lock()
@@ -193,7 +194,7 @@ class DetectorPvString:
def __init__(self, pvname, name=None, elog=None):
self.name = name
self.pvname = pvname
self._pv = PV(pvname, connection_timeout=CA_CONNECTION_TIMEOUT, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname, connection_timeout=CA_CONNECTION_TIMEOUT)
self._elog = elog
self.alias = Alias(name, channel=self.pvname, channeltype="CA")
@@ -230,7 +231,7 @@ class DetectorPvDataStream(Assembly):
super().__init__(name=name)
self.Id = pvname
self.pvname = pvname
self._pv = PV(pvname, auto_monitor=False)
self._pv = ca_tuning.make_pv(pvname)
self.alias = Alias(self.name, channel=self.pvname, channeltype="CA")
if has_fields:
self._append(
+76 -8
View File
@@ -22,21 +22,26 @@ client talks to.
## 1. Start it
```bash
eco-status-server start -b # detached, logs to ~/.eco/status_server_<host>.log
eco-status-server wait # block until it reports ready, printing progress
eco-status-server status # state, init progress, memory, running recordings
eco-status-server stop
eco-dev-status-server start -b # detached, logs to ~/.eco/status_server_<host>.log
eco-dev-status-server wait # block until it reports ready, printing progress
eco-dev-status-server status # state, init progress, memory, running recordings
eco-dev-status-server stop
```
`/sf/bernina/bin/eco-status-server` is installed from
[`bin/eco-status-server`](bin/eco-status-server) in this directory. Without
`/sf/bernina/bin/eco-dev-status-server` is a symlink to
[`scripts/eco-status-server`](../../scripts/eco-status-server) at the repo
root (alongside `eco-dev`) - editing the checkout takes effect immediately,
nothing to redeploy. Named `eco-dev-` like `eco-dev` itself: this always
runs the checkout it is symlinked into, not an installed package - see §11
for what that means and how it could become a real, installed `eco-status-
server` command instead. Without
`-b` it runs in the foreground, which is what the systemd unit uses — the
service and the interactive command run exactly the same thing. It reads
`/sf/bernina/config/eco_status_server/env` for the checkout, config and
interpreter to use, and each of those is overridable per invocation:
```bash
ECO_STATUS_SERVER_CHECKOUT=~/my-eco eco-status-server start -b
ECO_STATUS_SERVER_CHECKOUT=~/my-eco eco-dev-status-server start -b
```
The port is bound immediately; `namespace.init_all()` then runs on a
@@ -50,10 +55,24 @@ Without the wrapper it is just:
python -m eco.status_server --mode namespace --config /path/to/bernina_namespace.json
```
### GUI
```bash
eco-dev-status-server gui # small Qt window: status, reinit, query stats
```
Polls `/health` and `/stats` every couple of seconds. Shows state, init
progress, a bold-red banner the moment any *required* component is missing
from `initialized_names` (`failed_required` - see §8), buttons to reinitialize
(`failed`/`full`/`restart`) with a progress bar and ETA while it runs, and a
table of the last `/status/snapshot` and `/status/capture` calls this server
has served - duration, entry count, and any error. Needs a desktop/X session;
`--url` picks a server other than the site default.
### As a systemd user service
```bash
eco-status-server-install-user-service # run this from a shell where CA works
eco-dev-status-server-install-user-service # run this from a shell where CA works
loginctl enable-linger $USER
systemctl --user daemon-reload
systemctl --user enable --now eco-status-server
@@ -396,3 +415,52 @@ Recording numbers and the downthrottling analysis are in DESIGN.md §15.
a path rather than reading the file locally, but it is why a freshly
written `status.json` can `stat` as missing from the console you are
sitting at.
## 11. `eco-dev-status-server` vs a real installed `eco-status-server`
The command is named `eco-dev-status-server`, not `eco-status-server`,
because it currently only *can* mean "run from a development checkout" — the
status-server code lives in a personal checkout
(`/sf/bernina/config/personal/lemke_h/eco`), not the shared gac-bernina one,
so there is no meaningfully different "production, installed" version to
distinguish it from yet. `eco-dev` (this repo's other script) draws exactly
that line already: it always runs the checkout it is symlinked into, ignoring
whatever `eco` package is `pip`/`pixi`-installed in the environment, for the
same reason.
**The two scripts are not the same thing.** `eco-status-server` is the
day-to-day tool: start/stop/status/wait/stats/gui/logs — one running server,
managed. `eco-status-server-install-user-service` is a one-shot generator,
run once (or again with `--force`) to *produce* a `systemd --user` unit file
and environment file for that server — after which `systemctl` manages it,
not this script again. Confusingly similar names for two different jobs, kept
separate on purpose: the daily driver stays a small, dependency-free shell
script, while unit-file generation (capturing `EPICS_CA_*`, writing to
`~/.config/systemd/user/`) is templating logic that does not belong mixed
into it.
**Could either become a real `pyproject.toml` [project.scripts] entry**, so
`pip install eco` gives you an `eco-status-server` command directly (the way
`eco = "eco_cli:main"` already does for the main package)? Only after a
rewrite, not as-is:
- `[project.scripts]` entries are Python callables (`module:function`), not
arbitrary executables — pip generates a tiny wrapper that imports the
module and calls the function. `eco-status-server` is a genuine shell
script (`pgrep`, `systemctl`, `nohup`, signal handling for `stop`) with no
Python equivalent to point at.
- The install script is inherently host-filesystem-shaped (writes into
`~/.config/systemd/user/`, reads `$BASH_SOURCE` to find its sibling) — an
installed console-script would need that logic ported to Python
(`importlib.resources`/`shutil` instead of `dirname "$(readlink -f ...)"`),
which is a real, if mechanical, rewrite.
- The one piece that *is* already plain Python with an argparse `main()` is
the GUI (`eco.status_server.gui:main`) — that one could become
`[project.scripts]` today with a single `pyproject.toml` line, independent
of the other two.
Worth doing once the status-server code actually lands in a checkout meant to
be `pip install`ed rather than run from a specific path — not before, since
today every meaningful default (which checkout, which config) *is* "wherever
this script lives," which a `[project.scripts]` wrapper would have no way to
express.
+49
View File
@@ -17,9 +17,41 @@ from __future__ import annotations
import time
import colorama
import requests
def warn_failed_required(health):
"""Shout, in red, about required components the server could not build.
A component in ``required_names()`` is one the setup is not supposed to
fail. If one did, the status this server serves is missing something
that matters - silently, since every remaining channel still answers
fine - so a client taking status from it needs to be told at the moment
it starts relying on the server, not left to discover the gap in the
file afterwards. Non-required components failing is expected and stays
quiet.
Returns the list it warned about (empty if there was nothing to say),
so a caller can decide to do more than print.
"""
failed = list((health or {}).get("failed_required") or [])
if not failed:
return []
red, reset = colorama.Fore.RED + colorama.Style.BRIGHT, colorama.Style.RESET_ALL
print(
f"{red}!!! status server: {len(failed)} REQUIRED component(s) failed to "
f"initialize: {', '.join(failed)}{reset}",
flush=True,
)
print(
f"{red} Status recorded from this server is missing them. Inspect "
f"with client.failures(), or rebuild with client.reinit().{reset}",
flush=True,
)
return failed
class StatusServerError(RuntimeError):
pass
@@ -76,6 +108,22 @@ class StatusServerClient:
def names(self) -> dict:
return self._get("/names")
def stats(self, limit: int = None, kind: str = None) -> dict:
"""Recent /status/snapshot and /status/capture operations this
server has served: `{"summary": {...}, "recent": [...]}`. See
eco.status_server.query_stats - it answers "is the server serving
requests well", independent of `/health`'s "is the namespace
healthy"."""
path = "/stats"
params = []
if limit:
params.append(f"limit={int(limit)}")
if kind:
params.append(f"kind={kind}")
if params:
path += "?" + "&".join(params)
return self._get(path)
def failures(self) -> dict:
return self._get("/failures")["failures"]
@@ -120,6 +168,7 @@ class StatusServerClient:
f"{h.get('n_failed')} failed",
flush=True,
)
warn_failed_required(h)
return h
if progress:
line = (
+399
View File
@@ -0,0 +1,399 @@
"""A small Qt window for one status server: state, init progress with a
reinitialize button (progress bar + ETA), and a table of recent
/status/snapshot and /status/capture operations.
Standalone on purpose - it talks to the server over plain HTTP
(StatusServerClient), the same as any other client, so it needs no eco
namespace import and starts in under a second. Launch it with:
python -m eco.status_server.gui --url http://saresb-cons-04:8091
or `eco-status-server gui`, which does exactly that, detached.
All network calls run on a background QThread (_Poller) and report back via
Qt signals - never on the GUI thread, so a slow or unreachable server makes
the displayed state go stale (and the connection dot go red) instead of
freezing the window.
"""
from __future__ import annotations
import argparse
import sys
import time
from qtpy import QtCore, QtGui, QtWidgets
from .client import StatusServerClient, StatusServerError, StatusServerNotReady
POLL_HEALTH_S = 2.0
POLL_STATS_S = 5.0
# Palette: a small, fixed set of colours reused across state text, the
# connection dot and stats-table error rows, rather than picking new ones ad
# hoc per widget - see eco's dataviz guidance on consistent, limited colour
# use even outside chart contexts.
COLOR_OK = "#2e7d32"
COLOR_WARN = "#b26a00"
COLOR_BAD = "#b71c1c"
COLOR_MUTED = "#757575"
STATE_COLOR = {
"ready": COLOR_OK,
"importing": COLOR_WARN,
"initializing": COLOR_WARN,
"reinitializing": COLOR_WARN,
"failed": COLOR_BAD,
}
def _fmt_duration(s):
if s is None:
return "?"
if s < 120:
return f"{s:.0f}s"
return f"{s/60:.1f}min"
class _Poller(QtCore.QThread):
"""Background polling loop for one StatusServerClient.
Two independent intervals in one thread rather than two QTimers calling
into `requests` on the GUI thread - a hung/slow HTTP call must not freeze
the window. `health` is polled more often than `stats` since it is what
drives the progress bar during a reinit.
"""
health = QtCore.Signal(dict)
health_failed = QtCore.Signal(str)
stats = QtCore.Signal(dict)
def __init__(self, client: StatusServerClient, parent=None):
super().__init__(parent)
self.client = client
self._stop = False
def stop(self):
self._stop = True
def run(self):
last_stats = 0.0
while not self._stop:
try:
h = self.client.health()
self.health.emit(h)
except Exception as exc: # noqa: BLE001 - reported via signal
self.health_failed.emit(f"{type(exc).__name__}: {exc}")
now = time.time()
if now - last_stats >= POLL_STATS_S:
last_stats = now
try:
self.stats.emit(self.client.stats(limit=30))
except Exception:
pass # the health signal already reports connectivity
for _ in range(int(POLL_HEALTH_S * 10)):
if self._stop:
return
self.msleep(100)
class _ReinitWorker(QtCore.QThread):
"""Runs one blocking client call (reinit/restart) off the GUI thread."""
finished_ok = QtCore.Signal(dict)
finished_error = QtCore.Signal(str)
def __init__(self, fn, parent=None):
super().__init__(parent)
self._fn = fn
def run(self):
try:
result = self._fn()
except Exception as exc: # noqa: BLE001 - reported via signal
self.finished_error.emit(f"{type(exc).__name__}: {exc}")
return
self.finished_ok.emit(result or {})
class StatusServerMonitor(QtWidgets.QWidget):
def __init__(self, base_url: str, parent=None):
super().__init__(parent)
self.base_url = base_url
self.client = StatusServerClient(base_url, timeout=10.0, snapshot_timeout=60.0)
self.setWindowTitle(f"eco status server - {base_url}")
self.resize(720, 560)
self._last_health = {}
self._build_ui()
self._poller = _Poller(self.client, self)
self._poller.health.connect(self._on_health)
self._poller.health_failed.connect(self._on_health_failed)
self._poller.stats.connect(self._on_stats)
self._poller.start()
self._reinit_worker = None
# -- UI construction -----------------------------------------------
def _build_ui(self):
layout = QtWidgets.QVBoxLayout(self)
header = QtWidgets.QHBoxLayout()
self.dot = QtWidgets.QLabel("●") # filled circle
self.dot.setStyleSheet(f"color: {COLOR_MUTED}; font-size: 16px;")
header.addWidget(self.dot)
header.addWidget(QtWidgets.QLabel(f"<b>{self.base_url}</b>"))
header.addStretch(1)
self.updated_label = QtWidgets.QLabel("never updated")
self.updated_label.setStyleSheet(f"color: {COLOR_MUTED};")
header.addWidget(self.updated_label)
layout.addLayout(header)
self.state_label = QtWidgets.QLabel("-")
f = self.state_label.font()
f.setPointSize(f.pointSize() + 3)
f.setBold(True)
self.state_label.setFont(f)
layout.addWidget(self.state_label)
self.progress = QtWidgets.QProgressBar()
self.progress.setRange(0, 100)
self.progress.setTextVisible(True)
layout.addWidget(self.progress)
self.eta_label = QtWidgets.QLabel("")
self.eta_label.setStyleSheet(f"color: {COLOR_MUTED};")
layout.addWidget(self.eta_label)
# The banner the whole task is centred on: only visible while a
# REQUIRED component is missing from initialized_names.
self.required_banner = QtWidgets.QLabel("")
self.required_banner.setWordWrap(True)
self.required_banner.setStyleSheet(
f"background-color: {COLOR_BAD}; color: white; font-weight: bold; "
"padding: 6px; border-radius: 3px;"
)
self.required_banner.hide()
layout.addWidget(self.required_banner)
self.other_failed_label = QtWidgets.QLabel("")
self.other_failed_label.setWordWrap(True)
self.other_failed_label.setStyleSheet(f"color: {COLOR_WARN};")
layout.addWidget(self.other_failed_label)
grid = QtWidgets.QGridLayout()
self.detail_labels = {}
for i, key in enumerate((
"n_direct_read", "n_monitorable", "n_monitored",
"rss_mb", "n_threads", "cpu_seconds",
)):
title = QtWidgets.QLabel(key.replace("_", " "))
title.setStyleSheet(f"color: {COLOR_MUTED};")
value = QtWidgets.QLabel("-")
grid.addWidget(title, i // 3, (i % 3) * 2)
grid.addWidget(value, i // 3, (i % 3) * 2 + 1)
self.detail_labels[key] = value
layout.addLayout(grid)
buttons = QtWidgets.QHBoxLayout()
self.btn_failed = QtWidgets.QPushButton("Reinit failed")
self.btn_full = QtWidgets.QPushButton("Reinit full")
self.btn_restart = QtWidgets.QPushButton("Restart process")
self.btn_failed.clicked.connect(lambda: self._start_reinit("failed"))
self.btn_full.clicked.connect(lambda: self._start_reinit("full"))
self.btn_restart.clicked.connect(lambda: self._start_reinit("restart"))
for b in (self.btn_failed, self.btn_full, self.btn_restart):
buttons.addWidget(b)
buttons.addStretch(1)
layout.addLayout(buttons)
self.action_status = QtWidgets.QLabel("")
self.action_status.setStyleSheet(f"color: {COLOR_MUTED};")
layout.addWidget(self.action_status)
layout.addWidget(QtWidgets.QLabel("<b>Recent queries</b>"))
self.summary_label = QtWidgets.QLabel("no requests served yet")
self.summary_label.setStyleSheet(f"color: {COLOR_MUTED};")
layout.addWidget(self.summary_label)
self.table = QtWidgets.QTableWidget(0, 5)
self.table.setHorizontalHeaderLabels(
["age", "kind", "duration", "entries", "error"]
)
self.table.horizontalHeader().setStretchLastSection(True)
self.table.verticalHeader().setVisible(False)
self.table.setEditTriggers(QtWidgets.QAbstractItemView.NoEditTriggers)
layout.addWidget(self.table, stretch=1)
# -- polling callbacks -----------------------------------------------
def _on_health(self, h: dict):
self._last_health = h
self.updated_label.setText(time.strftime("updated %H:%M:%S"))
self.dot.setStyleSheet(f"color: {COLOR_OK}; font-size: 16px;")
state = h.get("state", "?")
self.state_label.setText(
f"{state} (generation {h.get('generation')}, "
f"up {h.get('uptime_s', 0)/60:.1f} min)"
)
self.state_label.setStyleSheet(
f"color: {STATE_COLOR.get(state, COLOR_MUTED)};"
)
n_init, n_target = h.get("n_initialized"), h.get("n_target_names")
elapsed = h.get("state_seconds") or 0
if n_target:
frac = (n_init or 0) / n_target
self.progress.setRange(0, 100)
self.progress.setValue(int(round(frac * 100)))
self.progress.setFormat(f"{n_init}/{n_target} components %p%")
busy = state in ("importing", "initializing", "reinitializing")
if busy and frac > 0:
eta = elapsed * (1 - frac) / frac
self.eta_label.setText(
f"elapsed {_fmt_duration(elapsed)}, eta ~{_fmt_duration(eta)}"
)
elif busy:
self.eta_label.setText(f"elapsed {_fmt_duration(elapsed)}, eta unknown")
else:
self.eta_label.setText("")
else:
self.progress.setRange(0, 0) # indeterminate: nothing to project yet
self.eta_label.setText(f"elapsed {_fmt_duration(elapsed)}")
failed_required = h.get("failed_required") or []
if failed_required:
self.required_banner.setText(
f"⚠ {len(failed_required)} REQUIRED component(s) failed to "
f"initialize: {', '.join(failed_required)}"
)
self.required_banner.show()
else:
self.required_banner.hide()
other_failed = [n for n in (h.get("failed_names") or [])
if n not in failed_required]
self.other_failed_label.setText(
f"failed (not required): {', '.join(other_failed)}" if other_failed else ""
)
for key, label in self.detail_labels.items():
v = h.get(key)
if key == "rss_mb" and v is not None:
label.setText(f"{v:.0f} MB")
elif key == "cpu_seconds" and v is not None:
label.setText(_fmt_duration(v))
else:
label.setText("-" if v is None else str(v))
busy = state in ("importing", "initializing", "reinitializing")
for b in (self.btn_failed, self.btn_full, self.btn_restart):
b.setEnabled(not busy and self._reinit_worker is None)
def _on_health_failed(self, message: str):
self.dot.setStyleSheet(f"color: {COLOR_BAD}; font-size: 16px;")
self.updated_label.setText(f"unreachable: {message}")
def _on_stats(self, d: dict):
s = d.get("summary") or {}
if not s.get("n"):
self.summary_label.setText("no requests served yet")
else:
avg = s.get("avg_duration_s")
self.summary_label.setText(
f"{s['n']} operation(s), {s['n_errors']} error(s)"
+ (f", avg {avg:.2f}s" if avg is not None else "")
)
rows = list(reversed(d.get("recent") or []))
self.table.setRowCount(len(rows))
now = time.time()
for r, entry in enumerate(rows):
age = now - entry.get("at", now)
dur = entry.get("duration_s")
n = entry.get("n_entries")
err = entry.get("error") or ""
values = [
f"{age:.0f}s ago", entry.get("kind", "?"),
f"{dur:.2f}s" if dur is not None else "?",
"" if n is None else str(n), err,
]
for col, val in enumerate(values):
item = QtWidgets.QTableWidgetItem(val)
if err:
item.setForeground(QtGui.QColor(COLOR_BAD))
self.table.setItem(r, col, item)
self.table.resizeColumnsToContents()
# -- actions -----------------------------------------------------------
def _start_reinit(self, mode: str):
if self._reinit_worker is not None:
return
for b in (self.btn_failed, self.btn_full, self.btn_restart):
b.setEnabled(False)
self.action_status.setText(f"reinit ({mode}) requested ...")
self.action_status.setStyleSheet(f"color: {COLOR_WARN};")
def call():
return self.client.reinit(mode=mode, wait=True, timeout=1800,
progress=False)
self._reinit_worker = _ReinitWorker(call, self)
self._reinit_worker.finished_ok.connect(self._on_reinit_ok)
self._reinit_worker.finished_error.connect(self._on_reinit_error)
self._reinit_worker.start()
def _on_reinit_ok(self, health: dict):
self._reinit_worker = None
self.action_status.setText(
f"reinit finished: {health.get('n_initialized')}/"
f"{health.get('n_target_names')} initialized, "
f"{health.get('n_failed')} failed"
)
self.action_status.setStyleSheet(f"color: {COLOR_OK};")
def _on_reinit_error(self, message: str):
self._reinit_worker = None
self.action_status.setText(f"reinit failed: {message}")
self.action_status.setStyleSheet(f"color: {COLOR_BAD};")
def closeEvent(self, event):
# A reinit is a blocking network call that can run for minutes
# (client.reinit(wait=True)); destroying the QThread while it is
# still running is a hard crash (observed directly, not
# hypothetically), so the window refuses to close under it instead.
# No modal dialog here on purpose - that would block the whole
# process on a click while the reinit keeps running regardless;
# the status label already says why nothing happened.
if self._reinit_worker is not None and self._reinit_worker.isRunning():
self.action_status.setText(
"cannot close: a reinit/restart is still running on the server"
)
self.action_status.setStyleSheet(f"color: {COLOR_BAD};")
event.ignore()
return
self._poller.stop()
self._poller.wait(2000)
super().closeEvent(event)
def main(argv=None):
parser = argparse.ArgumentParser(
description="Qt status/reinit monitor for one eco.status_server"
)
parser.add_argument("--url", required=True,
help="server base URL, e.g. http://saresb-cons-04:8091")
args = parser.parse_args(argv)
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(sys.argv)
win = StatusServerMonitor(args.url)
win.show()
return app.exec_()
if __name__ == "__main__":
raise SystemExit(main())
+103 -69
View File
@@ -27,6 +27,7 @@ from flask import Flask, jsonify, request
from .config import NamespaceServerConfig
from .namespace_store import READY, NamespaceMonitorStore, NotReady, ReinitInProgress
from .query_stats import QueryStats
from .storage import json_default, write_monitor_recording, write_status_snapshot
logger = logging.getLogger(__name__)
@@ -127,6 +128,10 @@ def create_namespace_app(
app.config["ECO_INSTANCE_ID"] = uuid.uuid4().hex
jobs = {}
jobs_lock = threading.Lock()
# Recent /status/snapshot and /status/capture operations, for the CLI's
# `stats` command and the GUI's query-stats table - see query_stats.py's
# module docstring for why this is separate from /health.
stats = QueryStats()
def _health_body():
report = store.connection_report()
@@ -147,6 +152,20 @@ def create_namespace_app(
def health():
return jsonify(_health_body())
@app.get("/stats")
def query_stats_endpoint():
"""Recent /status/snapshot, /status/capture and async /status/job
write operations - separate from /health, which is about the
namespace, not about whether requests are being served well. See
query_stats.py's module docstring.
"""
limit = request.args.get("limit", type=int)
kind = request.args.get("kind")
return jsonify({
"summary": stats.summary(),
"recent": stats.recent(limit=limit, kind=kind),
})
@app.get("/names")
def names():
ns = store.namespace
@@ -177,89 +196,95 @@ def create_namespace_app(
@app.post("/status/snapshot")
def status_snapshot():
body = request.get_json(force=True, silent=True) or {}
t0 = time.time()
fields = {"save": bool(body.get("save", False))}
error = None
try:
snap = store.snapshot(
allow_stale=bool(body.get("allow_stale", False)),
max_workers=body.get("max_workers"),
)
except NotReady as exc:
# health first, then the error keys: _health_body() carries its
# own "status" field, which would otherwise clobber "error".
return (
jsonify(
{
**_health_body(),
"status": "error",
"state": exc.state,
"message": str(exc),
}
),
503,
)
response = {"namespace": config.module_name, **snap}
if body.get("save", False):
try:
pgroup = body["pgroup"]
run_number = int(body["run_number"])
except (KeyError, TypeError, ValueError):
snap = store.snapshot(
allow_stale=bool(body.get("allow_stale", False)),
max_workers=body.get("max_workers"),
)
except NotReady as exc:
error = f"NotReady: {exc}"
# health first, then the error keys: _health_body() carries
# its own "status" field, which would otherwise clobber
# "error".
return (
jsonify(
{
**_health_body(),
"status": "error",
"message": "save=true requires 'pgroup' and 'run_number'",
"state": exc.state,
"message": str(exc),
}
),
400,
503,
)
key = body.get("key", "status_run_start")
directory = config.data_dir(pgroup, run_number)
payload = _status_payload(snap)
if body.get("write_async", False):
job_id = uuid.uuid4().hex
with jobs_lock:
jobs[job_id] = {
"state": "running",
"path": str(directory / "status.json"),
"started_at": time.time(),
}
fields["n_entries"] = len(snap.get("status", {}))
response = {"namespace": config.module_name, **snap}
def _write():
if body.get("save", False):
try:
pgroup = body["pgroup"]
run_number = int(body["run_number"])
except (KeyError, TypeError, ValueError):
error = "save=true requires 'pgroup' and 'run_number'"
return (
jsonify({"status": "error", "message": error}),
400,
)
key = body.get("key", "status_run_start")
fields.update(pgroup=pgroup, run_number=run_number, key=key)
directory = config.data_dir(pgroup, run_number)
payload = _status_payload(snap)
if body.get("write_async", False):
job_id = uuid.uuid4().hex
with jobs_lock:
jobs[job_id] = {
"state": "running",
"path": str(directory / "status.json"),
"started_at": time.time(),
}
def _write():
t0w = time.time()
werror = None
try:
path = write_status_snapshot(directory, payload, key=key)
rec = {"state": "done", "path": str(path)}
except Exception as exc: # noqa: BLE001 - via HTTP
logger.error("async status write failed", exc_info=True)
werror = f"{type(exc).__name__}: {exc}"
rec = {"state": "error", "error": werror}
rec["finished_at"] = time.time()
with jobs_lock:
jobs[job_id].update(rec)
stats.record("write", time.time() - t0w, error=werror,
pgroup=pgroup, run_number=run_number, key=key)
threading.Thread(
target=_write, name=f"status-write-{job_id[:8]}", daemon=True
).start()
response["write_job_id"] = job_id
response["saved_to"] = str(directory / "status.json")
else:
try:
path = write_status_snapshot(directory, payload, key=key)
rec = {"state": "done", "path": str(path)}
except Exception as exc: # noqa: BLE001 - reported via HTTP
logger.error("async status write failed", exc_info=True)
rec = {"state": "error", "error": f"{type(exc).__name__}: {exc}"}
rec["finished_at"] = time.time()
with jobs_lock:
jobs[job_id].update(rec)
except Exception as exc: # noqa: BLE001
logger.error("status write failed", exc_info=True)
error = f"could not write status file: {type(exc).__name__}: {exc}"
return (
jsonify({"status": "error", "message": error}),
500,
)
response["saved_to"] = str(path)
threading.Thread(
target=_write, name=f"status-write-{job_id[:8]}", daemon=True
).start()
response["write_job_id"] = job_id
response["saved_to"] = str(directory / "status.json")
else:
try:
path = write_status_snapshot(directory, payload, key=key)
except Exception as exc: # noqa: BLE001
logger.error("status write failed", exc_info=True)
return (
jsonify(
{
"status": "error",
"message": f"could not write status file: "
f"{type(exc).__name__}: {exc}",
}
),
500,
)
response["saved_to"] = str(path)
return jsonify(response)
return jsonify(response)
finally:
stats.record("snapshot", time.time() - t0, error=error, **fields)
def _append_aux(pgroup, run_number, files):
"""Hand files to sf_daq_broker's copy_user_files, the same call
@@ -327,6 +352,7 @@ def create_namespace_app(
def _capture():
rec = {}
t0_total = time.time()
try:
t0 = time.time()
snap = store.snapshot()
@@ -361,6 +387,14 @@ def create_namespace_app(
rec["finished_at"] = time.time()
with jobs_lock:
jobs[job_id].update(rec)
stats.record(
"capture", time.time() - t0_total, error=rec.get("error"),
pgroup=pgroup, run_number=run_number, key=key,
n_entries=rec.get("n_status"),
snapshot_s=rec.get("snapshot_seconds"),
write_s=rec.get("write_seconds"),
upload_s=rec.get("upload_seconds"),
)
threading.Thread(
target=_capture, name=f"status-capture-{job_id[:8]}", daemon=True
+10
View File
@@ -542,11 +542,19 @@ class NamespaceMonitorStore:
ns = self.namespace
n_init = n_failed = None
failed = []
failed_required = []
if ns is not None:
try:
n_init = len(self._target_names & set(ns.initialized_names))
failed = sorted(self._target_names & set(ns.failed_names))
n_failed = len(failed)
# A failed component that is in required_names() is the one
# a client has to be told about loudly: the setup is not
# supposed to fail those, so one of them missing means the
# status this server serves is incomplete in a way that
# matters, not merely in a way that is expected.
required = set(ns.required_names())
failed_required = sorted(set(failed) & required)
except Exception:
logger.debug("could not compute init progress", exc_info=True)
return {
@@ -563,6 +571,8 @@ class NamespaceMonitorStore:
"n_initialized": n_init,
"n_failed": n_failed,
"failed_names": failed,
"failed_required": failed_required,
"n_failed_required": len(failed_required),
"last_error": self.last_error,
"last_init_seconds": self.last_init_seconds,
"last_init_finished": self.last_init_finished,
+102
View File
@@ -0,0 +1,102 @@
"""A small ring buffer of recent status-server operations.
Exists for exactly one question, asked from the CLI/GUI rather than by
grepping the log: "is this server actually serving fast, and did anything
just fail?" `/health` answers "is the namespace healthy"; this answers "is
answering requests healthy" - the two are independent (a server can hold a
perfectly good namespace and still be slow or erroring on writes because the
NFS mount it writes status.json to is having a bad day).
Kept in-process, not persisted: a few hundred entries is enough for "what
just happened", and surviving a restart is not the point - `/health`'s
`generation`/`last_init_*` already covers "what happened across restarts".
"""
from __future__ import annotations
import threading
import time
from collections import deque
class QueryStats:
def __init__(self, maxlen: int = 200):
self._lock = threading.Lock()
self._history = deque(maxlen=maxlen)
def record(self, kind: str, duration_s: float, error: str | None = None,
**fields) -> dict:
"""Add one completed operation. `fields` is whatever is worth
showing for that `kind` - e.g. `n_entries`, `pgroup`, `run_number`,
`key`, `snapshot_s`/`write_s`/`upload_s` for a capture job."""
entry = {
"kind": kind,
"at": time.time(),
"duration_s": duration_s,
"error": error,
**fields,
}
with self._lock:
self._history.append(entry)
return entry
def recent(self, limit: int | None = None, kind: str | None = None) -> list[dict]:
with self._lock:
items = list(self._history)
if kind:
items = [i for i in items if i["kind"] == kind]
if limit:
items = items[-limit:]
return items
def summary(self) -> dict:
with self._lock:
items = list(self._history)
if not items:
return {"n": 0, "n_errors": 0}
durations = [i["duration_s"] for i in items if i.get("duration_s") is not None]
errors = [i for i in items if i.get("error")]
last = items[-1]
last_error = next((i for i in reversed(items) if i.get("error")), None)
return {
"n": len(items),
"n_errors": len(errors),
"last_at": last["at"],
"last_kind": last["kind"],
"last_duration_s": last.get("duration_s"),
"last_error": last.get("error"),
"last_error_at": last_error["at"] if last_error else None,
"avg_duration_s": sum(durations) / len(durations) if durations else None,
"max_duration_s": max(durations) if durations else None,
"min_duration_s": min(durations) if durations else None,
}
class timed:
"""Context manager: measure a block and record it on `stats` when done,
whichever way it ends.
with timed(stats, "snapshot", n_entries=len(snap["status"])) as t:
snap = store.snapshot()
t.fields["n_entries"] = len(snap["status"])
An exception inside the block is recorded as the operation's error
(str(exc)) and re-raised unchanged - this never swallows anything.
"""
def __init__(self, stats: QueryStats, kind: str, **fields):
self.stats = stats
self.kind = kind
self.fields = fields
self.error = None
def __enter__(self):
self._t0 = time.time()
return self
def __exit__(self, exc_type, exc, tb):
if exc is not None:
self.error = f"{exc_type.__name__}: {exc}"
self.stats.record(self.kind, time.time() - self._t0, error=self.error,
**self.fields)
return False
+87 -17
View File
@@ -1,4 +1,6 @@
from epics import caget_many
import time
from epics import PV, caget_many
from ..elements.adjustable import AdjustableMemory, AdjustableVirtual
from ..elements.detector import DetectorVirtual
from ..epics_utils.adjustable import (
@@ -217,7 +219,8 @@ class MasterEventSystem(Assembly):
)
self.event_codes[code] = self.__dict__[f"code{code:03d}"]
def _get_slot_codes(self, slots=range(1, 257), attempts=3, timeout=3.0):
def _get_slot_codes(self, slots=range(1, 257), attempts=3, timeout=3.0,
connect_timeout=1.0):
"""Read the master's slot->event-code table.
`caget_many` reports a slot it could not read as `None`, and those
@@ -228,28 +231,68 @@ class MasterEventSystem(Assembly):
the life of the session - after which every EVR pulser wired to a
dropped code came up without its delay/frequency chain, blaming "code
missing in Timing Master" for what was really a timed-out read.
Retry the missing ones (they are a handful, so this is cheap) and say
so if any are still missing, rather than quietly shipping a partial
table. Dependency-ordered init makes this much less likely to trigger
in the first place, but the silent-drop path is worth closing anyway.
Retry the missing ones and say so if any are still missing, rather
than quietly shipping a partial table.
**A slot that is simply not configured is not a failure.** Measured
on SIN-TIMAST-TMA (2026-09-06): exactly 74 of the 256 slots exist;
the other 182 have no record on the IOC at all - their PVs do not
connect even given 5 s on a completely idle network, and the count
is identical inside and outside `init_all()`. Retrying those is
pointless (three `caget_many` passes over 182 non-existent channels
cost ~15 s of every namespace init) and warning about them is a false
alarm that has been firing at every session start.
So the two cases are separated by *connection*, which is the only
thing that distinguishes them - `caget_many` reports both as `None`.
A channel that does not connect is an unconfigured slot: skipped
silently. A channel that connects but whose read came back empty is
a genuine timed-out read, which is what the retry and the warning
are for.
"""
slots = list(slots)
pvs = [f"{self.pvname}:Evt-{slot}-Code-SP" for slot in slots]
codes = list(caget_many(pvs, timeout=timeout))
missing = [i for i, c in enumerate(codes) if c is None]
unconfigured = set()
if missing:
# Create the channels non-blockingly and let libca resolve them
# in the background, then ask once - far cheaper than a
# wait_for_connection() per channel.
probes = {i: PV(pvs[i], connection_timeout=connect_timeout,
auto_monitor=False) for i in missing}
deadline = time.time() + connect_timeout
while time.time() < deadline and not all(
p.connected for p in probes.values()
):
time.sleep(0.05)
unconfigured = {i for i, p in probes.items() if not p.connected}
retryable = [i for i in missing if i not in unconfigured]
for _ in range(max(int(attempts) - 1, 0)):
missing = [i for i, c in enumerate(codes) if c is None]
if not missing:
if not retryable:
break
retried = caget_many([pvs[i] for i in missing], timeout=timeout)
for i, c in zip(missing, retried):
retried = caget_many([pvs[i] for i in retryable], timeout=timeout)
for i, c in zip(retryable, retried):
codes[i] = c
still_missing = [slots[i] for i, c in enumerate(codes) if c is None]
retryable = [i for i in retryable if codes[i] is None]
if unconfigured:
logger.debug(
"timing master %s: %d of %d event-code slots are not "
"configured (no record on the IOC) and were skipped; %d in "
"use.",
self.pvname, len(unconfigured), len(slots),
len(slots) - len(unconfigured),
)
still_missing = [slots[i] for i in retryable]
if still_missing:
logger.warning(
"timing master %s: %d of %d event-code slots could not be "
"read after %d attempts (slots %s%s); event codes served by "
"them will look missing to every EVR pulser using them.",
"timing master %s: %d of %d event-code slots connected but "
"could not be read after %d attempts (slots %s%s); event "
"codes served by them will look missing to every EVR pulser "
"using them.",
self.pvname,
len(still_missing),
len(slots),
@@ -465,6 +508,27 @@ class DummyPulser(Assembly):
self._append(AdjustableMemory, None, name="width")
_shared_dummy_pulser = None
def _get_shared_dummy_pulser():
"""The one `DummyPulser` instance for the whole process.
An out-of-range pulser number is a routine, expected IOC state (an unwired
output) rather than a per-output failure, so there is nothing output- or
EVR-specific to preserve by giving each affected output its own instance.
A full `init_all()` can hit this on a few dozen outputs at once (as it does
on the real Bernina EVR0, all wired to the sentinel 65535), and each fresh
`DummyPulser()` builds and appends eight `AdjustableMemory` children for no
behavioural difference from any other dummy -- one shared, lazily-built
instance avoids that multiplied-by-outputs construction cost.
"""
global _shared_dummy_pulser
if _shared_dummy_pulser is None:
_shared_dummy_pulser = DummyPulser()
return _shared_dummy_pulser
class EvrOutput(Assembly):
def __init__(self, pv_base, pulsers=None, name=None):
super().__init__(name=name)
@@ -650,7 +714,11 @@ class EvrOutput(Assembly):
try:
return self._pulsers[number]
except (IndexError, TypeError):
logger.warning(
# An unwired output (number outside the EVR's pulser range, e.g.
# the 65535 sentinel) is routine IOC state, not a failure worth
# surfacing by default -- see the docstring above and the shared
# dummy singleton this returns.
logger.debug(
"output %s (%s): %s number %r does not address any of the %d "
"pulsers of this EVR; using a dummy pulser.",
self.name,
@@ -659,7 +727,7 @@ class EvrOutput(Assembly):
number,
len(self._pulsers or ()),
)
return DummyPulser()
return _get_shared_dummy_pulser()
def update_pulsers(self):
"""Re-read which pulsers this output is wired to (they are otherwise
@@ -684,12 +752,14 @@ class EventReceiver(Assembly):
n_pulsers=24,
n_output_front=8,
n_output_rear=16,
has_evr_sequencer=True,
name=None,
):
super().__init__(name=name)
self.pvname = pvname
self._append(EvrSequencer,self.pvname,name='sequencer', is_display=True, is_setting=True)
if has_evr_sequencer:
self._append(EvrSequencer,self.pvname,name='sequencer', is_display=True, is_setting=True)
pulsers = []
+46 -19
View File
@@ -132,21 +132,50 @@ class NamespaceComponent:
return obj
def replace_NamespaceComponents(*args, **kwargs):
args_out = []
kwargs_out = {}
def _replace_NamespaceComponent(value):
"""Recurse into plain list/tuple/dict containers so a NamespaceComponent
stays replaceable even when passed as e.g. ``default_counters=[NamespaceComponent(...)]``
rather than directly as a keyword value -- see `replace_NamespaceComponents`,
which previously only checked one level of args/kwargs and silently left
any NamespaceComponent nested inside a container unresolved (a plain,
useless NamespaceComponent instance ends up where a real Proxy'd object
was expected). Only these three container types are unwrapped/rebuilt;
anything else (including other iterables, e.g. numpy arrays) is passed
through untouched.
"""
if isinstance(value, NamespaceComponent):
return Proxy(value.get)
elif isinstance(value, list):
return [_replace_NamespaceComponent(v) for v in value]
elif isinstance(value, tuple):
return tuple(_replace_NamespaceComponent(v) for v in value)
elif isinstance(value, dict):
return {k: _replace_NamespaceComponent(v) for k, v in value.items()}
else:
return value
for arg in args:
if isinstance(arg, NamespaceComponent):
args_out.append(Proxy(arg.get))
else:
args_out.append(arg)
pass
for name, value in kwargs.items():
if isinstance(value, NamespaceComponent):
kwargs_out[name] = Proxy(value.get)
else:
kwargs_out[name] = value
def _find_NamespaceComponents(value):
"""Collect every NamespaceComponent in `value`, recursing into plain
list/tuple/dict containers the same way `_replace_NamespaceComponent`
does -- used for declared-dependency tracking, which had the same
top-level-only blind spot as the resolution it mirrors.
"""
found = []
if isinstance(value, NamespaceComponent):
found.append(value)
elif isinstance(value, (list, tuple)):
for v in value:
found.extend(_find_NamespaceComponents(v))
elif isinstance(value, dict):
for v in value.values():
found.extend(_find_NamespaceComponents(v))
return found
def replace_NamespaceComponents(*args, **kwargs):
args_out = [_replace_NamespaceComponent(arg) for arg in args]
kwargs_out = {name: _replace_NamespaceComponent(value) for name, value in kwargs.items()}
return args_out, kwargs_out
@@ -2080,11 +2109,9 @@ class Namespace(Assembly):
"module_name": module_name,
"obj_factory": obj_factory,
}
self._declared_dependencies[name] = [
a
for a in list(args) + list(kwargs.values())
if isinstance(a, NamespaceComponent)
]
self._declared_dependencies[name] = _find_NamespaceComponents(
list(args) + list(kwargs.values())
)
if lazy:
def init_local():
+59 -1
View File
@@ -122,6 +122,17 @@ def _name_of_gid(gid):
return str(gid)
_NOGROUP_RE = re.compile(r"(?:^|[-_])nogroup$", re.IGNORECASE)
def _is_nogroup_sentinel(gid):
"""True if `gid` names a ``nogroup``-shaped group (``nogroup``,
``unx-nogroup``, ...) -- the fallback primary group a personal account's
files land in under a directory that lost its setgid bit, never a group
anyone chose on purpose. See `target_group_of_path`."""
return bool(_NOGROUP_RE.search(_name_of_gid(gid)))
def pgroup_of_path(path):
"""The pgroup owning `path`, or None if it isn't inside a pgroup tree.
@@ -179,24 +190,71 @@ def target_group_of_path(path):
Only the **nearest existing ancestor directory** is consulted, and only if
this process is a member of its group: a grandparent's group is not the
local convention, and a group we are not in cannot be set anyway.
One exception: if that nearest ancestor's group is itself a ``nogroup``
sentinel, one further level up is checked for a real group before falling
back to it. ``nogroup`` is never a deliberately-chosen shared group -- it
is the literal symptom of the very corruption this function exists to stop
propagating (see above): a directory that a personal account happened to
create under a non-setgid parent lands owned by that account's own
``nogroup``-shaped primary group. Accepting it as "the local convention"
would keep spreading it onto every new sibling/child written next to the
already-broken directory -- and would do so silently, since most accounts
on this beamline are themselves members of ``unx-nogroup``, so the
ordinary "not a member" guard above never catches it. The one-level
lookup recovers the real, intentionally-set group directly above the
broken directory (e.g. ``eco_cnf_bernina/memory`` itself, one level above
a device directory that lost it) -- exactly the group
`ensure_group_writable`/`repair_tree` are trying to restore on that
directory anyway. It deliberately does not walk further than one extra
level: a tree that is genuinely, uniformly ``nogroup`` all the way up (an
ordinary personal ``/tmp``, say) keeps exactly its previous behaviour.
"""
pgroup = pgroup_of_path(path)
if pgroup is not None:
return pgroup
for parent in Path(path).absolute().parents:
ancestors = list(Path(path).absolute().parents)
for idx, parent in enumerate(ancestors):
try:
st = os.stat(parent)
except OSError:
continue # does not exist yet (mkdir -p is about to create it)
if not stat.S_ISDIR(st.st_mode):
continue
if _is_nogroup_sentinel(st.st_gid):
better = _real_group_one_level_up(ancestors[idx + 1 :])
if better is not None:
return better
if st.st_gid in _process_gids():
return _name_of_gid(st.st_gid)
return None
return None
def _real_group_one_level_up(remaining_ancestors):
"""The group of the nearest existing, non-``nogroup`` directory in
`remaining_ancestors` that this process is a member of, or None.
Only consulted from `target_group_of_path` when the nearest ancestor is
itself a `nogroup` sentinel -- see there for why a single extra level is
enough.
"""
if not remaining_ancestors:
return None
try:
st = os.stat(remaining_ancestors[0])
except OSError:
return None
if not stat.S_ISDIR(st.st_mode):
return None
if _is_nogroup_sentinel(st.st_gid):
return None
if st.st_gid in _process_gids():
return _name_of_gid(st.st_gid)
return None
def acl_grants_group_write(path, group=None):
"""Best-effort "is group write already granted by a POSIX ACL?".
+3
View File
@@ -102,6 +102,9 @@ class TimeoutPath:
def get_path(self) -> Path:
return self._path
def __fspath__(self) -> str:
return self._path.__fspath__()
def __getattr__(self, name: str) -> Any:
return getattr(self._path, name)
+106 -7
View File
@@ -1,8 +1,9 @@
#!/bin/bash
# eco status server (namespace mode) - start/stop/status wrapper.
# eco status server (namespace mode) - start/stop/status/gui wrapper.
#
# Canonical copy lives in the eco repo at eco/status_server/bin/; the copy in
# /sf/bernina/bin is installed from there (see install-user-service).
# Canonical copy lives in the eco repo at scripts/, alongside eco-dev;
# /sf/bernina/bin/eco-status-server is a symlink to this file, so editing
# the checkout takes effect immediately - nothing to redeploy.
#
# Runs in the FOREGROUND by default, which is what systemd's Type=simple
# wants and what the accompanying user unit calls. Use `-b` to detach for
@@ -46,6 +47,8 @@ usage: $(basename "$0") <command> [options]
status query $URL/health
wait [timeout] block until the server reports ready
logs [-f] show the detached-mode log
stats recent /status/snapshot & /status/capture calls
gui launch the Qt status/reinit GUI (detached)
config print the resolved configuration
Environment (current values):
@@ -68,7 +71,7 @@ check_prereqs() {
if [ -z "${EPICS_CA_ADDR_LIST:-}" ]; then
echo "$(basename "$0"): warning: EPICS_CA_ADDR_LIST is not set;" >&2
echo " most components will fail to connect. See the EnvironmentFile" >&2
echo " written by eco-status-server-install-user-service." >&2
echo " written by the install-user-service script alongside this one." >&2
fi
[ -r "$CONFIG" ] || die "no server config at $CONFIG (set ECO_STATUS_SERVER_CONFIG)"
[ -x "$PYTHON" ] || die "no interpreter at $PYTHON (set ECO_STATUS_SERVER_PYTHON)"
@@ -174,6 +177,14 @@ cmd_status() {
"$PYTHON" - "$URL" <<'PYEOF' || true
import json, sys, urllib.request
RED_BOLD = "\033[1;31m"
YELLOW = "\033[33m"
RESET = "\033[0m"
color = sys.stdout.isatty()
def c(s, code):
return f"{code}{s}{RESET}" if color else s
url = sys.argv[1].rstrip("/") + "/health"
try:
with urllib.request.urlopen(url, timeout=15) as r:
@@ -190,8 +201,20 @@ print("serving: {d} status detectors, {m} monitorable".format(
d=h["n_direct_read"], m=h.get("n_monitorable")))
print("process: {r:.0f} MB, {t} threads, {c:.0f} s cpu".format(
r=h.get("rss_mb") or 0, t=h.get("n_threads"), c=h.get("cpu_seconds") or 0))
if h.get("failed_names"):
print("failed: " + ", ".join(h["failed_names"]))
# failed_required: components that are BOTH in required_names() and in
# failed_names(), i.e. the setup is not supposed to fail these - see
# NamespaceMonitorStore.connection_report(). A merely-optional failure
# (failed_names but not required) stays in the plain line below, unhighlighted.
failed_required = h.get("failed_required") or []
other_failed = [n for n in (h.get("failed_names") or []) if n not in failed_required]
if failed_required:
print(c(
"!!! {n} REQUIRED component(s) failed to initialize: {names}".format(
n=len(failed_required), names=", ".join(failed_required)),
RED_BOLD,
))
if other_failed:
print("failed: " + ", ".join(other_failed))
for rec in h.get("recordings", []):
if rec.get("running"):
print("recording {id}: {u} updates, {s} stored".format(
@@ -207,6 +230,36 @@ with urllib.request.urlopen(sys.argv[1].rstrip("/") + "/health", timeout=15) as
PYEOF
}
# One progress+ETA line, from /health alone: elapsed time in the current
# state (state_seconds) and how far init has gotten (n_initialized of
# n_target_names) linearly project how much longer it needs. Rough on
# purpose - components do not all cost the same - but far better than a
# bare spinner for something that can take several minutes.
progress_line() {
"$PYTHON" - "$URL" <<'PYEOF' 2>/dev/null || true
import json, sys, urllib.request
with urllib.request.urlopen(sys.argv[1].rstrip("/") + "/health", timeout=15) as r:
h = json.load(r)
state, elapsed = h["state"], h.get("state_seconds") or 0
i, t = h.get("n_initialized") or 0, h.get("n_target_names") or 0
line = f"{state}: {i}/{t} initialized ({h.get('n_failed')} failed)"
if t and i:
frac = i / t
eta = elapsed * (1 - frac) / frac if frac > 0 else None
bar_width = 24
filled = int(round(bar_width * min(frac, 1.0)))
bar = "#" * filled + "-" * (bar_width - filled)
line += f" [{bar}] {frac*100:5.1f}% elapsed {elapsed:.0f}s"
if eta is not None:
line += f" eta ~{eta:.0f}s"
else:
line += f" elapsed {elapsed:.0f}s"
print(line)
PYEOF
}
cmd_wait() {
# split, not `local a=.. b=$((a))`: under `set -u` bash evaluates the
# arithmetic before the first assignment is visible, and the command
@@ -215,12 +268,56 @@ cmd_wait() {
local end=$((SECONDS + timeout))
while [ $SECONDS -lt $end ]; do
if is_ready; then cmd_status; return 0; fi
cmd_status | sed -n '2p'
progress_line
sleep 15
done
die "not ready after ${timeout}s"
}
cmd_stats() {
"$PYTHON" - "$URL" <<'PYEOF' || true
import json, sys, urllib.request
url = sys.argv[1].rstrip("/") + "/stats"
try:
with urllib.request.urlopen(url, timeout=15) as r:
d = json.load(r)
except Exception as exc:
print("stats: no answer from %s (%s)" % (url, exc))
raise SystemExit(1)
s = d["summary"]
if not s.get("n"):
print("no requests served yet")
raise SystemExit(0)
print(f"{s['n']} operation(s) recorded, {s['n_errors']} error(s)")
if s.get("avg_duration_s") is not None:
print(f"duration: avg {s['avg_duration_s']:.2f}s "
f"min {s['min_duration_s']:.2f}s max {s['max_duration_s']:.2f}s")
if s.get("last_error"):
print(f"last error: {s['last_error']}")
print()
print(f"{'when':>8s} {'kind':<9s} {'dur(s)':>7s} {'entries':>7s} error")
import time
for e in d["recent"][-20:]:
age = time.time() - e["at"]
dur = e.get("duration_s")
dur_s = f"{dur:.2f}" if dur is not None else "?"
n = e.get("n_entries")
err = e.get("error") or ""
print(f"{age:7.0f}s {e['kind']:<9s} {dur_s:>7s} {str(n) if n is not None else '-':>7s} {err}")
PYEOF
}
cmd_gui() {
check_prereqs
export PYTHONPATH="$CHECKOUT${PYTHONPATH:+:$PYTHONPATH}"
nohup "$PYTHON" -m eco.status_server.gui --url "$URL" "$@" </dev/null >/dev/null 2>&1 &
disown
echo "launched (pid $!)"
}
case "${1:-}" in
start) shift; cmd_start "$@" ;;
stop) cmd_stop ;;
@@ -234,6 +331,8 @@ case "${1:-}" in
status) cmd_status ;;
wait) shift; cmd_wait "$@" ;;
logs) shift; [ "${1:-}" = "-f" ] && tail -f "$LOG" || tail -n 100 "$LOG" ;;
stats) cmd_stats ;;
gui) shift; cmd_gui "$@" ;;
config) usage ;;
""|-h|--help|help) usage ;;
*) usage; exit 2 ;;
+9 -4
View File
@@ -70,7 +70,10 @@ def test_multi_source_and_stepscan_binning(bs_worker):
fake_scan = FakeScan(n_steps)
for cb in ctr.callbacks_start_scan:
cb(scan=fake_scan)
assert len(ctr._scan._values) == n_steps
# Bins are open-ended (escape-fel 0.2.7 -- see bs_counter.py's module
# docstring) and created on demand, so none exist yet right after
# start; they grow to n_steps once every step has been visited below.
assert len(ctr._scan._values) == 0
all_pulse_ids = {name: [] for name in ctr._channels}
for step in range(n_steps):
@@ -81,14 +84,16 @@ def test_multi_source_and_stepscan_binning(bs_worker):
assert len(pids) == 10
all_pulse_ids[name].extend(pids)
assert len(ctr._scan._values) == n_steps
for cb in ctr.callbacks_end_scan:
cb(scan=fake_scan)
# Pulse ids strictly increasing and non-overlapping across steps, for
# every channel -- confirms the per-step bins are a clean partition of
# the live stream (this is exactly the case that silently lost data
# with an un-pre-declared/open-ended escape.stream.Scan -- see
# bs_counter.py's module docstring).
# the live stream. This is exactly the case that used to silently lose
# data for a late-joining channel on a shared, open-ended
# escape.stream.Scan before escape-fel 0.2.7.
for name, ids in all_pulse_ids.items():
assert ids == sorted(ids)
assert len(ids) == len(set(ids))
+281
View File
@@ -0,0 +1,281 @@
"""The adaptive channel-access monitor policy in eco.epics_utils.ca_tuning.
Background, from pyepics' own `PV.get_with_metadata`::
if not self.wait_for_connection(timeout=timeout):
return None
if ((not use_monitor) or (not self.auto_monitor) or ...):
metad = ca.get_with_metadata(...)
if metad is None:
return
so `auto_monitor=False` - the old eco-wide default - makes every read a
network round trip with two ways to silently return `None`, while
`auto_monitor=True` with a cached value skips that block entirely. The
policy here monitors by default and demotes only channels that prove too
fast, rather than every caller hand-rolling its own monitor cache.
"""
import json
import time
import types
import pytest
from eco.epics_utils import ca_tuning
class FakePV:
"""Enough of epics.PV for the policy: callbacks, auto_monitor, pvname."""
def __init__(self, pvname, auto_monitor=False, **kwargs):
self.pvname = pvname
self.auto_monitor = auto_monitor
self.kwargs = kwargs
self.callbacks = {}
self._next_index = 0
def add_callback(self, cb, with_ctrlvars=True, **kw):
self.with_ctrlvars = with_ctrlvars
self._next_index += 1
self.callbacks[self._next_index] = cb
return self._next_index
def remove_callback(self, index):
self.callbacks.pop(index, None)
def fire(self, n=1):
for _ in range(n):
for cb in list(self.callbacks.values()):
cb()
@pytest.fixture
def policy(tmp_path, monkeypatch):
"""ca_tuning with its module state isolated per test."""
monkeypatch.setattr(ca_tuning, "AUTO_MONITOR_STATE_FILE",
tmp_path / "fast.json")
monkeypatch.setattr(ca_tuning, "_fast_channels", set())
monkeypatch.setattr(ca_tuning, "_update_counts", {})
monkeypatch.setattr(ca_tuning, "_tracked", {})
monkeypatch.setattr(ca_tuning, "_sensitive_depth", 0)
monkeypatch.setattr(ca_tuning, "_sweeper", object()) # never start a thread
fake_epics = types.ModuleType("epics")
fake_epics.PV = FakePV
monkeypatch.setitem(__import__("sys").modules, "epics", fake_epics)
return ca_tuning
# --------------------------------------------------------------------------
# default policy
def test_pvs_are_monitored_by_default(policy):
pv = policy.make_pv("TEST:SLOW")
assert pv.auto_monitor is True
assert "TEST:SLOW" in policy._tracked
def test_the_rate_counter_does_not_trigger_a_ctrlvars_storm(policy):
"""pyepics defaults add_callback(with_ctrlvars=True), which issues a
blocking get_ctrlvars() per connected PV - across a namespace that is
the get-storm this policy exists to avoid."""
pv = policy.make_pv("TEST:SLOW")
assert pv.with_ctrlvars is False
def test_an_explicit_auto_monitor_still_wins(policy):
pv = policy.make_pv("TEST:X", auto_monitor=False)
assert pv.auto_monitor is False
assert policy._tracked == {}
# --------------------------------------------------------------------------
# learning which channels are too fast
def test_a_fast_channel_is_demoted_and_remembered(policy):
pv = policy.make_pv("TEST:FAST")
pv.fire(100) # 100 updates ...
policy._sweep_once(interval=1.0) # ... in one second = 100 Hz
assert pv.auto_monitor is False
assert pv.callbacks == {}, "the rate counter should be detached too"
assert policy.is_known_fast("TEST:FAST")
assert "TEST:FAST" not in policy._tracked
def test_a_slow_channel_keeps_its_monitor(policy):
pv = policy.make_pv("TEST:SLOW")
pv.fire(3)
policy._sweep_once(interval=1.0)
assert pv.auto_monitor is True
assert not policy.is_known_fast("TEST:SLOW")
def test_a_channel_someone_else_monitors_is_never_demoted(policy):
"""A recording, a Monitor() or a CallbackEpics registers its own
callback; clearing the subscription under it would silently stop its
data."""
pv = policy.make_pv("TEST:FAST")
pv.add_callback(lambda **kw: None) # somebody else's monitor
pv.fire(100)
policy._sweep_once(interval=1.0)
assert pv.auto_monitor is True
assert not policy.is_known_fast("TEST:FAST")
def test_a_known_fast_channel_is_never_subscribed_again(policy):
pv = policy.make_pv("TEST:FAST")
pv.fire(100)
policy._sweep_once(interval=1.0)
again = policy.make_pv("TEST:FAST")
assert again.auto_monitor is False
assert "TEST:FAST" not in policy._tracked
def test_the_learned_list_survives_a_restart(policy, tmp_path):
pv = policy.make_pv("TEST:FAST")
pv.fire(100)
policy._sweep_once(interval=1.0)
written = json.loads((tmp_path / "fast.json").read_text())
assert written == ["TEST:FAST"]
policy._fast_channels.clear()
policy._load_fast_channels()
assert policy.is_known_fast("TEST:FAST")
def test_clearing_forgets_everything(policy):
pv = policy.make_pv("TEST:FAST")
pv.fire(100)
policy._sweep_once(interval=1.0)
assert policy.is_known_fast("TEST:FAST")
policy.clear_fast_channels()
assert not policy.is_known_fast("TEST:FAST")
def test_counts_are_zeroed_not_lost_between_sweeps(policy):
pv = policy.make_pv("TEST:SLOW")
pv.fire(3)
policy._sweep_once(interval=1.0)
assert policy._update_counts["TEST:SLOW"] == 0
pv.fire(2)
assert policy._update_counts["TEST:SLOW"] == 2
# --------------------------------------------------------------------------
# sensitive periods
def test_the_sweeper_leaves_subscriptions_alone_during_an_acquisition(policy):
"""Reconfiguring a monitor mid-acquisition is exactly the wrong moment,
even for a channel that deserves demotion."""
pv = policy.make_pv("TEST:FAST")
pv.fire(100)
with policy.sensitive_period("scan step"):
policy._sweep_once(interval=1.0)
assert pv.auto_monitor is True, "demoted mid-acquisition"
# and it still gets demoted once the window closes
policy._sweep_once(interval=1.0)
assert pv.auto_monitor is False
def test_reads_get_a_more_patient_budget_while_sensitive(policy):
assert policy.read_retries() == policy.CA_READ_RETRIES
with policy.sensitive_period("scan step"):
assert policy.read_retries() == policy.CA_READ_RETRIES_SENSITIVE
assert policy.read_retries() == policy.CA_READ_RETRIES
def test_sensitive_periods_nest(policy):
with policy.sensitive_period("outer"):
with policy.sensitive_period("inner"):
assert policy.in_sensitive_period()
assert policy.in_sensitive_period(), "the inner exit ended both"
assert not policy.in_sensitive_period()
def test_a_raising_body_still_ends_the_period(policy):
with pytest.raises(ValueError):
with policy.sensitive_period("boom"):
raise ValueError("boom")
assert not policy.in_sensitive_period()
# --------------------------------------------------------------------------
# _read_pv: the single chokepoint that stops the next caller needing its own
# private cache
class FlakyPV:
"""A PV that returns None for the first `n_none` reads, then a value."""
def __init__(self, n_none, value=42.0, connected=True,
pvname="TEST:FLAKY"):
self.pvname = pvname
self.connected = connected
self._left = n_none
self._value = value
self.reads = 0
def get(self, timeout=None):
self.reads += 1
if self._left > 0:
self._left -= 1
return None
return self._value
@pytest.fixture
def read_pv(monkeypatch):
from eco.epics_utils import adjustable
monkeypatch.setattr(ca_tuning, "_last_ok", {})
monkeypatch.setattr(ca_tuning, "CA_READ_RETRY_DELAY", 0.0)
monkeypatch.setattr(ca_tuning, "_sensitive_depth", 0)
return adjustable._read_pv
def test_a_transient_none_is_retried_into_a_real_value(read_pv):
"""The recurring bug: pyepics returns None on a momentary CA hiccup
instead of raising, and the caller uses it as data (`int(None)`,
`100 / None`, `event_codes[None]`)."""
ca_tuning.note_successful_read("TEST:FLAKY") # it has worked before
pv = FlakyPV(n_none=1)
assert read_pv(pv, name="flaky") == 42.0
assert pv.reads == 2
def test_retries_are_bounded(read_pv):
ca_tuning.note_successful_read("TEST:FLAKY")
pv = FlakyPV(n_none=99)
assert read_pv(pv, name="flaky") is None
assert pv.reads == 1 + ca_tuning.CA_READ_RETRIES
def test_a_channel_that_never_worked_is_not_retried(read_pv):
"""An absent PV is read on every get_status() fan-out, i.e. once per
scan step - it must not pay for retries it will never win."""
pv = FlakyPV(n_none=99, connected=False, pvname="TEST:ABSENT")
assert read_pv(pv, name="absent") is None
assert pv.reads == 1
def test_a_sensitive_period_buys_more_attempts(read_pv):
ca_tuning.note_successful_read("TEST:FLAKY")
pv = FlakyPV(n_none=3)
with ca_tuning.sensitive_period("scan step"):
assert read_pv(pv, name="flaky") == 42.0
assert pv.reads == 4 # would have given up at 3 outside the period
def test_a_successful_read_needs_no_retry(read_pv):
ca_tuning.note_successful_read("TEST:FLAKY")
pv = FlakyPV(n_none=0)
assert read_pv(pv, name="flaky") == 42.0
assert pv.reads == 1
+8 -2
View File
@@ -78,7 +78,11 @@ def test_never_connected_pv_keeps_the_short_budget():
def test_a_disconnected_but_previously_working_pv_gets_the_full_budget():
"""A dropped virtual circuit is exactly what the longer budget is for."""
"""A dropped virtual circuit is exactly what the longer budget is for -
and, since this channel demonstrably works, also worth retrying: the
retry lives in `_read_pv` so no individual caller has to grow its own
cache the way `Daq.get_pulse_id` and the event-code frequency did."""
from eco.epics_utils import ca_tuning
from eco.epics_utils.adjustable import _read_pv
pv = FakePV([2.0], pvname="SOME:FLAKY:PV", connected=True)
@@ -86,7 +90,9 @@ def test_a_disconnected_but_previously_working_pv_gets_the_full_budget():
pv.connected = False
pv.timeouts.clear()
_read_pv(pv, name="flaky")
assert pv.timeouts == [None], "should use the PV's own connection_timeout"
assert pv.timeouts, "no read was attempted at all"
assert set(pv.timeouts) == {None}, "should use the PV's own connection_timeout"
assert len(pv.timeouts) == 1 + ca_tuning.CA_READ_RETRIES
def test_falsy_values_are_not_treated_as_failures(caplog):
+149
View File
@@ -537,3 +537,152 @@ def test_only_the_start_block_keeps_its_values_on_the_server():
daq.append_status_to_scan_and_store(RunTableScan(runno=11))
assert client.captures[0]["keep_status"] is True
assert client.captures[1]["keep_status"] is False
# --------------------------------------------------------------------------
# rate_multiplicator / get_detector_code_frequency: never a raw CA get, never
# a silent None reaching arithmetic (the pulse_id bug, 4th occurrence)
class FakeFreqMonitor:
"""Stands in for CallbackEpics(func="latest"): .data is the same dict
object the real one mutates in place, so a cached reference stays live."""
def __init__(self, initial=None):
self.data = {"value": initial}
self.started = False
def start(self, add_current_value=True):
self.started = True
def push(self, value):
self.data["value"] = value
class FreqDetector:
def __init__(self, value, pvname="TEST:Evt-1-Freq-I", monitorable=True):
self._value = value
self.pvname = pvname
self._monitor = FakeFreqMonitor(value) if monitorable else None
def get_current_value(self):
return self._value
def set_current_value_callback(self, func="accumulate", **kwargs):
if self._monitor is None:
raise AttributeError("not monitorable")
return self._monitor
def _event_master(freq_detector, code=50):
em = types.SimpleNamespace()
em.__dict__[f"code{code:03d}"] = types.SimpleNamespace(frequency=freq_detector)
return em
def _freq_daq(freq_detector, code=50):
daq = Daq.__new__(Daq)
daq._event_master = None
daq._detectors_event_code = None
daq._frequency_detector = None
daq._frequency_monitor = None
daq._frequency_latest = {"value": None}
event_master = _event_master(freq_detector, code=code)
# replicate the __init__ snippet directly, since Daq.__new__ skips it
try:
daq._frequency_detector = event_master.__dict__[
f"code{code:03d}"
].frequency
mon = daq._frequency_detector.set_current_value_callback(func="latest")
mon.start()
daq._frequency_monitor = mon
daq._frequency_latest = mon.data
except Exception:
pass
daq._detectors_event_code = code
return daq
def test_frequency_is_read_from_the_monitor_cache_not_a_fresh_get():
det = FreqDetector(50.0)
daq = _freq_daq(det)
assert daq._frequency_monitor.started is True
assert daq.get_detector_code_frequency() == 50.0
assert daq.rate_multiplicator == 2
def test_frequency_cache_follows_live_monitor_updates():
det = FreqDetector(50.0)
daq = _freq_daq(det)
det._monitor.push(25.0)
assert daq.get_detector_code_frequency() == 25.0
def test_a_transient_none_from_the_monitor_falls_back_to_a_direct_read():
"""The exact failure mode observed live: pyepics silently returns None
on a transient CA hiccup instead of raising."""
det = FreqDetector(50.0)
daq = _freq_daq(det)
det._monitor.push(None)
# get_current_value() still works even though the monitor cache is
# momentarily empty
assert daq.get_detector_code_frequency() == 50.0
def test_no_value_anywhere_raises_instead_of_dividing_by_none():
"""Before this fix: int(100 / freq) with freq=None -> TypeError, deep
inside retrieve(), killing a real scan mid-run."""
det = FreqDetector(None)
daq = _freq_daq(det)
with pytest.raises(TimeoutError, match="TEST:Evt-1-Freq-I"):
daq.get_detector_code_frequency()
with pytest.raises(TimeoutError):
daq.rate_multiplicator
def test_a_non_monitorable_frequency_falls_back_to_direct_reads():
"""MasterEventCodeFix (fixed CTA sequencer codes) has no PV to monitor
at all - set_current_value_callback isn't there, __init__'s attach must
not blow up, and reads should still work via get_current_value()."""
det = FreqDetector(50.0, monitorable=False)
daq = _freq_daq(det)
assert daq._frequency_monitor is None
assert daq.get_detector_code_frequency() == 50.0
def test_missing_event_master_does_not_crash_init():
daq = Daq.__new__(Daq)
daq._frequency_detector = None
daq._frequency_monitor = None
daq._frequency_latest = {"value": None}
daq._detectors_event_code = None
with pytest.raises(TimeoutError):
daq.get_detector_code_frequency()
# --------------------------------------------------------------------------
# _create_runtable_metadata_append_status_to_runtable must honor
# append_status_info=False (it silently didn't - a regression from the old
# combined elog+run_table callback, which had the guard)
def test_runtable_metadata_callback_skips_when_append_status_info_is_false():
rt = RunTableSpy()
daq = _runtable_daq(HealthClient(), rt)
scan = RunTableScan(runno=99)
daq._create_runtable_metadata_append_status_to_runtable(
scan, append_status_info=False
)
assert rt.calls == [], (
"run_table.append_run() ran despite append_status_info=False - this "
"is what makes a session's first scan pay for Run_Table2's Google "
"Sheets authentication even when status collection was disabled"
)
def test_runtable_metadata_callback_still_runs_by_default():
rt = RunTableSpy()
daq = _runtable_daq(HealthClient(), rt)
scan = RunTableScan(runno=100)
daq._create_runtable_metadata_append_status_to_runtable(scan)
assert len(rt.calls) == 1
+32
View File
@@ -333,6 +333,38 @@ def test_target_group_of_path_uses_the_nearest_existing_ancestor(tmp_path):
)
def test_target_group_of_path_skips_a_nogroup_ancestor(tmp_path, monkeypatch):
"""A directory that lost its setgid bit and was recreated by a personal
account lands owned by that account's own nogroup-shaped primary group --
the eco_cnf_bernina/memory incident described in the module docstring. A
new file written next to/inside that broken directory must not inherit
the broken group; the walk should recover the real group one level up
instead (fully mocked here, since the real gid a `/tmp`-shaped path gets
on a given machine is itself sometimes nogroup, which would make this
indistinguishable from the bug it is testing for)."""
broken = tmp_path / "broken"
broken.mkdir()
real_stat = os.stat
broken_gid, real_group_gid = 999998, 999999
names = {broken_gid: "unx-nogroup", real_group_gid: "unx-sf_bernina_bs"}
def fake_stat(path, *a, **k):
st = real_stat(path, *a, **k)
gid = broken_gid if Path(path) == broken else (
real_group_gid if Path(path) == tmp_path else None
)
if gid is None:
return st
return type("FakeStat", (), {"st_gid": gid, "st_mode": st.st_mode})()
monkeypatch.setattr(df.os, "stat", fake_stat)
monkeypatch.setattr(df, "_name_of_gid", lambda gid: names.get(gid, str(gid)))
monkeypatch.setattr(df, "_process_gids", lambda: frozenset(names))
result = df.target_group_of_path(broken / "new_file.json")
assert result == "unx-sf_bernina_bs"
def test_target_group_of_path_ignores_a_group_the_process_is_not_in(
tmp_path, monkeypatch
):
+59
View File
@@ -0,0 +1,59 @@
"""EvrOutput._resolve_pulser: an out-of-range pulser number is routine IOC
state (an unwired output), not a per-output failure.
`init_all()` on the real Bernina EVR0 hits this on a few dozen outputs at
once, all wired to the sentinel 65535 -- see the module docstring in
`eco.timing.event_timing_new_new`. Two things follow from that being routine:
the message should not be shown by default (WARNING -> DEBUG, matching the
"unconfigured slot" treatment in `MasterEventSystem._get_slot_codes`), and it
should not cost a fresh `DummyPulser` (eight `AdjustableMemory` children) per
occurrence -- one shared, lazily-built instance is enough.
"""
import logging
import pytest
evt = pytest.importorskip("eco.timing.event_timing_new_new")
@pytest.fixture(autouse=True)
def _reset_shared_dummy(monkeypatch):
monkeypatch.setattr(evt, "_shared_dummy_pulser", None)
def _make_output(monkeypatch, number=65535):
monkeypatch.setattr(
evt, "read_pv_value", lambda *a, **k: number
)
output = evt.EvrOutput.__new__(evt.EvrOutput)
output.name = "output_test"
output.pv_base = "TEST:Output"
output._pulsers = [object() for _ in range(24)]
return output
def test_out_of_range_pulser_returns_the_shared_dummy(monkeypatch):
output_a = _make_output(monkeypatch)
output_b = _make_output(monkeypatch)
pulser_a = output_a._resolve_pulser(None, "pulserA")
pulser_b = output_b._resolve_pulser(None, "pulserB")
assert isinstance(pulser_a, evt.DummyPulser)
assert pulser_a is pulser_b, "one shared instance, not one per output"
assert pulser_a is evt._get_shared_dummy_pulser()
def test_out_of_range_pulser_is_not_shown_by_default(monkeypatch, caplog):
output = _make_output(monkeypatch)
with caplog.at_level(logging.WARNING, logger=evt.__name__):
output._resolve_pulser(None, "pulserA")
assert caplog.text == "", "should be invisible at the default log level"
with caplog.at_level(logging.DEBUG, logger=evt.__name__):
output._resolve_pulser(None, "pulserA")
assert "does not address any of the" in caplog.text
+44
View File
@@ -1192,3 +1192,47 @@ def test_capture_is_refused_while_not_ready(fake_module):
)
assert resp.status_code == 503
assert store.wait_ready(timeout=20)
def test_health_reports_failed_required_components_separately(fake_module):
"""A failed component that is in required_names() is the one a client has
to be told about loudly - the setup is not supposed to fail those."""
name, mod = fake_module(names=["a", "b", "c"], required=["a", "b"], fail=["b", "c"])
store = _store(name, init_required_only=False)
assert store.wait_ready(timeout=10)
report = store.connection_report()
assert report["failed_names"] == ["b", "c"]
assert report["failed_required"] == ["b"], "c is not required, b is"
assert report["n_failed_required"] == 1
app = create_namespace_app(NamespaceServerConfig(module_name=name), store=store)
body = app.test_client().get("/health").get_json()
assert body["failed_required"] == ["b"]
def test_no_failed_required_is_reported_as_empty(fake_module):
name, _ = fake_module(names=["a", "b"], required=["a"], fail=["b"])
store = _store(name, init_required_only=False)
assert store.wait_ready(timeout=10)
report = store.connection_report()
assert report["failed_names"] == ["b"]
assert report["failed_required"] == []
def test_the_client_warns_in_red_about_failed_required(capsys):
from eco.status_server.client import warn_failed_required
warned = warn_failed_required({"failed_required": ["mon_und", "scilog"]})
out = capsys.readouterr().out
assert warned == ["mon_und", "scilog"]
assert "REQUIRED" in out and "mon_und" in out and "scilog" in out
assert "\x1b[" in out, "should be colourised"
def test_the_client_stays_quiet_when_nothing_required_failed(capsys):
from eco.status_server.client import warn_failed_required
assert warn_failed_required({"failed_required": [], "failed_names": ["x"]}) == []
assert warn_failed_required({}) == []
assert capsys.readouterr().out == ""
+197
View File
@@ -0,0 +1,197 @@
"""eco.status_server.gui: the display/state logic, driven directly (no real
poller thread, no real server) - _on_health/_on_stats/_start_reinit are
plain slots, so they can be called synchronously with fake payloads."""
import pytest
pytest.importorskip("qtpy")
from qtpy import QtGui, QtWidgets
from eco.status_server.gui import COLOR_BAD, COLOR_OK, StatusServerMonitor
@pytest.fixture(scope="module")
def qapp():
return QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
@pytest.fixture
def win(qapp, monkeypatch):
# Never let the real background poller touch the network during a test.
from eco.status_server import gui as gui_module
monkeypatch.setattr(gui_module._Poller, "start", lambda self: None)
w = StatusServerMonitor("http://fake-host:8091")
yield w
w._poller.stop()
def _health(**overrides):
h = {
"state": "ready", "generation": 3, "uptime_s": 120.0,
"n_initialized": 80, "n_target_names": 86, "n_failed": 6,
"state_seconds": 5.0, "n_direct_read": 16000, "n_monitorable": 10000,
"n_monitored": 0, "rss_mb": 2000.0, "n_threads": 100,
"cpu_seconds": 300.0, "failed_names": [], "failed_required": [],
}
h.update(overrides)
return h
# --------------------------------------------------------------------------
# health rendering
def test_ready_state_is_shown_in_ok_color(win):
win._on_health(_health())
assert "ready" in win.state_label.text()
assert COLOR_OK in win.state_label.styleSheet()
assert win.required_banner.isHidden()
def test_progress_bar_reflects_init_fraction(win):
win._on_health(_health(n_initialized=43, n_target_names=86))
assert win.progress.value() == 50
def test_required_failure_shows_the_red_banner(win):
win._on_health(_health(
state="ready", failed_names=["mon_und", "scilog"],
failed_required=["mon_und", "scilog"],
))
assert not win.required_banner.isHidden()
assert "mon_und" in win.required_banner.text()
assert "scilog" in win.required_banner.text()
assert "REQUIRED" in win.required_banner.text()
# a required failure must not also be double-listed as "other"
assert win.other_failed_label.text() == ""
def test_non_required_failure_does_not_trigger_the_red_banner(win):
win._on_health(_health(failed_names=["xrd"], failed_required=[]))
assert win.required_banner.isHidden()
assert "xrd" in win.other_failed_label.text()
def test_mixed_failures_split_correctly(win):
win._on_health(_health(
failed_names=["mon_und", "xrd"], failed_required=["mon_und"],
))
assert "mon_und" in win.required_banner.text()
assert "xrd" not in win.required_banner.text()
assert "xrd" in win.other_failed_label.text()
assert "mon_und" not in win.other_failed_label.text()
def test_eta_is_shown_while_initializing(win):
win._on_health(_health(
state="initializing", n_initialized=50, n_target_names=100,
state_seconds=10.0,
))
# 50% done in 10s -> another ~10s projected
assert "eta" in win.eta_label.text()
def test_no_eta_once_ready(win):
win._on_health(_health(state="ready"))
assert "eta" not in win.eta_label.text()
def test_buttons_disabled_while_busy(win):
win._on_health(_health(state="initializing"))
assert not win.btn_failed.isEnabled()
win._on_health(_health(state="ready"))
assert win.btn_failed.isEnabled()
def test_unreachable_server_turns_the_dot_red_and_keeps_last_state(win):
win._on_health(_health(state="ready"))
win._on_health_failed("ConnectionError: refused")
assert COLOR_BAD in win.dot.styleSheet()
assert "ready" in win.state_label.text() # stale, but still shown
# --------------------------------------------------------------------------
# stats table
def test_stats_summary_line(win):
win._on_stats({"summary": {"n": 5, "n_errors": 1, "avg_duration_s": 2.5},
"recent": []})
assert "5 operation" in win.summary_label.text()
assert "1 error" in win.summary_label.text()
def test_empty_stats_says_so(win):
win._on_stats({"summary": {"n": 0, "n_errors": 0}, "recent": []})
assert "no requests" in win.summary_label.text()
def test_stats_table_populates_and_colors_errors(win):
win._on_stats({
"summary": {"n": 2, "n_errors": 1},
"recent": [
{"at": 1000.0, "kind": "snapshot", "duration_s": 1.2, "n_entries": 99},
{"at": 1001.0, "kind": "capture", "duration_s": 0.5, "error": "boom"},
],
})
assert win.table.rowCount() == 2
# most recent first
assert win.table.item(0, 1).text() == "capture"
assert win.table.item(0, 4).text() == "boom"
assert win.table.item(0, 4).foreground().color().name() == QtGui.QColor(COLOR_BAD).name()
assert win.table.item(1, 1).text() == "snapshot"
assert win.table.item(1, 4).text() == ""
# --------------------------------------------------------------------------
# reinit action + close guard
def test_reinit_click_disables_buttons_and_calls_the_client(win, qapp):
calls = []
win.client.reinit = lambda **kw: calls.append(kw) or {
"n_initialized": 80, "n_target_names": 86, "n_failed": 6,
}
win.btn_failed.click()
win._reinit_worker.wait(2000)
qapp.processEvents()
assert calls and calls[0]["mode"] == "failed"
assert "reinit finished" in win.action_status.text()
def test_reinit_error_is_shown(win, qapp):
def boom(**kw):
raise RuntimeError("server unreachable")
win.client.reinit = boom
win.btn_full.click()
win._reinit_worker.wait(2000)
qapp.processEvents()
assert "reinit failed" in win.action_status.text()
assert "server unreachable" in win.action_status.text()
def test_window_refuses_to_close_while_a_reinit_is_running(win, qapp):
import threading
release = threading.Event()
win.client.reinit = lambda **kw: (release.wait(2), {})[1]
win.btn_failed.click()
qapp.processEvents()
assert win._reinit_worker.isRunning()
ev = QtGui.QCloseEvent()
win.closeEvent(ev)
assert not ev.isAccepted()
assert "cannot close" in win.action_status.text()
release.set()
win._reinit_worker.wait(2000)
def test_window_closes_normally_when_idle(win):
ev = QtGui.QCloseEvent()
win.closeEvent(ev)
assert ev.isAccepted()
+114
View File
@@ -0,0 +1,114 @@
"""eco.status_server.query_stats: the ring buffer behind GET /stats."""
import pytest
from eco.status_server.query_stats import QueryStats, timed
def test_empty_summary():
s = QueryStats()
assert s.summary() == {"n": 0, "n_errors": 0}
assert s.recent() == []
def test_record_and_recent():
s = QueryStats()
s.record("snapshot", 1.5, n_entries=100)
s.record("capture", 2.5, pgroup="p1")
recent = s.recent()
assert len(recent) == 2
assert recent[0]["kind"] == "snapshot"
assert recent[0]["n_entries"] == 100
assert recent[1]["pgroup"] == "p1"
def test_recent_filters_by_kind():
s = QueryStats()
s.record("snapshot", 1.0)
s.record("capture", 1.0)
s.record("snapshot", 1.0)
assert len(s.recent(kind="snapshot")) == 2
assert len(s.recent(kind="capture")) == 1
def test_recent_respects_limit():
s = QueryStats()
for i in range(10):
s.record("snapshot", float(i))
assert len(s.recent(limit=3)) == 3
assert s.recent(limit=3)[-1]["duration_s"] == 9.0
def test_ring_buffer_drops_oldest():
s = QueryStats(maxlen=3)
for i in range(5):
s.record("snapshot", float(i))
recent = s.recent()
assert len(recent) == 3
assert [r["duration_s"] for r in recent] == [2.0, 3.0, 4.0]
def test_summary_aggregates_durations_and_errors():
s = QueryStats()
s.record("snapshot", 1.0)
s.record("snapshot", 3.0, error="boom")
s.record("snapshot", 2.0)
summary = s.summary()
assert summary["n"] == 3
assert summary["n_errors"] == 1
assert summary["avg_duration_s"] == pytest.approx(2.0)
assert summary["min_duration_s"] == 1.0
assert summary["max_duration_s"] == 3.0
# "last_error" tracks the *last entry's* error (None here - the last
# recorded op succeeded); test_summary_last_error_is_the_most_recent_one
# covers the separate "did anything fail recently" question below.
assert summary["last_error"] is None
assert summary["last_error_at"] is not None # "boom" is still visible via this
def test_summary_last_error_is_the_most_recent_one():
s = QueryStats()
s.record("snapshot", 1.0, error="first")
s.record("snapshot", 1.0) # no error
s.record("snapshot", 1.0, error="second")
s.record("snapshot", 1.0) # no error again
summary = s.summary()
assert summary["last_error"] is None # the very last entry had none
assert summary["last_error_at"] is not None # but we remember the last one that did
def test_timed_context_manager_records_success():
s = QueryStats()
with timed(s, "snapshot", n_entries=5) as t:
t.fields["n_entries"] = 42
entry = s.recent()[0]
assert entry["kind"] == "snapshot"
assert entry["n_entries"] == 42
assert entry["error"] is None
assert entry["duration_s"] >= 0
def test_timed_context_manager_records_and_reraises_errors():
s = QueryStats()
with pytest.raises(ValueError, match="boom"):
with timed(s, "capture"):
raise ValueError("boom")
entry = s.recent()[0]
assert entry["error"] == "ValueError: boom"
def test_thread_safety_of_concurrent_records():
import threading
s = QueryStats(maxlen=1000)
def worker():
for _ in range(100):
s.record("snapshot", 0.1)
threads = [threading.Thread(target=worker) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert s.summary()["n"] == 800
+113
View File
@@ -0,0 +1,113 @@
"""MasterEventSystem._get_slot_codes: an unconfigured slot is not a failure.
Measured on SIN-TIMAST-TMA (2026-09-06): exactly 74 of 256 event slots
exist. The other 182 have no record on the IOC - their PVs do not connect
even given 5 s on an idle network, and the count is identical inside and
outside `init_all()`. The old code could not tell that apart from a
timed-out read, so it retried all 182 three times (~15 s of every namespace
init) and then warned alarmingly about a completely normal configuration.
"""
import logging
import types
import pytest
evt = pytest.importorskip("eco.timing.event_timing_new_new")
class ProbePV:
"""A PV that connects only if its slot is in `configured`."""
configured = set()
def __init__(self, pvname, connection_timeout=None, auto_monitor=None):
self.pvname = pvname
slot = int(pvname.split("Evt-")[1].split("-")[0])
self.connected = slot in ProbePV.configured
@pytest.fixture
def master(monkeypatch):
m = evt.MasterEventSystem.__new__(evt.MasterEventSystem)
m.pvname = "TEST-TMA"
monkeypatch.setattr(evt, "PV", ProbePV)
monkeypatch.setattr(evt.time, "sleep", lambda s: None)
return m
def _install_caget(monkeypatch, table, flaky=()):
"""table: slot -> code (absent slot => None). `flaky` slots return None
on the first read and their code afterwards."""
seen = {"calls": 0}
state = {s: 0 for s in flaky}
def caget_many(pvs, timeout=None):
seen["calls"] += 1
out = []
for pv in pvs:
slot = int(pv.split("Evt-")[1].split("-")[0])
if slot in state:
state[slot] += 1
out.append(table.get(slot) if state[slot] > 1 else None)
else:
out.append(table.get(slot))
return out
monkeypatch.setattr(evt, "caget_many", caget_many)
return seen
def test_unconfigured_slots_are_skipped_without_retry_or_warning(
master, monkeypatch, caplog
):
table = {1: 10, 2: 20, 3: 30} # only 3 of 6 exist
ProbePV.configured = set(table)
seen = _install_caget(monkeypatch, table)
with caplog.at_level(logging.DEBUG, logger=evt.__name__):
slots, codes = master._get_slot_codes(slots=range(1, 7))
assert sorted(codes) == [10, 20, 30]
assert seen["calls"] == 1, "non-existent slots must not be retried"
text = caplog.text
assert "not configured" in text and "3 in use" in text
assert "could not be read" not in text, "that would be the false alarm"
def test_a_connected_but_unread_slot_is_retried_and_warned_about(
master, monkeypatch, caplog
):
"""The case the retry and the warning actually exist for."""
table = {1: 10, 2: 20}
ProbePV.configured = {1, 2, 3} # slot 3 exists but never reads
seen = _install_caget(monkeypatch, table)
with caplog.at_level(logging.INFO):
slots, codes = master._get_slot_codes(slots=range(1, 4), attempts=3)
assert sorted(codes) == [10, 20]
assert seen["calls"] == 3, "a connected slot should be retried"
assert "connected but" in caplog.text and "could not be read" in caplog.text
def test_a_transient_read_recovers_on_retry(master, monkeypatch, caplog):
table = {1: 10, 2: 20, 3: 30}
ProbePV.configured = set(table)
seen = _install_caget(monkeypatch, table, flaky=[3])
with caplog.at_level(logging.INFO):
slots, codes = master._get_slot_codes(slots=range(1, 4), attempts=3)
assert sorted(codes) == [10, 20, 30], "the retry should have recovered slot 3"
assert seen["calls"] == 2
assert "could not be read" not in caplog.text
def test_everything_present_needs_no_probing_at_all(master, monkeypatch):
table = {1: 10, 2: 20, 3: 30}
ProbePV.configured = set(table)
seen = _install_caget(monkeypatch, table)
slots, codes = master._get_slot_codes(slots=range(1, 4))
assert seen["calls"] == 1
assert sorted(codes) == [10, 20, 30]