diff --git a/docs/examples/archiver_stripchart.md b/docs/examples/archiver_stripchart.md index 4b9756f..0c4527e 100644 --- a/docs/examples/archiver_stripchart.md +++ b/docs/examples/archiver_stripchart.md @@ -47,9 +47,10 @@ df = archiver.get_data_time_range( ) # Relative starts: a number of seconds, a timedelta, or a dict of timedelta -# kwargs — all relative to `end` (which defaults to now): +# kwargs — all counted back from `end` (which defaults to now). Negative reads +# naturally, but the sign is ignored: 1800 and {"minutes": 30} mean the same. df = archiver.get_data_time_range(channels=[chan], start=-1800) # last 30 min -df = archiver.get_data_time_range(channels=[chan], start={"minutes": 30}) +df = archiver.get_data_time_range(channels=[chan], start={"minutes": -30}) ``` Useful options: @@ -60,6 +61,82 @@ Useful options: - `convert_timezone=True` converts the index from UTC to `Europe/Zurich`. - `labels=[...]` sets nicer legend labels for the plot. +Each call prints the range it actually asked the archiver for (UTC and local) +and, per channel, whether data came back or why not (`verbose=False` silences +this and skips the extra requests below). A CA channel is only stored when it +changes, so a quiet one has no events in a short window. One extra request per +such channel tells you which case you are in: + +``` +target_stages.x.offset (SARES20-MF2:MOT_1.OFF): no events in range; last before it: 8.3875 (set 2026-10-05 13:03:37 UTC) +target_stages.x.velocity (SARES20-MF2:MOT_1.VELO): not found in sf-archiver (not archived under this name) +``` + +The first exists and was simply quiet (the value it had is shown); the second +is not archived at all. + +### Several eco objects at once + +Instead of channel names you can pass eco objects — an Assembly, an adjustable, +a detector. Every archived channel below the object's alias is retrieved. +Objects, plain channel ids and lists of both can be mixed, and a channel that +several objects share is fetched once: + +```python +df = archiver.get_data_time_range(prof_kb.target_stages, mon_opt, start=-1e5) + +# extra channels: plain ids are read from the DataBuffer (BS) unless you say +# otherwise, either per channel ... +df = archiver.get_data_time_range( + prof_kb.target_stages, + channels=["SARES20-MF1:MOT_1.RBV"], channel_types=["CA"], + start=-3600, +) +# ... or for everything at once with force_type= +``` + +Objects bring their own channel types and legend labels, so leave `force_type` +unset when passing them. Legend labels carry each channel's full alias +(`prof_kb.target_stages.x.offset (SARES20-MF2:MOT_1.OFF)`). The start (and end) +may also follow the objects positionally if it is a number, timedelta, dict or +datetime — `archiver.get_data_time_range(prof_kb.target_stages, -3600)`; a date +*string* must be passed as `start=`, since it could as well be a channel id. + +By default the DataFrame columns are channel ids, in the order you asked. +`column_names="alias"` names them by the full alias instead +(`prof_kb.target_stages.x.offset`), `column_names="label"` by the whole legend +label. + +### Long ranges of BS channels: `bins` + +A BS channel runs at up to 100 Hz: a day of one channel is 8.6 million +samples, roughly 20 s to fetch. A warning (with that estimate) is printed +whenever a request would be that big. Downsample **on the server** instead: + +```python +# a bin count (the server rounds to a sensible bin width) ... +df = archiver.get_data_time_range(gasmon, start=-86400, bins=500) +# ... or a bin width: "10s", "1m", "1h", ... +df = archiver.get_data_time_range(gasmon, start=-86400, bins="10m") +``` + +Each channel comes as its bin average plus ` min`, ` max` and +` count` columns, stamped at the bin centre; empty bins are NaN. +`bins` works for CA channels and for `get_data_pulse_id_range` too. + +### The value at the start of the window: `last_before` + +A quiet CA channel has no events in a short window, and a step plot of it has +no starting value. `last_before=True` also fetches each channel's last event +before the window (the archiver's "one before range") and puts it *at* the +start of the window: + +```python +df = archiver.get_data_time_range(prof_kb.target_stages, start=-600, last_before=True) +``` + +It cannot be combined with `bins`. + ### Finding channel names If you are not sure of the exact channel name, search with a glob pattern: @@ -79,6 +156,12 @@ For beam-synchronous channels you can query a pulse-id range instead. With no df = archiver.get_data_pulse_id_range(channels=[chan], start=-1000) ``` +It takes everything `get_data_time_range` does — eco objects, `bins`, +`last_before`, `column_names` — and positional `start`/`end`: +`archiver.get_data_pulse_id_range(gasmon, 28549932558, 28549938558)`. A negative +`start` is an offset back from `end`. (Pulse ids are mapped to times with the +service's own map; datahub's built-in linear formula was days off.) + ## Getting archiver data into a plot Every retrieval method takes `plot=True`, which draws the returned DataFrame as @@ -96,6 +179,13 @@ archiver.get_data_time_range( ) ``` +With many curves, the figure window's toolbar has a **Select curves** button +(right after *Save*, Qt backends): it opens a window with a checkbox per curve, +a filter box and All/None, and hides or shows curves immediately — the legend +and the y-range follow. The button comes from escape (`escape >= 0.2.14`, +`escape.plot_utilities.attach_select_button`, which also covers the Jupyter +widget backend); with an older escape a built-in Qt-only version is used. + Since you also get the DataFrame back, you can just as easily plot or analyse it yourself with pandas/matplotlib. diff --git a/eco/dbase/archiver.py b/eco/dbase/archiver.py index afa0360..348bee9 100644 --- a/eco/dbase/archiver.py +++ b/eco/dbase/archiver.py @@ -5,13 +5,15 @@ import multiprocessing import re import subprocess import sys +import contextlib import threading import time -from collections import deque +from collections import Counter, deque from numbers import Number import pandas as pd from matplotlib import pyplot as plt +from matplotlib import colors as mcolors from matplotlib import dates as mdates from matplotlib.animation import FuncAnimation from matplotlib.widgets import Button, CheckButtons @@ -20,6 +22,7 @@ from datahub import Consumer, Daqbuf, Dispatcher, Enum, Epics, Table, pulse_id_t from .. import ecocnf from ..elements.assembly import Assembly +from ..epics_utils import get_archiver_channels from ..epics_utils.detector import DetectorPvDataStream _logger = logging.getLogger(__name__) @@ -48,6 +51,17 @@ RETRY_WINDOW = datetime.timedelta(seconds=60) RETRY_COUNT = 6 RETRY_DELAY = 1.5 +# A "BS" (DataBuffer) channel is recorded at up to 100 Hz: a request that +# would return more than BS_LARGE_ROWS samples (a bit under 40 MB per channel +# as float64, but pandas and the merge of many channels multiply that) gets a +# warning, pointing at `bins=` (see `DataHub.get_data_time_range`). +BS_RATE_HZ = 100 +BS_LARGE_ROWS = 5_000_000 +BS_ROWS_PER_SECOND = 4e5 # raw retrieval speed measured against data-api.psi.ch + +NO_DATA = "no data in the requested range" +COLUMN_NAME_MODES = ("channel", "alias", "label") + # A single, facility-wide pulse-id counter (the injector gun EVR's raw # pulse-id readback) that is always live at high rate, regardless of which # beamline or experiment is running. Pulse ids are one shared global counter @@ -130,35 +144,100 @@ def datetime2str(datetime_date): def local2utc(datetime_date): - return datetime_date.replace( - tzinfo=None, - ).astimezone( - tz=datetime.timezone.utc, - ) + """Naive datetimes are taken as local time; aware ones keep their zone.""" + return datetime_date.astimezone(tz=datetime.timezone.utc) + + +def _relative_delta(start): + """`start` as a timedelta *before* the end time, or None if `start` is not + a relative specification (timedelta, dict of timedelta kwargs, number of + seconds). The sign is ignored: -3600, 3600 and hours=1 all mean "one hour + before end".""" + if isinstance(start, datetime.timedelta): + return abs(start) + if isinstance(start, dict): + return abs(datetime.timedelta(**start)) + if isinstance(start, Number) and not isinstance(start, bool): + return abs(datetime.timedelta(seconds=start)) + return None def _resolve_range(start, end, kwargs): - """Turn (start, end) into a pair of absolute datetimes. `end` defaults to - now; `start` can be a datetime, an ISO string, a timedelta, a dict of - timedelta kwargs, a number of seconds (all relative to `end`), or - - if none of the above is given - is built from **kwargs (e.g. hours=1).""" - if not end: - end = datetime.datetime.now() - if isinstance(start, datetime.timedelta): - start = end + start - elif isinstance(start, dict): - start = end + datetime.timedelta(**start) - elif isinstance(start, Number): - start = end + datetime.timedelta(seconds=start) - elif start is None: - start = end + datetime.timedelta(**kwargs) - if isinstance(start, str): - start = dateutil.parser.parse(start) + """Turn (start, end) into a pair of absolute, naive-local or aware + datetimes. `end` defaults to now and can be a datetime or an ISO string. + `start` can be a datetime, an ISO string, or - relative to `end`, with the + sign ignored - a timedelta, a dict of timedelta kwargs or a number of + seconds. If `start` is None it is built from **kwargs (e.g. hours=1 for + the hour before `end`).""" if isinstance(end, str): end = dateutil.parser.parse(end) + elif isinstance(end, Number) and not isinstance(end, bool): + raise ValueError( + f"end={end!r}: an end time is a datetime or an ISO string " + "(numbers only make sense for `start`, as seconds before `end`)" + ) + elif not end: + end = datetime.datetime.now() + if start is None: + if not kwargs: + raise ValueError( + "no start time given: pass start=... as a keyword (e.g. " + "start=-3600, start='2026-10-01 08:00') or a delta such as hours=1" + ) + try: + start = datetime.timedelta(**kwargs) + except TypeError as e: + raise ValueError( + f"unknown time-range keyword in {sorted(kwargs)}: use " + "days/hours/minutes/seconds/weeks (or pass start=...)" + ) from e + delta = _relative_delta(start) + if delta is not None: + start = end - delta + elif isinstance(start, str): + start = dateutil.parser.parse(start) + elif not isinstance(start, datetime.datetime): + raise ValueError( + f"cannot interpret start={start!r} ({type(start).__name__}): use a " + "datetime, an ISO string, a number of seconds, a timedelta or a dict" + ) + if start >= end: + raise ValueError( + f"start ({start}) is not before end ({end}); relative starts " + "count back from end" + ) return start, end +def _index_to_datetime(index): + """Convert a numeric timestamp index to a UTC DatetimeIndex, whatever unit + it is in. datahub's `Table` delivers nanoseconds unless it is created with + a different `timetype`, while older code assumed seconds; guessing the unit + from the magnitude makes this independent of that setting.""" + values = pd.Index(index) + if values.empty: + return pd.to_datetime(values, utc=True) + magnitude = abs(float(values.astype("float64").max())) + if magnitude >= 1e17: + unit = "ns" + elif magnitude >= 1e14: + unit = "us" + elif magnitude >= 1e11: + unit = "ms" + else: + unit = "s" + return pd.to_datetime(values, unit=unit, utc=True) + + +def _channel_type(channel_type, force_type=None): + """The history backend type ("CA" or "BS") a channel is retrieved as: + `force_type` if given, else its own type, else "BS" (also for anything + unrecognised or unknown).""" + if force_type: + return force_type + return channel_type if channel_type in HISTORY_BACKENDS else "BS" + + def _group_by_type(channels, force_type=None, channel_types=None): """Group `channels` by their channel type ("CA" or "BS"): either the same type for all of them (`force_type`) or individually (`channel_types`, a @@ -171,13 +250,214 @@ def _group_by_type(channels, force_type=None, channel_types=None): groups = {} types = channel_types or [] for i, channel in enumerate(channels): - channel_type = types[i] if i < len(types) else None - if channel_type not in HISTORY_BACKENDS: - channel_type = "BS" + channel_type = _channel_type(types[i] if i < len(types) else None) groups.setdefault(channel_type, []).append(channel) return groups +def _flatten(items): + """Yield the leaves of arbitrarily nested lists/tuples.""" + for item in items: + if isinstance(item, (list, tuple)): + yield from _flatten(item) + else: + yield item + + +def _object_name(obj): + try: + return obj.alias.get_full_name() + except Exception: + return getattr(obj, "name", None) or repr(obj) + + +def _resolve_targets(targets, channels=None, channel_types=None, labels=None): + """Turn what a caller wants archived data for into parallel lists + `(channel_ids, channel_types, labels)`, plus the names of any objects + that contributed no channel at all. + + `targets` are eco objects (anything with an alias tree - an Assembly, an + adjustable, a detector, ...: every channel below it is taken, see + `eco.epics_utils.get_archiver_channels`), plain channel-id strings, or + lists/tuples of those. `channels` takes the same, a single string + included; only for plain channel ids in it can `channel_types` ("CA"/"BS") + and `labels` be given, parallel to `channels`. A channel that turns up + more than once is requested once (first label wins).""" + channels = [channels] if isinstance(channels, str) else list(channels or []) + if (channel_types or labels) and not all(isinstance(c, str) for c in channels): + raise ValueError( + "channel_types/labels run parallel to plain channel-id strings in " + "`channels`; for eco objects they come from the objects' aliases" + ) + ids, types, names, seen, empty = [], [], [], set(), [] + + def add(channel, channel_type, label): + if channel not in seen: + seen.add(channel) + ids.append(channel) + types.append(channel_type) + names.append(label) + + def add_item(item, channel_type=None, label=None): + if isinstance(item, str): + add(item, channel_type, label or item) + return + try: + item_ids, item_types, item_labels, _ = get_archiver_channels(item) + except Exception as e: + raise TypeError( + f"cannot get archiver channels of {item!r} ({type(e).__name__}: " + f"{e}): expected an eco object with an alias tree, a channel-id " + "string, or a list of these" + ) from e + if not item_ids: + empty.append(_object_name(item)) + for j, channel in enumerate(item_ids): + add(channel, item_types[j] if item_types else None, item_labels[j]) + + for item in _flatten(targets): + add_item(item) + for i, item in enumerate(_flatten(channels)): + add_item( + item, + channel_types[i] if channel_types and i < len(channel_types) else None, + labels[i] if labels and i < len(labels) else None, + ) + return ids, types, names, empty + + +@contextlib.contextmanager +def _mute_channel_not_found(): + """datahub fetches the channels of a request in one thread each and lets a + refused one ("channel not found") die there, which prints a traceback to + stderr per channel. `DataHub._probe_missing` reports those channels itself, + so keep that noise - and only that - out while a request runs.""" + original = getattr(threading, "excepthook", None) + if original is None: # Python < 3.8 + yield + return + + def hook(args): + message = str(args.exc_value).lower() + if isinstance(args.exc_value, RuntimeError) and ( + "not found" in message or "notfound" in message + ): + return + original(args) + + threading.excepthook = hook + try: + yield + finally: + threading.excepthook = original + + +def _alias_from_label(label, channel): + """The alias part of an object channel's label ("alias (channel) [unit]"), + or the channel id itself for a plain channel id, whose label has none.""" + head, sep, _ = label.partition(f" ({channel})") + return head if sep else channel + + +def _display_names(channels, labels, mode): + """{channel: column name} for `mode` ("channel", "alias" or "label"). Two + channels that would end up under one name keep their channel id in it.""" + if mode not in COLUMN_NAME_MODES: + raise ValueError(f"column_names must be one of {COLUMN_NAME_MODES}, not {mode!r}") + if mode == "channel": + return {c: c for c in channels} + if mode == "label": + names = dict(zip(channels, labels)) + else: + names = {c: _alias_from_label(l, c) for c, l in zip(channels, labels)} + counts = Counter(names.values()) + return {c: n if counts[n] == 1 else f"{n} ({c})" for c, n in names.items()} + + +def _rename_columns(data, mapping): + """Rename `data`'s columns by {channel: name}. The " max"/" min"/" count" + columns that a binned request adds follow their channel.""" + + def new_name(column): + if column in mapping: + return mapping[column] + for suffix in (" max", " min", " count"): + if column.endswith(suffix) and column[: -len(suffix)] in mapping: + return mapping[column[: -len(suffix)]] + suffix + return column + + return data.rename(columns=new_name) + + +def _peel_range_args(targets, accepted): + """`start`/`end` may follow the targets positionally, e.g. + get_data_time_range(myassembly, -3600): peel up to two trailing arguments + of an `accepted` type off `targets`. Never strings: one could as well be a + channel id. Returns (targets, [start, end][:n]).""" + targets = list(targets) + peeled = [] + while ( + targets + and len(peeled) < 2 + and isinstance(targets[-1], accepted) + and not isinstance(targets[-1], bool) + ): + peeled.insert(0, targets.pop()) + return targets, peeled + + +def _positional_range(targets, start, end, accepted): + targets, peeled = _peel_range_args(targets, accepted) + if peeled: + if start is not None: + raise TypeError("start was given twice: positionally and as a keyword") + start = peeled[0] + if len(peeled) > 1: + if end is not None: + raise TypeError("end was given twice: positionally and as a keyword") + end = peeled[1] + return targets, start, end + + +def _iso_to_ns(iso): + return int(pd.Timestamp(iso).value) + + +def _ts_text(ns): + return pd.Timestamp(ns, tz="UTC").strftime("%Y-%m-%d %H:%M:%S UTC") + + +def _carry_in(table, start_ns): + """Reduce each channel's events before `start_ns` (what a `last` request + adds) to the latest one, moved *to* `start_ns`: the state the window opens + with. Returns {channel: (original timestamp ns, value, has events in range)}.""" + carried = {} + for name, records in table.data.items(): + before = [r for r in records if r[Table.TIMESTAMP] < start_ns] + if not before: + continue + last = before[-1] + rest = [r for r in records if r[Table.TIMESTAMP] >= start_ns] + if rest and rest[0][Table.TIMESTAMP] == start_ns: + records[:] = rest # an event right at the start supersedes it + else: + records[:] = [dict(last, **{Table.TIMESTAMP: start_ns})] + rest + carried[name] = (last[Table.TIMESTAMP], last[name], bool(rest)) + return carried + + +def _mask_empty_bins(table, channels): + """The server returns an empty bin as 0.0 for average/min/max with count 0: + make those NaN.""" + for channel in channels: + counts = table.data.get(f"{channel} count") + for i, record in enumerate(counts or []): + if record[f"{channel} count"] == 0: + for key in (channel, f"{channel} max", f"{channel} min"): + if i < len(table.data.get(key, [])): + table.data[key][i][key] = float("nan") + + def _glob_to_regex(pattern): """Convert a simple unix glob pattern ('*', '?') into a regex understood by the channel search endpoint. `fnmatch.translate` is not usable here: @@ -206,6 +486,282 @@ def _show_figure(fig): plt.show(block=False) +def _antialiasing(QtGui): + return getattr(QtGui.QPainter, "Antialiasing", None) or QtGui.QPainter.RenderHint.Antialiasing + + +def _checklist_icon(toolbar, QtGui): + """A small "list of checkboxes" icon, drawn in the toolbar's own text + colour so it matches the built-in monochrome buttons in light and dark + themes.""" + size, ratio = 24, 2 + pixmap = QtGui.QPixmap(size * ratio, size * ratio) + pixmap.setDevicePixelRatio(ratio) + pixmap.fill(QtGui.QColor(0, 0, 0, 0)) + painter = QtGui.QPainter(pixmap) + try: + painter.setRenderHint(_antialiasing(QtGui)) + painter.setPen(QtGui.QPen(toolbar.palette().color(toolbar.foregroundRole()), 1.6)) + for row, y in enumerate((3, 10, 17)): + painter.drawRect(3, y, 5, 5) + painter.drawLine(11, y + 2, 21, y + 2) + if row != 1: # the middle one is the unticked one + painter.drawLine(4, y + 3, 5, y + 4) + painter.drawLine(5, y + 4, 8, y) + finally: + painter.end() + return QtGui.QIcon(pixmap) + + +def _swatch_icon(QtGui, color): + """A short line with a dot in `color`: the curve's legend handle.""" + pixmap = QtGui.QPixmap(28, 12) + pixmap.fill(QtGui.QColor(0, 0, 0, 0)) + painter = QtGui.QPainter(pixmap) + try: + painter.setRenderHint(_antialiasing(QtGui)) + qcolor = QtGui.QColor(mcolors.to_hex(color)) + painter.setPen(QtGui.QPen(qcolor, 2)) + painter.drawLine(1, 6, 27, 6) + painter.setBrush(qcolor) + painter.drawEllipse(10, 3, 6, 6) + finally: + painter.end() + return QtGui.QIcon(pixmap) + + +def _add_toolbar_button(fig, tooltip, callback): + """Add a button to `fig`'s own window toolbar - right after "Save" - if + its backend is a Qt one. Returns whether it did.""" + if "qt" not in type(fig.canvas).__module__.lower(): + return False + try: + from qtpy import QtGui, QtWidgets + + toolbar = getattr(fig.canvas.manager, "toolbar", None) + if not isinstance(toolbar, QtWidgets.QToolBar): + return False + QAction = getattr(QtGui, "QAction", None) or QtWidgets.QAction # Qt6: QtGui + + def run(*_): + # an exception escaping a Qt slot can take the whole session down + try: + callback() + except Exception: + _logger.exception("%s failed", tooltip) + + action = QAction(_checklist_icon(toolbar, QtGui), tooltip, toolbar) + action.setToolTip(tooltip) + action.triggered.connect(run) + actions = toolbar.actions() + # The toolbar ends in a stretchy coordinates label: appending would + # put the button far right of it, so insert in front of whatever + # follows "Save" (or, without one, in front of that label). + anchor = getattr(toolbar, "_actions", {}).get("save_figure") + label = getattr(toolbar, "locLabel", None) + if anchor is not None and actions.index(anchor) + 1 < len(actions): + before = actions[actions.index(anchor) + 1] + else: + before = next((a for a in actions if toolbar.widgetForAction(a) is label), None) + if before is None: + toolbar.addAction(action) + else: + toolbar.insertAction(before, action) + return True + except Exception: + _logger.debug("no toolbar button for %r", tooltip, exc_info=True) + return False + + +def _make_select_dialog(selector): + """The Qt window behind the toolbar button: a scrollable list with one + checkbox per curve (with the curve's legend handle), a filter box, and + All/None for the rows currently shown by the filter. Ticking applies + immediately.""" + from qtpy import QtCore, QtGui, QtWidgets + + lines = selector.lines + dialog = QtWidgets.QDialog(selector.fig.canvas.manager.window) + dialog.setWindowTitle("Select curves") + layout = QtWidgets.QVBoxLayout(dialog) + filter_box = QtWidgets.QLineEdit() + filter_box.setPlaceholderText("filter, e.g. readback") + filter_box.setClearButtonEnabled(True) + layout.addWidget(filter_box) + + holder = QtWidgets.QWidget() + rows = QtWidgets.QVBoxLayout(holder) + boxes = [] + for line in lines: + box = QtWidgets.QCheckBox(line.get_label()) + box.setIcon(_swatch_icon(QtGui, line.get_color())) + box.setIconSize(QtCore.QSize(28, 12)) + box.setChecked(line.get_visible()) + rows.addWidget(box) + boxes.append(box) + rows.addStretch(1) + scroll = QtWidgets.QScrollArea() + scroll.setWidgetResizable(True) + scroll.setWidget(holder) + layout.addWidget(scroll) + + def guarded(fn): + def run(*_): + try: + fn() + except Exception: + _logger.exception("curve selection failed") + + return run + + def sync(): + selector.apply([box.isChecked() for box in boxes]) + + def set_shown(state): + for box in boxes: + if not box.isHidden(): # i.e. matches the filter + box.blockSignals(True) + box.setChecked(state) + box.blockSignals(False) + sync() + + def refilter(): + needle = filter_box.text().strip().lower() + for box in boxes: + box.setVisible(needle in box.text().lower()) + + buttons = QtWidgets.QHBoxLayout() + for text, state in (("All", True), ("None", False)): + button = QtWidgets.QPushButton(text) + button.clicked.connect(guarded(lambda state=state: set_shown(state))) + buttons.addWidget(button) + buttons.addStretch(1) + layout.addLayout(buttons) + for box in boxes: + box.toggled.connect(guarded(sync)) + filter_box.textChanged.connect(guarded(refilter)) + + metrics = dialog.fontMetrics() + advance = getattr(metrics, "horizontalAdvance", None) or metrics.width + widest = max(advance(box.text()) for box in boxes) + dialog.resize(min(widest + 130, 1000), min(28 * len(boxes) + 140, 760)) + return dialog + + +def _legend_kwargs(legend): + """What is worth carrying over from `legend` to its replacement.""" + kwargs = {"frameon": legend.get_frame_on()} + loc = getattr(legend, "_loc", None) # private, but the only place it is kept + if loc is not None: + kwargs["loc"] = loc + ncol = getattr(legend, "_ncols", None) or getattr(legend, "_ncol", None) + if ncol: + kwargs["ncol"] = ncol + title = legend.get_title().get_text() + if title: + kwargs["title"] = title + return kwargs + + +class _CurveSelector: + """Show/hide a figure's curves from a Qt checkbox window (see + `_attach_curve_selector`). Hiding a curve also drops it from the legend + and re-fits the y-range to what is left, since the usual reason to hide + one is that its scale dwarfs the others.""" + + def __init__(self, fig, ax, lines): + self.fig = fig + self.ax = ax + self.lines = list(lines) + self._dialog = None + # None: there is no legend to maintain. Kept apart from the live one + # so it survives the state where every curve is hidden and it is gone. + legend = ax.get_legend() + self._legend_look = None if legend is None else _legend_kwargs(legend) + fig.canvas.mpl_connect("close_event", self._close_dialog) + + def _close_dialog(self, _event=None): + try: + if self._dialog is not None: + self._dialog.close() + except RuntimeError: # already deleted along with the figure window + pass + self._dialog = None + + def apply(self, states): + """Show exactly the curves whose entry in `states` (parallel to the + curves) is true.""" + for line, state in zip(self.lines, states): + line.set_visible(bool(state)) + shown = [line for line in self.lines if line.get_visible()] + legend = self.ax.get_legend() + if legend is not None: + self._legend_look = _legend_kwargs(legend) # as the user left it + legend.remove() + if shown and self._legend_look is not None: + # rebuilt without the hidden curves; other labelled artists (bands, + # scatter, ...) keep their entries + entries = [ + (handle, label) + for handle, label in zip(*self.ax.get_legend_handles_labels()) + if getattr(handle, "get_visible", lambda: True)() + ] + if entries: + self.ax.legend(*zip(*entries), **self._legend_look) + # relim() knows lines, patches and images but not collections (scatter, + # fill_between, ...): with those present it would shrink the range to + # the lines alone, so leave the limits alone then + if shown and not self.ax.collections: + self.ax.relim(visible_only=True) + self.ax.autoscale(enable=True, axis="y") # zooming had switched it off + self.ax.autoscale_view(scalex=False) + self.fig.canvas.draw_idle() + + def open_dialog(self): + """Show the checkbox window, creating it on first use; a closed one + is just shown again (it is a child of the figure window, so it goes + away with that).""" + dialog = self._dialog + try: + if dialog is not None: + dialog.isVisible() # raises RuntimeError once the window is gone + except RuntimeError: + dialog = None + if dialog is None: + dialog = self._dialog = _make_select_dialog(self) + dialog.show() + dialog.raise_() + dialog.activateWindow() + + +def _attach_select_curves(fig, ax, lines): + """The "Select curves" toolbar button: escape's own + (`escape.plot_utilities.attach_select_button`, escape >= 0.2.14, which also + covers ipympl) when the installed escape has it, else the Qt-only copy + below. That copy - `_antialiasing` ... `_attach_curve_selector` - can go + once the beamline environments have a new enough escape.""" + if len(lines) < 2: + return None + try: + from escape.plot_utilities import attach_select_button + except Exception: # no escape, or one without the selector + return _attach_curve_selector(fig, ax, lines) + attach_select_button(fig) + + +def _attach_curve_selector(fig, ax, lines): + """Give `fig` a "select curves" button in its Qt window toolbar, which + opens a checkbox window to show/hide `lines`. Does nothing for a single + curve, or on backends without a Qt toolbar (inline, ipympl, non-GUI).""" + if len(lines) < 2: + return None + selector = _CurveSelector(fig, ax, lines) + if not _add_toolbar_button(fig, "Select curves to show", selector.open_dialog): + return None + fig._eco_curve_selector = selector + return selector + + def _ensure_esc_h5(path): """Normalise `path` to an escape results filename (needs both `.esc` and `.h5` in its suffixes).""" @@ -267,10 +823,11 @@ def _plot_dataframe(data, channels, labels=None): # silently blows up the visible x-range to spans of decades and leaves # the tick labels showing that bogus range's raw internal day-numbers # instead of the actual dates. - _, ah = plt.subplots() + fig, ah = plt.subplots() if not labels: labels = channels plotted = False + lines = [] for chan, label in zip(channels, labels): if chan not in data: continue @@ -284,7 +841,7 @@ def _plot_dataframe(data, channels, labels=None): x = data.index[sel].to_pydatetime() y = column[sel] try: - ah.step(x, y, ".-", label=label, where="post") + lines.extend(ah.step(x, y, ".-", label=label, where="post")) plotted = True except (TypeError, ValueError) as e: _logger.warning("Skipping %r in the plot: not numeric (%s)", chan, e) @@ -297,6 +854,7 @@ def _plot_dataframe(data, channels, labels=None): plt.tight_layout() plt.xlabel(data.index.name) ah.figure.tight_layout() + _attach_select_curves(fig, ah, lines) _show_figure(ah.figure) @@ -1115,7 +1673,35 @@ class DataHub(Assembly): raise Exception(f"force_type must be one of {list(LIVE_SOURCES)}") return cls(time_type="sec") - def get_data(self, channels, start, end, force_type=None, channel_types=None): + def get_data( + self, + channels, + start, + end, + force_type=None, + channel_types=None, + status=None, + bins=None, + last_before=False, + ): + """Raw retrieval behind `get_data_time_range` / `get_data_pulse_id_range`: + a DataFrame on the Table's own (nanosecond) timestamp index, or None. + + `start`/`end`: ISO strings. `bins`: downsample on the server - a number + of bins or a bin width such as "10s", "1m", "1h": each channel then + comes as its bin average plus " min", " max" and + " count" columns, stamped at the bin centre (empty bins are + NaN). `last_before`: also fetch each channel's last event before + `start` and put it *at* `start` - the state the window opens with. + `status`, if a dict, is filled with {channel: "ok" | why there is no + data}.""" + if status is None: + status = {} + extra = {} + if bins: + extra["bins"] = bins + if last_before: + extra["last"] = True # the archiver's "oneBeforeRange" groups = _group_by_type(channels, force_type, channel_types) table = Table() for channel_type, group_channels in groups.items(): @@ -1128,10 +1714,31 @@ class DataHub(Assembly): # frequent outcome there (the channel just hasn't changed) and is # not retried. retry = backend == DEFAULT_HISTORY_BACKEND and _is_recent(end) - self._request_into(source, group_channels, start, end, table, retry) + self._request_into( + source, group_channels, start, end, table, retry, status, extra + ) + carried = _carry_in(table, _iso_to_ns(start)) if last_before else {} + if bins: + _mask_empty_bins(table, channels) + for channel in channels: + if channel in status: # failed + continue + counts = table.data.get(f"{channel} count") + if channel in carried and not carried[channel][2]: + original, value, _ = carried[channel] + status[channel] = ( + "no events in range; its value at the start is the last one " + f"before it: {value} (set {_ts_text(original)})" + ) + elif counts is not None and not any(r[f"{channel} count"] for r in counts): + status[channel] = NO_DATA # binned, and every bin empty + elif table.data.get(channel): + status[channel] = "ok" + else: + status[channel] = NO_DATA return table.as_dataframe() - def _request_into(self, source, channels, start, end, table, retry): + def _request_into(self, source, channels, start, end, table, retry, status, extra): """Request `channels` on `source`, feeding `table`. Channels that are just empty (no error) are retried up to `retry`'s policy. A batch request that errors out - e.g. a component's ".EGU"/status/enum @@ -1142,69 +1749,283 @@ class DataHub(Assembly): for attempt in range(attempts): source.add_listener(table) try: - source.request(dict(channels=channels, start=start, end=end)) + with _mute_channel_not_found(): + source.request(dict(channels=channels, start=start, end=end, **extra)) except Exception as e: source.remove_listeners() if len(channels) == 1: + reason = " ".join(str(e).split("\n")[:2]) or type(e).__name__ + status[channels[0]] = f"retrieval failed: {reason}" _logger.warning( "Skipping %r: could not retrieve historical data (%s)", channels[0], - e, + reason, ) else: for channel in channels: - self._request_into(source, [channel], start, end, table, retry) + self._request_into( + source, [channel], start, end, table, retry, status, extra + ) return source.remove_listeners() - if any(ch in table.data for ch in channels): - return + if any(table.data.get(ch) for ch in channels): + break if attempt < attempts - 1: time.sleep(RETRY_DELAY) - def get_data_time_range( + def _probe_missing(self, status, channel_ids, ctypes, force_type, start, end): + """Why do some channels have no data in the range? Asks, for each one + and one at a time, for the last event before it (the archiver's + "oneBeforeRange") on a source that does *not* fetch in parallel: only + then does a request error - "channel not found" - reach us instead of + dying unseen in a worker thread. A channel that answers with a value + exists and was just quiet; one that is refused is not archived under + that name. Refines `status` in place. About 0.07 s per channel.""" + by_backend = {} + for channel, channel_type in zip(channel_ids, ctypes): + if status.get(channel) == NO_DATA: + backend = HISTORY_BACKENDS[_channel_type(channel_type, force_type)] + by_backend.setdefault(backend, []).append(channel) + for backend, channels in by_backend.items(): + source = Daqbuf(backend=backend, time_type="sec", parallel=False) + for channel in channels: + table = Table() + source.add_listener(table) + try: + source.request(dict(channels=[channel], start=start, end=end, last=True)) + except Exception as e: + message = str(e) + if "not found" in message.lower() or "notfound" in message.lower(): + status[channel] = f"not found in {backend} (not archived under this name)" + else: + reason = " ".join(message.split("\n")[:2]) or type(e).__name__ + status[channel] = f"retrieval failed: {reason}" + continue + finally: + source.remove_listeners() + records = table.data.get(channel) + if records: + last = records[-1] + status[channel] = ( + "no events in range; last before it: " + f"{last[channel]} (set {_ts_text(last[Table.TIMESTAMP])})" + ) + else: + status[channel] = "no events in or before the range" + + def _retrieve( self, - channels=[], - start=None, - end=None, - plot=False, - force_type=None, - channel_types=None, - labels=None, - convert_timezone=False, - **kwargs, + targets, + channels, + start_utc, + end_utc, + *, + plot, + force_type, + channel_types, + labels, + convert_timezone, + verbose, + bins, + last_before, + column_names, ): - """Retrieve historical data for `channels` between `start` and `end` - (datetimes, ISO date strings, or - for `start` - a timedelta/dict of - timedelta kwargs/number of seconds relative to `end`). `end` defaults - to now; if `start` is not given either, it is derived from **kwargs - (e.g. `hours=1`). - - `force_type` ("CA" or "BS") selects the backend for all channels; - `channel_types` instead gives the type of each channel individually - (a list running parallel to `channels`, as found e.g. in Alias - entries). - """ - start, end = _resolve_range(start, end, kwargs) - - data = self.get_data( - channels, - start=datetime2str(local2utc(start)), - end=datetime2str(local2utc(end)), - force_type=force_type, - channel_types=channel_types, + """What `get_data_time_range` and `get_data_pulse_id_range` share, once + they have turned their kind of range into `start_utc`/`end_utc`.""" + if column_names not in COLUMN_NAME_MODES: + raise ValueError( + f"column_names must be one of {COLUMN_NAME_MODES}, not {column_names!r}" + ) + if bins and last_before: + raise ValueError("bins (downsampling) and last_before cannot be combined") + channel_ids, ctypes, names, empty = _resolve_targets( + targets, channels, channel_types, labels ) + span = end_utc - start_utc + start_iso, end_iso = datetime2str(start_utc), datetime2str(end_utc) + if verbose: + how = f", binned ({bins})" if bins else "" + how += ", plus the last value before the start" if last_before else "" + print( + f"archiver request: {start_iso} -> {end_iso} UTC (span {span}; " + f"local {start_utc.astimezone():%Y-%m-%d %H:%M:%S} -> " + f"{end_utc.astimezone():%Y-%m-%d %H:%M:%S}), " + f"{len(channel_ids)} channel(s){how}" + ) + if empty: + print(f" no archiver channels found under: {', '.join(empty)}") + n_bs = sum(_channel_type(t, force_type) == "BS" for t in ctypes) + rows = n_bs * span.total_seconds() * BS_RATE_HZ + if n_bs and not bins and rows > BS_LARGE_ROWS: + message = ( + f"{n_bs} DataBuffer (BS) channel(s) over {span} can mean up to " + f"{rows:.1e} samples (~{rows * 8 / 1e9:.1f} GB as float64, roughly " + f"{rows / BS_ROWS_PER_SECOND:.0f} s to fetch): " + "beam-synchronous data comes at up to 100 Hz. Consider a " + "shorter range, or downsample on the server with bins=' + ) + if verbose: + print(f" warning: {message}") + else: + _logger.warning(message) + if not channel_ids: + if verbose: + print(" nothing to retrieve: no channels") + return None + + status = {} + data = self.get_data( + channel_ids, + start=start_iso, + end=end_iso, + force_type=force_type, + channel_types=ctypes, + status=status, + bins=bins, + last_before=last_before, + ) + if verbose: + if any(reason == NO_DATA for reason in status.values()): + self._probe_missing(status, channel_ids, ctypes, force_type, start_iso, end_iso) + failed = {c: r for c, r in status.items() if r != "ok"} + if not failed: + print(f" all {len(channel_ids)} channel(s) returned data") + else: + print( + f" {len(channel_ids) - len(failed)} of {len(channel_ids)} " + "channel(s) had events in the range; the others:" + ) + label_of = dict(zip(channel_ids, names)) + type_of = dict(zip(channel_ids, ctypes)) + for c, r in failed.items(): + backend = HISTORY_BACKENDS[_channel_type(type_of[c], force_type)] + print(f" {label_of[c]} [{backend}]: {r}") if data is not None: - data.index = pd.to_datetime(data.index, unit="s", utc=True) + data.index = _index_to_datetime(data.index) data.index.name = "timestamp" if convert_timezone: data.index = data.index.tz_convert("Europe/Zurich") + # columns in the order asked for (the channels arrive in whatever + # order their threads finish), a binned channel's extras behind it + ordered = [ + column + for c in channel_ids + for column in (c, f"{c} max", f"{c} min", f"{c} count") + if column in data.columns + ] + data = data[ordered + [c for c in data.columns if c not in ordered]] if plot and data is not None: - _plot_dataframe(data, channels, labels) + _plot_dataframe(data, channel_ids, names) + if data is not None and column_names != "channel": + data = _rename_columns(data, _display_names(channel_ids, names, column_names)) return data + def get_data_time_range( + self, + *targets, + channels=None, + start=None, + end=None, + plot=False, + force_type=None, + channel_types=None, + labels=None, + convert_timezone=False, + verbose=True, + bins=None, + last_before=False, + column_names="channel", + **kwargs, + ): + """Retrieve historical data as a DataFrame indexed by UTC timestamp, + one column per channel (or None if nothing came back). + + What: any mix of eco objects (an Assembly, an adjustable, a detector: + every channel below its alias is taken), plain channel-id strings and + lists of those, positionally and/or in `channels`: + archiver.get_data_time_range(myassembly, myadj, mydet, start=-1e5) + archiver.get_data_time_range("SARFE10-PBPG050:HAMP-INTENSITY-CAL", + hours=6) + archiver.get_data_time_range(myassembly, start=-3600, + channels=["SARES20-MF1:MOT_1.RBV"], channel_types=["CA"]) + A plain channel id carries no type: it is read from the DataBuffer + ("BS") unless `force_type` is given, or - in `channels` only - + `channel_types` (parallel to `channels`, "CA"/"BS") says otherwise; + `labels` (also parallel to `channels`) names them in a plot. Objects + bring their own types and labels. `force_type` ("CA" or "BS") instead + overrides the type of *all* channels, objects' included. + + When: `end` defaults to now. `start` can be + - a number of seconds before `end`: start=-3600 + - a timedelta / dict of its kwargs: start=timedelta(days=-2), + start=dict(hours=-6) + - keyword shortcuts (no `start`): hours=6, days=2, minutes=30 + - an absolute ISO string or datetime: start="2026-10-01 08:00" + (naive = local time; "...Z" or a tz-aware datetime = as given) + (negative = back in time from `end`; the sign is ignored, so + start=3600 or dict(days=2) mean the same as the negative forms) + With an explicit `end` (same absolute forms), relative starts count + back from it: start=-600, end="2026-10-01 09:00" is 08:50-09:00. + A number, timedelta, dict or datetime may also follow the targets + positionally (`get_data_time_range(myassembly, -3600)`); a date + *string* must be a keyword, since it could as well be a channel id. + + Columns: channel ids by default. `column_names="alias"` names them by + the object's full alias (e.g. "prof_kb.target_stages.x.readback"), + "label" by the full legend label ("prof_kb.target_stages.x.readback + (SARES20-MF2:MOT_1.RBV) [mm]"). A plain channel id keeps its id as + alias. + + Downsampling and the window's opening state: + - `bins=500` or `bins="10s"` (a bin width: "10s", "1m", "1h", ...) + downsamples on the *server*: each channel comes as its bin average + plus " min", " max" and " count" columns + at the bin centre, empty bins NaN. Use it for long ranges of BS + channels (100 Hz!). The server picks round bin widths, so the + number of bins is only approximately what you asked for. + - `last_before=True` also fetches each channel's last event before + `start` and places it *at* `start`: the value the channel had when + the window opened - for a slowly changing channel (CA) the state to + start a step plot from. Not combinable with `bins`. + + With `verbose` (default) the range actually requested (UTC and local) + and the per-channel outcome are printed: ok, or why there is no data. + For a channel without events in the range one extra request per channel + tells why: "no events in range; last before it: (set