Logo Search packages:      
Sourcecode: obex-data-server version File versions

ods-server-test.py

#!/usr/bin/python

import dbus
import dbus.decorators
import dbus.glib
import gobject
import sys
from signal import *
import time

# enable server on all adapters
bt_address = '00:00:00:00:00:00'
allow_write = True
auto_accept = True
accept = True

class ServerSession:
    
    def __init__(self, session_object_path):
        self.prefix = '>>'+session_object_path+'<< '
        self.bus = dbus.SessionBus()
        
        # get org.openobex.ServerSession object
        session_obj = self.bus.get_object('org.openobex', session_object_path)
        self.session = dbus.Interface(session_obj, 'org.openobex.ServerSession')
        
        # connect signals
        self.session.connect_to_signal('Disconnected', self.disconnected_cb)
        self.session.connect_to_signal('Cancelled', self.cancelled_cb)
        self.session.connect_to_signal('TransferStarted', self.transfer_started_cb)
        self.session.connect_to_signal('TransferProgress', self.transfer_progress_cb)
        self.session.connect_to_signal('TransferCompleted', self.transfer_completed_cb)
        self.session.connect_to_signal('ErrorOccurred', self.error_occurred_cb)
    
    # emitted when session is disconnected
    def disconnected_cb(self):
        print self.prefix,
        print 'Disconnected'
    
    # emitted when transfer is cancelled
    def cancelled_cb(self):
        print self.prefix,
        print 'Transfer cancelled'

    # emitted when transfer begins
    def transfer_started_cb(self, filename, local_path, total_bytes):
        print self.prefix,
        print 'Transfer started (%s, %s, %d)' % (filename, local_path, total_bytes)
        self.total_bytes = total_bytes
        if auto_accept == False:
            print self.prefix, 'Sleeping for 3 secs'
            time.sleep(3)
            print self.prefix,
            if accept == True:
                print 'Accepting'
                self.session.Accept()
            else:
                print 'Rejecting'
                self.session.Reject()   
    
    # emitted constantly during transfer
    def transfer_progress_cb(self, bytes_transferred):
        print self.prefix,
        if self.total_bytes != -1:
            print 'Progress: %d %%' % int(float(bytes_transferred)/self.total_bytes*100)
        else:
            print 'Progress'
    
    # emitted when transfer is completed
    def transfer_completed_cb(self):
        print self.prefix,
        print 'Transfer completed'
    
    # emitted when error occurs (for instance link error)
    def error_occurred_cb(self, error_name, error_message):
        print self.prefix,
        print 'Error occurred: %s: %s' % (error_name, error_message)


class Tester:
    
    def __init__(self, server_type, root_path):
        self.server_type = server_type
        self.root_path = root_path
        
        # get Session bus
        self.bus = dbus.SessionBus()
        
        # get org.openobex.Manager object
        manager_obj = self.bus.get_object('org.openobex', '/org/openobex')
        self.manager = dbus.Interface(manager_obj, 'org.openobex.Manager')
        
        # call CreateBluetoothServer with specified server type 
        # (opp - Object Push Profile, ftp - File Transfer Profile)
        # returns Server object path
        require_pairing = (self.server_type == 'ftp')
        server_path = self.manager.CreateBluetoothServer(bt_address, self.server_type, require_pairing)
        print 'Server object: ', server_path
        # get org.openobex.Server object
        server_obj = self.bus.get_object('org.openobex', server_path)
        self.server = dbus.Interface(server_obj, 'org.openobex.Server')

        # connect signals
        self.server.connect_to_signal('Started', self.started_cb)
        self.server.connect_to_signal('Stopped', self.stopped_cb)
        self.server.connect_to_signal('Closed', self.closed_cb)
        self.server.connect_to_signal('ErrorOccurred', self.error_occurred_cb)
        self.server.connect_to_signal('SessionCreated', self.session_created_cb)
        self.server.connect_to_signal('SessionRemoved', self.session_removed_cb)
        
        # start server with specified options
        self.server.Start(self.root_path, allow_write, auto_accept)
        
        self.main_loop = gobject.MainLoop()
        self.main_loop.run()
    
    # emitted when Server is started
    def started_cb(self):
        print 'Started'
    
    # emitted when Server is stopped
    def stopped_cb(self):
        print 'Stopped'
    
    # emitted when Server is closed
    def closed_cb(self):
        print 'Closed'
    
    def error_occurred_cb(self, error_name, error_message):
        print 'Error occurred: %s: %s' % (error_name, error_message)
    
    # emitted when client connects to server (server session is established)    
    def session_created_cb(self, session_object_path):
        print 'Session created: %s' % session_object_path
      session_info = self.server.GetServerSessionInfo(session_object_path)
      print 'Session Bluetooth address: %s' % session_info['BluetoothAddress']
        session = ServerSession(session_object_path)
    
    # emitted when client disconnects
    def session_removed_cb(self, session_object_path):
        print 'Session removed: %s' % session_object_path

if __name__ == '__main__':
    gobject.threads_init()
    dbus.glib.init_threads()

    tester = Tester(sys.argv[1], sys.argv[2])
    

Generated by  Doxygen 1.6.0   Back to index