656 lines
22 KiB
Python
656 lines
22 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
import copy
|
|
from datetime import datetime
|
|
from inspect import isclass
|
|
import json
|
|
from pathlib import Path
|
|
from tkinter import W
|
|
import weakref
|
|
from markdown import markdown
|
|
|
|
from numpy import isin
|
|
import numpy as np
|
|
|
|
|
|
from eco.acquisition.scan_data import run_status_convenience
|
|
from eco.elements.protocols import Detector, InitialisationWaitable
|
|
from eco.epics import get_from_archive
|
|
|
|
from ..aliases import Alias
|
|
from ..utilities.tables import format_table
|
|
import colorama
|
|
from . import memory
|
|
from enum import Enum
|
|
import os
|
|
import subprocess
|
|
from rich.progress import track
|
|
from eco import Adjustable, Detector
|
|
|
|
import eco
|
|
|
|
|
|
_initializing_assemblies = []
|
|
|
|
|
|
class StatusCollection:
|
|
def __init__(self, parent, name="status_collection"):
|
|
self.parent = weakref.ref(parent)
|
|
self.selections = {}
|
|
|
|
if name is None:
|
|
raise Exception("A name of collection is required")
|
|
self.name = name
|
|
self._list = []
|
|
|
|
def get_list(self, selection=None, **kwargs):
|
|
rec_list_items = kwargs.get("rec_list_items", [])
|
|
ls = []
|
|
for witem in self._list:
|
|
item = witem()
|
|
if item is None:
|
|
continue
|
|
|
|
if item is self.parent:
|
|
continue
|
|
|
|
if item in ls:
|
|
continue
|
|
|
|
if selection is not None:
|
|
if selection not in self.selections.keys():
|
|
continue
|
|
item_name = item.alias.get_full_name(base=self.parent())
|
|
if item_name not in self.selections[selection].keys():
|
|
continue
|
|
recurse = self.selections[selection][item_name]["recurse"]
|
|
else:
|
|
recurse = True
|
|
|
|
ls.append(item)
|
|
|
|
# important to get field in case no recursion is defined.
|
|
if item is self.parent():
|
|
recurse = False
|
|
|
|
if hasattr(item, f"{self.name}") and isinstance(
|
|
item.__dict__[self.name], self.__class__
|
|
):
|
|
if recurse:
|
|
# if hasattr(item, "recursing") and item.recursing:
|
|
# print(
|
|
# f"recursing detected loop at {item.alias.get_full_name()}"
|
|
# )
|
|
# item.recursing = True
|
|
|
|
for titem in item.__dict__[self.name].get_list(
|
|
selection=selection, ls=[]
|
|
):
|
|
|
|
if titem not in ls:
|
|
ls.append(titem)
|
|
|
|
else:
|
|
|
|
if item not in ls:
|
|
ls.append(item)
|
|
|
|
else:
|
|
if item not in ls:
|
|
ls.append(item)
|
|
|
|
return ls
|
|
|
|
def get_names(self, selection=None):
|
|
return [
|
|
item.alias.get_full_name(base=self.parent())
|
|
for item in self.get_list(selection=selection)
|
|
]
|
|
|
|
def get_selections_names(self):
|
|
return self.selections.keys()
|
|
|
|
def append(self, obj, selection=None, recursive=True):
|
|
|
|
if selection is not None:
|
|
if selection not in self.selections:
|
|
self.selections[selection] = {}
|
|
obj_name = obj.alias.get_full_name(base=self.parent())
|
|
self.selections[selection][obj_name] = {"recurse": recursive}
|
|
if obj not in [tl() for tl in self._list]:
|
|
self._list.append(weakref.ref(obj))
|
|
|
|
def remove(self, obj, selection=None):
|
|
"""Remove an object from the collection. If selection is given, only remove from that selection."""
|
|
if obj in [wobj() for wobj in self._list]:
|
|
obj_name = obj.alias.get_full_name(base=self.parent())
|
|
if selection is None:
|
|
ix = [wobj() for wobj in self._list].index(obj)
|
|
self._list.remove(self._list[ix])
|
|
else:
|
|
raise ValueError("Item not in list")
|
|
if selection is not None:
|
|
selections = [selection]
|
|
else:
|
|
selections = self.selections.keys()
|
|
for selection in selections:
|
|
if obj_name in self.selections[selection]:
|
|
del self.selections[selection][obj_name]
|
|
|
|
def __call__(self):
|
|
return self.get_list()
|
|
|
|
|
|
class NumpyEncoder(json.JSONEncoder):
|
|
"""Special json encoder for numpy types"""
|
|
|
|
def default(self, obj):
|
|
if isinstance(obj, np.integer):
|
|
return int(obj)
|
|
elif isinstance(obj, np.floating):
|
|
return float(obj)
|
|
elif isinstance(obj, np.ndarray):
|
|
return obj.tolist()
|
|
return json.JSONEncoder.default(self, obj)
|
|
|
|
|
|
@get_from_archive
|
|
@run_status_convenience
|
|
class Assembly:
|
|
def __init__(self, name=None, parent=None, is_alias=True, elog=None):
|
|
self.name = name
|
|
self.alias = Alias(name, parent=parent)
|
|
# self.settings = []
|
|
# self.status_indicators = []
|
|
self.status_collection = StatusCollection(self, name="status_collection")
|
|
|
|
if memory.global_memory_dir:
|
|
self.memory = memory.Memory(self)
|
|
if elog:
|
|
self.__elog = elog
|
|
# else:
|
|
# self.__class__.__elog = property(lambda dum: ELOG)
|
|
|
|
# TODO: Lazy an threaded append! (for PVs, should be quite a speedup).
|
|
def _append(
|
|
self,
|
|
foo_obj_init,
|
|
*args,
|
|
name=None,
|
|
is_setting=False,
|
|
is_display=True,
|
|
is_status=True,
|
|
# recursive=None,
|
|
call_obj=True,
|
|
overwrite=False,
|
|
**kwargs,
|
|
):
|
|
"""This hidden method appends an object to the assembly. It can take either an object instance, or a class (in which case it will be called with the provided args and kwargs).
|
|
Parameters
|
|
----------
|
|
foo_obj_init : Adjustable, Detector, Assembly, class, callable
|
|
The object to append, or a class/callable to instantiate.
|
|
name : str, optional
|
|
The name of the object within the assembly. If None, the name attribute of the object will be used.
|
|
is_setting : bool or str "recursive", optional"""
|
|
if overwrite:
|
|
|
|
if name in self.__dict__:
|
|
old = self.__dict__[name]
|
|
self.status_collection.remove(old)
|
|
self.alias.pop_object(old.alias)
|
|
del old
|
|
|
|
if isinstance(foo_obj_init, Adjustable) and not isclass(foo_obj_init):
|
|
# adj_copy = copy.copy(foo_obj_init)
|
|
adj_copy = foo_obj_init
|
|
self.__dict__[name] = adj_copy
|
|
elif isinstance(foo_obj_init, Detector) and not isclass(foo_obj_init):
|
|
self.__dict__[name] = foo_obj_init
|
|
elif isinstance(foo_obj_init, Assembly) and not isclass(foo_obj_init):
|
|
self.__dict__[name] = foo_obj_init
|
|
elif call_obj and callable(foo_obj_init):
|
|
self.__dict__[name] = foo_obj_init(*args, **kwargs, name=name)
|
|
else:
|
|
self.__dict__[name] = foo_obj_init
|
|
self.alias.append(self.__dict__[name].alias)
|
|
|
|
self.status_collection.append(self.__dict__[name])
|
|
# if is_status == "auto":
|
|
# is_status = isinstance(self.__dict__[name], Detector)
|
|
if is_setting:
|
|
if isinstance(is_setting, str):
|
|
recursive = is_setting.lower() == "recursive"
|
|
else:
|
|
recursive = True
|
|
self.status_collection.append(
|
|
self.__dict__[name], selection="settings", recursive=recursive
|
|
)
|
|
# self.status_collection.append(
|
|
# self.__dict__[name], selection="settings", recursive=True
|
|
# )
|
|
if is_display:
|
|
if isinstance(is_display, str):
|
|
recursive = is_display.lower() == "recursive"
|
|
else:
|
|
recursive = False
|
|
self.status_collection.append(
|
|
self.__dict__[name], selection="display", recursive=recursive
|
|
)
|
|
|
|
def get_status(
|
|
self,
|
|
base="self",
|
|
verbose=False,
|
|
print_times=False,
|
|
channeltypes=None,
|
|
selections=[],
|
|
threads=True,
|
|
max_workers=20,
|
|
# print_name=False,
|
|
):
|
|
if base == "self":
|
|
base = self
|
|
# settings = {}
|
|
# settings_channels = {}
|
|
# settings_times = {}
|
|
status = {}
|
|
status_channels = {}
|
|
status_times = {}
|
|
nodet = []
|
|
geterror = []
|
|
|
|
def get_stat_one_detector(ts):
|
|
tstart = time.time()
|
|
try:
|
|
if (not channeltypes) or (ts.alias.channeltype in channeltypes):
|
|
status[ts.alias.get_full_name(base=base)] = ts.get_current_value()
|
|
try:
|
|
status_channels[ts.alias.get_full_name(base=base)] = (
|
|
ts.alias.channel
|
|
)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
geterror.append(ts.alias.get_full_name(base=base))
|
|
finally:
|
|
status_times[ts.alias.get_full_name(base=base)] = time.time() - tstart
|
|
|
|
detectors = []
|
|
for ts in self.status_collection.get_list():
|
|
if isinstance(ts, Detector):
|
|
detectors.append(ts)
|
|
else:
|
|
nodet.append(ts.alias.get_full_name(base=base))
|
|
|
|
if threads and max_workers > 1 and len(detectors) > 1:
|
|
with ThreadPoolExecutor(max_workers=max_workers) as exc:
|
|
list(
|
|
track(
|
|
exc.map(get_stat_one_detector, detectors),
|
|
transient=True,
|
|
description="Reading status indicators ...",
|
|
total=len(detectors),
|
|
)
|
|
)
|
|
else:
|
|
for ts in track(
|
|
detectors,
|
|
transient=True,
|
|
description="Reading status indicators ...",
|
|
):
|
|
get_stat_one_detector(ts)
|
|
|
|
if verbose:
|
|
if nodet:
|
|
print("Could not retrieve status from:\n " + ",\n ".join(nodet))
|
|
if geterror:
|
|
print(
|
|
"Retrieved error while running get_current_value from:\n "
|
|
+ ",\n ".join(geterror)
|
|
)
|
|
|
|
if print_times:
|
|
from ascii_graph import Pyasciigraph
|
|
|
|
gr = Pyasciigraph()
|
|
for line in gr.graph(
|
|
"Times required to get status",
|
|
sorted(status_times.items(), key=lambda w: w[1]),
|
|
):
|
|
print(line)
|
|
|
|
sel_dict = {}
|
|
for selection_name in selections:
|
|
sel = self.status_collection.get_names(selection=selection_name)
|
|
sel_dict[selection_name] = {
|
|
tname: status[tname] for tname in sel if tname in status.keys()
|
|
}
|
|
|
|
return {
|
|
# "settings": settings,
|
|
"status": status,
|
|
# "settings_channels": settings_channels,
|
|
"status_channels": status_channels,
|
|
# "settings_times": settings_times,
|
|
"status_times": status_times,
|
|
"selections": sel_dict,
|
|
}
|
|
def get_tree(self, level=1, print_tree=True):
|
|
"""Return nested status_collection member names as a dictionary."""
|
|
|
|
def build_tree(node, depth):
|
|
result = {}
|
|
for wref in node.status_collection._list:
|
|
item = wref()
|
|
if item is None or item is node:
|
|
continue
|
|
try:
|
|
name = item.alias.get_full_name(base=node)
|
|
except Exception:
|
|
name = getattr(item, 'name', str(item))
|
|
if hasattr(item, 'status_collection') and (depth > 1 or depth <= 0):
|
|
next_depth = depth - 1 if depth > 0 else depth
|
|
result[name] = build_tree(item, next_depth)
|
|
else:
|
|
result[name] = {}
|
|
return result
|
|
|
|
def format_tree(tree, prefix=''):
|
|
lines = []
|
|
items = list(tree.items())
|
|
for index, (name, subtree) in enumerate(items):
|
|
connector = '└── ' if index == len(items) - 1 else '├── '
|
|
lines.append(f"{prefix}{connector}{name}")
|
|
if subtree:
|
|
extension = ' ' if index == len(items) - 1 else '│ '
|
|
lines.extend(format_tree(subtree, prefix + extension))
|
|
return lines
|
|
|
|
tree = build_tree(self, level)
|
|
if print_tree:
|
|
if tree:
|
|
for line in format_tree(tree):
|
|
print(line)
|
|
else:
|
|
print('(empty)')
|
|
return tree
|
|
|
|
def status(self, get_string=False):
|
|
stat = self.get_status()
|
|
s = format_table([[name, value] for name, value in stat["status"].items()])
|
|
if get_string:
|
|
return s
|
|
else:
|
|
print(s)
|
|
|
|
def settings(self, get_string=False):
|
|
stat = self.get_status()
|
|
s = format_table(
|
|
[
|
|
[colorama.Style.BRIGHT + name + colorama.Style.RESET_ALL, value]
|
|
for name, value in stat["settings"].items()
|
|
]
|
|
)
|
|
if get_string:
|
|
return s
|
|
else:
|
|
print(s)
|
|
|
|
def get_status_str(self, base=None, stat_fields=["settings"]):
|
|
stat = self.get_status(base=base)
|
|
stat_filt = {}
|
|
for stat_field in stat_fields:
|
|
tstat = stat[stat_field]
|
|
for to in self.view_toplevel_only:
|
|
tname = to.alias.get_full_name(base=base)
|
|
tstat = filter_names(tname, tstat)
|
|
stat_filt[stat_field] = tstat
|
|
s = format_table([[name, value] for name, value in stat_filt[stat_field].items()])
|
|
return s
|
|
|
|
def get_display_str(
|
|
self,
|
|
tablefmt="simple",
|
|
with_base_name=False,
|
|
maxcolwidths=[None, 50, None, None, None],
|
|
):
|
|
main_name = self.name
|
|
stats = self.status_collection.get_list(selection="display")
|
|
# stats_dict = {}
|
|
tab = []
|
|
for to in stats:
|
|
name = to.alias.get_full_name(base=self)
|
|
|
|
is_adjustable = isinstance(to, Adjustable)
|
|
is_detector = isinstance(to, Detector)
|
|
typechar = ""
|
|
if is_adjustable:
|
|
typechar += "✏️"
|
|
elif is_detector:
|
|
typechar += "👁️"
|
|
if hasattr(to, "status_collection"):
|
|
typechar += " ↳"
|
|
|
|
try:
|
|
value = to.get_current_value()
|
|
except AttributeError:
|
|
if hasattr(to, "status_collection"):
|
|
value = "\x1b[3mhas lower level items\x1b[0m"
|
|
|
|
if isinstance(value, Enum):
|
|
value = f"{value.value} ({value.name})"
|
|
try:
|
|
unit = to.unit.get_current_value()
|
|
except:
|
|
unit = ""
|
|
try:
|
|
description = to.description.get_current_value()
|
|
except:
|
|
description = ""
|
|
|
|
if value is None:
|
|
value = ""
|
|
if with_base_name:
|
|
tab.append(
|
|
[".".join([main_name, name]), value, unit, typechar, description]
|
|
)
|
|
else:
|
|
tab.append([name, value, unit, typechar, description])
|
|
if tab:
|
|
s = format_table(tab, tablefmt=tablefmt, maxcolwidths=maxcolwidths)
|
|
else:
|
|
s = ""
|
|
|
|
return s
|
|
|
|
def status_to_elog(
|
|
self,
|
|
text="",
|
|
elog=None,
|
|
files=None,
|
|
text_encoding="markdown",
|
|
auto_title=True,
|
|
attach_display=True,
|
|
attach_status_file=True,
|
|
):
|
|
if elog is None:
|
|
elog = self._get_elog()
|
|
message = ""
|
|
|
|
if auto_title:
|
|
message += markdown(f"#### Status {self.alias.get_full_name()}")
|
|
|
|
if text:
|
|
if text_encoding == "markdown":
|
|
message += markdown(text)
|
|
if attach_display:
|
|
message += self.get_display_str(tablefmt="html")
|
|
if files is None:
|
|
files = []
|
|
if attach_status_file:
|
|
stat = self.get_status()
|
|
tmppath = Path("/tmp")
|
|
filepath = tmppath / Path(
|
|
f"status_{self.alias.get_full_name}_{datetime.now().isoformat()}.json"
|
|
)
|
|
with open(filepath, "w") as f:
|
|
# json.dump(stat, f, cls=NumpyEncoder, indent=4)
|
|
json.dump(stat, f, indent=4, cls=NumpyEncoder)
|
|
files.append(filepath)
|
|
|
|
if len(files) > 1:
|
|
print(files)
|
|
|
|
return elog.post(
|
|
message,
|
|
*files,
|
|
text_encoding="html",
|
|
)
|
|
# tags=[],
|
|
|
|
def __repr__(self):
|
|
label = self.alias.get_full_name() + " display\n"
|
|
return label + self.get_display_str()
|
|
|
|
# def _wait_for_initialisation(self, timeout=2):
|
|
# for ton, to in self.__dict__.items():
|
|
# try:
|
|
# iswaitable = isinstance(to, InitialisationWaitable)
|
|
# if iswaitable:
|
|
# to._wait_for_initialisation()
|
|
# except:
|
|
# pass
|
|
|
|
def _wait_for_initialisation(self):
|
|
for item in self.status_collection.get_list():
|
|
if isinstance(item, Assembly) and (item in _initializing_assemblies):
|
|
continue
|
|
if isinstance(item, InitialisationWaitable):
|
|
if isinstance(item, Assembly):
|
|
_initializing_assemblies.append(item)
|
|
item._wait_for_initialisation()
|
|
|
|
def _run_cmd(self, line, silent=True):
|
|
if silent:
|
|
print(f"Starting following commandline silently:\n" + line)
|
|
with open(os.devnull, "w") as FNULL:
|
|
subprocess.Popen(
|
|
line, shell=True, stdout=FNULL, stderr=subprocess.STDOUT
|
|
)
|
|
else:
|
|
subprocess.Popen(line, shell=True)
|
|
|
|
def _get_elog(self):
|
|
if hasattr(self, "_elog") and self._elog:
|
|
return self._elog
|
|
elif hasattr(self, "__elog") and self.__elog:
|
|
return self.__elog
|
|
elif eco.defaults.ELOG:
|
|
return eco.defaults.ELOG
|
|
else:
|
|
return None
|
|
|
|
def widget(self):
|
|
from eco.utilities.utilities import is_notebook
|
|
|
|
if is_notebook():
|
|
from eco.widgets.display_widget import make_assembly_widget
|
|
|
|
return make_assembly_widget(self)
|
|
else:
|
|
try:
|
|
from eco.widgets.display_qt import make_assembly_qt_window
|
|
|
|
return make_assembly_qt_window(self)
|
|
except ImportError:
|
|
from eco.widgets.display_tk import make_assembly_tk_window
|
|
|
|
return make_assembly_tk_window(self)
|
|
|
|
def _ipython_display_(self):
|
|
"""Show the interactive widget when this object is the result of a
|
|
Jupyter cell, or passed to `display()`. Falls back to the normal
|
|
repr for a terminal/non-notebook IPython session, or if building
|
|
the widget fails.
|
|
"""
|
|
from eco.utilities.utilities import is_notebook
|
|
from IPython.display import display
|
|
|
|
if is_notebook():
|
|
try:
|
|
display(self.widget())
|
|
return
|
|
except Exception as e:
|
|
print(f"Could not build widget for {self.alias.get_full_name()}: {e}")
|
|
print(repr(self))
|
|
|
|
def show(self, in_window=False, exclude_group_ids=None):
|
|
"""Opens the interactive SVG viewer for `_show_svg`, if defined on this instance.
|
|
|
|
Commands clicked in the SVG run against this assembly's own full
|
|
alias name as namespace prefix, e.g. clicking a device labelled
|
|
"slit_att" inside an assembly aliased "bernina.optics" runs
|
|
"bernina.optics.slit_att" in the IPython session.
|
|
"""
|
|
svg_path = getattr(self, "_show_svg", None)
|
|
if not svg_path:
|
|
print(f"No _show_svg defined for {self.alias.get_full_name()}.")
|
|
return
|
|
|
|
from eco.utilities.svg_interactor import launch_svg_viewer
|
|
|
|
launch_svg_viewer(
|
|
svg_path,
|
|
in_window=in_window,
|
|
namespace_prefix=self.alias.get_full_name(),
|
|
exclude_group_ids=exclude_group_ids,
|
|
)
|
|
|
|
|
|
import epics.pv
|
|
import time
|
|
|
|
|
|
class Monitor:
|
|
def __init__(self, assembly):
|
|
self.assembly = assembly
|
|
self.data = {}
|
|
self.callbacks = {}
|
|
self.pvs = {}
|
|
|
|
def start_monitoring(self):
|
|
o = self.assembly.get_status(channeltypes=["CA"])
|
|
# self.data = {k: [v] for k, v in o["status"].items()}
|
|
self.channelkeys = {v: k for k, v in o["status_channels"].items()}
|
|
self.pvs = {k: epics.pv.PV(v) for k, v in o["status_channels"].items()}
|
|
# for cik, civ in epics.pv._PVcache_.items():
|
|
# if cik[0] in o["status_channels"].keys():
|
|
# tname = self.channelkeys[cik[0]]
|
|
# tpv = civ
|
|
for tname, tpv in self.pvs.items():
|
|
self.callbacks[tname] = tpv.add_callback(self.append)
|
|
|
|
def stop_monitoring(self):
|
|
for tname in self.pvs:
|
|
self.pvs[tname].remove_callback(index=self.callbacks[tname])
|
|
|
|
def append(self, pvname=None, value=None, timestamp=None, **kwargs):
|
|
if not (self.channelkeys[pvname] in self.data):
|
|
self.data[self.channelkeys[pvname]] = []
|
|
ts_local = time.time()
|
|
self.data[self.channelkeys[pvname]].append(
|
|
{"value": value, "timestamp": timestamp, "timestamp_local": ts_local}
|
|
)
|
|
|
|
|
|
def filter_names(name, stat_dict):
|
|
out = {}
|
|
for key, value in stat_dict.items():
|
|
keys = key.split(".")
|
|
if keys[0] == name:
|
|
if len(keys) == 1:
|
|
out[key] = value
|
|
else:
|
|
out[key] = value
|
|
return out
|