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
|
#!/usr/bin/python
import gobject
import sys
import os
import dbus
import dbus.service
import dbus.mainloop.glib
class Transfer:
def __init__(self, callback_func):
self.callback_func = callback_func
self.path = None
self.filename = None
class PbapClient:
def __init__(self, session_path):
self.transfers = 0
self.props = dict()
self.flush_func = None
bus = dbus.SessionBus()
obj = bus.get_object("org.bluez.obex.client", session_path)
self.session = dbus.Interface(obj, "org.bluez.obex.Session")
self.pbap = dbus.Interface(obj,
"org.bluez.obex.PhonebookAccess")
bus.add_signal_receiver(self.transfer_complete,
dbus_interface="org.bluez.obex.Transfer",
signal_name="Complete",
path_keyword="path")
bus.add_signal_receiver(self.transfer_error,
dbus_interface="org.bluez.obex.Transfer",
signal_name="Error",
path_keyword="path")
def register(self, reply, transfer):
(path, properties) = reply
transfer.path = path
transfer.filename = properties["Filename"]
self.props[path] = transfer
print "Transfer created: %s (file %s)" % (path,
transfer.filename)
def error(self, err):
print err
mainloop.quit()
def transfer_complete(self, path):
req = self.props.get(path)
if req == None:
return
self.transfers -= 1
print "Transfer %s finished" % path
f = open(req.filename, "r")
os.remove(req.filename)
lines = f.readlines()
del self.props[path]
req.callback_func(lines)
if (len(self.props) == 0) and (self.transfers == 0):
if self.flush_func != None:
f = self.flush_func
self.flush_func = None
f()
def transfer_error(self, code, message, path):
req = self.props.get(path)
if req == None:
return
print "Transfer finished with error %s: %s" % (code, message)
mainloop.quit()
def pull(self, vcard, params, func):
req = Transfer(func)
self.pbap.Pull(vcard, "", params,
reply_handler=lambda r: self.register(r, req),
error_handler=self.error)
self.transfers += 1
def pull_all(self, params, func):
req = Transfer(func)
self.pbap.PullAll("", params,
reply_handler=lambda r: self.register(r, req),
error_handler=self.error)
self.transfers += 1
def flush_transfers(self, func):
if (len(self.props) == 0) and (self.transfers == 0):
return
self.flush_func = func
def interface(self):
return self.pbap
if __name__ == '__main__':
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
mainloop = gobject.MainLoop()
client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"),
"org.bluez.obex.Client")
if (len(sys.argv) < 2):
print "Usage: %s <device>" % (sys.argv[0])
sys.exit(1)
print "Creating Session"
session_path = client.CreateSession(sys.argv[1], { "Target": "PBAP" })
pbap_client = PbapClient(session_path)
def process_result(lines, header):
if header != None:
print header
for line in lines:
print line,
print
def test_paths(paths):
if len(paths) == 0:
print
print "FINISHED"
mainloop.quit()
return
path = paths[0]
print "\n--- Select Phonebook %s ---\n" % (path)
pbap_client.interface().Select("int", path)
print "\n--- GetSize ---\n"
ret = pbap_client.interface().GetSize()
print "Size = %d\n" % (ret)
print "\n--- List vCard ---\n"
ret = pbap_client.interface().List(dbus.Dictionary())
params = dbus.Dictionary({ "Format" : "vcard30",
"Fields" : [ "VERSION", "FN", "TEL"] })
for item in ret:
print "%s : %s" % (item[0], item[1])
pbap_client.pull(item[0], params,
lambda x: process_result(x, None))
pbap_client.pull_all(params, lambda x: process_result(x,
"\n--- PullAll ---\n"))
pbap_client.flush_transfers(lambda: test_paths(paths[1:]))
test_paths(["PB", "ICH", "OCH", "MCH", "CCH"])
mainloop.run()
|
