Re: [PATCH BlueZ v3] test: Add unified test for mesh node example app

["Gix, Brian" <brian.gix@intel.com>]

Applied

On Sat, 2019-03-09 at 23:53 -0800, Inga Stotland wrote:
> This adds one script, test-mesh, to replace three test-join,
> example-onoff-server and example-onoff-client.
> This is menu driven test that allows provisioning (join) and/or
> connecting existing (attach) nodes.
> ---
>  Makefile.tools            |   2 +-
>  test/agent.py             |  14 +-
>  test/example-onoff-client | 288 -------------
>  test/example-onoff-server | 365 -----------------
>  test/test-mesh            | 842 ++++++++++++++++++++++++++++++++++++++
>  5 files changed, 854 insertions(+), 657 deletions(-)
>  delete mode 100644 test/example-onoff-client
>  delete mode 100644 test/example-onoff-server
>  create mode 100755 test/test-mesh
> 
> diff --git a/Makefile.tools b/Makefile.tools
> index 0f94bbbe7..379e127b6 100644
> --- a/Makefile.tools
> +++ b/Makefile.tools
> @@ -463,7 +463,7 @@ test_scripts += test/sap_client.py test/bluezutils.py \
>  		test/test-hfp test/opp-client test/ftp-client \
>  		test/pbap-client test/map-client test/example-advertisement \
>  		test/example-gatt-server test/example-gatt-client \
> -		test/test-gatt-profile
> +		test/test-gatt-profile test/test-mesh test/agent.py
>  
>  if BTPCLIENT
>  noinst_PROGRAMS += tools/btpclient
> diff --git a/test/agent.py b/test/agent.py
> index 22c92f952..778dbe092 100755
> --- a/test/agent.py
> +++ b/test/agent.py
> @@ -1,9 +1,16 @@
> -#!/usr/bin/python
> +#!/usr/bin/python3
>  
>  import sys
>  import dbus
>  import dbus.service
> -import dbus.mainloop.glib
> +
> +try:
> +  from termcolor import colored, cprint
> +  set_green = lambda x: colored(x, 'green', attrs=['bold'])
> +  set_cyan = lambda x: colored(x, 'cyan', attrs=['bold'])
> +except ImportError:
> +  set_green = lambda x: x
> +  set_cyan = lambda x: x
>  
>  AGENT_IFACE = 'org.bluez.mesh.ProvisionAgent1'
>  AGENT_PATH = "/mesh/test/agent"
> @@ -37,4 +44,5 @@ class Agent(dbus.service.Object):
>  
>  	@dbus.service.method(AGENT_IFACE, in_signature="su", out_signature="")
>  	def DisplayNumeric(self, type, value):
> -		print("DisplayNumeric type=", type, " number=", value)
> +		print(set_cyan('DisplayNumeric ('), type,
> +				set_cyan(') number ='), set_green(value))
> diff --git a/test/example-onoff-client b/test/example-onoff-client
> deleted file mode 100644
> index e4a87eb12..000000000
> --- a/test/example-onoff-client
> +++ /dev/null
> @@ -1,288 +0,0 @@
> -#!/usr/bin/env python3
> -
> -import sys
> -import struct
> -import numpy
> -import dbus
> -import dbus.service
> -import dbus.exceptions
> -
> -try:
> -  from gi.repository import GObject
> -except ImportError:
> -  import gobject as GObject
> -from dbus.mainloop.glib import DBusGMainLoop
> -
> -MESH_SERVICE_NAME = 'org.bluez.mesh'
> -DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> -DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> -
> -MESH_NETWORK_IFACE = 'org.bluez.mesh.Network1'
> -MESH_NODE_IFACE = 'org.bluez.mesh.Node1'
> -MESH_ELEMENT_IFACE = 'org.bluez.mesh.Element1'
> -
> -VENDOR_ID_NONE = 0xffff
> -
> -app = None
> -bus = None
> -mainloop = None
> -node = None
> -token = numpy.uint64(0x76bd4f2372477600)
> -
> -def unwrap(item):
> -	if isinstance(item, dbus.Boolean):
> -		return bool(item)
> -	if isinstance(item, (dbus.UInt16, dbus.Int16, dbus.UInt32, dbus.Int32,
> -						dbus.UInt64, dbus.Int64)):
> -		return int(item)
> -	if isinstance(item, dbus.Byte):
> -		return bytes([int(item)])
> -	if isinstance(item, dbus.String):
> -			return item
> -	if isinstance(item, (dbus.Array, list, tuple)):
> -		return [unwrap(x) for x in item]
> -	if isinstance(item, (dbus.Dictionary, dict)):
> -		return dict([(unwrap(x), unwrap(y)) for x, y in item.items()])
> -
> -	print('Dictionary item not handled')
> -	print(type(item))
> -	return item
> -
> -def attach_app_cb(node_path, dict_array):
> -	print('Mesh application registered ', node_path)
> -	print(type(node_path))
> -	print(type(dict_array))
> -	print(dict_array)
> -
> -	els = unwrap(dict_array)
> -	print("Get Elements")
> -	for el in els:
> -		print(el)
> -		idx = struct.unpack('b', el[0])[0]
> -		print('Configuration for Element ', end='')
> -		print(idx)
> -		models = el[1]
> -
> -		element = app.get_element(idx)
> -		element.set_model_config(models)
> -
> -	obj = bus.get_object(MESH_SERVICE_NAME, node_path)
> -	global node
> -	node = dbus.Interface(obj, MESH_NODE_IFACE)
> -
> -def error_cb(error):
> -	print('D-Bus call failed: ' + str(error))
> -
> -def generic_reply_cb():
> -	print('D-Bus call done')
> -
> -def interfaces_removed_cb(object_path, interfaces):
> -	if not mesh_net:
> -		return
> -
> -	if object_path == mesh_net[2]:
> -		print('Service was removed')
> -		mainloop.quit()
> -
> -class Application(dbus.service.Object):
> -
> -	def __init__(self, bus):
> -		self.path = '/example'
> -		self.elements = []
> -		dbus.service.Object.__init__(self, bus, self.path)
> -
> -	def get_path(self):
> -		return dbus.ObjectPath(self.path)
> -
> -	def add_element(self, element):
> -		self.elements.append(element)
> -
> -	def get_element(self, idx):
> -		for ele in self.elements:
> -			if ele.get_index() == idx:
> -				return ele
> -
> -	@dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
> -	def GetManagedObjects(self):
> -		response = {}
> -		print('GetManagedObjects')
> -		for element in self.elements:
> -			response[element.get_path()] = element.get_properties()
> -		return response
> -
> -class Element(dbus.service.Object):
> -	PATH_BASE = '/example/ele'
> -
> -	def __init__(self, bus, index):
> -		self.path = self.PATH_BASE + format(index, '02x')
> -		print(self.path)
> -		self.models = []
> -		self.bus = bus
> -		self.index = index
> -		dbus.service.Object.__init__(self, bus, self.path)
> -
> -	def _get_sig_models(self):
> -		ids = []
> -		for model in self.models:
> -			id = model.get_id()
> -			vendor = model.get_vendor()
> -			if vendor == VENDOR_ID_NONE:
> -				ids.append(id)
> -		return ids
> -
> -	def get_properties(self):
> -		return {
> -				MESH_ELEMENT_IFACE: {
> -				'Index': dbus.Byte(self.index),
> -				'Models': dbus.Array(
> -					self._get_sig_models(), signature='q')
> -				}
> -		}
> -
> -	def add_model(self, model):
> -		model.set_path(self.path)
> -		self.models.append(model)
> -
> -	def get_index(self):
> -		return self.index
> -
> -	def set_model_config(self, config):
> -		print('Set element models config')
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="qqbay", out_signature="")
> -	def MessageReceived(self, source, key, is_sub, data):
> -		print('Message Received on Element ', end='')
> -		print(self.index)
> -		for model in self.models:
> -			model.process_message(source, key, data)
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="qa{sv}", out_signature="")
> -
> -	def UpdateModelConfiguration(self, model_id, config):
> -		print('UpdateModelConfig ', end='')
> -		print(hex(model_id))
> -		for model in self.models:
> -			if model_id == model.get_id():
> -				model.set_config(config)
> -				return
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="", out_signature="")
> -	def get_path(self):
> -		return dbus.ObjectPath(self.path)
> -
> -class Model():
> -	def __init__(self, model_id):
> -		self.cmd_ops = []
> -		self.model_id = model_id
> -		self.vendor = VENDOR_ID_NONE
> -		self.path = None
> -
> -	def set_path(self, path):
> -		self.path = path
> -
> -	def get_id(self):
> -		return self.model_id
> -
> -	def get_vendor(self):
> -		return self.vendor
> -
> -	def process_message(self, source, key, data):
> -		print('Model process message')
> -
> -	def set_publication(self, period):
> -		self.period = period
> -
> -	def set_bindings(self, bindings):
> -		self.bindings = bindings
> -
> -	def set_config(self, config):
> -		if 'Bindings' in config:
> -			self.bindings = config.get('Bindings')
> -			print('Bindings: ', end='')
> -			print(self.bindings)
> -		if 'PublicationPeriod' in config:
> -			self.set_publication(config.get('PublicationPeriod'))
> -			print('Model publication period ', end='')
> -			print(self.pub_period, end='')
> -			print(' ms')
> -
> -class OnOffClient(Model):
> -	def __init__(self, model_id):
> -		Model.__init__(self, model_id)
> -		self.cmd_ops = { 0x8201, # get
> -						 0x8202, # set
> -						 0x8203 } # set unacknowledged
> -		print('OnOff Client')
> -
> -	def _reply_cb(state):
> -		print('State ', end='');
> -		print(state)
> -
> -	def _send_message(self, dest, key, data, reply_cb):
> -		print('OnOffClient send data')
> -		node.Send(self.path, dest, key, data, reply_handler=reply_cb,
> -				  error_handler=error_cb)
> -
> -	def get_state(self, dest, key):
> -		opcode = 0x8201
> -		data = struct.pack('<H', opcode)
> -		self._send_message(dest, key, data, self._reply_cb)
> -
> -	def set_state(self, dest, key, state):
> -		opcode = 0x8202
> -		data = struct.pack('<HB', opcode, state)
> -		self._send_message(dest, key, data, self._reply_cb)
> -
> -	def process_message(self, source, key, data):
> -		print('OnOffClient process message len ', end = '')
> -		datalen = len(data)
> -		print(datalen)
> -
> -		if datalen!=3:
> -			return
> -
> -		opcode, state=struct.unpack('<HB',bytes(data))
> -		if opcode != 0x8202 :
> -			print('Bad opcode ', end='')
> -			print(hex(opcode))
> -			return
> -
> -		print('Got state ', end = '')
> -		print(hex(state))
> -
> -def attach_app_error_cb(error):
> -	print('Failed to register application: ' + str(error))
> -	mainloop.quit()
> -
> -def main():
> -
> -	DBusGMainLoop(set_as_default=True)
> -
> -	global bus
> -	bus = dbus.SystemBus()
> -	global mainloop
> -	global app
> -
> -	mesh_net = dbus.Interface(bus.get_object(MESH_SERVICE_NAME,
> -							"/org/bluez/mesh"),
> -							MESH_NETWORK_IFACE)
> -	mesh_net.connect_to_signal('InterfacesRemoved', interfaces_removed_cb)
> -
> -	app = Application(bus)
> -	first_ele = Element(bus, 0x00)
> -	first_ele.add_model(OnOffClient(0x1001))
> -	app.add_element(first_ele)
> -
> -	mainloop = GObject.MainLoop()
> -
> -	print('Attach')
> -	mesh_net.Attach(app.get_path(), token,
> -					reply_handler=attach_app_cb,
> -					error_handler=attach_app_error_cb)
> -	mainloop.run()
> -
> -if __name__ == '__main__':
> -	main()
> diff --git a/test/example-onoff-server b/test/example-onoff-server
> deleted file mode 100644
> index 131b6415c..000000000
> --- a/test/example-onoff-server
> +++ /dev/null
> @@ -1,365 +0,0 @@
> -#!/usr/bin/env python3
> -
> -import sys
> -import struct
> -import numpy
> -import dbus
> -import dbus.service
> -import dbus.exceptions
> -
> -from threading import Timer
> -import time
> -
> -
> -try:
> -  from gi.repository import GObject
> -except ImportError:
> -  import gobject as GObject
> -from dbus.mainloop.glib import DBusGMainLoop
> -
> -MESH_SERVICE_NAME = 'org.bluez.mesh'
> -DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> -DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> -
> -MESH_NETWORK_IFACE = 'org.bluez.mesh.Network1'
> -MESH_NODE_IFACE = 'org.bluez.mesh.Node1'
> -MESH_APPLICATION_IFACE = 'org.bluez.mesh.Application1'
> -MESH_ELEMENT_IFACE = 'org.bluez.mesh.Element1'
> -
> -APP_COMPANY_ID = 0x05f1
> -APP_PRODUCT_ID = 0x0001
> -APP_VERSION_ID = 0x0001
> -
> -VENDOR_ID_NONE = 0xffff
> -
> -app = None
> -bus = None
> -mainloop = None
> -node = None
> -
> -token = numpy.uint64(0x76bd4f2372476578)
> -
> -def generic_error_cb(error):
> -	print('D-Bus call failed: ' + str(error))
> -
> -def generic_reply_cb():
> -	print('D-Bus call done')
> -
> -def unwrap(item):
> -	if isinstance(item, dbus.Boolean):
> -		return bool(item)
> -	if isinstance(item, (dbus.UInt16, dbus.Int16, dbus.UInt32, dbus.Int32,
> -						dbus.UInt64, dbus.Int64)):
> -		return int(item)
> -	if isinstance(item, dbus.Byte):
> -		return bytes([int(item)])
> -	if isinstance(item, dbus.String):
> -			return item
> -	if isinstance(item, (dbus.Array, list, tuple)):
> -		return [unwrap(x) for x in item]
> -	if isinstance(item, (dbus.Dictionary, dict)):
> -		return dict([(unwrap(x), unwrap(y)) for x, y in item.items()])
> -
> -	print('Dictionary item not handled')
> -	print(type(item))
> -	return item
> -
> -def attach_app_cb(node_path, dict_array):
> -	print('Mesh application registered ', node_path)
> -
> -	obj = bus.get_object(MESH_SERVICE_NAME, node_path)
> -
> -	global node
> -	node = dbus.Interface(obj, MESH_NODE_IFACE)
> -
> -	els = unwrap(dict_array)
> -	print("Get Elements")
> -
> -	for el in els:
> -		idx = struct.unpack('b', el[0])[0]
> -		print('Configuration for Element ', end='')
> -		print(idx)
> -
> -		models = el[1]
> -		element = app.get_element(idx)
> -		element.set_model_config(models)
> -
> -def interfaces_removed_cb(object_path, interfaces):
> -	if not mesh_net:
> -		return
> -
> -	if object_path == mesh_net[2]:
> -		print('Service was removed')
> -		mainloop.quit()
> -
> -def send_response(path, dest, key, data):
> -		print('send response ', end='')
> -		print(data)
> -		node.Send(path, dest, key, data, reply_handler=generic_reply_cb,
> -						error_handler=generic_error_cb)
> -
> -def send_publication(path, model_id, data):
> -		print('send publication ', end='')
> -		print(data)
> -		node.Publish(path, model_id, data,
> -						reply_handler=generic_reply_cb,
> -						error_handler=generic_error_cb)
> -
> -class PubTimer():
> -	def __init__(self):
> -		self.seconds = None
> -		self.func = None
> -		self.thread = None
> -		self.busy = False
> -
> -	def _timeout_cb(self):
> -		self.func()
> -		self.busy = True
> -		self._schedule_timer()
> -		self.busy =False
> -
> -	def _schedule_timer(self):
> -		self.thread = Timer(self.seconds, self._timeout_cb)
> -		self.thread.start()
> -
> -	def start(self, seconds, func):
> -		self.func = func
> -		self.seconds = seconds
> -		if not self.busy:
> -			self._schedule_timer()
> -
> -	def cancel(self):
> -		print('Cancel timer')
> -		if self.thread is not None:
> -			print('Cancel thread')
> -			self.thread.cancel()
> -			self.thread = None
> -
> -class Application(dbus.service.Object):
> -
> -	def __init__(self, bus):
> -		self.path = '/example'
> -		self.elements = []
> -		dbus.service.Object.__init__(self, bus, self.path)
> -
> -	def get_path(self):
> -		return dbus.ObjectPath(self.path)
> -
> -	def add_element(self, element):
> -		self.elements.append(element)
> -
> -	def get_element(self, idx):
> -		for ele in self.elements:
> -			if ele.get_index() == idx:
> -				return ele
> -
> -	def get_properties(self):
> -		return {
> -			MESH_APPLICATION_IFACE: {
> -				'CompanyID': dbus.UInt16(APP_COMPANY_ID),
> -				'ProductID': dbus.UInt16(APP_PRODUCT_ID),
> -				'VersionID': dbus.UInt16(APP_VERSION_ID)
> -			}
> -		}
> -
> -	@dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
> -	def GetManagedObjects(self):
> -		response = {}
> -		print('GetManagedObjects')
> -		response[self.path] = self.get_properties()
> -		for element in self.elements:
> -			response[element.get_path()] = element.get_properties()
> -		return response
> -
> -class Element(dbus.service.Object):
> -	PATH_BASE = '/example/ele'
> -
> -	def __init__(self, bus, index):
> -		self.path = self.PATH_BASE + format(index, '02x')
> -		print(self.path)
> -		self.models = []
> -		self.bus = bus
> -		self.index = index
> -		dbus.service.Object.__init__(self, bus, self.path)
> -
> -	def _get_sig_models(self):
> -		ids = []
> -		for model in self.models:
> -			id = model.get_id()
> -			vendor = model.get_vendor()
> -			if vendor == VENDOR_ID_NONE:
> -				ids.append(id)
> -		return ids
> -
> -	def get_properties(self):
> -		return {
> -			MESH_ELEMENT_IFACE: {
> -				'Index': dbus.Byte(self.index),
> -				'Models': dbus.Array(
> -					self._get_sig_models(), signature='q')
> -			}
> -		}
> -
> -	def add_model(self, model):
> -		model.set_path(self.path)
> -		self.models.append(model)
> -
> -	def get_index(self):
> -		return self.index
> -
> -	def set_model_config(self, configs):
> -		print('Set element models config')
> -		for config in configs:
> -			mod_id = config[0]
> -			self.UpdateModelConfiguration(mod_id, config[1])
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="qqbay", out_signature="")
> -	def MessageReceived(self, source, key, is_sub, data):
> -		print('Message Received on Element ', end='')
> -		print(self.index)
> -		for model in self.models:
> -			model.process_message(source, key, data)
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="qa{sv}", out_signature="")
> -
> -	def UpdateModelConfiguration(self, model_id, config):
> -		print('UpdateModelConfig ', end='')
> -		print(hex(model_id))
> -		for model in self.models:
> -			if model_id == model.get_id():
> -				model.set_config(config)
> -				return
> -
> -	@dbus.service.method(MESH_ELEMENT_IFACE,
> -					in_signature="", out_signature="")
> -
> -	def get_path(self):
> -		return dbus.ObjectPath(self.path)
> -
> -class Model():
> -	def __init__(self, model_id):
> -		self.cmd_ops = []
> -		self.model_id = model_id
> -		self.vendor = VENDOR_ID_NONE
> -		self.bindings = []
> -		self.pub_period = 0
> -		self.pub_id = 0
> -		self.path = None
> -
> -	def set_path(self, path):
> -		self.path = path
> -
> -	def get_id(self):
> -		return self.model_id
> -
> -	def get_vendor(self):
> -		return self.vendor
> -
> -	def process_message(self, source, key, data):
> -		print('Model process message')
> -
> -	def set_publication(self, period):
> -		self.pub_period = period
> -
> -	def set_config(self, config):
> -		if 'Bindings' in config:
> -			self.bindings = config.get('Bindings')
> -			print('Bindings: ', end='')
> -			print(self.bindings)
> -		if 'PublicationPeriod' in config:
> -			self.set_publication(config.get('PublicationPeriod'))
> -			print('Model publication period ', end='')
> -			print(self.pub_period, end='')
> -			print(' ms')
> -
> -class OnOffServer(Model):
> -	def __init__(self, model_id):
> -		Model.__init__(self, model_id)
> -		self.cmd_ops = { 0x8201, # get
> -						 0x8202, # set
> -						 0x8203 } # set unacknowledged
> -
> -		print("OnOff Server ", end="")
> -		self.state = 0
> -		print('State ', end='')
> -		self.timer = PubTimer()
> -
> -	def process_message(self, source, key, data):
> -		datalen = len(data)
> -		print('OnOff Server process message len ', datalen)
> -
> -		if datalen!=2 and datalen!=3:
> -			return
> -
> -		if datalen==2:
> -			op_tuple=struct.unpack('<H',bytes(data))
> -			opcode = op_tuple[0]
> -			if opcode != 0x8201:
> -				print(hex(opcode))
> -				return
> -			print('Get state')
> -		elif datalen==3:
> -			opcode,self.state=struct.unpack('<HB',bytes(data))
> -			if opcode != 0x8202 and opcode != 0x8203:
> -				print(hex(opcode))
> -				return
> -			print('Set state: ', end='')
> -			print(self.state)
> -
> -		rsp_data = struct.pack('<HB', 0x8204, self.state)
> -		send_response(self.path, source, key, rsp_data)
> -
> -	def publish(self):
> -		print('Publish')
> -		data = struct.pack('B', self.state)
> -		send_publication(self.path, self.model_id, data)
> -
> -	def set_publication(self, period):
> -		if period == 0:
> -			self.pub_period = 0
> -			self.timer.cancel()
> -			return
> -
> -		# We do not handle ms in this example
> -		if period < 1000:
> -			return
> -
> -		self.pub_period = period
> -		self.timer.start(period/1000, self.publish)
> -
> -def attach_app_error_cb(error):
> -	print('Failed to register application: ' + str(error))
> -	mainloop.quit()
> -
> -def main():
> -
> -	DBusGMainLoop(set_as_default=True)
> -
> -	global bus
> -	bus = dbus.SystemBus()
> -	global mainloop
> -	global app
> -
> -	mesh_net = dbus.Interface(bus.get_object(MESH_SERVICE_NAME,
> -						"/org/bluez/mesh"),
> -						MESH_NETWORK_IFACE)
> -	mesh_net.connect_to_signal('InterfacesRemoved', interfaces_removed_cb)
> -
> -	app = Application(bus)
> -	first_ele = Element(bus, 0x00)
> -	first_ele.add_model(OnOffServer(0x1000))
> -	app.add_element(first_ele)
> -
> -	mainloop = GObject.MainLoop()
> -
> -	print('Attach')
> -	mesh_net.Attach(app.get_path(), token,
> -					reply_handler=attach_app_cb,
> -					error_handler=attach_app_error_cb)
> -
> -	mainloop.run()
> -
> -if __name__ == '__main__':
> -	main()
> diff --git a/test/test-mesh b/test/test-mesh
> new file mode 100755
> index 000000000..fd02207bc
> --- /dev/null
> +++ b/test/test-mesh
> @@ -0,0 +1,842 @@
> +#!/usr/bin/env python3
> +
> +###################################################################
> +#
> +# This is a unified test sample for BT Mesh
> +#
> +# To run the test:
> +#     test-mesh [token]
> +#
> +#            'token' is an optional argument. It must be a 16-digit
> +#            hexadecimal number. The token must be associated with
> +#            an existing node. The token is generated and assigned
> +#            to a node as a result of successful provisioning (see
> +#            explanation of "join" option).
> +#            When the token is set, the menu operations "attach"
> +#            and "remove" may be performed on a node specified
> +#            by this token.
> +#
> +#      The test imitates a device with 2 elements:
> +#            element 0: OnOff Server model
> +#            element 1: OnOff Client model
> +#
> +# The main menu:
> +#       1 - set node ID (token)
> +#       2 - join mesh network
> +#       3 - attach mesh node
> +#       4 - remove node
> +#       5 - client menu
> +#       6 - exit
> +#
> +# The main menu options explained:
> +#     1 - set token
> +#            Set the unique node token.
> +#            The token can be set from command line arguments as
> +#            well.
> +#
> +#     2 - join
> +#            Request provisioning of a device to become a node
> +#            on a mesh network. The test generates device UUID
> +#            which is displayed and will need to be provided to
> +#            an outside entity that acts as a Provisioner. Also,
> +#            during the provisioning process, an agent that is
> +#            part of the test, will request (or will be requested)
> +#            to perform a specified operation, e.g., a number will
> +#            be displayed and this number will need to be  entered
> +#            on the Provisioner's side.
> +#            In case of successful provisioning, the application
> +#            automatically attaches as a node to the daemon. A node
> +#            'token' is returned to the application and is used
> +#            for the runtime of the test.
> +#
> +#     3 - attach
> +#            Attach the application to bluetoothd-daemon as a node.
> +#            For the call to be successful, the valid node token must
> +#            be already set, either from command arguments or by
> +#            executing "set token" operation or automatically after
> +#            successfully executing "join" operation in the same test
> +#            run.
> +#
> +#     4 - remove
> +#           Permanently removes any node configuration from daemon
> +#           and persistent storage. After this operation, the node
> +#           is permanently forgotten by the daemon and the associated
> +#           node token is no longer valid.
> +#
> +#     5 - client menu
> +#           Enter On/Off client submenu.
> +#
> +#     6 - exit
> +#
> +###################################################################
> +import sys
> +import struct
> +import fcntl
> +import os
> +import numpy
> +import random
> +import dbus
> +import dbus.service
> +import dbus.exceptions
> +
> +from threading import Timer
> +import time
> +
> +try:
> +  from gi.repository import GLib
> +except ImportError:
> +  import glib as GLib
> +from dbus.mainloop.glib import DBusGMainLoop
> +
> +try:
> +  from termcolor import colored, cprint
> +  set_error = lambda x: colored('!' + x, 'red', attrs=['bold'])
> +  set_cyan = lambda x: colored(x, 'cyan', attrs=['bold'])
> +  set_green = lambda x: colored(x, 'green', attrs=['bold'])
> +  set_yellow = lambda x: colored(x, 'yellow', attrs=['bold'])
> +except ImportError:
> +  print('!!! Install termcolor module for better experience !!!')
> +  set_error = lambda x: x
> +  set_cyan = lambda x: x
> +  set_green = lambda x: x
> +  set_yellow = lambda x: x
> +
> +# Provisioning agent
> +try:
> +  import agent
> +except ImportError:
> +  agent = None
> +
> +MESH_SERVICE_NAME = 'org.bluez.mesh'
> +DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
> +DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
> +
> +MESH_NETWORK_IFACE = 'org.bluez.mesh.Network1'
> +MESH_NODE_IFACE = 'org.bluez.mesh.Node1'
> +MESH_APPLICATION_IFACE = 'org.bluez.mesh.Application1'
> +MESH_ELEMENT_IFACE = 'org.bluez.mesh.Element1'
> +
> +APP_COMPANY_ID = 0x05f1
> +APP_PRODUCT_ID = 0x0001
> +APP_VERSION_ID = 0x0001
> +
> +VENDOR_ID_NONE = 0xffff
> +
> +app = None
> +bus = None
> +mainloop = None
> +node = None
> +mesh_net = None
> +
> +menu_level = 0
> +dst_addr = 0x0000
> +app_idx = 0
> +
> +# Node token housekeeping
> +token = None
> +have_token = False
> +
> +user_input = 0
> +
> +
> +def app_exit():
> +	global mainloop
> +	global app
> +
> +	for el in app.elements:
> +		for model in el.models:
> +			if model.timer != None:
> +				model.timer.cancel()
> +	mainloop.quit()
> +
> +def array_to_string(b_array):
> +	str = ""
> +	for b in b_array:
> +		str += "%02x" % b
> +	return str
> +
> +def generic_error_cb(error):
> +	print(set_error('D-Bus call failed: ') + str(error))
> +
> +def generic_reply_cb():
> +	return
> +
> +def attach_app_error_cb(error):
> +	print(set_error('Failed to register application: ') + str(error))
> +
> +def attach(token):
> +	print('Attach mesh node to bluetooth-meshd daemon')
> +
> +	mesh_net.Attach(app.get_path(), token,
> +					reply_handler=attach_app_cb,
> +					error_handler=attach_app_error_cb)
> +
> +def join_cb():
> +	print('Join procedure started')
> +
> +def join_error_cb(reason):
> +	print('Join procedure failed: ', reason)
> +
> +def unwrap(item):
> +	if isinstance(item, dbus.Boolean):
> +		return bool(item)
> +	if isinstance(item, (dbus.UInt16, dbus.Int16, dbus.UInt32, dbus.Int32,
> +						dbus.UInt64, dbus.Int64)):
> +		return int(item)
> +	if isinstance(item, dbus.Byte):
> +		return bytes([int(item)])
> +	if isinstance(item, dbus.String):
> +			return item
> +	if isinstance(item, (dbus.Array, list, tuple)):
> +		return [unwrap(x) for x in item]
> +	if isinstance(item, (dbus.Dictionary, dict)):
> +		return dict([(unwrap(x), unwrap(y)) for x, y in item.items()])
> +
> +	print(set_error('Dictionary item not handled: ') + type(item))
> +
> +	return item
> +
> +def attach_app_cb(node_path, dict_array):
> +	print('Mesh application registered ', node_path)
> +
> +	obj = bus.get_object(MESH_SERVICE_NAME, node_path)
> +
> +	global node
> +	node = dbus.Interface(obj, MESH_NODE_IFACE)
> +
> +	els = unwrap(dict_array)
> +
> +	for el in els:
> +		idx = struct.unpack('b', el[0])[0]
> +
> +		models = el[1]
> +		element = app.get_element(idx)
> +		element.set_model_config(models)
> +
> +def interfaces_removed_cb(object_path, interfaces):
> +	print('Removed')
> +	if not mesh_net:
> +		return
> +
> +	print(object_path)
> +	if object_path == mesh_net[2]:
> +		print('Service was removed')
> +		app_exit()
> +
> +def send_response(path, dest, key, data):
> +		node.Send(path, dest, key, data, reply_handler=generic_reply_cb,
> +						error_handler=generic_error_cb)
> +
> +def send_publication(path, model_id, data):
> +		print('Send publication ', end='')
> +		print(data)
> +		node.Publish(path, model_id, data,
> +						reply_handler=generic_reply_cb,
> +						error_handler=generic_error_cb)
> +
> +def print_state(state):
> +	print('State is ', end='')
> +	if state == 0:
> +		print('OFF')
> +	elif state == 1:
> +		print('ON')
> +	else:
> +		print('UNKNOWN')
> +class PubTimer():
> +	def __init__(self):
> +		self.seconds = None
> +		self.func = None
> +		self.thread = None
> +		self.busy = False
> +
> +	def _timeout_cb(self):
> +		self.func()
> +		self.busy = True
> +		self._schedule_timer()
> +		self.busy =False
> +
> +	def _schedule_timer(self):
> +		self.thread = Timer(self.seconds, self._timeout_cb)
> +		self.thread.start()
> +
> +	def start(self, seconds, func):
> +		self.func = func
> +		self.seconds = seconds
> +		if not self.busy:
> +			self._schedule_timer()
> +
> +	def cancel(self):
> +		if self.thread is not None:
> +			self.thread.cancel()
> +			self.thread = None
> +
> +class Application(dbus.service.Object):
> +
> +	def __init__(self, bus):
> +		self.path = '/example'
> +		self.agent = None
> +		self.elements = []
> +		dbus.service.Object.__init__(self, bus, self.path)
> +
> +	def set_agent(self, agent):
> +		self.agent = agent
> +
> +	def get_path(self):
> +		return dbus.ObjectPath(self.path)
> +
> +	def add_element(self, element):
> +		self.elements.append(element)
> +
> +	def get_element(self, idx):
> +		for ele in self.elements:
> +			if ele.get_index() == idx:
> +				return ele
> +
> +	def get_properties(self):
> +		return {
> +			MESH_APPLICATION_IFACE: {
> +				'CompanyID': dbus.UInt16(APP_COMPANY_ID),
> +				'ProductID': dbus.UInt16(APP_PRODUCT_ID),
> +				'VersionID': dbus.UInt16(APP_VERSION_ID)
> +			}
> +		}
> +
> +	@dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
> +	def GetManagedObjects(self):
> +		response = {}
> +		response[self.path] = self.get_properties()
> +		response[self.agent.get_path()] = self.agent.get_properties()
> +		for element in self.elements:
> +			response[element.get_path()] = element.get_properties()
> +		return response
> +
> +	@dbus.service.method(MESH_APPLICATION_IFACE,
> +					in_signature="t", out_signature="")
> +	def JoinComplete(self, value):
> +		global token
> +		global have_token
> +
> +		print('JoinComplete with token ' + set_green(hex(value)))
> +
> +		token = value
> +		have_token = True
> +
> +		attach(token)
> +
> +	@dbus.service.method(MESH_APPLICATION_IFACE,
> +					in_signature="s", out_signature="")
> +	def JoinFailed(self, value):
> +		print(set_error('JoinFailed '), value)
> +
> +
> +class Element(dbus.service.Object):
> +	PATH_BASE = '/example/ele'
> +
> +	def __init__(self, bus, index):
> +		self.path = self.PATH_BASE + format(index, '02x')
> +		self.models = []
> +		self.bus = bus
> +		self.index = index
> +		dbus.service.Object.__init__(self, bus, self.path)
> +
> +	def _get_sig_models(self):
> +		ids = []
> +		for model in self.models:
> +			id = model.get_id()
> +			vendor = model.get_vendor()
> +			if vendor == VENDOR_ID_NONE:
> +				ids.append(id)
> +		return ids
> +
> +	def get_properties(self):
> +		return {
> +			MESH_ELEMENT_IFACE: {
> +				'Index': dbus.Byte(self.index),
> +				'Models': dbus.Array(
> +					self._get_sig_models(), signature='q')
> +			}
> +		}
> +
> +	def add_model(self, model):
> +		model.set_path(self.path)
> +		self.models.append(model)
> +
> +	def get_index(self):
> +		return self.index
> +
> +	def set_model_config(self, configs):
> +		for config in configs:
> +			mod_id = config[0]
> +			self.UpdateModelConfiguration(mod_id, config[1])
> +
> +	@dbus.service.method(MESH_ELEMENT_IFACE,
> +					in_signature="qqbay", out_signature="")
> +	def MessageReceived(self, source, key, is_sub, data):
> +		print('Message Received on Element ', end='')
> +		print(self.index)
> +		for model in self.models:
> +			model.process_message(source, key, data)
> +
> +	@dbus.service.method(MESH_ELEMENT_IFACE,
> +					in_signature="qa{sv}", out_signature="")
> +
> +	def UpdateModelConfiguration(self, model_id, config):
> +		print('UpdateModelConfig ', end='')
> +		print(hex(model_id))
> +		for model in self.models:
> +			if model_id == model.get_id():
> +				model.set_config(config)
> +				return
> +
> +	@dbus.service.method(MESH_ELEMENT_IFACE,
> +					in_signature="", out_signature="")
> +
> +	def get_path(self):
> +		return dbus.ObjectPath(self.path)
> +
> +class Model():
> +	def __init__(self, model_id):
> +		self.cmd_ops = []
> +		self.model_id = model_id
> +		self.vendor = VENDOR_ID_NONE
> +		self.bindings = []
> +		self.pub_period = 0
> +		self.pub_id = 0
> +		self.path = None
> +		self.timer = None
> +
> +	def set_path(self, path):
> +		self.path = path
> +
> +	def get_id(self):
> +		return self.model_id
> +
> +	def get_vendor(self):
> +		return self.vendor
> +
> +	def process_message(self, source, key, data):
> +		return
> +
> +	def set_publication(self, period):
> +		self.pub_period = period
> +
> +	def set_config(self, config):
> +		if 'Bindings' in config:
> +			self.bindings = config.get('Bindings')
> +			print('Bindings: ', end='')
> +			print(self.bindings)
> +		if 'PublicationPeriod' in config:
> +			self.set_publication(config.get('PublicationPeriod'))
> +			print('Model publication period ', end='')
> +			print(self.pub_period, end='')
> +			print(' ms')
> +
> +	def print_bindings(self):
> +		print(set_cyan('Model'), set_cyan('%04x' % self.model_id),
> +			set_cyan('is bound to application key(s): '), end = '')
> +
> +		if len(self.bindings) == 0:
> +			print(set_cyan('** None **'))
> +		for b in self.bindings:
> +			print(set_cyan('%04x' % b), set_cyan(', '))
> +
> +########################
> +# On Off Server Model
> +########################
> +class OnOffServer(Model):
> +	def __init__(self, model_id):
> +		Model.__init__(self, model_id)
> +		self.cmd_ops = { 0x8201,  # get
> +				 0x8202,  # set
> +				 0x8203,  # set unacknowledged
> +				 0x8204 } # status
> +
> +		print("OnOff Server ")
> +		self.state = 0
> +		print_state(self.state)
> +		self.timer = PubTimer()
> +
> +	def process_message(self, source, key, data):
> +		datalen = len(data)
> +		print('OnOff Server process message len: ', datalen)
> +
> +		if datalen != 2 and datalen != 3:
> +			# The opcode is not recognized by this model
> +			return
> +
> +		if datalen == 2:
> +			op_tuple=struct.unpack('<H',bytes(data))
> +			opcode = op_tuple[0]
> +			if opcode != 0x8201:
> +				# The opcode is not recognized by this model
> +				return
> +			print('Get state')
> +		elif datalen == 3:
> +			opcode,self.state=struct.unpack('<HB',bytes(data))
> +			if opcode != 0x8202 and opcode != 0x8203:
> +				# The opcode is not recognized by this model
> +				return
> +			print_state(self.state)
> +
> +		rsp_data = struct.pack('<HB', 0x8204, self.state)
> +		send_response(self.path, source, key, rsp_data)
> +
> +	def set_publication(self, period):
> +
> +		# We do not handle ms in this example
> +		if period < 1000:
> +			return
> +
> +		self.pub_period = period
> +		if period == 0:
> +			self.timer.cancel()
> +			return
> +
> +		self.timer.start(period/1000, self.publish)
> +
> +
> +	def publish(self):
> +		print('Publish')
> +		data = struct.pack('<HB', 0x8204, self.state)
> +		send_publication(self.path, self.model_id, data)
> +
> +########################
> +# On Off Client Model
> +########################
> +class OnOffClient(Model):
> +	def __init__(self, model_id):
> +		Model.__init__(self, model_id)
> +		self.cmd_ops = { 0x8201,  # get
> +				 0x8202,  # set
> +				 0x8203,  # set unacknowledged
> +				 0x8204 } # status
> +		print('OnOff Client')
> +
> +	def _reply_cb(state):
> +		print('State ', end='');
> +		print(state)
> +
> +	def _send_message(self, dest, key, data, reply_cb):
> +		print('OnOffClient send data')
> +		node.Send(self.path, dest, key, data, reply_handler=reply_cb,
> +				  error_handler=generic_error_cb)
> +
> +	def get_state(self, dest, key):
> +		opcode = 0x8201
> +		data = struct.pack('<H', opcode)
> +		self._send_message(dest, key, data, self._reply_cb)
> +
> +	def set_state(self, dest, key, state):
> +		opcode = 0x8202
> +		print('State:', state)
> +		data = struct.pack('<HB', opcode, state)
> +		self._send_message(dest, key, data, self._reply_cb)
> +
> +	def process_message(self, source, key, data):
> +		print('OnOffClient process message len = ', end = '')
> +		datalen = len(data)
> +		print(datalen)
> +
> +		if datalen != 3:
> +			# The opcode is not recognized by this model
> +			return
> +
> +		opcode, state=struct.unpack('<HB',bytes(data))
> +
> +		if opcode != 0x8204 :
> +			# The opcode is not recognized by this model
> +			return
> +
> +		print(set_yellow('Got state '), end = '')
> +
> +		state_str = "ON"
> +		if state == 0:
> +			state_str = "OFF"
> +
> +		print(set_green(state_str), set_yellow('from'),
> +						set_green('%04x' % source))
> +
> +########################
> +# Menu functions
> +########################
> +class MenuHandler(object):
> +	def __init__(self, callback):
> +		self.cb = callback
> +		flags = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL)
> +		flags |= os.O_NONBLOCK
> +		fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, flags)
> +		sys.stdin.flush()
> +		GLib.io_add_watch(sys.stdin, GLib.IO_IN, self.input_callback)
> +
> +	def input_callback(self, fd, condition):
> +		chunk = fd.read()
> +		buffer = ''
> +		for char in chunk:
> +			buffer += char
> +			if char == '\n':
> +				self.cb(buffer)
> +
> +		return True
> +
> +def process_input(input_str):
> +	if menu_level == 0:
> +		process_main_menu(input_str)
> +	elif menu_level == 1:
> +		process_client_menu(input_str)
> +	else:
> +		print(set_error('BUG: bad menu level'))
> +
> +def switch_menu(level):
> +	global menu_level
> +
> +	if level > 1:
> +		return
> +
> +	if level == 0:
> +		main_menu()
> +	elif level == 1:
> +		client_menu()
> +
> +	menu_level = level
> +
> +########################
> +# Main menu functions
> +########################
> +def process_main_menu(input_str):
> +	global token
> +	global user_input
> +	global have_token
> +
> +	str = input_str.strip()
> +
> +	if user_input == 1:
> +		res = set_token(str)
> +		user_input = 0
> +
> +		if res == False:
> +			main_menu()
> +
> +		return
> +
> +	# Allow entering empty lines for better output visibility
> +	if len(str) == 0:
> +		return
> +
> +	if str.isdigit() == False:
> +		main_menu()
> +		return
> +
> +	opt = int(str)
> +
> +	if opt > 6:
> +		print(set_error('Unknown menu option: '), opt)
> +		main_menu()
> +	elif opt == 1:
> +		if have_token:
> +			print('Token already set')
> +			return
> +
> +		user_input = 1;
> +		print(set_cyan('Enter 16-digit hex node ID:'))
> +	elif opt == 2:
> +		if agent == None:
> +			print(set_error('Provisioning agent not found'))
> +			return
> +
> +		join_mesh()
> +	elif opt == 3:
> +		if have_token == False:
> +			print(set_error('Token is not set'))
> +			main_menu()
> +			return
> +
> +		attach(token)
> +	elif opt == 4:
> +		if have_token == False:
> +			print(set_error('Token is not set'))
> +			main_menu()
> +			return
> +
> +		print('Remove mesh node')
> +		mesh_net.Leave(token, reply_handler=generic_reply_cb,
> +					error_handler=generic_error_cb)
> +		have_token = False
> +	elif opt == 5:
> +		switch_menu(1)
> +	elif opt == 6:
> +		app_exit()
> +
> +
> +def main_menu():
> +	print(set_cyan('*** MAIN MENU ***'))
> +	print(set_cyan('1 - set node ID (token)'))
> +	print(set_cyan('2 - join mesh network'))
> +	print(set_cyan('3 - attach mesh node'))
> +	print(set_cyan('4 - remove node'))
> +	print(set_cyan('5 - client menu'))
> +	print(set_cyan('6 - exit'))
> +
> +def set_token(str):
> +	global token
> +	global have_token
> +
> +	if len(str) != 16:
> +		print(set_error('Expected 16 digits'))
> +		return False
> +
> +	try:
> +		input_number = int(str, 16)
> +	except ValueError:
> +		print(set_error('Not a valid hexadecimal number'))
> +		return False
> +
> +	token = numpy.uint64(input_number)
> +	have_token = True
> +
> +	return True
> +
> +def join_mesh():
> +	uuid = bytearray.fromhex("0a0102030405060708090A0B0C0D0E0F")
> +
> +	caps = ["out-numeric"]
> +	oob = ["other"]
> +
> +	random.shuffle(uuid)
> +	uuid_str = array_to_string(uuid)
> +	print('Joining with UUID ' + set_green(uuid_str))
> +
> +	mesh_net.Join(app.get_path(), uuid,
> +			reply_handler=join_cb,
> +			error_handler=join_error_cb)
> +
> +##############################
> +# On/Off Client menu functions
> +##############################
> +def process_client_menu(input_str):
> +	global user_input
> +	global dst_addr
> +	global app_idx
> +
> +	res = -1
> +	str = input_str.strip()
> +
> +	if user_input == 1:
> +		res = set_value(str)
> +		if res != -1:
> +			dst_addr = res
> +	elif user_input == 2:
> +		res = set_value(str)
> +		if res != -1:
> +			app_idx = res
> +
> +	if user_input != 0:
> +		user_input = 0
> +		if res == -1:
> +			client_menu()
> +		return
> +
> +	# Allow entering empty lines for better output visibility
> +	if len(str) == 0:
> +		return
> +
> +	if str.isdigit() == False:
> +		client_menu()
> +		return
> +
> +	opt = int(str)
> +
> +	if opt > 7:
> +		print(set_error('Unknown menu option: '), opt)
> +		client_menu()
> +		return
> +
> +	if opt >= 3 and opt <= 5 and dst_addr == 0x0000:
> +		print(set_error('Destination address not set!'))
> +		return
> +
> +	if opt == 1:
> +		user_input = 1;
> +		print(set_cyan('Enter 4-digit hex destination address:'))
> +	elif opt == 2:
> +		user_input = 2;
> +		app.elements[1].models[0].print_bindings()
> +		print(set_cyan('Choose application key index:'))
> +	elif opt == 3:
> +		app.elements[1].models[0].get_state(dst_addr, app_idx)
> +	elif opt == 4 or opt == 5:
> +		app.elements[1].models[0].set_state(dst_addr, app_idx, opt - 4)
> +	elif opt == 6:
> +		switch_menu(0)
> +	elif opt == 7:
> +		app_exit()
> +
> +def client_menu():
> +	print(set_cyan('*** ON/OFF CLIENT MENU ***'))
> +	print(set_cyan('1 - set destination address'))
> +	print(set_cyan('2 - set application key index'))
> +	print(set_cyan('3 - get state'))
> +	print(set_cyan('4 - set state OFF'))
> +	print(set_cyan('5 - set state ON'))
> +	print(set_cyan('6 - back to main menu'))
> +	print(set_cyan('7 - exit'))
> +
> +def set_value(str):
> +
> +	if len(str) != 4:
> +		print(set_error('Expected 4 digits'))
> +		return -1
> +
> +	try:
> +		value = int(str, 16)
> +	except ValueError:
> +		print(set_error('Not a valid hexadecimal number'))
> +		return -1
> +
> +	return value
> +
> +########################
> +# Main entry
> +########################
> +def main():
> +
> +	DBusGMainLoop(set_as_default=True)
> +
> +	global bus
> +	bus = dbus.SystemBus()
> +	global mainloop
> +	global app
> +	global mesh_net
> +
> +	if len(sys.argv) > 1 :
> +		set_token(sys.argv[1])
> +
> +	mesh_net = dbus.Interface(bus.get_object(MESH_SERVICE_NAME,
> +						"/org/bluez/mesh"),
> +						MESH_NETWORK_IFACE)
> +	mesh_net.connect_to_signal('InterfacesRemoved', interfaces_removed_cb)
> +
> +	app = Application(bus)
> +
> +	# Provisioning agent
> +	if agent != None:
> +		app.set_agent(agent.Agent(bus))
> +
> +	first_ele = Element(bus, 0x00)
> +	second_ele = Element(bus, 0x01)
> +
> +	print(set_yellow('Register OnOff Server model on element 0'))
> +	first_ele.add_model(OnOffServer(0x1000))
> +
> +	print(set_yellow('Register OnOff Client model on element 1'))
> +	second_ele.add_model(OnOffClient(0x1001))
> +	app.add_element(first_ele)
> +	app.add_element(second_ele)
> +
> +	mainloop = GLib.MainLoop()
> +
> +	main_menu()
> +	event_catcher = MenuHandler(process_input);
> +	mainloop.run()
> +
> +if __name__ == '__main__':
> +	main()