/usr/lib/python2.7/dist-packages/sardana/sardanaevent.py is in python-sardana 1.2.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | #!/usr/bin/env python
##############################################################################
##
## This file is part of Sardana
##
## http://www.tango-controls.org/static/sardana/latest/doc/html/index.html
##
## Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
##
## Sardana 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.
##
## Sardana is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU Lesser General Public License for more details.
##
## You should have received a copy of the GNU Lesser General Public License
## along with Sardana. If not, see <http://www.gnu.org/licenses/>.
##
##############################################################################
"""This module is part of the Python Pool libray. It defines the base classes
for pool event mechanism"""
__all__ = ["EventGenerator", "EventReceiver", "EventType"]
__docformat__ = 'restructuredtext'
import weakref
import collections
from .sardanautils import is_callable
from taurus.core.util import CallableRef, BoundMethodWeakref
def _get_callable_ref(listener, callback=None):
"""Returns a callable reference for the given listener"""
meth = getattr(listener, 'event_received', None)
if meth is not None and is_callable(meth):
return weakref.ref(listener, callback)
else:
return CallableRef(listener, callback)
class EventGenerator(object):
"""A class capable of generating events to their listeners"""
def __init__(self, max_queue_len=10, listeners=None):
self._listeners = []
self._event_queue = collections.deque(maxlen=max_queue_len)
if listeners is not None:
if not isinstance(listeners, collections.Sequence):
listeners = listeners,
for listener in listeners:
self.add_listener(listener)
def _listener_died(self, weak_listener):
"""Callback executed when a listener dies"""
if self._listeners is None:
return
try:
self._listeners.remove(weak_listener)
except ValueError:
pass
def add_listener(self, listener):
"""Adds a new listener for this object.
:param listener: a listener"""
if self._listeners is None or listener is None:
return False
weak_listener = _get_callable_ref(listener, self._listener_died)
if weak_listener in self._listeners:
return False
self._listeners.append(weak_listener)
return True
def remove_listener(self, listener):
"""Removes an existing listener for this object.
:param listener: the listener to be removed
:return: True is succeeded or False otherwise"""
if self._listeners is None:
return
weak_listener = _get_callable_ref(listener)
try:
self._listeners.remove(weak_listener)
except ValueError:
return False
return True
def has_listeners(self):
"""Returns True if anybody is listening to events from this object
:return: True is at least one listener is listening or False otherwise
"""
if self._listeners is None:
return False
return len(self._listeners) > 0
def fire_event(self, event_type, event_value, listeners=None):
self.flush_queue()
self._fire_event(event_type, event_value, listeners=listeners)
def _fire_event(self, event_type, event_value, listeners=None):
"""Sends an event to all listeners or a specific one"""
if listeners is None:
listeners = self._listeners
if listeners is None:
return
if not isinstance(listeners, collections.Sequence):
listeners = listeners,
for listener in listeners:
if isinstance(listener, weakref.ref) or \
isinstance(listener, BoundMethodWeakref):
real_listener = listener()
else:
real_listener = listener
if real_listener is None:
continue
meth = getattr(real_listener, 'event_received', None)
if meth is not None and is_callable(meth):
real_listener.event_received(self, event_type, event_value)
elif is_callable(real_listener):
real_listener(self, event_type, event_value)
def queue_event(self, event_type, event_value, listeners=None):
queue = self._event_queue
queue.append((event_type, event_value, listeners))
def flush_queue(self):
queue = self._event_queue
n = len(queue)
while n>0:
self._fire_event(*queue.pop())
n = n - 1
class EventReceiver(object):
"""A simple class that implements useful features for a class which is
an event receiver. The actual class may inherit from this EventReceiver class
and may choose to use just a part of the API provided by this class, the
whole API or none of the API."""
def __init__(self):
self._events_blocked = False
def block_events(self):
self._events_blocked = True
def unblock_events(self):
self._events_blocked = False
def are_events_blocked(self):
return self._events_blocked
class EventType(object):
"""Definition of an event type"""
def __init__(self, name, priority=0):
self.name = name
self.priority = priority
def __str__(self):
return "EventType(name=%s, priority=%s)" % (self.name, self.priority)
def __repr__(self):
return "EventType(name=%s, priority=%s)" % (self.name, self.priority)
def get_name(self):
"""Returns this event name
:return: this event name
:rtype: str"""
return self.name
def get_priority(self):
"""Returns this event priority
:return: this event priority
:rtype: str"""
return self.priority
|