#!/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.
#
# ###########################################################################
"""Read-only diagnostic snapshots for live Taurus models."""
from dataclasses import dataclass
from typing import Optional
from taurus.core.taurusbasetypes import TaurusEventType
__all__ = [
"AttributeDiagnostics",
"DeviceDiagnostics",
"FactoryDiagnostics",
"collect_attribute_diagnostics",
"collect_device_diagnostics",
"collect_factory_diagnostics",
]
[docs]
@dataclass(frozen=True)
class AttributeDiagnostics:
"""Diagnostic information for an attribute model."""
name: str
listener_count: int
listeners: tuple
subscription_state: str
polling_active: bool
polling_period: int
event_counts: dict
last_event_time: Optional[float]
[docs]
@dataclass(frozen=True)
class DeviceDiagnostics:
"""Diagnostic information for a device model.
``idl_version`` and ``events_support`` are optional, backend-specific
capabilities. They are currently provided by Tango devices.
"""
name: str
idl_version: Optional[int]
events_support: str
listener_count: int
listeners: tuple
event_counts: dict
@property
def event_count(self):
"""Return the total number of attribute events for this device."""
return sum(self.event_counts.values())
[docs]
@dataclass(frozen=True)
class FactoryDiagnostics:
"""Diagnostic information provided by a Taurus factory."""
name: str
schemes: tuple
details: dict
def _listener_name(listener):
cls = type(listener)
try:
owner = getattr(listener, "__self__", None)
if owner is not None:
cls = type(owner)
callback = getattr(listener, "__name__", type(listener).__qualname__)
return "{}.{}.{} [object 0x{:x}]".format(
cls.__module__, cls.__qualname__, callback, id(owner)
)
callback = getattr(listener, "eventReceived", None)
suffix = ".eventReceived" if callable(callback) else ""
return "{}.{}{} [object 0x{:x}]".format(
cls.__module__, cls.__qualname__, suffix, id(listener)
)
except RuntimeError:
# A Qt wrapper may remain alive briefly after its C++ object has been
# deleted. Keep the stale registration visible without aborting the
# complete diagnostics refresh.
return "{}.{} [deleted Qt object 0x{:x}]".format(
cls.__module__, cls.__qualname__, id(listener)
)
def _event_name(event_type):
try:
return TaurusEventType.whatis(event_type)
except (KeyError, TypeError):
return str(event_type)
def _subscription_state(attribute):
get_state = getattr(attribute, "getSubscriptionState", None)
if get_state is None:
return "N/A"
state = get_state()
try:
from taurus.core import SubscriptionState
return SubscriptionState.whatis(state)
except (KeyError, TypeError):
return str(state)
def _iter_factories(factory):
"""Yield one requested factory, or one instance of every backend."""
if factory is not None:
return (factory,)
from taurus import Manager
factories = []
seen = set()
for factory_class in Manager().getPlugins().values():
if factory_class in seen:
continue
seen.add(factory_class)
try:
factories.append(factory_class())
except Exception:
# A diagnostics window must not fail because an optional backend
# cannot be initialized in the current environment.
continue
return tuple(factories)
[docs]
def collect_attribute_diagnostics(factory=None):
"""Return snapshots for attributes already created by Taurus factories.
This function only inspects existing models. It never creates attributes,
registers listeners, starts polling, or changes event subscriptions.
:param factory: optional factory providing ``getExistingAttributes``. If
omitted, all available Taurus factories are inspected.
:return: tuple of :class:`AttributeDiagnostics`, sorted by model name
"""
snapshots = []
for current_factory in _iter_factories(factory):
for attribute in current_factory.getExistingAttributes().values():
get_listeners = getattr(attribute, "getListeners", None)
listeners = get_listeners() if get_listeners is not None else ()
get_counts = getattr(attribute, "getEventCounts", None)
counts = get_counts() if get_counts is not None else {}
counts = {_event_name(key): value for key, value in counts.items()}
snapshots.append(
AttributeDiagnostics(
name=attribute.getFullName(),
listener_count=len(listeners),
listeners=tuple(_listener_name(item) for item in listeners),
subscription_state=_subscription_state(attribute),
polling_active=attribute.isPollingActive(),
polling_period=attribute.getPollingPeriod(),
event_counts=counts,
last_event_time=attribute.getLastEventTime(),
)
)
return tuple(sorted(snapshots, key=lambda item: item.name))
[docs]
def collect_factory_diagnostics(factory=None):
"""Return runtime information for the available Taurus factories.
Optional factories which cannot be initialized are omitted. A failure in
one factory's diagnostic hook does not prevent reporting the others.
"""
snapshots = []
for current_factory in _iter_factories(factory):
try:
details = dict(current_factory.getDiagnosticInfo())
except Exception:
details = {}
snapshots.append(
FactoryDiagnostics(
name=type(current_factory).__name__,
schemes=tuple(getattr(current_factory, "schemes", ())),
details=details,
)
)
return tuple(sorted(snapshots, key=lambda item: item.name.lower()))
def _read_device_capabilities(factory, device):
try:
info = factory.getDeviceDiagnosticInfo(device)
except Exception:
return None, "Unknown"
return info.get("idl_version"), info.get("events_support", "N/A")
[docs]
def collect_device_diagnostics(factory=None, device_info_cache=None):
"""Return snapshots for devices already created by Taurus factories.
If *factory* is omitted, all available Taurus factories are inspected.
``device_info_cache`` avoids repeating backend capability queries on every
UI refresh. Event count is the total number of events dispatched while
event statistics were enabled for attributes belonging to the device.
"""
if device_info_cache is None:
device_info_cache = {}
snapshots = []
for current_factory in _iter_factories(factory):
attributes = tuple(current_factory.getExistingAttributes().values())
for device in current_factory.getExistingDevices().values():
name = device.getFullName()
if name not in device_info_cache:
device_info_cache[name] = _read_device_capabilities(
current_factory, device
)
idl_version, events_support = device_info_cache[name]
listeners = device.getListeners()
event_counts = {}
for attribute in attributes:
if attribute.getParentObj() is not device:
continue
for event_type, count in attribute.getEventCounts().items():
event_name = _event_name(event_type)
event_counts[event_name] = event_counts.get(event_name, 0) + count
snapshots.append(
DeviceDiagnostics(
name=name,
idl_version=idl_version,
events_support=events_support,
listener_count=len(listeners),
listeners=tuple(_listener_name(item) for item in listeners),
event_counts=event_counts,
)
)
return tuple(sorted(snapshots, key=lambda item: item.name))