initial import, no crypto

1 files changed, 210 insertions, 0 deletions

diff --git a/scripts/test-hfp-ag b/scripts/test-hfp-ag
new file mode 100755
index 0000000..3152339
--- /dev/null
+++ b/scripts/test-hfp-ag
@@ -0,0 +1,210 @@
+#!/usr/bin/python
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+from optparse import OptionParser, make_option
+import os
+from socket import SOCK_SEQPACKET, socket
+import sys
+import dbus
+import dbus.service
+import dbus.mainloop.glib
+import glib
+try:
+  from gi.repository import GObject
+except ImportError:
+  import gobject as GObject
+
+mainloop = None
+audio_supported = True
+
+try:
+	from socket import AF_BLUETOOTH, BTPROTO_SCO
+except:
+	print("WARNING: python compiled without Bluetooth support"
+					" - audio will not be available")
+	audio_supported = False
+
+BUF_SIZE = 1024
+
+BDADDR_ANY = '00:00:00:00:00:00'
+
+HF_NREC			= 0x0001
+HF_3WAY			= 0x0002
+HF_CLI			= 0x0004
+HF_VOICE_RECOGNITION	= 0x0008
+HF_REMOTE_VOL		= 0x0010
+HF_ENHANCED_STATUS	= 0x0020
+HF_ENHANCED_CONTROL	= 0x0040
+HF_CODEC_NEGOTIATION	= 0x0080
+
+AG_3WAY			= 0x0001
+AG_NREC			= 0x0002
+AG_VOICE_RECOGNITION	= 0x0004
+AG_INBAND_RING		= 0x0008
+AG_VOICE_TAG		= 0x0010
+AG_REJECT_CALL		= 0x0020
+AG_ENHANCED_STATUS	= 0x0040
+AG_ENHANCED_CONTROL	= 0x0080
+AG_EXTENDED_RESULT	= 0x0100
+AG_CODEC_NEGOTIATION	= 0x0200
+
+HF_FEATURES = (HF_3WAY | HF_CLI | HF_VOICE_RECOGNITION |
+			HF_REMOTE_VOL | HF_ENHANCED_STATUS |
+			HF_ENHANCED_CONTROL | HF_CODEC_NEGOTIATION)
+
+AVAIL_CODECS = "1,2"
+
+class HfpConnection:
+	slc_complete = False
+	fd = None
+	io_id = 0
+	version = 0
+	features = 0
+	pending = None
+
+	def disconnect(self):
+		if (self.fd >= 0):
+			os.close(self.fd)
+			self.fd = -1
+			glib.source_remove(self.io_id)
+			self.io_id = 0
+
+	def slc_completed(self):
+		print("SLC establisment complete")
+		self.slc_complete = True
+
+	def slc_next_cmd(self, cmd):
+		print("Unknown SLC command completed: %s" % (cmd))
+
+	def io_cb(self, fd, cond):
+		buf = os.read(fd, BUF_SIZE)
+		buf = buf.strip()
+
+		print("Received: %s" % (buf))
+
+		return True
+
+	def send_cmd(self, cmd):
+		if (self.pending):
+			print("ERROR: Another command is pending")
+			return
+
+		print("Sending: %s" % (cmd))
+
+		os.write(self.fd, cmd + "\r\n")
+		self.pending = cmd
+
+	def __init__(self, fd, version, features):
+		self.fd = fd
+		self.version = version
+		self.features = features
+
+		print("Version 0x%04x Features 0x%04x" % (version, features))
+
+		self.io_id = glib.io_add_watch(fd, glib.IO_IN, self.io_cb)
+
+class HfpProfile(dbus.service.Object):
+	sco_socket = None
+	io_id = 0
+	conns = {}
+
+	def sco_cb(self, sock, cond):
+		(sco, peer) = sock.accept()
+		print("New SCO connection from %s" % (peer))
+
+	def init_sco(self, sock):
+		self.sco_socket = sock
+		self.io_id = glib.io_add_watch(sock, glib.IO_IN, self.sco_cb)
+
+	def __init__(self, bus, path, sco):
+		dbus.service.Object.__init__(self, bus, path)
+
+		if sco:
+			self.init_sco(sco)
+
+	@dbus.service.method("org.bluez.Profile1",
+					in_signature="", out_signature="")
+	def Release(self):
+		print("Release")
+		mainloop.quit()
+
+	@dbus.service.method("org.bluez.Profile1",
+					in_signature="", out_signature="")
+	def Cancel(self):
+		print("Cancel")
+
+	@dbus.service.method("org.bluez.Profile1",
+				in_signature="o", out_signature="")
+	def RequestDisconnection(self, path):
+		conn = self.conns.pop(path)
+		conn.disconnect()
+
+	@dbus.service.method("org.bluez.Profile1",
+				in_signature="oha{sv}", out_signature="")
+	def NewConnection(self, path, fd, properties):
+		fd = fd.take()
+		version = 0x0105
+		features = 0
+		print("NewConnection(%s, %d)" % (path, fd))
+		for key in properties.keys():
+			if key == "Version":
+				version = properties[key]
+			elif key == "Features":
+				features = properties[key]
+
+		#conn = HfpConnection(fd, version, features)
+
+		#self.conns[path] = conn
+
+if __name__ == '__main__':
+	dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+	bus = dbus.SystemBus()
+
+	manager = dbus.Interface(bus.get_object("org.bluez",
+				"/org/bluez"), "org.bluez.ProfileManager1")
+
+	option_list = [
+			make_option("-p", "--path", action="store",
+					type="string", dest="path",
+					default="/bluez/test/hfp"),
+			make_option("-n", "--name", action="store",
+					type="string", dest="name",
+					default=None),
+			make_option("-C", "--channel", action="store",
+					type="int", dest="channel",
+					default=None),
+			]
+
+	parser = OptionParser(option_list=option_list)
+
+	(options, args) = parser.parse_args()
+
+	mainloop = GObject.MainLoop()
+
+	opts = {
+			"Version" : dbus.UInt16(0x0106),
+			"Features" : dbus.UInt16(HF_FEATURES),
+		}
+
+	if (options.name):
+		opts["Name"] = options.name
+
+	if (options.channel is not None):
+		opts["Channel"] = dbus.UInt16(options.channel)
+
+	if audio_supported:
+		sco = socket(AF_BLUETOOTH, SOCK_SEQPACKET, BTPROTO_SCO)
+		sco.bind(BDADDR_ANY)
+		sco.listen(1)
+	else:
+		sco = None
+
+	profile = HfpProfile(bus, options.path, sco)
+
+	manager.RegisterProfile(options.path, "hfp-ag", opts)
+
+	print("Profile registered - waiting for connections")
+
+	mainloop.run()