/usr/lib/python2.7/dist-packages/framework/subsystems/oeventsd/trigger.py is in fso-frameworkd 0.10.1-3.
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 | # -*- coding: UTF-8 -*-
"""
The freesmartphone Events Module - Python Implementation
(C) 2008 Guillaume 'Charlie' Chereau
(C) 2008 Openmoko, Inc.
GPLv2 or later
Package: oeventsd
Module: trigger
"""
__version__ = "0.2.0"
MODULE_NAME = "oeventsd.trigger"
from parser import AutoFunction
import dbus
import logging
logger = logging.getLogger( MODULE_NAME )
#============================================================================#
class Trigger(AutoFunction):
#============================================================================#
"""
A trigger is the initial event that can activate a rule.
When a trigger is activated, it call the rule `trigger` method,
giving a set of keywork arguments (the signal attributes to the method)
Then the rule can decide to start or not its actions.
A trigger can also optionaly have a an `untrigger` method. This method will
call the `untrigger` method of the connected rules.
"""
def __init__(self):
"""
Create a new trigger
The trigger need to be initialized with the `init` method before
it can trigger the connected rules
"""
self.__listeners = [] # List of rules that are triggered by this trigger
def connect(self, action):
"""
Connect the trigger to an action
This method should only be called by the Rule class
"""
self.__listeners.append(action)
def _trigger(self, **kargs):
"""
Trigger all the connected rules
"""
logger.debug("trigger %s", self)
for action in self.__listeners:
action.trigger(**kargs)
def _untrigger(self, **kargs):
"""
Untrigger all the connected rules
"""
logger.debug("untrigger %s", self)
for action in self.__listeners:
action.untrigger(**kargs)
def enable(self):
"""
Enable the trigger
The trigger won't trigger the connect rules before this
method has been called
"""
pass
def disable(self):
"""
Disable the trigger
"""
pass
#============================================================================#
class DBusTrigger(Trigger):
#============================================================================#
"""
A special trigger that waits for a given DBus signal to trigger its rules
"""
function_name = 'DbusTrigger'
def __init__(self, bus, service, obj, interface, signal):
"""Create the DBus trigger
arguments:
- bus the DBus bus name (or a string : 'system' | 'session')
- service the DBus name of the service
- obj the DBus path of the object
- interface the Dbus interface of the signal
- signal the DBus name of the signal
"""
Trigger.__init__( self )
# some arguments checking
if isinstance(bus, str):
if bus == 'system':
bus = dbus.SystemBus()
elif bus == 'session':
bus = dbus.SessionBus()
else:
raise TypeError("Bad dbus bus : %s" % bus)
if not obj:
obj = None
assert isinstance(service, str), "service is not str"
assert obj is None or isinstance(obj, str), "obj is not str or None"
assert isinstance(interface, str), "interface is not str"
assert isinstance(signal, str), "signal is not str"
self.bus = bus
self.service = service
self.obj = obj
self.interface = interface
self.signal = signal
self.dbus_match = None
def __repr__(self):
return "DBusTrigger(%s %s.%s)" % (self.service, self.obj, self.signal)
def enable(self):
if self.dbus_match is not None: # if the rule is already enabled we do nothing
return
# Connect to the DBus signal
logger.debug("connect trigger to dbus signal %s %s", self.obj, self.signal)
self.dbus_match = self.bus.add_signal_receiver(
self.on_signal,
dbus_interface=self.interface,
signal_name=self.signal
)
def disable(self):
if self.dbus_match is None: # if the rule is already disabled we do nothing
return
self.dbus_match.remove()
self.dbus_match = None
def on_signal(self, *args):
kargs = dict( ('arg%d' % i, v) for i,v in enumerate(args) )
self._trigger( **kargs )
#============================================================================#
class TestTrigger( Trigger ):
#============================================================================#
"""
This trigger can be used ot debug the events system.
It is triggered when oeventsd.TriggerTest method is called
"""
function_name = "Test"
def __init__( self, name ):
Trigger.__init__( self )
self.name = name
def __repr__( self ):
return "Test(%s)" % self.name
|