/usr/lib/python2.7/dist-packages/framework/subsystems/ophoned/ophoned.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 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 | # -*- coding: UTF-8 -*-
"""
The Open Phone Daemon - Python Implementation
(C) 2008 Guillaume "Charlie" Chereau
(C) 2008 Openmoko, Inc.
GPLv2 or later
Package: ophoned
Implementation of the dbus objects.
"""
MODULE_NAME = "ophoned"
__version__ = "0.1.0.1"
from framework import helpers
from framework.controller import Controller
from framework.patterns import dbuscache
from protocol import Protocol, ProtocolUnusable
from test import TestProtocol
from gsm import GSMProtocol
from headset import HeadsetManager
import dbus
import dbus.service
from dbus import DBusException
import gobject
import logging
logger = logging.getLogger( MODULE_NAME )
class Phone(dbus.service.Object):
"""The Phone object is used to create Call objects using different protocols"""
def __init__(self, bus):
# Those two attributes are needed by the framwork
self.bus = bus
self.path = '/org/freesmartphone/Phone'
self.interface = "org.freesmartphone.Phone"
super(Phone, self).__init__(bus, self.path)
self.protocols = {}
self.active_call = None
self.gsm = bus.get_object(
'org.freesmartphone.ousaged', '/org/freesmartphone/Usage',
introspect = False,
follow_name_owner_changes = True
)
# FIXME
# FxH disbled because no clients are there yet to remove to call objects (see remark in gsm.py)
# Moreover, the removal of the GSM protocol when suspending causes exceptions after resume (stale references?)
# self.gsm.connect_to_signal( 'ResourceChanged', self.on_resource_changed )
self.headset = HeadsetManager( self.bus, self.on_bt_answer_requested, self.on_bt_connection_status )
gobject.idle_add( self.on_startup )
def on_startup( self ):
opreferencesd = Controller.object( "/org/freesmartphone/Preferences" )
self.preferences = opreferencesd.GetService( "phone" )
address = self.preferences.GetValue( "bt-headset-address" )
self.headset.setAddress( address )
self.bus.add_signal_receiver(
self.on_preferences_changed, 'Notify', 'org.freesmartphone.Preferences.Service',
'org.freesmartphone.opreferencesd', '/org/freesmartphone/Preferences/phone'
)
def on_preferences_changed (self, key, value ):
if key == "bt-headset-address":
self.headset.setAddress( value )
def on_resource_changed( self, resourcename, state, attributes ):
if resourcename == "GSM":
if state and not "GSM" in self.protocols:
self.protocols["GSM"] = GSMProtocol( self )
elif not state and "GSM" in self.protocols:
self.protocols["GSM"].fini()
del self.protocols["GSM"]
def on_bt_answer_requested( self ):
logger.info( "BT-Headset: AnswerRequested (active call = %s)", self.active_call )
if self.active_call:
if self.active_call.GetStatus() in ['incoming']:
self.Accept( dbus_ok = helpers.drop_dbus_result, dbus_error = helpers.log_dbus_error )
else:
self.Hangup( dbus_ok = helpers.drop_dbus_result, dbus_error = helpers.log_dbus_error )
def on_bt_connection_status( self, connected ):
logger.info("BT-Headset: ConnectionStatus = %s", connected)
self.BTHeadsetConnected( connected )
@dbus.service.method('org.freesmartphone.Phone', in_signature='ssb', out_signature='o')
def CreateCall(self, number, protocol = None, force = True):
""" Return a new Call targeting the given number, with an optional protocol.
If the protocol is not provided, the service will determine the best protocol to use.
if force is set to true, then we kill the channel if it is already opened
parameters:
number -- A string representing the number of the peer
protocol -- The name of the protocol as returned by InitProtocols, if None the best protocol will be used. Default to None
force -- If true, we destroy any already present call object to this number. Default to True
"""
if self.protocols is None:
self.InitProtocols()
number = str(number)
# first we guess the best protocol to use
if protocol:
protocol = self.protocols[str(protocol)]
else:
# Here we need to guess the best protocol for the number
protocol = self.protocols["Test"]
# Then we just ask the protocol class
ret = protocol.CreateCall(number, force = force)
return ret
@dbus.service.method('org.freesmartphone.Phone', in_signature='b', out_signature='')
def SetBTHeadsetPlaying( self, playing ):
self.headset.setPlaying( playing )
@dbus.service.signal('org.freesmartphone.Phone', signature='b')
def BTHeadsetConnected(self, connected):
pass
# FIXME handle multiple calls correctly
@dbus.service.method(
'org.freesmartphone.Phone', in_signature='', out_signature='',
async_callbacks=("dbus_ok","dbus_error")
)
def Accept(self, dbus_ok, dbus_error):
logger.info( "Accept (active call = %s)", self.active_call )
if self.active_call:
self.active_call.Activate( dbus_ok = dbus_ok, dbus_error = dbus_error )
@dbus.service.method(
'org.freesmartphone.Phone', in_signature='', out_signature='',
async_callbacks=("dbus_ok","dbus_error")
)
def Hangup(self, dbus_ok, dbus_error):
logger.info( "Hangup (active call = %s)", self.active_call )
if self.active_call:
self.active_call.Release( dbus_ok = dbus_ok, dbus_error = dbus_error )
@dbus.service.signal('org.freesmartphone.Phone', signature='o')
def CallCreated(self, call):
"""Emitted when a new call has been created"""
logger.info( "CallCreated (%s)", call )
self.active_call = call
@dbus.service.signal('org.freesmartphone.Phone', signature='o')
def CallReleased(self, call):
"""Emitted when a call has been released"""
logger.info( "CallReleased (%s)", call )
if self.active_call == call:
self.active_call = None
def factory(prefix, controller):
"""This is the magic function that will be called bye the framework module manager"""
try: # We use a try because the module manager ignores the exceptions in the factory
phone = Phone(controller.bus)
return [phone]
except Exception, e:
# XXX: remove that
logger.error("%s", e) # Just so that if an exception is raised, we can at least see the error message
raise
def generate_doc():
"""This function can be used to generate a wiki style documentation for the DBus API
It should be replaced by doxygen
"""
from protocol import Call
objects = [Phone, Call]
services = {}
for obj in objects:
for attr_name in dir(obj):
attr = getattr(obj, attr_name)
if hasattr(attr, '_dbus_interface'):
if hasattr(attr, '_dbus_is_method'):
func = {}
func['name'] = attr_name
func['args'] = ','.join(attr._dbus_args)
func['in_sig'] = attr._dbus_in_signature
func['out_sig'] = attr._dbus_out_signature
func['doc'] = attr.__doc__
funcs, sigs = services.setdefault(attr._dbus_interface, [[],[]])
funcs.append(func)
if hasattr(attr, '_dbus_is_signal'):
sig = {}
sig['name'] = attr_name
sig['args'] = ','.join(attr._dbus_args)
sig['sig'] = attr._dbus_signature
sig['doc'] = attr.__doc__
funcs, sigs = services.setdefault(attr._dbus_interface, [[],[]])
sigs.append(sig)
for name, funcs in services.items():
print '= %s =' % name
for func in funcs[0]:
print """
== method %(name)s(%(args)s) ==
* in: %(in_sig)s
* out: %(out_sig)s
* %(doc)s""" % func
for sig in funcs[1]:
print """
== signal %(name)s(%(args)s) ==
* out: %(sig)s
* %(doc)s""" % sig
print
if __name__ == '__main__':
import sys
generate_doc()
sys.exit(0)
import gobject
import dbus.mainloop.glib
dbus.mainloop.glib.DBusGMainLoop( set_as_default=True )
mainloop = gobject.MainLoop()
bus = dbus.SystemBus()
name = dbus.service.BusName("org.freesmartphone.ophoned", bus)
phone = Phone(bus)
mainloop.run()
|