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 AutoAcceptCall(object):
  13
  14         def __init__(self, bus, conn, chan):
  15                 self._sessionBus = bus
  16                 self._conn = conn
  17                 self._chan = chan
  18
  19                 self._selfHandle = None
  20                 self._initiatorHandle = None
  21                 self._initiatorID = None
  22                 self._targetHandle = None
  23                 self._targetID = None
  24                 self._pendingHandles = None
  25
  26                 if False:
  27                         # @bug Unsure why this isn't working
  28                         self._conn[DBUS_PROPERTIES].Get(
  29                                 telepathy.interfaces.CONNECTION_INTERFACE,
  30                                 'SelfHandle',
  31                                 reply_handler = self._on_got_self_handle,
  32                                 error_handler = self._on_nothing,
  33                         )
  34                 else:
  35                         self._conn[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].GetSelfHandle(
  36                                 reply_handler = self._on_got_self_handle,
  37                                 error_handler = self._on_nothing,
  38                         )
  39
  40                 self._chan[DBUS_PROPERTIES].GetAll(
  41                         telepathy.interfaces.CHANNEL_INTERFACE,
  42                         reply_handler = self._on_got_all,
  43                         error_handler = self._on_nothing,
  44                 )
  45
  46                 if False:
  47                         # @bug Unsure why this isn't working
  48                         self._chan[DBUS_PROPERTIES].Get(
  49                                 telepathy.interfaces.CHANNEL_INTERFACE_GROUP,
  50                                 'LocalPendingMembers',
  51                                 reply_handler = self._on_got_pending_members,
  52                                 error_handler = self._on_nothing,
  53                         )
  54                 else:
  55                         self._chan[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].GetLocalPendingMembersWithInfo(
  56                                 reply_handler = self._on_got_pending_members,
  57                                 error_handler = self._on_nothing,
  58                         )
  59
  60         def _pickup_if_ready(self):
  61                 if None in (
  62                         self._selfHandle,
  63                         self._initiatorHandle,
  64                         self._initiatorID,
  65                         self._targetHandle,
  66                         self._targetID,
  67                         self._pendingHandles,
  68                 ):
  69                         # Note ready yet, still some outstanding requests
  70                         return
  71
  72                 if self._selfHandle != self._targetHandle:
  73                         # Turns out it was an inbound call
  74                         return
  75
  76                 # @bug does not distinguish between who the call is from for use for TOR auto-pickup
  77                 self._chan[telepathy.interfaces.CHANNEL_INTERFACE_GROUP].AddMembers(
  78                         reply_handler = self._on_members_added,
  79                         error_handler = self._on_nothing,
  80                 )
  81
  82         def _on_got_self_handle(self, selfHandle):
  83                 self._pickup_if_ready()
  84
  85         def _on_got_all(self, properties):
  86                 self._initiatorID = properties["InitiatorID"]
  87                 self._initiatorHandle = properties["InitiatorHandle"]
  88                 self._targetID = properties["InitiatorID"]
  89                 self._targetHandle = properties["InitiatorHandle"]
  90
  91                 self._pickup_if_ready()
  92
  93         def _on_got_pending_members(self, pendings):
  94                 for pendingHandle, instigatorHandle, reason, message in pendings:
  95                         print pendingHandle, instigatorHandle, reason, message
  96
  97                 self._pendingHandles = [pendingWithInfo[0] for pendingWithInfo in pendings]
  98                 self._pickup_if_ready()
  99
 100         def _on_members_added(self):
 101                 print "Should be picked up now"
 102
 103         def _on_nothing(*args):
 104                 print "ERROR", args
 105
 106
 107 class NewChannelSignaller(object):
 108
 109         def __init__(self, on_new_channel):
 110                 self._sessionBus = dbus.SessionBus()
 111                 self._on_new_channel = on_new_channel
 112
 113         def start(self):
 114                 self._sessionBus.add_signal_receiver(
 115                         self._on_new_channel,
 116                         "NewChannel",
 117                         "org.freedesktop.Telepathy.Connection",
 118                         None,
 119                         None
 120                 )
 121
 122         def stop(self):
 123                 self._sessionBus.remove_signal_receiver(
 124                         self._on_new_channel,
 125                         "NewChannel",
 126                         "org.freedesktop.Telepathy.Connection",
 127                         None,
 128                         None
 129                 )
 130
 131         def _on_new_channel(
 132                 self, channelObjectPath, channelType, handleType, handle, supressHandler
 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                 self._on_new_channel(self._sessionBus, conn, chan, channelType)
 139
 140
 141 class Manager(object):
 142
 143         def __init__(self):
 144                 self._newChannelSignaller = NewChannelSignaller(self._on_new_channel)
 145                 self._activeAttempts = []
 146
 147         def _on_new_channel(self, bus, conn, chan, channelType):
 148                 if channelType != telepathy.interfaces.CHANNEL_TYPE_STREAMED_MEDIA:
 149                         return
 150
 151                 # @bug does not distinguish between preferred CMs
 152                 # @todo Need a way to be notified on error, ignored, or if picked up
 153                 attemptPickup = AutoAcceptCall(bus, conn, chan)
 154                 self._activeAttempts.append(attemptPickup)
 155
 156
 157 if __name__ == "__main__":
 158         l = dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 159         autoaccept = Manager()
 160
 161         gobject.threads_init()
 162         gobject.idle_add(autoaccept.start)
 163
 164         mainloop = gobject.MainLoop(is_running=True)
 165         mainloop.run()