Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Generic PyDM Widget for testing plugins. Contains a more thorough waveform
Channel.
Lots of copy/paste because pyqt multiple inheritance is broken.
"""
__pyqtSignals__ = ("send_waveform_signal(np.ndarray)",
"waveform_updated_signal()",
"conn_updated_signal()",
"sevr_updated_signal()",
"rwacc_updated_signal()",
"enums_updated_signal()",
"units_updated_signal()",
"prec_updated_signal()",)
send_waveform_signal = pyqtSignal(np.ndarray)
waveform_updated_signal = pyqtSignal()
def __init__(self, channel=None, parent=None):
super(WaveformWidget, self).__init__(channel=channel, parent=parent)
@pyqtSlot(np.ndarray)
def recv_waveform(self, waveform):
"""
Recieve waveform from plugin signal. Store in self.value.
:param waveform: waveform
:type waveform: np.ndarray
"""
self.value = waveform
self.waveform_updated_signal.emit()
Check that we can put to the waveform set function
"""
array = np.asarray((5, 5, 6))
self.set_widget.send_waveform(array)
self.event_loop(0)
result = self.obj.get()
self.assertEqual(tuple(result), tuple(array),
msg="waveform function has wrong result: was {0}, expected {1}".format(
result, array))
class SignalHolder(QObject):
"""
Dummy QObject to let us use pyqtSignal
"""
value_sig = pyqtSignal([int], [float], [str])
empty_sig = pyqtSignal()
class SignalTestCase(LocalPluginTest):
def test_signal_update(self):
"""
Check that we can update t=0 widgets intelligently with signals
"""
sig = SignalHolder()
self.plugin.connect_to_update("x?t=0", sig.empty_sig)
self.stale_widget.send_value(4)
self.event_loop(0.01)
self.assertEqual(self.stale_widget.value, 4,
msg="t=0 widgets do not update on puts")
self.obj.x = 10
self.event_loop(2)
self.assertEqual(self.stale_widget.value, 4,
# One more process step just in case (this actually matters I think)
self.app.processEvents()
signal.disconnect(queue.put_to_queue)
# Get value if we have one (otherwise, this is None)
value = queue.get()
return value
class QueueSlots(QObject):
"""
Dummy object that contains a slot that puts signal results to a queue.
Exists for the implementation of PyDMTest.signal_wait.
"""
__pyqtSignals__ = ("got_value()",)
got_value = pyqtSignal()
def __init__(self, parent=None):
"""
Set up QObject and create internal queue.
"""
super(QueueSlots, self).__init__(parent=parent)
self.queue = Queue.Queue()
def get(self, timeout=None):
"""
Retrieve queue value or None. Waits for timeout seconds.
:param timeout: get timeout in seconds
:type timeout: float or int
:rtyp: object or None
"""
from .channel import PyDMChannel
import warnings
class PyDMIndicator(QWidget):
# Tell Designer what signals are available.
__pyqtSignals__ = ("connected_signal()",
"disconnected_signal()",
"no_alarm_signal()",
"minor_alarm_signal()",
"major_alarm_signal()",
"invalid_alarm_signal()")
# Internal signals, used by the state machine
connected_signal = pyqtSignal()
disconnected_signal = pyqtSignal()
no_alarm_signal = pyqtSignal()
minor_alarm_signal = pyqtSignal()
major_alarm_signal = pyqtSignal()
invalid_alarm_signal = pyqtSignal()
# Usually, this widget will get this from its parent pydm application. However, in Designer, the parent isnt a pydm application, and doesn't know what a color map is. The following two color maps are provided for that scenario.
local_alarm_severity_color_map = {
0: QColor(0, 0, 0), # NO_ALARM
1: QColor(200, 200, 20), # MINOR_ALARM
2: QColor(240, 0, 0), # MAJOR_ALARM
3: QColor(240, 0, 240) # INVALID_ALARM
}
local_connection_status_color_map = {
False: QColor(0, 0, 0),
True: QColor(0, 0, 0,)
}
from numpy import ndarray
from ..PyQt.QtCore import pyqtSlot, pyqtSignal, QObject, Qt
class PyDMConnection(QObject):
new_value_signal = pyqtSignal([float],[int],[str])
new_waveform_signal = pyqtSignal(ndarray)
connection_state_signal = pyqtSignal(bool)
new_severity_signal = pyqtSignal(int)
write_access_signal = pyqtSignal(bool)
enum_strings_signal = pyqtSignal(tuple)
unit_signal = pyqtSignal([str],[bytes])
prec_signal = pyqtSignal(int)
def __init__(self, channel, address, parent=None):
super(PyDMConnection, self).__init__(parent)
self.listener_count = 0
def add_listener(self, channel):
self.listener_count = self.listener_count + 1
if channel.connection_slot is not None:
self.connection_state_signal.connect(channel.connection_slot, Qt.QueuedConnection)
if channel.value_slot is not None:
from numpy import ndarray
from ..PyQt.QtCore import pyqtSlot, pyqtSignal, QObject, Qt
class PyDMConnection(QObject):
new_value_signal = pyqtSignal([float],[int],[str])
new_waveform_signal = pyqtSignal(ndarray)
connection_state_signal = pyqtSignal(bool)
new_severity_signal = pyqtSignal(int)
write_access_signal = pyqtSignal(bool)
enum_strings_signal = pyqtSignal(tuple)
unit_signal = pyqtSignal([str],[bytes])
prec_signal = pyqtSignal(int)
def __init__(self, channel, address, parent=None):
super(PyDMConnection, self).__init__(parent)
self.listener_count = 0
def add_listener(self, channel):
self.listener_count = self.listener_count + 1
if channel.connection_slot is not None:
self.connection_state_signal.connect(channel.connection_slot, Qt.QueuedConnection)
if channel.value_slot is not None:
try:
self.new_value_signal[int].connect(channel.value_slot, Qt.QueuedConnection)
from numpy import ndarray
from ..PyQt.QtCore import pyqtSlot, pyqtSignal, QObject, Qt
class PyDMConnection(QObject):
new_value_signal = pyqtSignal([float],[int],[str])
new_waveform_signal = pyqtSignal(ndarray)
connection_state_signal = pyqtSignal(bool)
new_severity_signal = pyqtSignal(int)
write_access_signal = pyqtSignal(bool)
enum_strings_signal = pyqtSignal(tuple)
unit_signal = pyqtSignal([str],[bytes])
prec_signal = pyqtSignal(int)
def __init__(self, channel, address, parent=None):
super(PyDMConnection, self).__init__(parent)
self.listener_count = 0
def add_listener(self, channel):
self.listener_count = self.listener_count + 1
if channel.connection_slot is not None:
self.connection_state_signal.connect(channel.connection_slot, Qt.QueuedConnection)
if channel.value_slot is not None:
try:
self.new_value_signal[int].connect(channel.value_slot, Qt.QueuedConnection)
except TypeError:
from numpy import ndarray
from ..PyQt.QtCore import pyqtSlot, pyqtSignal, QObject, Qt
class PyDMConnection(QObject):
new_value_signal = pyqtSignal([float],[int],[str])
new_waveform_signal = pyqtSignal(ndarray)
connection_state_signal = pyqtSignal(bool)
new_severity_signal = pyqtSignal(int)
write_access_signal = pyqtSignal(bool)
enum_strings_signal = pyqtSignal(tuple)
unit_signal = pyqtSignal([str],[bytes])
prec_signal = pyqtSignal(int)
def __init__(self, channel, address, parent=None):
super(PyDMConnection, self).__init__(parent)
self.listener_count = 0
def add_listener(self, channel):
self.listener_count = self.listener_count + 1
if channel.connection_slot is not None:
self.connection_state_signal.connect(channel.connection_slot, Qt.QueuedConnection)
if channel.value_slot is not None:
try:
from numpy import ndarray
from ..PyQt.QtCore import pyqtSlot, pyqtSignal, QObject, Qt
class PyDMConnection(QObject):
new_value_signal = pyqtSignal([float],[int],[str])
new_waveform_signal = pyqtSignal(ndarray)
connection_state_signal = pyqtSignal(bool)
new_severity_signal = pyqtSignal(int)
write_access_signal = pyqtSignal(bool)
enum_strings_signal = pyqtSignal(tuple)
unit_signal = pyqtSignal([str],[bytes])
prec_signal = pyqtSignal(int)
def __init__(self, channel, address, parent=None):
super(PyDMConnection, self).__init__(parent)
self.listener_count = 0
def add_listener(self, channel):
self.listener_count = self.listener_count + 1
if channel.connection_slot is not None:
self.connection_state_signal.connect(channel.connection_slot, Qt.QueuedConnection)
if channel.value_slot is not None:
try:
self.new_value_signal[int].connect(channel.value_slot, Qt.QueuedConnection)
except TypeError:
pass
try:
class PyDMIndicator(QWidget):
# Tell Designer what signals are available.
__pyqtSignals__ = ("connected_signal()",
"disconnected_signal()",
"no_alarm_signal()",
"minor_alarm_signal()",
"major_alarm_signal()",
"invalid_alarm_signal()")
# Internal signals, used by the state machine
connected_signal = pyqtSignal()
disconnected_signal = pyqtSignal()
no_alarm_signal = pyqtSignal()
minor_alarm_signal = pyqtSignal()
major_alarm_signal = pyqtSignal()
invalid_alarm_signal = pyqtSignal()
# Usually, this widget will get this from its parent pydm application. However, in Designer, the parent isnt a pydm application, and doesn't know what a color map is. The following two color maps are provided for that scenario.
local_alarm_severity_color_map = {
0: QColor(0, 0, 0), # NO_ALARM
1: QColor(200, 200, 20), # MINOR_ALARM
2: QColor(240, 0, 0), # MAJOR_ALARM
3: QColor(240, 0, 240) # INVALID_ALARM
}
local_connection_status_color_map = {
False: QColor(0, 0, 0),
True: QColor(0, 0, 0,)
}
def __init__(self, parent=None, init_channel=None):
warnings.warn("PyDMIndicator is deprecated. Please replace it with"