git.maemo.org Git - theonering/blob - hand_tests/dbus_signals.py

   1 #!/usr/bin/env python
   2
   3 import gobject
   4 import dbus
   5 import dbus.mainloop.glib
   6 import telepathy
   7
   8
   9 DBUS_PROPERTIES = 'org.freedesktop.DBus.Properties'
  10
  11
  12 class AutoAcceptAttempt(object):
  13         # @todo Make this more composable by just checking for incoming call.  Why
  14         # incoming rather than just call?  Because it has more demands on what
  15         # properties to get which we can then get them in parallel.  The callback
  16         # would then chose to pickup based on the caller's number, wait to see if
  17         # the call is ignored/rejected, etc
  18
  19         def __init__(self, bus, conn, chan):
  20                 self._sessionBus = bus
  21                 self._conn = conn
  22                 self._chan = chan
  23
  24                 self._selfHandle = None
  25                 self._initiatorHandle = None
  26                 self._initiatorID = None
  27                 self._targetHandle = None
  28                 self._targetID = None
  29                 self._pendingHandles = None
  30
  31                 if False:
  32                         # @bug Unsure why this isn't working
  33                         self._conn[DBUS_PROPERTIES].Get(
  34                                 telepathy.interfaces.CONNECTION_INTERFACE,
  35                                 'SelfHandle',
  36                                 reply_handler = self._on_got_self_handle,
  37                                 error_handler = self._on_nothing,
  38                         )
  39                 else:
  40                         self._conn[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].GetSelfHandle(
  41                                 reply_handler = self._on_got_self_handle,
  42                                 error_handler = self._on_nothing,
  43                         )
  44
  45                 self._chan[DBUS_PROPERTIES].GetAll(
  46                         telepathy.interfaces.CHANNEL_INTERFACE,
  47                         reply_handler = self._on_got_all,
  48                         error_handler = self._on_nothing,
  49                 )
  50
  51                 if False:
  52                         # @bug Unsure why this isn't working
  53                         self._chan[DBUS_PROPERTIES].Get(
  54                                 telepathy.interfaces.CHANNEL_INTERFACE_GROUP,
  55                                 'LocalPendingMembers',
  56                                 reply_handler = self._on_got_pending_members,
  57                                 error_handler = self._on_nothing,
  58                         )
  59                 else:
  60                         self._chan[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].GetLocalPendingMembersWithInfo(
  61                                 reply_handler = self._on_got_pending_members,
  62                                 error_handler = self._on_nothing,
  63                         )
  64
  65         def _pickup_if_ready(self):
  66                 if None in (
  67                         self._selfHandle,
  68                         self._initiatorHandle,
  69                         self._initiatorID,
  70                         self._targetHandle,
  71                         self._targetID,
  72                         self._pendingHandles,
  73                 ):
  74                         # Note ready yet, still some outstanding requests
  75                         return
  76
  77                 if self._selfHandle != self._targetHandle:
  78                         # Turns out it was an inbound call
  79                         return
  80
  81                 # @bug does not distinguish between who the call is from for use for TOR auto-pickup
  82                 self._chan[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].AddMembers(
  83                         reply_handler = self._on_members_added,
  84                         error_handler = self._on_nothing,
  85                 )
  86
  87         def _on_got_self_handle(self, selfHandle):
  88                 self._pickup_if_ready()
  89
  90         def _on_got_all(self, properties):
  91                 self._initiatorID = properties["InitiatorID"]
  92                 self._initiatorHandle = properties["InitiatorHandle"]
  93                 self._targetID = properties["InitiatorID"]
  94                 self._targetHandle = properties["InitiatorHandle"]
  95
  96                 self._pickup_if_ready()
  97
  98         def _on_got_pending_members(self, pendings):
  99                 for pendingHandle, instigatorHandle, reason, message in pendings:
 100                         print pendingHandle, instigatorHandle, reason, message
 101
 102                 self._pendingHandles = [pendingWithInfo[0] for pendingWithInfo in pendings]
 103                 self._pickup_if_ready()
 104
 105         def _on_members_added(self):
 106                 print "Should be picked up now"
 107
 108         def _on_nothing(*args):
 109                 print "ERROR", args
 110
 111
 112 class AutoAcceptCall(object):
 113         # @todo Make this more composable by switchig it to just handle monitoring
 114         # for new channels.  Other the callback on a new channel will filter for
 115         # channel type.
 116
 117         def __init__(self):
 118                 self._sessionBus = dbus.SessionBus()
 119                 self._activeAttempts = []
 120
 121         def start(self):
 122                 self._sessionBus.add_signal_receiver(
 123                         self._on_new_channel,
 124                         "NewChannel",
 125                         "org.freedesktop.Telepathy.Connection",
 126                         None,
 127                         None
 128                 )
 129
 130         def _on_new_channel(self, channelObjectPath, channelType, handleType, handle, supressHandler):
 131                 if channelType != telepathy.interfaces.CHANNEL_TYPE_STREAMED_MEDIA:
 132                         return
 133
 134                 connObjectPath = channelObjectPath.rsplit("/", 1)[0][1:]
 135                 serviceName = connObjectPath.replace("/", ".")
 136                 conn = telepathy.client.Channel(serviceName, connObjectPath)
 137                 chan = telepathy.client.Channel(serviceName, channelObjectPath)
 138                 # @bug does not distinguish between preferred CMs
 139                 # @todo Need a way to be notified on error, ignored, or if picked up
 140                 attemptPickup = AutoAcceptAttempt(self._sessionBus, conn, chan)
 141                 self._activeAttempts.append(attemptPickup)
 142
 143         def _on_nothing(*args):
 144                 print "ERROR", args
 145
 146
 147 if __name__ == "__main__":
 148         l = dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 149         autoaccept = AutoAcceptCall()
 150
 151         gobject.threads_init()
 152         gobject.idle_add(autoaccept.start)
 153
 154         mainloop = gobject.MainLoop(is_running=True)
 155         mainloop.run()