From 8b73ee2d8d3258a3e519aa2cbeac1ff1365f22d2 Mon Sep 17 00:00:00 2001 From: lemke_h Date: Sat, 5 Sep 2026 17:34:09 +0200 Subject: [PATCH] fix again on init all, hopefully --- eco/acquisition/counters.py | 74 ++++++++++++ eco/epics_utils/adjustable.py | 47 ++++++++ eco/timing/event_timing_new_new.py | 184 +++++++++++++++++++++++++---- eco/utilities/config.py | 156 +++++++++++++++++++----- tests/test_config_lazy_init.py | 131 ++++++++++++++++++++ 5 files changed, 542 insertions(+), 50 deletions(-) diff --git a/eco/acquisition/counters.py b/eco/acquisition/counters.py index 4d81e1e..d4748fc 100644 --- a/eco/acquisition/counters.py +++ b/eco/acquisition/counters.py @@ -4,6 +4,7 @@ import weakref from eco.acquisition.utilities import Acquisition from eco.elements.protocols import Detector, MonitorableValueUpdate, resolve_lazy from eco.utilities.datafiles import ensure_dir, ensure_group_writable +from eco.utilities.utilities import is_notebook from collections import namedtuple from escape import ArrayTimestamps from matplotlib.animation import FuncAnimation @@ -18,6 +19,42 @@ DEFAULT_STORAGE_DIR = Path("./") StepTime = namedtuple("StepTime", "start stop") +def _build_fit_icon(size=24): + """A small "scatter + fit line" QIcon for the toolbar button, drawn with + QPainter instead of shipping a bitmap asset.""" + from qtpy import QtCore, QtGui + + pixmap = QtGui.QPixmap(size, size) + pixmap.fill(QtCore.Qt.transparent) + painter = QtGui.QPainter(pixmap) + painter.setRenderHint(QtGui.QPainter.Antialiasing) + + pen = QtGui.QPen(QtGui.QColor("black")) + pen.setWidthF(max(1.0, size / 10)) + pen.setCapStyle(QtCore.Qt.RoundCap) + painter.setPen(pen) + painter.drawLine( + QtCore.QPointF(0.08 * size, 0.85 * size), + QtCore.QPointF(0.92 * size, 0.15 * size), + ) + + painter.setPen(QtCore.Qt.NoPen) + painter.setBrush(QtGui.QColor("black")) + dot_r = size * 0.06 + dots = [ + (0.10, 0.55), (0.14, 0.72), (0.18, 0.40), (0.22, 0.62), (0.26, 0.30), + (0.30, 0.50), (0.34, 0.68), (0.38, 0.35), (0.42, 0.55), (0.46, 0.25), + (0.54, 0.62), (0.58, 0.30), (0.62, 0.50), (0.66, 0.20), + (0.70, 0.40), (0.74, 0.58), (0.78, 0.28), (0.82, 0.45), (0.86, 0.15), + (0.90, 0.35), (0.20, 0.20), (0.60, 0.65), (0.35, 0.15), + ] + for fx, fy in dots: + painter.drawEllipse(QtCore.QPointF(fx * size, fy * size), dot_r, dot_r) + + painter.end() + return QtGui.QIcon(pixmap) + + class CounterValue: def __init__(self, *detectors, name="value_counter"): @@ -199,6 +236,7 @@ class CounterValue: scan.fig = f if isinstance(axs, plt.Axes): axs = [axs] + scan.axs = axs def plotdat(n, *args): for ma, ax in zip(scan.monitor_scan_arrays.values(), axs): @@ -209,11 +247,47 @@ class CounterValue: fig=f, func=plotdat, cache_frame_data=False, interval=500 ) plt.show(block=False) + self._add_fit_button(scan, axs) else: scan.fig.tight_layout() scan.fig.canvas.draw() scan.fig.canvas.flush_events() + def _add_fit_button(self, scan, axs): + """Add a UI trigger that opens `escape.fit_gui.AxesFitter` on demand, + one per plotted axis -- a Qt toolbar icon if the figure has a Qt + navigation toolbar, else an ipywidgets button if running in a + Jupyter kernel. Nothing is added otherwise (e.g. inline backend). + + The fit GUI draws its span selector/preview on top of `axs`, which + the live animation clears every frame (`ax.cla()`), so starting a + fit also stops that animation -- same effect `stopani` already has + once the scan itself ends. + """ + + def start_fitters(*_): + from escape.fit_gui import AxesFitter + + if hasattr(scan, "animation"): + scan.animation.event_source.stop() + for ax in axs: + if ax.lines: + AxesFitter(ax) + + toolbar = getattr(scan.fig.canvas.manager, "toolbar", None) + if toolbar is not None and toolbar.__class__.__name__.startswith( + "NavigationToolbar2" + ): + action = toolbar.addAction(_build_fit_icon(), "Fit", start_fitters) + action.setToolTip("Open interactive fit (escape.fit_gui.AxesFitter)") + elif is_notebook(): + import ipywidgets + from IPython.display import display + + button = ipywidgets.Button(description="Fit", icon="line-chart") + button.on_click(start_fitters) + display(button) + def store_arrays( self, scan, filename="auto", directory="auto", elog=None, **kwargs ): diff --git a/eco/epics_utils/adjustable.py b/eco/epics_utils/adjustable.py index d34d9b2..ae45387 100644 --- a/eco/epics_utils/adjustable.py +++ b/eco/epics_utils/adjustable.py @@ -59,6 +59,53 @@ def wait_for_enum_strs(pv, retries=10, delay=0.05): return pv.enum_strs +def read_pv_value(adjustable, timeout=3.0, attempts=3, description=None): + """Read `adjustable`'s current value, treating a timed-out get as a + failure instead of as the value `None`. + + Why this is needed at all: pyepics' ``PV.get()`` returns ``None`` when + ``wait_for_connection()`` fails - it never raises (see + ``epics/pv.py``'s ``get_with_metadata``). eco's ``AdjustablePv`` builds + every PV with ``connection_timeout=0.05`` and its readback with + ``auto_monitor=False``, so a read issued shortly after construction has + a 50 ms budget on a brand-new channel and no monitor cache to fall back + on. During a concurrent ``Namespace.init_all()`` pass - thousands of + channels connecting at once - that budget is regularly missed, and the + ``None`` then flows on as if it were data: an event code of ``None``, a + pulser number of ``None``. The resulting failure surfaces far from its + cause, or does not surface at all (see ``EvrPulser``/``EvrOutput``). + + This is the same failure shape as `wait_for_enum_strs` above (a value + that is simply not there yet under load), and the same shape as the + ``Daq.get_pulse_id()`` fix - both boil down to "never let a timed-out + CA get become a value". + + Uses the underlying PV directly so a real timeout can be given, rather + than inheriting the 50 ms connection timeout. Raises ``TimeoutError`` + naming the PV once the attempts are used up, so the caller fails with + something readable instead of an ``AttributeError`` three frames later. + """ + pv = getattr(adjustable, "_pvreadback", None) + if pv is None: + pv = getattr(adjustable, "_pv", None) + what = description or getattr(adjustable, "name", None) or repr(adjustable) + pvname = getattr(pv, "pvname", None) or getattr(adjustable, "pvname", what) + for _ in range(max(int(attempts), 1)): + if pv is not None: + pv.wait_for_connection(timeout=timeout) + value = pv.get(timeout=timeout) + else: + value = adjustable.get_current_value() + if value is not None: + return value + raise TimeoutError( + f"could not read '{what}' ({pvname}) after {attempts} attempts of " + f"{timeout}s each - the channel did not connect or did not return a " + f"value. pyepics reports this as None rather than raising, so it is " + f"raised here to keep it from being used as if it were data." + ) + + # Work in progress! TODO @spec_convenience @get_from_archive diff --git a/eco/timing/event_timing_new_new.py b/eco/timing/event_timing_new_new.py index 365c6bd..61b7d21 100644 --- a/eco/timing/event_timing_new_new.py +++ b/eco/timing/event_timing_new_new.py @@ -1,7 +1,12 @@ from epics import caget_many from ..elements.adjustable import AdjustableMemory, AdjustableVirtual from ..elements.detector import DetectorVirtual -from ..epics_utils.adjustable import AdjustablePv, AdjustablePvEnum, AdjustablePvString +from ..epics_utils.adjustable import ( + AdjustablePv, + AdjustablePvEnum, + AdjustablePvString, + read_pv_value, +) from ..epics_utils.detector import DetectorPvData, DetectorPvDataStream from ..detector.detectors_psi import DetectorBsStream from eco.epics_utils.utilities_epics import EpicsString @@ -11,6 +16,8 @@ from ..utilities.tables import format_table logging.getLogger("cta_lib").setLevel(logging.WARNING) +logger = logging.getLogger(__name__) + class TimingSystem(Assembly): """This is a wrapper object for the global timing system at SwissFEL""" @@ -210,9 +217,46 @@ class MasterEventSystem(Assembly): ) self.event_codes[code] = self.__dict__[f"code{code:03d}"] - def _get_slot_codes(self, slots=range(1, 257)): + def _get_slot_codes(self, slots=range(1, 257), attempts=3, timeout=3.0): + """Read the master's slot->event-code table. + + `caget_many` reports a slot it could not read as `None`, and those + used to be dropped silently. Whichever init worker builds this + component does so while the rest of the pass saturates channel + access, so under `init_all()` a handful of slots could time out and + the resulting `event_codes` dict stayed permanently incomplete for + 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. + """ + slots = list(slots) pvs = [f"{self.pvname}:Evt-{slot}-Code-SP" for slot in slots] - codes = caget_many(pvs) + codes = list(caget_many(pvs, timeout=timeout)) + + for _ in range(max(int(attempts) - 1, 0)): + missing = [i for i, c in enumerate(codes) if c is None] + if not missing: + break + retried = caget_many([pvs[i] for i in missing], timeout=timeout) + for i, c in zip(missing, retried): + codes[i] = c + still_missing = [slots[i] for i, c in enumerate(codes) if c is None] + 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.", + self.pvname, + len(still_missing), + len(slots), + attempts, + ", ".join(str(x) for x in still_missing[:10]), + ", ..." if len(still_missing) > 10 else "", + ) slots_out = [] codes_out = [] @@ -326,16 +370,26 @@ class EvrPulser(Assembly): ) self.description = EpicsString(pv_base + "-Name-I") - if self._eventcode is not None: + # Resolve the event code ONCE and keep it. It used to be a property + # re-issuing a CA get on every access, and __init__ touched it three + # times (the `is not None` guard, then .frequency, then .delay): under + # a concurrent init_all() the guard could pass and the next access + # come back None (timed-out get -> None -> KeyError -> None), giving + # `AttributeError: 'NoneType' object has no attribute 'frequency'` on + # a pulser that initializes fine on its own. One read, one value, no + # window between the check and the use. + self._eventcode_resolved = self._resolve_eventcode() + + if self._eventcode_resolved is not None: self._append( DetectorVirtual, - [self._eventcode.frequency], + [self._eventcode_resolved.frequency], lambda x: x, name="frequency", ) self._append( DetectorVirtual, - [self._eventcode.delay], + [self._eventcode_resolved.delay], lambda x: x, name="delay_eventcode", ) @@ -347,17 +401,55 @@ class EvrPulser(Assembly): name="delay", ) else: - print(f"Error initializing pulser {self.name} of EVR {self.pv_base}: Event code {self.eventcode.get_current_value()} is missing in Timing Master") + logger.warning( + "pulser %s of EVR %s: event code %s is missing in the timing " + "master; delay/frequency are not available on it.", + self.name, + self.pv_base, + self._eventcode_number, + ) + + def _resolve_eventcode(self, timeout=3.0, attempts=3): + """The timing-master event code object this pulser is wired to. + + The read is done through `read_pv_value` rather than + `get_current_value()` so a channel that has not connected yet raises + instead of yielding `None` - a `None` here used to be looked up in + `event_codes`, miss, and silently leave the pulser without its + delay/frequency chain, which then broke every EvrOutput pointing at + it. Returns None only when the code is genuinely absent from the + master. + """ + try: + self._eventcode_number = read_pv_value( + self.eventcode, + timeout=timeout, + attempts=attempts, + description=f"event code of pulser {self.name} ({self.pv_base})", + ) + except TimeoutError as e: + # Degrade the way this always did (no delay/frequency chain) + # rather than failing the whole pulser - a PV that genuinely + # isn't there must not turn into a wall of new failures. The + # difference is that it now says so instead of silently looking + # up event code `None` and reporting it as missing from the + # master. + logger.warning("%s", e) + self._eventcode_number = None + return None + if self._eventcode_number == 27: + return self._parent_evr.sequencer + return self._event_master.event_codes.get(self._eventcode_number) + + def update_eventcode(self): + """Re-read the event code from the IOC (it is otherwise resolved once, + at construction). Use after re-wiring the pulser externally.""" + self._eventcode_resolved = self._resolve_eventcode() + return self._eventcode_resolved @property def _eventcode(self): - if self.eventcode.get_current_value() == 27: - return self._parent_evr.sequencer - else: - try: - return self._event_master.event_codes[self.eventcode.get_current_value()] - except KeyError: - return None + return self._eventcode_resolved class DummyPulser(Assembly): @@ -398,6 +490,8 @@ class EvrOutput(Assembly): name="pulserA_number", is_setting=True, ) + # resolve once, before the eight virtuals below reach for it + self._pulserA_resolved = self._resolve_pulser(self.pulserA_number, "pulserA") self._append( AdjustableVirtual, [self.pulserA.delay], @@ -466,6 +560,9 @@ class EvrOutput(Assembly): name="pulserB_number", is_setting=True, ) + # NB: this used to read pulserA_number - every pulserB_* virtual on + # every output was silently tracking pulser A. + self._pulserB_resolved = self._resolve_pulser(self.pulserB_number, "pulserB") self._append( AdjustableVirtual, [self.pulserB.delay], @@ -521,19 +618,62 @@ class EvrOutput(Assembly): name="pulserB_width", ) + def _resolve_pulser(self, number_adj, which, timeout=3.0, attempts=3): + """The pulser object this output is wired to, read once. + + Two bugs used to live here. The read went through + `get_current_value()`, which returns None on a timed-out CA get, and + `self._pulsers[None]` raised TypeError, which was caught and turned + into a *fresh* `DummyPulser()`. And because this was a property, + `__init__` re-read it once per virtual it builds - eight times for + pulserA, eight for pulserB - so under a concurrent init_all() a + single output could end up with its eight `pulserA_*` virtuals bound + to several different objects, some of them throwaway dummies whose + values are permanently None. That mis-wiring never failed and never + printed anything. + + `read_pv_value` raises rather than yielding None, and the result is + resolved once and cached; an out-of-range number still degrades to a + DummyPulser, since that is a real (if odd) IOC state rather than a + transport failure. + """ + try: + number = read_pv_value( + number_adj, + timeout=timeout, + attempts=attempts, + description=f"{which} number of output {self.name} ({self.pv_base})", + ) + except TimeoutError as e: + logger.warning("%s", e) + number = None + try: + return self._pulsers[number] + except (IndexError, TypeError): + logger.warning( + "output %s (%s): %s number %r does not address any of the %d " + "pulsers of this EVR; using a dummy pulser.", + self.name, + self.pv_base, + which, + number, + len(self._pulsers or ()), + ) + return DummyPulser() + + def update_pulsers(self): + """Re-read which pulsers this output is wired to (they are otherwise + resolved once, at construction). Use after re-wiring the output.""" + self._pulserA_resolved = self._resolve_pulser(self.pulserA_number, "pulserA") + self._pulserB_resolved = self._resolve_pulser(self.pulserB_number, "pulserB") + @property def pulserA(self): - try: - return self._pulsers[self.pulserA_number.get_current_value()] - except (IndexError, TypeError): - return DummyPulser() + return self._pulserA_resolved @property def pulserB(self): - try: - return self._pulsers[self.pulserA_number.get_current_value()] - except (IndexError, TypeError): - return DummyPulser() + return self._pulserB_resolved class EventReceiver(Assembly): diff --git a/eco/utilities/config.py b/eco/utilities/config.py index 93fb88a..e25ca9b 100644 --- a/eco/utilities/config.py +++ b/eco/utilities/config.py @@ -837,6 +837,14 @@ class Namespace(Assembly): self._initialisation_start_time = {} self._init_priority = {} + # name -> [NamespaceComponent, ...] found among the args/kwargs it + # was registered with, i.e. the other namespace items it declares a + # dependency on. Recorded at append_obj time (the only moment the + # arguments are visible as objects rather than already-substituted + # proxies) so init_all can order the pass by dependency instead of + # letting every consumer race to build the same shared component. + self._declared_dependencies = {} + # name -> {"module_name": str|None, "obj_factory": callable|str}, as # passed to append_obj. Lets reinitialize(reload_modules=True) find # which module to importlib.reload() for a given name; see @@ -1381,6 +1389,71 @@ class Namespace(Assembly): print(f"Captured init log no longer exists at {path}.") return "" + def get_dependencies(self, name): + """Names this item declared a dependency on, via a + `NamespaceComponent` among its `append_obj` arguments. + + Only *declared* dependencies are visible here. A component that + reaches into the namespace from inside its own `__init__` (rather + than being handed a NamespaceComponent) is invisible to this, which + is why dependency ordering is a scheduling hint and not a guarantee - + see `_dependency_layers`. + """ + out = set() + for dep in self._declared_dependencies.get(name, ()): + try: + dep._resolve() + except KeyError: + # points at something never registered - not our problem + # here, it will fail loudly when the item is actually built. + continue + if dep.obj_name is not None and dep.obj_name != name: + out.add(dep.obj_name) + return out + + def _dependency_layers(self, names, log=None): + """Split `names` into successive layers such that everything a name + declares a dependency on is in an earlier layer. + + The point is the shared-dependency stampede: at Bernina all nine + EventReceivers take the same `event_master` NamespaceComponent, so a + flat concurrent pass has up to `max_workers` of them reaching for it + at once. One wins and builds it (a 256-PV `caget_many`); the rest sit + in the lazy-init wait until `init_timeout` expires and come back as + IsInitialisingError, to be retried a cycle later - which is both slow + and, because all of that happens while channel access is saturated, + the condition under which the EVRs' own reads start timing out. + Building `event_master` in an earlier layer removes the collision + entirely. + + Dependencies pointing outside `names` (already initialized, excluded, + or never registered) impose no constraint. A dependency cycle cannot + be layered, so the members of the cycle are emitted together as one + layer and left to the existing retry loop - the same behaviour as + before this method existed, applied to just the cyclic group. + """ + remaining = set(names) + edges = {n: self.get_dependencies(n) & remaining for n in remaining} + layers = [] + placed = set() + while remaining: + ready = {n for n in remaining if not (edges[n] - placed)} + if not ready: + # cycle (or a dependency on something that can never be + # placed): stop layering and hand the rest over as one group. + if log is not None: + log( + f"dependency cycle among {len(remaining)} name(s), " + f"initializing them together: " + + ", ".join(sorted(remaining)) + ) + layers.append(set(remaining)) + break + layers.append(ready) + placed |= ready + remaining -= ready + return layers + def _select_names_to_init(self, required_only, exclude_names, log): """Which names this init_all() call needs to build, honoring required_only/exclude_names; warns (via `log`) about pre-existing @@ -1468,37 +1541,59 @@ class Namespace(Assembly): init_fn() first_exception = None - pending = set(names_to_init) - cycles_left = max(int(N_cycles), 1) + # Dependency-ordered: a name is only submitted once everything it + # declared a NamespaceComponent dependency on has been built, so + # consumers of a shared component no longer race each other to build + # it. Names with no declared dependencies all land in the first + # layer, so this costs essentially no parallelism - at Bernina it + # moves the nine EventReceivers behind `event_master` and leaves the + # other ~70 components exactly as concurrent as before. + layers = self._dependency_layers(names_to_init, log=log) + if len(layers) > 1: + log( + f"Initializing in {len(layers)} dependency layers: " + + ", ".join(str(len(la)) for la in layers) + + " name(s)" + ) + givenup = set() with cap, ThreadPoolExecutor( max_workers=max_workers, initializer=_thread_initializer ) as exc: - while pending and cycles_left: - cycles_left -= 1 - futs = { - exc.submit( - self.init_name, - name, - verbose=verbose, - raise_errors=True, - quiet=bool(silent), - ): name - for name in pending - } - retry = set() - for fut in as_completed(futs): - name = futs[fut] - try: - fut.result() - except IsInitialisingError: - # Lost a race for a name someone else (this pass - # itself, or the session directly) is already - # initializing as a dependency; retry next round. - retry.add(name) - except Exception as exc_: - if first_exception is None: - first_exception = exc_ - pending = retry & (self.all_names - self.initialized_names) + for layer in layers: + pending = set(layer) + cycles_left = max(int(N_cycles), 1) + while pending and cycles_left: + cycles_left -= 1 + futs = { + exc.submit( + self.init_name, + name, + verbose=verbose, + raise_errors=True, + quiet=bool(silent), + ): name + for name in pending + } + retry = set() + for fut in as_completed(futs): + name = futs[fut] + try: + fut.result() + except IsInitialisingError: + # Lost a race for a name someone else (this pass + # itself, or the session directly) is already + # initializing as a dependency; retry next round. + retry.add(name) + except Exception as exc_: + if first_exception is None: + first_exception = exc_ + pending = retry & (self.all_names - self.initialized_names) + # Whatever this layer could not finish is not allowed to hold + # up the layers behind it: the dependents are attempted + # anyway (they may not really need it), exactly as they were + # before layering existed. + givenup |= pending + pending = givenup if pending: # A warning rather than log(): log() is silenced by silent=True, @@ -1985,6 +2080,11 @@ 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) + ] if lazy: def init_local(): diff --git a/tests/test_config_lazy_init.py b/tests/test_config_lazy_init.py index e9dab02..1801412 100644 --- a/tests/test_config_lazy_init.py +++ b/tests/test_config_lazy_init.py @@ -317,3 +317,134 @@ def test_output_capture_survives_a_thread_that_outlives_the_pass(): cap._write_sink("late output from a straggler thread") assert "late output" in real.getvalue() + + +class _DependentThing: + def __init__(self, dep=None, name=None): + self.name = name + self.dep = dep + + +def test_declared_dependencies_are_recorded_and_layered(): + """init_all used to submit every name flat, so all thirteen consumers of + bernina's `event_master` reached for it at once: one built it, the rest + waited out init_timeout and came back as IsInitialisingError. Names are + now ordered behind what they declare a NamespaceComponent dependency on. + """ + from eco.utilities.config import NamespaceComponent + + ns = Namespace(name="test") + ns.append_obj(_DependentThing, lazy=True, name="master") + for n in range(3): + ns.append_obj( + _DependentThing, + NamespaceComponent(ns, "master"), + lazy=True, + name=f"consumer{n}", + ) + ns.append_obj(_DependentThing, lazy=True, name="unrelated") + # a sub-attribute reference still resolves to the registered item + ns.append_obj( + _DependentThing, + NamespaceComponent(ns, "master.some.child"), + lazy=True, + name="deep", + ) + + assert ns.get_dependencies("consumer0") == {"master"} + assert ns.get_dependencies("deep") == {"master"} + assert ns.get_dependencies("master") == set() + + layers = ns._dependency_layers(ns.all_names) + assert layers[0] == {"master", "unrelated"} + assert layers[1] == {"consumer0", "consumer1", "consumer2", "deep"} + + +def test_init_all_builds_a_dependency_before_its_consumers(tmp_path): + from eco.utilities.config import NamespaceComponent + + ns = Namespace(name="test") + ns.append_obj(_DependentThing, lazy=True, name="master") + for n in range(4): + ns.append_obj( + _DependentThing, + NamespaceComponent(ns, "master"), + lazy=True, + name=f"consumer{n}", + ) + + order = [] + original = ns.init_name + + def spy(name, **kwargs): + order.append(name) + return original(name, **kwargs) + + ns.init_name = spy + ns.init_all( + required_only=False, + background=False, + silent=str(tmp_path / "i.log"), + max_workers=4, + ) + + assert order[0] == "master" + assert set(order[1:]) == {f"consumer{n}" for n in range(4)} + + +def test_dependency_cycle_does_not_hang_and_is_emitted_as_one_layer(): + from eco.utilities.config import NamespaceComponent + + ns = Namespace(name="test") + ns.append_obj( + _DependentThing, NamespaceComponent(ns, "b"), lazy=True, name="a" + ) + ns.append_obj( + _DependentThing, NamespaceComponent(ns, "a"), lazy=True, name="b" + ) + + assert ns._dependency_layers(ns.all_names) == [{"a", "b"}] + + +def test_read_pv_value_retries_and_refuses_to_return_none(): + """pyepics' PV.get() returns None on a timed-out get instead of raising, + and eco's AdjustablePv gives every channel a 50 ms connection budget with + auto_monitor=False on the readback - so under a concurrent init_all() a + None routinely flowed on as if it were data (an event code of None, a + pulser number of None).""" + import pytest + + from eco.epics_utils.adjustable import read_pv_value + + class FakePV: + def __init__(self, values): + self.values = list(values) + self.pvname = "FAKE:PV" + self.gets = 0 + + def wait_for_connection(self, timeout=None): + return True + + def get(self, timeout=None): + self.gets += 1 + return self.values.pop(0) if self.values else None + + class FakeAdjustable: + def __init__(self, pv): + self._pvreadback = pv + self.name = "fake" + + # a transient timeout is retried, not taken at face value + adj = FakeAdjustable(FakePV([None, None, 27])) + assert read_pv_value(adj, timeout=0.01, attempts=3) == 27 + assert adj._pvreadback.gets == 3 + + # a persistent one raises with the PV named, instead of returning None + adj = FakeAdjustable(FakePV([])) + with pytest.raises(TimeoutError) as excinfo: + read_pv_value(adj, timeout=0.01, attempts=2) + assert "FAKE:PV" in str(excinfo.value) + + # a legitimate falsy value is a value, not a failure + adj = FakeAdjustable(FakePV([0])) + assert read_pv_value(adj, timeout=0.01, attempts=2) == 0