#!/usr/bin/python

# -*- coding: utf-8

#######################################################################

# Copyright 2010 Signove Corporation - All rights reserved.

# Contact: Signove Corporation (contact@signove.com)

#

# This library is free software; you can redistribute it and/or modify

# it under the terms of version 2.1 of the GNU Lesser General Public

# License as published by the Free Software Foundation.

#

# This library is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Lesser General Public License for more details.

#

# You should have received a copy of the GNU Lesser General Public

# License along with this library; if not, write to the

# Free Software Foundation, Inc., 59 Temple Place, Suite 330,

# Boston, MA 02111-1307  USA

#

# If you have questions regarding the use of this file, please contact

# Signove at contact@signove.com.

#######################################################################

import os.path

import sys

import time

import termios

import socket

import glib

import gobject

import dbus

import dbus.mainloop.glib

import dbus.exceptions

import dbus.service

from fcntl import ioctl

from ctypes import c_long

from hdp.dummy_ieee10404 import *

# PTS suite specs

# FIXME sometimes destroy_channel causes a 'Bailing out test' and not

# destroying the application. This MAY be the cause of some funny

# problems. Also, failing upon channel destruction is more or less

# expected in some tests, because PTS default config is to take initiative

# in destroying channels. I suggest you create a 'destroy_channel_whatever'

# and use it in tests where PTS gives Pass veredict but it is spitting an

# error at IUT side. (No point in yelling errors where PTS finds none!)

# FIXME what is the point of asking create_app every time? I think that

# it should happen as soon as we enter the test. Destroy_app may be asked

# about, because it effectively ceases communication.

"""

Notes (HDP.SRC.CON)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SRC_CON_BV_01_I

	- TSPC_HDP_4_1 (Source: Supports advertisement of HDP Service Record)

- TC_SRC_CON_BV_07_I

	- TSPC_HDP_2_7 (Source: Supports General Inquiry (C.5))

- TC_SRC_CON_BV_09_I

	- TSPC_HDP_2_9 (Source: Supports Initiation of Bonding (O))

- TC_SRC_CON_BV_10_I

	- TSPC_HDP_2_11 (Source: Supports use of Health Class of Device (O))

"""

hdp_src_con = { 'TC_SRC_CON_BV_01_I': ( 'create_app_source',

										'destroy_app'),

				'TC_SRC_CON_BV_02_I': ( 'set_discoverable', ),

				'TC_SRC_CON_BV_03_I': (	'create_app_source',

										'destroy_app'),

				'TC_SRC_CON_BV_04_I': ( 'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_CON_BV_05_I': (	'create_app_source',

										'destroy_app'),

				'TC_SRC_CON_BV_06_I': (	'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_CON_BV_07_I': ( 'discover', ),

				'TC_SRC_CON_BV_08_I': ( 'delete_pairing',

										'create_app_source',

										'destroy_app'),

				'TC_SRC_CON_BV_09_I': ( 'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_CON_BV_10_I': ( 'set_discoverable', ) }

"""

Notes (HDP.SRC.CC)

==================

PICS flags for enabling tests (set to True)

-------------------------------------------

- Tests BV_01_C and BV_03_C:

	- TSPC_HDP_6_2 (Source: Supports Inititiate creation of Control and Data Channel)

- Tests BV_02_C, BV_05_C and BI_12_C:

	- TSPC_HDP_6_3 (Source: SUpports Accept creation of Control and Data Channel)

- Tests BV_07_C and BV_09_C:

	- TSPC_HDP_3_7 (Source: Supports Streaming Data Channel (M,C.4)[Mandatory])

"""

hdp_src_cc = { 	'TC_SRC_CC_BV_01_C': (	'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

			 	'TC_SRC_CC_BV_02_C': (  'create_app_source',

										'destroy_app'),

				'TC_SRC_CC_BV_03_C': (  'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_CC_BV_05_C': (	'create_app_source',

										'destroy_app'),

				'TC_SRC_CC_BV_07_C': (	'create_app_source',

										'create_channel_reliable',

										'create_channel_streaming',

										'destroy_channel',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_CC_BV_09_C': (	'create_app_source_streaming',

										'destroy_app'),

				'TC_SRC_CC_BI_12_C': ( 	'create_app_source',

										'destroy_app') }

"""

Notes (HDP.SRC.HCT)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- Tests BV_03_I and BV_05_I:

	- TSPC_HDP_6_4 (Source: Supports Inititiate Reconnection of MDL (O)

- Tests BV_04_I, BV_06_C and BV_07_C:

	- TSPC_HDP_6_5 (Source: Supports Accept Reconnection of MDL (C.4)[Mandatory])

"""

hdp_src_hct = { 'TC_SRC_HCT_BV_01_I': ( 'create_app_source',

										'create_channel_reliable',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_HCT_BV_02_I': ( 'create_app_source',

										'destroy_app'),

				'TC_SRC_HCT_BV_03_I': ( 'create_app_source_reliable',

										'create_channel_reliable',

										'schedule_disconnection',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_HCT_BV_04_I': (	'create_app_source',

										'destroy_app'),

				'TC_SRC_HCT_BV_05_C': ( 'create_app_source_reliable',

										'create_channel_reliable',

										'schedule_disconnection',

										'destroy_channel',

										'destroy_app'),

				'TC_SRC_HCT_BV_06_C': (	'create_app_source',

									  	'destroy_app'),

				'TC_SRC_HCT_BV_07_C': (	'create_app_source_reliable',

										'destroy_app')

			}

"""

Notes (HDP.SRC.DE)

==================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SRC_DE_BV_01_I:

	- TSPC_HDP_7_1 (Source: Supports Initiation of Echo Test (O))

"""

hdp_src_de = {	'TC_SRC_DE_BV_01_I': (	'create_app_source',

										'echo', # no need to create a channel

										'destroy_app'),

				'TC_SRC_DE_BV_02_I': (	'create_app_source',

										'destroy_app') }

"""

Notes (HDP.SRC.DEP)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SRC_DEP_BV_01_I and TC_SRC_DEP_BV_02_I:

	- TSPC_HDP_7_4 (Source: Supports IEEE 11073-20601 Agent Role (C.1))

Broken tests

------------

- All tests require an "associate request", which ATM is unknown.

"""

hdp_src_dep = {'TC_SRC_DEP_BV_01_I': ( 'create_app_source',

					'acquire_channel_src',

					'send_assoc',

					'send_config',

					'send_sample',

					'send_dissoc',

					'destroy_app'),

		'TC_SRC_DEP_BV_02_I': ( 'create_app_source',

					'acquire_channel_src',

					'send_assoc',

					'send_config',

					'send_sample',

					'send_dissoc',

					'send_assoc',

					'send_sample',

					'send_dissoc',

					'destroy_app')

		}

"""

Notes (HDP.SNK.CON)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SNK_CON_BV_09_I

	- TSPC_HDP_8_9 (Sink: Supports Initiation of Bonding (O))

- TC_SNK_CON_BV_10_I

	- TSPC_HDP_8_11 (Sink: Supports use of Health Class of Device (O))

"""

hdp_snk_con = { 'TC_SNK_CON_BV_01_I': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_CON_BV_02_I': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_CON_BV_03_I': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_CON_BV_04_I': ( 'delete_pairing',

										'pair',

										'create_app_sink',

										'create_channel_any',

										'destroy_channel',

										'destroy_app'),

				'TC_SNK_CON_BV_05_I': ('create_app_sink', 'destroy_app'),

				'TC_SNK_CON_BV_06_I': ( 'create_app_sink',

										'create_channel_any'),

				'TC_SNK_CON_BV_07_I': ( 'discover', ),

				'TC_SNK_CON_BV_08_I': ( 'delete_pairing',

										'create_app_sink',

										'destroy_app'),

				'TC_SNK_CON_BV_09_I': ( 'create_app_sink',

										'create_channel_any',

										'destroy_channel',

										'destroy_app'),

				'TC_SNK_CON_BV_10_I': ( 'set_discoverable', ) }

hdp_snk_cc = { 	'TC_SNK_CC_BV_01_C': ( 	'create_app_sink',

										'create_channel_any',

										'destroy_channel',

										'destroy_app' ),

				'TC_SNK_CC_BV_02_C': (	'create_app_sink',

										'destroy_app'),

				'TC_SNK_CC_BV_04_C': (	'create_app_sink',

										'create_channel_any',

										'destroy_app'),

				'TC_SNK_CC_BV_06_C': (	'create_app_sink',

										'destroy_app'),

				'TC_SNK_CC_BV_08_C': (	'create_app_sink',

										'create_channel_any',

										'create_channel_any',

										'destroy_app'),

				'TC_SNK_CC_BV_10_C': (	'create_app_sink',

										'destroy_app'),

				'TC_SNK_CC_BI_11_C': (	'create_app_sink',

										'destroy_app')}

"""

Notes (HDP.SNK.HCT)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SNK_HCT_BV_03_I and TC_SNK_HCT_BV_05_C

	- TSPC_HDP_12_4 (Sink: SUpports Initiate Reconnection of MDL (O))

"""

hdp_snk_hct = { 'TC_SNK_HCT_BV_01_I': (	'create_app_sink',

										'create_channel_any',

										'destroy_channel',

										'destroy_app'),

				'TC_SNK_HCT_BV_02_I': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_HCT_BV_03_I': ( 'create_app_sink',

										'create_channel_any',

										'schedule_disconnection',

										'destroy_channel',

										'destroy_app'),

				'TC_SNK_HCT_BV_04_I': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_HCT_BV_05_C': ( 'create_app_sink',

										'create_channel_any',

										'schedule_disconnection',

										'destroy_channel',

										'destroy_app'),

				'TC_SNK_HCT_BV_06_C': ( 'create_app_sink',

										'destroy_app'),

				'TC_SNK_HCT_BV_07_C': ( 'create_app_sink',

										'destroy_app'),

			}

"""

Notes (HDP.SNK.DE)

==================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SNK_DE_BV_01_I

	- TSPC_HDP_13_1 (Sink: Supports Initiation of Echo Test (O))

"""

hdp_snk_de = { 	'TC_SNK_DE_BV_01_I': (	'create_app_sink',

										'echo',

										'destroy_app'),

				'TC_SNK_DE_BV_02_I': (	'create_app_sink',

										'destroy_app') }

"""

Notes (HDP.SNK.DEP)

===================

PICS flags for enabling tests (set to True)

-------------------------------------------

- TC_SNK_DEP_BV_03_I and TC_SNK_DEP_BV_04_I:

	- TSPC_HDP_13_5 (Sink: Supports IEEE 11073-20601 Manager Role (C.1))

Broken tests

------------

- Missing IEEE stack to be released

"""

hdp_snk_dep = {	'TC_SNK_DEP_BV_03_I': ('create_app_sink',

					'set_unknown_config',

					'create_channel_any',

					'acquire_channel',

					'destroy_app'),

		'TC_SNK_DEP_BV_04_I': ('create_app_sink',

					'set_unknown_config',

					'create_channel_any',

					'acquire_channel',

					'destroy_app'),

		}

hdp_snk_suites = {'SNK_CON': hdp_snk_con,

				  'SNK_CC': hdp_snk_cc,

				  'SNK_HCT': hdp_snk_hct,

				  'SNK_DE': hdp_snk_de,

				  'SNK_DEP': hdp_snk_dep}

hdp_src_suites = {'SRC_CON': hdp_src_con,

				  'SRC_CC': hdp_src_cc,

				  'SRC_HCT': hdp_src_hct,

				  'SRC_DE': hdp_src_de,

				  'SRC_DEP': hdp_src_dep}

groups = {'SRC': hdp_src_suites,

		  'SNK': hdp_snk_suites}

# Dictates if we report to the other side that IEEE configuration is unknown

unknown_config = False

def set_unknown_config():

	global unknown_config

	unknown_config = True

def is_unknown_config(data):

	global unknown_config

	res = unknown_config

	unknown_config = False # next time, it is not unknown anymore

	return res

watch_bitmap = glib.IO_IN | glib.IO_ERR | glib.IO_HUP | glib.IO_NVAL

def data_received_src(sk, evt):

	return data_received(sk, evt, mode=1)

def data_received(sk, evt, mode=0):

	data = None

	if evt & glib.IO_IN:

		try:

			data = sk.recv(1024)

		except IOError:

			data = ""

		if data:

			if not mode:

				print "Data received (sink mode)"

				response = parse_message_str(data, is_unknown_config(data))

				if response:

					sk.send(response)

					print "Response sent"

			else:

				print "Data received (src mode)"

	more = (evt == glib.IO_IN and data)

	if not more:

		print "EOF"

		try:

			sk.shutdown(2)

		except IOError:

			pass

		print "Closing"

		sk.close()

	return more

class Rejected(dbus.DBusException):

	_dbus_error_name = "org.bluez.Error.Rejected"

class PairAgent(dbus.service.Object):

	def __init__(self, bus, path, obs):

		dbus.service.Object.__init__(self, bus, path)

		self.obs = obs

	@dbus.service.method("org.bluez.Agent",

					in_signature="", out_signature="")

	def Release(self):

		print 'Release'

	@dbus.service.method("org.bluez.Agent",

					in_signature="os", out_signature="")

	def Authorize(self, device, uuid):

		print "Authorize (%s, %s)" % (device, uuid)

		authorize = raw_input("Authorize connection (Yes/no): ")

		if (authorize.lower() == "yes"):

			return

		raise Rejected("Connection rejected by user")

	@dbus.service.method("org.bluez.Agent",

					in_signature="o", out_signature="s")

	def RequestPinCode(self, device):

		print "RequestPinCode (%s)" % (device)

		pin = raw_input("Enter PIN Code: ")

		return pin

	@dbus.service.method("org.bluez.Agent",

					in_signature="o", out_signature="u")

	def RequestPasskey(self, device):

		print "RequestPasskey (%s)" % (device)

		passkey = raw_input("Enter passkey: ")

		return dbus.UInt32(passkey)

	@dbus.service.method("org.bluez.Agent",

					in_signature="ou", out_signature="")

	def DisplayPasskey(self, device, passkey):

		print "DisplayPasskey (%s, %d)" % (device, passkey)

	@dbus.service.method("org.bluez.Agent",

					in_signature="ou", out_signature="")

	def RequestConfirmation(self, device, passkey):

		print "RequestConfirmation (%s, %d)" % (device, passkey)

		confirm = raw_input("Confirm passkey (Yes/no): ")

		if (confirm.lower() == "yes"):

			return

		raise Rejected("Passkey doesn't match")

	@dbus.service.method("org.bluez.Agent",

					in_signature="s", out_signature="")

	def ConfirmModeChange(self, mode):

		print "ConfirmModeChange (%s)" % (mode)

		authorize = raw_input("Authorize mode change (Yes/no): ")

		if (authorize.lower() == "yes"):

			return

		raise Rejected("Mode change by user")

	@dbus.service.method("org.bluez.Agent",

					in_signature="", out_signature="")

	def Cancel(self):

		print "Cancel"

class HDPAutoPTS:

	""" Runs an interactive prompt for automatically test BlueZ against PTS.

	"""

	def __init__(self, bdaddr, tests, mainloop):

		glib.io_add_watch(sys.stdin, glib.IO_IN | glib.IO_PRI, self.tick)

		self.mainloop = mainloop

		self.device_bdaddr = bdaddr

		self.data_type = 4103

		# Test context

		self.app = None

		self.device = None

		self.channels = []

		self.sk = None

		# Test suite context

		self.tests = tests

		self.test = None

		self.cmd = None

		self.cmd_idx = 0

		self._device_found = False

		self.discovering = False

		self.bus = dbus.SystemBus()

		self.manager = dbus.Interface(self.bus.get_object("org.bluez", "/"),

				"org.bluez.Manager")

		self.hdp_manager = dbus.Interface(self.bus.get_object("org.bluez",

				"/org/bluez"), "org.bluez.HealthManager")

		self.agent = PairAgent(self.bus, '/test/agent', self)

		self.select_adapter()

	def select_adapter(self):

		""" Prompts user to select the adapter used for tests.

		"""

		sel = self.manager.DefaultAdapter()

		adapters = self.manager.ListAdapters()

		if not adapters:

			print "No adapters, test can not be run"

			sys.exit(1)

		print "Select an adapter [%s]:" % sel 

		for ad in adapters:

			print "\t%d. %s" % (adapters.index(ad) + 1, ad)

		while True:

			if len(adapters) <= 1:

				sel = adapters[0]

				print 'Auto-selected adapter %s.' % sel

				break

			try:

				print 'Select: ',

				cmd = sys.stdin.readline()

				if cmd == '\n':

					print

					break

				pos = int(cmd) - 1

				if pos < 0:

					raise TypeError

				sel = adapters[pos]

				print

				print 'Selected adapter', sel

				break

			except (TypeError, IndexError, ValueError):

				print "Wrong selection, choose a device from the list."

			except KeyboardInterrupt:

				sys.exit()

		self.adapter = dbus.Interface(self.bus.get_object("org.bluez", sel),

				"org.bluez.Adapter")

		self.bus.add_signal_receiver(self.device_found,

				dbus_interface="org.bluez.Adapter",

				signal_name="DeviceFound")

		self.bus.add_signal_receiver(self.property_changed,

				dbus_interface="org.bluez.Adapter",

				signal_name="PropertyChanged")

		self.bus.add_signal_receiver(self.channel_connected,

			signal_name="ChannelConnected",

			bus_name="org.bluez",

			path_keyword="device",

			interface_keyword="interface",

			dbus_interface="org.bluez.HealthDevice")

	def set_discoverable(self):

		self.adapter.SetProperty('Discoverable', dbus.Boolean(1))

		print 'Adapter set to General Discoverable mode...'

		cls = int(self.adapter.GetProperties()['Class'])

		if cls != 0x4a0914:

			print 'Make sure that device class is Health (0x4a0914).' \

			      'Configure with \'sudo hciconfig <IFACE> class 0x4a0914\'' \

			      'Check with \'sudo hciconfig -a\''

			print "0x%x" % cls

	def select_device(self):

		devices = self.adapter.ListDevices()

		select = None

		if len(devices) == 0:

			print "No devices available"

			sys.exit()

		for dev in devices:

			odev = dbus.Interface(self.bus.get_object("org.bluez", dev),

							"org.bluez.Device")

			addr = odev.GetProperties()["Address"].lower()

			if self.device_bdaddr == addr:

				select = dev

				break

		else:

			print "PTS device not found"

			sys.exit()

		print

		print "Connecting to", select

		self.device = dbus.Interface(self.bus.get_object("org.bluez", select),

						"org.bluez.HealthDevice")

	def echo(self):

		print 'Echoing...'

		self.select_device()

		ans = self.device.Echo()

		print 'Press OK on PTS side now. (Echo answer was %s)' % ans

	def pair(self):

		print 'Pairing..'

		self.pairing = True

		self.adapter.CreatePairedDevice(self.device_bdaddr, '/test/agent',

			'DisplayYesNo', reply_handler=self.pair_reply, error_handler=self.pair_error)

		slept = 0

		while self.pairing:

			time.sleep(1)

			slept += 1

			if slept >= 10:

				break

		self.pairing = False

	def delete_pairing(self):

		print 'Deleting pairing with %s' % self.device_bdaddr

		try:

			device = self.adapter.FindDevice(self.device_bdaddr)

			if device:

				self.adapter.RemoveDevice(device)

		except:

			pass

		print 'Pairing deleted. Input the PIN number whenever requested.'

	def device_found(self, addr, props):

		""" Receives DeviceFound signals, used for discovery.

		"""

		if not self.discovering or self._device_found:

			return

		if addr.lower() == self.device_bdaddr:

			print 'Device found! Confirm on PTS that it was found to continue.'

			self.discovering = False

			self._device_found = True

		else:

			print 'Found device', addr, 'ignoring...'

	def discover(self):

		print 'Discovering device...'

		self.discovering = True

		self.adapter.StartDiscovery()

		# Block cmd input until finished discovering or timeout (10s)

		slept = 0

		while self.discovering:

			time.sleep(1)

			slept += 1

			if slept >= 15:

				break

		print 'Discovering timed out. Next test will be ready to execute.'

	def property_changed(self, name, value):

		if not self.discovering:

			return

		if name == 'Discovering' and not value:

			self.discovering = False

			if self._device_found:

				print 'Finished discovery, device was found. Confirm on PTS if'\

					  ' you haven\'t already.'

	def pair_reply(self, device):

		print 'Finished pairing with', device

		self.pairing = False

	def pair_error(self, error):

		print 'Failed to pair,', error

		self.pairing = False

	def create_app(self, data_type, role, chanpref=''):

		if not self.hdp_manager:

			return False

		conf = {'DataType': dbus.types.UInt16(data_type),

				'Role': role}

		if chanpref in ['Any', 'Reliable', 'Streaming']:

			conf['ChannelType'] = chanpref

		self.app = self.hdp_manager.CreateApplication(conf)

		return self.app

	def create_app_source_streaming(self):

		return self.create_app(self.data_type, 'Source', 'Streaming')

	def create_app_source_reliable(self):

		return self.create_app(self.data_type, 'Source', 'Reliable')

	def create_app_sink(self):

		return self.create_app(self.data_type, 'Sink')

	def create_app_source(self):

		return self.create_app(self.data_type, 'Source')

	def destroy_app(self):

		print 'Destroying app'

		self.hdp_manager.DestroyApplication(self.app)

		self.app = None

	def toggle_dc(self, channel):

		channel = dbus.Interface(self.bus.get_object("org.bluez", channel),

						"org.bluez.HealthChannel")

		channel.Release()

		glib.timeout_add(3000, self.toggle_dc_2, channel)

		return False

	def toggle_dc_2(self, channel):

		retries = 3

		while retries > 0:

			try:

				fd = channel.Acquire()

			except:

				print 'Acquire() failed, retrying'

			retries -= 1

		glib.timeout_add(3000, self.toggle_dc_3, channel, fd)

		return False

	def toggle_dc_3(self, channel, fd):

		os.close(fd.take())

		try:

			channel.Release()

		except:

			print 'Release failed (probably channel is gone)'

	def schedule_disconnection(self):

		glib.timeout_add(3000, self.toggle_dc, self.channels[-1])

	def send_data(self):

		if len(self.channels) == 0:

			return

		channel = self.channels[-1]

		chan = dbus.Interface(self.bus.get_object("org.bluez", channel),

						"org.bluez.HealthChannel")

		fd = chan.Acquire()

		fd = fd.take()

		fd.write('Hello PTS')

	def set_unknown_config(self):

		set_unknown_config()

	def send_assoc(self):

		if not self.sk:

			print "No socket to send data"

			return

		self.sk.send(make_assoc_str())

	def send_config(self):

		self.sk.send(make_config_str())

	def send_sample(self):

		self.sk.send(make_sample_str())

	def send_dissoc(self):

		self.sk.send(make_release_req_str())

	def create_channel(self, type):

		self.select_device()

		try:

			channel = self.device.CreateChannel(self.app, type)

			self.channels.append(channel)

		except dbus.exceptions.DBusException, e:

			print 'Failed to create channel. You might have selected the' \

				' wrong device or the test already passed, check PTS.', e

	def channel_connected(self, channel, interface, device):

		print "Channel connected signal"

		if channel not in self.channels:

			self.device = device

			self.channels.append(channel)

			print "	...appended"

	def destroy_channel(self, channel=None):

		if not self.device or not self.channels:

			return

		if not channel:

			if len(self.channels) == 0:

				return

			channel = self.channels[-1]

		self.device.DestroyChannel(channel)

		self.channels.remove(channel)

	def acquire_channel_src(self, channel=None):

		self.acquire_channel(channel, 1)

	def acquire_channel(self, channel=None, mode=0):

		if not self.device or not self.channels:

			print "No channel to acquire"

			return

		if not channel:

			if len(self.channels) == 0:

				return

			channel = self.channels[-1]

		chan = dbus.Interface(self.bus.get_object("org.bluez", channel),

						"org.bluez.HealthChannel")

		fd = chan.Acquire()

		fd = fd.take()

		# encapsulate numericfd in Python socket object

		self.sk = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)

		# fromfd() does dup() so we need to close the original

		os.close(fd)

		print "FD acquired"

		if mode:

			cb = data_received_src

		else:

			cb = data_received

		glib.io_add_watch(self.sk, watch_bitmap, cb)

	def release_channel(self, channel=None):

		self.sk = None

		if not self.device or not self.channels:

			return

		if not channel:

			if len(self.channels) == 0:

				return

			channel = self.channels[-1]

		chan = dbus.Interface(self.bus.get_object("org.bluez", channel),

						"org.bluez.HealthChannel")

		chan.Release()

	def create_channel_any(self):

		self.create_channel("Any")

	def create_channel_reliable(self):

		self.create_channel('Reliable')

	def create_channel_streaming(self):

		self.create_channel('Streaming')

	def tick(self, src, cond):

		available = c_long()

		ioctl(src, termios.FIONREAD, available)

		ch = src.read(available.value)

		if ch.lower().startswith('no'):

			self.skip()

		else:

			self.run()

		return True

	def next_cmd(self):

		cmds = self.test[1]

		if self.cmd_idx + 1 == len(cmds):

			print 'Test finished'

			self.cmd_idx = 0

			self.next_test()

		else:

			self.cmd_idx += 1

			self.cmd = cmds[self.cmd_idx]

			print 'Execute command %s ? (Yes/no) [Yes]' % self.cmd

	def next_test(self):

		if not self.test:

			self.mainloop.quit()

			return

		test_idx = self.tests.index(self.test)

		if test_idx + 1 == len(self.tests):

			print 'Finished all tests listed on file.'

			opt = raw_input('Do you want to start over? (Yes/no) [Yes]\n')

			if opt.lower().startswith('no'):

				self.mainloop.quit()

			else:

				self.test = self.tests[0]

				print 'Enter test %s (Yes/no) [Yes]' % self.test[0]

		else:

			self.test = self.tests[test_idx + 1]

			print 'Enter test %s ? (Yes/no) [Yes]' % self.test[0]

		self.cmd = None

	def run(self):

		if not self.test:

			self.test = self.tests[0]

			print 'Enter test %s ? (Yes/no) [Yes]' % self.test[0]

			return

		if not self.cmd:

			self.cmd = self.test[1][0]

			self.cmd_idx = 0

			print 'Execute command %s ? (Yes/no) [Yes]' % self.cmd

			return

		try:

			getattr(self, self.cmd)()

			self.next_cmd()

		except Exception, e:

			print 'Failed to call command', e, self.cmd

			self.skip()

	def skip(self):

		self.cmd_idx = 0

		self.cmd = None

		print 'Exiting test %s' % self.test[0]

		self.next_test()

def get_tests(fname):

	tests = []

	names = [l.replace('\n', '') for l in open(fname, 'r').readlines() if not

			l.startswith('[')]

	raw_tests = {}

	for g in groups.values():

		for s in g.values():

			for test_name, test in s.items():

				if test_name in names:

					raw_tests[test_name] = test

	for name in names:

		if name in raw_tests:

			tests.append((name, raw_tests[name]))

	return tests

def usage():

	print 'Usage: %s --list\t(list tests)' % sys.argv[0]

	print '       %s <PTS dongle bdaddr> <infile>\t(run tests listed on infile)' % sys.argv[0]

	sys.exit(0)

def list_tests():

	for gname, g in groups.items():

		for sname, s in g.items():

			print

			print '[Namespace [%s.%s]]' % (gname, sname)

			tests = s.keys()

			def test_sorter(a, b):

				if 'bi' in b.lower():

					return -1

				elif 'bi' in a.lower():

					return 1

				return cmp(a, b)

			tests.sort(test_sorter)

			print '\n'.join(tests)

	sys.exit(0)

def check_bdaddr(addr):

	if addr.count(':') == 5 and \

		not [c for c in addr.replace(':', '') if c.lower() > 'f']:

		return addr

if __name__ == '__main__':

	if len(sys.argv) < 2:

		usage()

	elif len(sys.argv) == 2:

		{'--list': list_tests, \

		 '--help': usage,}.get(sys.argv[1], usage)()

	else:

		pass

	bdaddr = check_bdaddr(sys.argv[1].lower())

	infile = sys.argv[2]

	if not bdaddr:

		print 'Invalid broadcast address:', sys.argv[1]

		sys.exit(0)

	if not os.path.exists(infile):

		print 'Input file %s does not exist.' % infile

		sys.exit(0)

	tests = get_tests(infile)

	if not tests:

		print 'Could not match any tests on input with our list.'

		sys.exit(0)

	print 'Current test suite: [', ', '.join([t[0] for t in tests]), ']'

	dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

	mainloop = gobject.MainLoop()

	suite = HDPAutoPTS(bdaddr, tests, mainloop)

	try:

		print 'Enter test suite? (Yes/no) [Yes]'

		mainloop.run()

	except KeyboardInterrupt:

		print 'Bye'