#!/usr/bin/env python
# ###########################################################################
#
# This file is part of Taurus
#
# http://taurus-scada.org
#
# Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
#
# Taurus is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# ###########################################################################
"""Widgets for inspecting diagnostics of the current Taurus process."""
import os
import platform
import time
from importlib import metadata
import taurus
from taurus.core.diagnostics import (
collect_attribute_diagnostics,
collect_device_diagnostics,
collect_factory_diagnostics,
)
from taurus.core.taurusmodel import TaurusModel
from taurus.external import qt as qt_compat
from taurus.external.qt import Qt
__all__ = ["TaurusDiagnosticsDialog"]
class _NumericTableWidgetItem(Qt.QTableWidgetItem):
"""Table item which displays text but sorts by a numeric value."""
def __init__(self, text, sort_value):
Qt.QTableWidgetItem.__init__(self, text)
self.sort_value = sort_value
def __lt__(self, other):
if isinstance(other, _NumericTableWidgetItem):
return self.sort_value < other.sort_value
return Qt.QTableWidgetItem.__lt__(self, other)
[docs]
class TaurusDiagnosticsDialog(Qt.QDialog):
"""Display live, read-only diagnostics for the current application."""
DEFAULT_REFRESH_INTERVAL_SECONDS = 3
EVENT_TYPES = ("Change", "Periodic", "Config", "Error")
DEVICE_COLUMNS = (
"Device name",
"IDL version",
"Events support",
"Listeners",
"Event count",
)
ATTRIBUTE_COLUMNS = (
"Attribute",
"Subscription",
"Listeners",
"Polling",
"Period (ms)",
"Event count",
"Change (Hz)",
"Periodic (Hz)",
"Errors",
"Last event",
)
DEVICE_NUMERIC_COLUMNS = (1, 3, 4)
ATTRIBUTE_NUMERIC_COLUMNS = (2, 4, 5, 6, 7, 8)
def __init__(self, parent=None, factory=None):
Qt.QDialog.__init__(self, parent)
self.setWindowFlags(
Qt.Qt.Window
| Qt.Qt.WindowTitleHint
| Qt.Qt.WindowSystemMenuHint
| Qt.Qt.WindowMinMaxButtonsHint
| Qt.Qt.WindowCloseButtonHint
)
self.setWindowTitle("Taurus diagnostics")
self.resize(1050, 600)
self._factory = factory
self._samples = {}
self._snapshots = {}
self._device_snapshots = {}
self._device_info_cache = {}
self._attribute_columns_sized = False
self._device_columns_sized = False
self._process = self._create_process()
self._tabs = Qt.QTabWidget(self)
self._tabs.addTab(self._create_process_page(), "Process")
self._tabs.addTab(self._create_devices_page(), "Devices")
self._tabs.addTab(self._create_attributes_page(), "Attributes")
self._tabs.currentChanged.connect(self._refresh_tab)
self._refresh_button = Qt.QPushButton("Refresh", self)
self._refresh_button.clicked.connect(self.refresh)
self._refresh_interval_spinbox = Qt.QSpinBox(self)
self._refresh_interval_spinbox.setRange(1, 3600)
self._refresh_interval_spinbox.setSuffix(" s")
self._refresh_interval_spinbox.setValue(self.DEFAULT_REFRESH_INTERVAL_SECONDS)
self._refresh_interval_spinbox.setToolTip(
"Interval between automatic updates of the visible tab"
)
close_button = Qt.QPushButton("Close", self)
close_button.clicked.connect(self.close)
buttons = Qt.QHBoxLayout()
buttons.addWidget(Qt.QLabel("Refresh interval:", self))
buttons.addWidget(self._refresh_interval_spinbox)
buttons.addStretch()
buttons.addWidget(self._refresh_button)
buttons.addWidget(close_button)
layout = Qt.QVBoxLayout(self)
layout.addWidget(self._tabs)
layout.addLayout(buttons)
self._timer = Qt.QTimer(self)
self._set_refresh_interval(self.DEFAULT_REFRESH_INTERVAL_SECONDS)
self._timer.timeout.connect(self._refresh_current_tab)
self._refresh_interval_spinbox.valueChanged.connect(self._set_refresh_interval)
self.refresh()
def _set_refresh_interval(self, seconds):
"""Set the automatic refresh interval from its UI value."""
self._timer.setInterval(seconds * 1000)
@staticmethod
def _create_process():
try:
import psutil
return psutil.Process()
except (ImportError, OSError):
return None
def _create_process_page(self):
page = Qt.QScrollArea(self)
page.setWidgetResizable(True)
contents = Qt.QWidget(page)
layout = Qt.QVBoxLayout(contents)
environment = Qt.QGroupBox("Application environment", contents)
environment_form = Qt.QFormLayout(environment)
self._environment_labels = {}
for name, title, value in self._environment_info():
self._environment_labels[name] = self._add_info_row(
environment_form, title, value, environment
)
layout.addWidget(environment)
backends = Qt.QGroupBox("Taurus Schemes", contents)
backends_form = Qt.QFormLayout(backends)
self._backend_labels = {}
factory_snapshots = collect_factory_diagnostics(self._factory)
schemes = sorted(
scheme for snapshot in factory_snapshots for scheme in snapshot.schemes
)
plugin_names = []
for snapshot in factory_snapshots:
plugin = snapshot.name
if snapshot.schemes:
plugin += " ({})".format(", ".join(snapshot.schemes))
plugin_names.append(plugin)
self._backend_labels["schemes"] = self._add_info_row(
backends_form, "Available schemes", ", ".join(schemes) or "-", backends
)
self._backend_labels["plugins"] = self._add_info_row(
backends_form, "Factory plugins", "; ".join(plugin_names) or "-", backends
)
for snapshot in factory_snapshots:
backend_name = snapshot.name
if backend_name.endswith("Factory"):
backend_name = backend_name[: -len("Factory")]
for detail_name, value in snapshot.details.items():
key = "{}:{}".format(snapshot.name, detail_name)
self._backend_labels[key] = self._add_info_row(
backends_form,
"{} / {}".format(backend_name, detail_name),
value,
backends,
)
layout.addWidget(backends)
process = Qt.QGroupBox("Process usage", contents)
process_form = Qt.QFormLayout(process)
self._process_labels = {}
for name, title in (
("pid", "PID"),
("started", "Process started"),
("cpu", "CPU"),
("memory", "Memory"),
("rss", "Resident memory"),
):
label = Qt.QLabel("-", process)
label.setTextInteractionFlags(Qt.Qt.TextSelectableByMouse)
self._process_labels[name] = label
process_form.addRow(title + ":", label)
layout.addWidget(process)
layout.addStretch()
page.setWidget(contents)
return page
@staticmethod
def _add_info_row(form, title, value, parent):
label = Qt.QLabel(str(value), parent)
label.setTextInteractionFlags(Qt.Qt.TextSelectableByMouse)
label.setWordWrap(True)
form.addRow(title + ":", label)
return label
@staticmethod
def _package_version(*distribution_names):
"""Return an installed distribution version without importing it."""
for distribution_name in distribution_names:
try:
return metadata.version(distribution_name)
except metadata.PackageNotFoundError:
continue
except Exception:
return "Unknown"
return None
def _environment_info(self):
application = Qt.QCoreApplication.instance()
application_name = application.applicationName() if application else ""
application_version = application.applicationVersion() if application else ""
qt_platform = Qt.QGuiApplication.platformName()
if qt_platform == "xcb":
window_system = "X11 (xcb)"
elif qt_platform.startswith("wayland"):
window_system = "Wayland ({})".format(qt_platform)
else:
window_system = qt_platform or "Unknown"
binding_version = qt_compat.PYQT_VERSION or qt_compat.PYSIDE_VERSION
info = []
if application_name:
info.append(("application", "Application", application_name))
if application_version:
info.append(
("application_version", "Application version", application_version)
)
info.extend(
(
("taurus", "Taurus", taurus.__version__),
(
"python",
"Python",
"{} {}".format(
platform.python_implementation(), platform.python_version()
),
),
("operating_system", "Operating system", platform.platform()),
("window_system", "Window system", window_system),
("qt", "Qt", qt_compat.QT_VERSION),
(
"qt_binding",
"Qt binding",
"{} {}".format(qt_compat.API_NAME, binding_version),
),
)
)
plotpy_version = self._package_version("plotpy") or "not installed"
info.append(("plotpy", "plotpy", plotpy_version))
taurus_pyqtgraph_version = self._package_version(
"taurus-pyqtgraph", "taurus_pyqtgraph"
)
if taurus_pyqtgraph_version is None:
plotting_version = "not installed"
else:
pyqtgraph_version = self._package_version("pyqtgraph") or "not installed"
plotting_version = "{} (pyqtgraph {})".format(
taurus_pyqtgraph_version, pyqtgraph_version
)
info.append(("taurus_pyqtgraph", "taurus_pyqtgraph", plotting_version))
return info
def _create_attributes_page(self):
page = Qt.QWidget(self)
layout = Qt.QVBoxLayout(page)
self._attribute_summary = Qt.QLabel(page)
layout.addWidget(self._attribute_summary)
splitter = Qt.QSplitter(Qt.Qt.Vertical, page)
self._attribute_table = Qt.QTableWidget(
0, len(self.ATTRIBUTE_COLUMNS), splitter
)
self._attribute_table.setHorizontalHeaderLabels(self.ATTRIBUTE_COLUMNS)
self._attribute_table.setEditTriggers(Qt.QAbstractItemView.NoEditTriggers)
self._attribute_table.setSelectionBehavior(Qt.QAbstractItemView.SelectRows)
self._attribute_table.setSelectionMode(Qt.QAbstractItemView.SingleSelection)
self._attribute_table.setSortingEnabled(True)
self._attribute_table.sortItems(0, Qt.Qt.AscendingOrder)
self._attribute_table.itemSelectionChanged.connect(self._show_listeners)
header = self._attribute_table.horizontalHeader()
for column in range(len(self.ATTRIBUTE_COLUMNS)):
header.setSectionResizeMode(column, Qt.QHeaderView.Interactive)
details = Qt.QGroupBox("Registered listeners", splitter)
details_layout = Qt.QVBoxLayout(details)
self._listeners = Qt.QPlainTextEdit(details)
self._listeners.setReadOnly(True)
self._listeners.setPlaceholderText(
"Select an attribute to inspect its listeners"
)
details_layout.addWidget(self._listeners)
splitter.addWidget(self._attribute_table)
splitter.addWidget(details)
splitter.setStretchFactor(0, 3)
splitter.setStretchFactor(1, 1)
layout.addWidget(splitter)
return page
def _create_devices_page(self):
page = Qt.QWidget(self)
layout = Qt.QVBoxLayout(page)
self._device_summary = Qt.QLabel(page)
layout.addWidget(self._device_summary)
splitter = Qt.QSplitter(Qt.Qt.Vertical, page)
self._device_table = Qt.QTableWidget(0, len(self.DEVICE_COLUMNS), splitter)
self._device_table.setHorizontalHeaderLabels(self.DEVICE_COLUMNS)
self._device_table.setEditTriggers(Qt.QAbstractItemView.NoEditTriggers)
self._device_table.setSelectionBehavior(Qt.QAbstractItemView.SelectRows)
self._device_table.setSelectionMode(Qt.QAbstractItemView.SingleSelection)
self._device_table.setSortingEnabled(True)
self._device_table.itemSelectionChanged.connect(self._show_device_listeners)
header = self._device_table.horizontalHeader()
for column in range(len(self.DEVICE_COLUMNS)):
header.setSectionResizeMode(column, Qt.QHeaderView.Interactive)
details = Qt.QGroupBox("Registered listeners", splitter)
details_layout = Qt.QVBoxLayout(details)
self._device_listeners = Qt.QPlainTextEdit(details)
self._device_listeners.setReadOnly(True)
self._device_listeners.setPlaceholderText(
"Select a device to inspect its listeners"
)
details_layout.addWidget(self._device_listeners)
splitter.addWidget(self._device_table)
splitter.addWidget(details)
splitter.setStretchFactor(0, 3)
splitter.setStretchFactor(1, 1)
layout.addWidget(splitter)
return page
[docs]
def showEvent(self, event):
TaurusModel.setEventStatisticsEnabled(True)
self.refresh()
self._timer.start()
return Qt.QDialog.showEvent(self, event)
[docs]
def hideEvent(self, event):
self._timer.stop()
TaurusModel.setEventStatisticsEnabled(False)
return Qt.QDialog.hideEvent(self, event)
[docs]
def refresh(self):
self._refresh_process()
self._refresh_devices()
self._refresh_attributes()
def _refresh_current_tab(self):
self._refresh_tab(self._tabs.currentIndex())
def _refresh_tab(self, index):
if index == 0:
self._refresh_process()
elif index == 1:
self._refresh_devices()
elif index == 2:
self._refresh_attributes()
def _refresh_process(self):
labels = self._process_labels
labels["pid"].setText(str(os.getpid()))
if self._process is None:
labels["started"].setText("Unavailable (install psutil)")
labels["cpu"].setText("Unavailable")
labels["memory"].setText("Unavailable")
labels["rss"].setText("Unavailable")
return
labels["started"].setText(
time.strftime(
"%Y-%m-%d %H:%M:%S", time.localtime(self._process.create_time())
)
)
labels["cpu"].setText("{:.1f} %".format(self._process.cpu_percent()))
labels["memory"].setText("{:.1f} %".format(self._process.memory_percent()))
rss = self._process.memory_info().rss / 2**20
labels["rss"].setText("{:.1f} MiB".format(rss))
def _refresh_attributes(self):
now = time.monotonic()
snapshots = collect_attribute_diagnostics(self._factory)
self._snapshots = {snapshot.name: snapshot for snapshot in snapshots}
self._attribute_summary.setText(
"{} existing attribute model(s). Counters advance while this window is open.".format(
len(snapshots)
)
)
selected_name = self._selected_attribute_name()
rows = []
for snapshot in snapshots:
previous_time, previous_counts = self._samples.get(
snapshot.name, (now, snapshot.event_counts)
)
elapsed = max(now - previous_time, 0.001)
change_rate = (
snapshot.event_counts.get("Change", 0)
- previous_counts.get("Change", 0)
) / elapsed
periodic_rate = (
snapshot.event_counts.get("Periodic", 0)
- previous_counts.get("Periodic", 0)
) / elapsed
last_event = "-"
if snapshot.last_event_time is not None:
last_event = time.strftime(
"%H:%M:%S", time.localtime(snapshot.last_event_time)
)
values = (
snapshot.name,
snapshot.subscription_state,
snapshot.listener_count,
"Yes" if snapshot.polling_active else "No",
snapshot.polling_period if snapshot.polling_active else "-",
sum(snapshot.event_counts.values()),
"{:.2f}".format(max(change_rate, 0)),
"{:.2f}".format(max(periodic_rate, 0)),
snapshot.event_counts.get("Error", 0),
last_event,
)
rows.append((snapshot.name, values))
self._samples[snapshot.name] = (now, snapshot.event_counts)
self._samples = {
name: sample
for name, sample in self._samples.items()
if name in self._snapshots
}
self._update_table(self._attribute_table, rows, self.ATTRIBUTE_NUMERIC_COLUMNS)
self._set_event_count_tooltips(self._attribute_table, 5, self._snapshots)
if not self._attribute_columns_sized and rows:
self._attribute_table.resizeColumnsToContents()
self._attribute_columns_sized = True
self._restore_selection(selected_name)
self._show_listeners()
def _refresh_devices(self):
snapshots = collect_device_diagnostics(self._factory, self._device_info_cache)
self._device_snapshots = {snapshot.name: snapshot for snapshot in snapshots}
self._device_summary.setText(
"{} existing device model(s). IDL is queried once; counters advance "
"only while this window is open.".format(len(snapshots))
)
selected_name = self._selected_device_name()
rows = []
for snapshot in snapshots:
if snapshot.idl_version is None:
idl_version = "N/A" if snapshot.events_support == "N/A" else "Unknown"
else:
idl_version = snapshot.idl_version
values = (
snapshot.name,
idl_version,
snapshot.events_support,
snapshot.listener_count,
snapshot.event_count,
)
rows.append((snapshot.name, values))
self._update_table(self._device_table, rows, self.DEVICE_NUMERIC_COLUMNS)
self._set_event_count_tooltips(self._device_table, 4, self._device_snapshots)
if not self._device_columns_sized and rows:
self._device_table.resizeColumnsToContents()
self._device_columns_sized = True
self._restore_device_selection(selected_name)
self._show_device_listeners()
@staticmethod
def _update_table(table, rows, numeric_columns=()):
"""Update table contents without rebuilding unchanged rows."""
sorting_enabled = table.isSortingEnabled()
signals_blocked = table.blockSignals(True)
updates_enabled = table.updatesEnabled()
table.setUpdatesEnabled(False)
table.setSortingEnabled(False)
try:
existing_rows = {}
for row in range(table.rowCount()):
item = table.item(row, 0)
if item is not None:
existing_rows[item.text()] = row
names = [name for name, values in rows]
if set(existing_rows) != set(names):
table.clearContents()
table.setRowCount(len(rows))
existing_rows = {name: row for row, (name, values) in enumerate(rows)}
for name, values in rows:
row = existing_rows[name]
for column, value in enumerate(values):
text = str(value)
item = table.item(row, column)
if column in numeric_columns:
try:
sort_value = float(value)
except (TypeError, ValueError):
sort_value = float("-inf")
if not isinstance(item, _NumericTableWidgetItem):
item = _NumericTableWidgetItem(text, sort_value)
table.setItem(row, column, item)
else:
item.sort_value = sort_value
if item.text() != text:
item.setText(text)
elif item is None:
table.setItem(row, column, Qt.QTableWidgetItem(text))
elif item.text() != text:
item.setText(text)
finally:
table.setSortingEnabled(sorting_enabled)
table.setUpdatesEnabled(updates_enabled)
table.blockSignals(signals_blocked)
@classmethod
def _event_count_tooltip(cls, event_counts):
return "\n".join(
"{} {}".format(event_counts.get(name, 0), name.lower())
for name in cls.EVENT_TYPES
)
@classmethod
def _set_event_count_tooltips(cls, table, column, snapshots):
for row in range(table.rowCount()):
name_item = table.item(row, 0)
count_item = table.item(row, column)
if name_item is None or count_item is None:
continue
snapshot = snapshots.get(name_item.text())
if snapshot is not None:
count_item.setToolTip(cls._event_count_tooltip(snapshot.event_counts))
def _selected_attribute_name(self):
items = self._attribute_table.selectedItems()
if not items:
return None
return self._attribute_table.item(items[0].row(), 0).text()
def _selected_device_name(self):
items = self._device_table.selectedItems()
if not items:
return None
return self._device_table.item(items[0].row(), 0).text()
def _restore_selection(self, name):
if name is None:
return
for row in range(self._attribute_table.rowCount()):
if self._attribute_table.item(row, 0).text() == name:
self._attribute_table.selectRow(row)
return
def _restore_device_selection(self, name):
if name is None:
return
for row in range(self._device_table.rowCount()):
if self._device_table.item(row, 0).text() == name:
self._device_table.selectRow(row)
return
def _show_listeners(self):
name = self._selected_attribute_name()
snapshot = self._snapshots.get(name)
if snapshot is None:
self._listeners.clear()
return
if snapshot.listeners:
self._listeners.setPlainText("\n".join(snapshot.listeners))
else:
self._listeners.setPlainText("No listeners registered")
def _show_device_listeners(self):
name = self._selected_device_name()
snapshot = self._device_snapshots.get(name)
if snapshot is None:
self._device_listeners.clear()
return
if snapshot.listeners:
self._device_listeners.setPlainText("\n".join(snapshot.listeners))
else:
self._device_listeners.setPlainText("No listeners registered")