Source code for channel

import logging

import pydm.data_plugins

logger = logging.getLogger(__name__)

def clear_channel_address(channel):
    # We must remove spaces, \n, \t and other crap from
    # channel address
    if channel is None:
        return None
    return str(channel).strip()

[docs]class PyDMChannel(object): """ Object to hold signals and slots for a PyDM Widget interface to an external plugin The purpose of this class is to create a templated slot and signals list that can be sent to an external plugin. The type of plugin is determined by the PyDMApplication based on the identifier placed at the beginning of the :attr:`.address` attribute. This allows a generic way to connect slots and signals to functionality within your PyDM Widget. Slots should be connected to functions on your created widget that perform actions upon changes. For instance, the :attr:`.value_slot` will be automatically called every time a new value is found by the plugin. This should probably linked to a function that updates the display to report the new value. Signals perform the reverse operation. These should be used to send new values back to the plugin to update the source. Using this structure to interface with plugins allows your created PyDM Widget a greater flexibility in choosing its underlying source. For instance, returning to the example of the :attr:`.value_slot`, getting a value to display from channel access or from the EPICS Archiver are very different operations. However, actually displaying the value should be identical. By simply attaching your PyDM Widget's display functionality to the :attr:`.value_slot` you have created a Widget that can do either interchangeably, all the user has to do is specify the correct address signature and the rest of the work is done by the underlying plugins. Parameters ---------- address : str, optional The name of the address to be used by the plugin. This should usually be a user inputted field when a specific PyDM widget is initialized connection_slot : Slot, optional A function to be run when the connection state changes value_slot : Slot, optional A function to be run when the value updates severity_slot : Slot, optional A function to be run when the severity changes write_access_slot : Slot, optional A function to be run when the write access changes enum_strings_slot : Slot, optional A function to be run when the enum_strings change unit_slot : Slot, optional A function to be run when the unit changes prec_slot : Slot, optional A function to be run when the precision value changes upper_alarm_limit_slot : Slot, optional A function to be run when the upper alarm limit changes lower_alarm_limit_slot : Slot, optional A function to be run when the lower alarm limit changes upper_warning_limit_slot : Slot, optional A function to be run when the upper warning limit changes lower_warning_limit_slot : Slot, optional A function to be run when the lower warning limit changes value_signal : Signal, optional Attach a signal here that emits a desired value to be sent through the plugin timestamp_slot : Slot, optional A function to be run when the timestamp updates """ def __init__( self, address=None, connection_slot=None, value_slot=None, severity_slot=None, write_access_slot=None, enum_strings_slot=None, unit_slot=None, prec_slot=None, upper_ctrl_limit_slot=None, lower_ctrl_limit_slot=None, upper_alarm_limit_slot=None, lower_alarm_limit_slot=None, upper_warning_limit_slot=None, lower_warning_limit_slot=None, value_signal=None, timestamp_slot=None, ): self._address = None self.address = address self.connection_slot = connection_slot self.value_slot = value_slot self.severity_slot = severity_slot self.write_access_slot = write_access_slot self.enum_strings_slot = enum_strings_slot self.unit_slot = unit_slot self.prec_slot = prec_slot self.upper_ctrl_limit_slot = upper_ctrl_limit_slot self.lower_ctrl_limit_slot = lower_ctrl_limit_slot self.upper_alarm_limit_slot = upper_alarm_limit_slot self.lower_alarm_limit_slot = lower_alarm_limit_slot self.upper_warning_limit_slot = upper_warning_limit_slot self.lower_warning_limit_slot = lower_warning_limit_slot self.timestamp_slot = timestamp_slot self.value_signal = value_signal @property def address(self): return self._address @address.setter def address(self, address): self._address = clear_channel_address(address)
[docs] def connect(self): """ Connect a PyDMChannel to the proper PyDMPlugin """ if not self.address: return logger.debug("Connecting %r", self.address) # Connect to proper PyDMPlugin try: pydm.data_plugins.establish_connection(self) except Exception: logger.exception("Unable to make proper connection " "for %r", self)
[docs] def disconnect(self, destroying=False): """ Disconnect a PyDMChannel """ try: plugin = pydm.data_plugins.plugin_for_address(self.address) if not plugin: return plugin.remove_connection(self, destroying=destroying) except Exception: logger.exception("Unable to remove connection " "for %r", self)
def __eq__(self, other): if isinstance(self, other.__class__): address_matched = self.address == other.address connection_slot_matched = self.connection_slot == other.connection_slot value_slot_matched = self.value_slot == other.value_slot severity_slot_matched = self.severity_slot == other.severity_slot enum_strings_slot_matched = self.enum_strings_slot == other.enum_strings_slot unit_slot_matched = self.unit_slot == other.unit_slot prec_slot_matched = self.prec_slot == other.prec_slot upper_ctrl_slot_matched = self.upper_ctrl_limit_slot == other.upper_ctrl_limit_slot lower_ctrl_slot_matched = self.lower_ctrl_limit_slot == other.lower_ctrl_limit_slot upper_alarm_slot_matched = self.upper_alarm_limit_slot == other.upper_alarm_limit_slot lower_alarm_slot_matched = self.lower_alarm_limit_slot == other.lower_alarm_limit_slot upper_warning_slot_matched = self.upper_warning_limit_slot == other.upper_warning_limit_slot lower_warning_slot_matched = self.lower_warning_limit_slot == other.lower_warning_limit_slot write_access_slot_matched = self.write_access_slot == other.write_access_slot timestamp_slot_matched = self.timestamp_slot == other.timestamp_slot value_signal_matched = self.value_signal is None and other.value_signal is None if self.value_signal and other.value_signal: value_signal_matched = self.value_signal.signal == other.value_signal.signal return ( address_matched and connection_slot_matched and value_slot_matched and severity_slot_matched and enum_strings_slot_matched and unit_slot_matched and prec_slot_matched and upper_ctrl_slot_matched and lower_ctrl_slot_matched and upper_alarm_slot_matched and lower_alarm_slot_matched and upper_warning_slot_matched and lower_warning_slot_matched and write_access_slot_matched and value_signal_matched and timestamp_slot_matched ) return NotImplemented def __ne__(self, other): equality_result = self.__eq__(other) if equality_result is not NotImplemented: return not equality_result return NotImplemented def __hash__(self): return id(self) def __repr__(self): return "<PyDMChannel ({:})>".format(self.address)