oneM2M air quality monitoring system

; acme.ini
;
; Simplified configuration file for the [ACME] CSE
;
; created: 2022-10-06 10:50:35
;
; CSE type: IN
; Environment: Development
;

[basic.config]
cseType=IN
cseID=id-in
cseName=cse-in
adminID=aid
dataDirectory=${baseDirectory}
networkInterface=0.0.0.0
cseHost=0.0.0.0
httpPort=8080
logLevel=info
databaseInMemory=False

[server.http]
enableUpperTesterEndpoint=true
enableStructureEndpoint=true

[server.http.security]
; Enable TLS for communications.
; This can be overridden by the command line arguments --http and --https.
; See oneM2M TS-0003 Clause 8.2.1 "Overview on Security Association Establishment Frameworks".
; Default: False
useTLS=true
; TLS version to be used in connections.
; Allowed versions: TLS1.1, TLS1.2, auto . Use "auto" to allow client-server certificate version negotiation.
; Default: auto
tlsVersion=auto
; Verify certificates in requests. Set to False when using self-signed certificates.
; Default: False
verifyCertificate=false
; Path and filename of the certificate file.
; Default: None
caCertificateFile=/opt/ACME-oneM2M-CSE/cert/mio_aid.crt
; Path and filename of the private key file.
; Default: None
caPrivateKeyFile=/opt/ACME-oneM2M-CSE/cert/mio_aid.key

[client.mqtt]
; Enable the MQTT binding.
; Default: False
enable=True
; The hostname of the MQTT broker.
; Default; 127.0.0.1
address=mosquitto
; Set the port for the MQTT broker. 
; Default: 1883, or 8883 for TLS
port=8883
; Value for the MQTT connection's keep-alive parameter in seconds.
; Default: 60 seconds
keepalive=60
; Interface to listen to. Use 0.0.0.0 for "all" interfaces. 
; Default: 127.0.0.1
listenIF=${basic.config:networkInterface}
; Optional prefix for topics.
; Default: empty string
topicPrefix=
; Timeout when sending MQTT requests and waiting for responses.
; Default: 5.0 seconds
timeout=5.0

[client.mqtt.security]
; The username for MQTT broker authentication if required by the broker.
; Default: None
; username=
; The password for MQTT broker authentication.
; Default: None
; password=
;Enable TLS for communications with the MQTT broker.
;Default: False
useTLS=true
; Verify certificates in requests. Set to False when using self-signed certificates..
; Default: False
verifyCertificate=false
; Path and filename of the certificate file.
; Default: None
caCertificateFile=/opt/ACME-oneM2M-CSE/cert/ca.crt
; List of credential-IDs that can be used to register an AE via MQTT. If this list is empty then all credential IDs are allowed.
; This is a comma-separated list. Wildcards (* and ?) are supported.
; Default: empty list
; allowedCredentialIDs = []

#
#	notificationServer.py
#
#	(c) 2020 by Andreas Kraft
#	License: BSD 3-Clause License. See the LICENSE file for further details.
#
#	Simple base implementation of a notification server to handle notifications 
#	from a CSE.
#

from __future__ import annotations
from http.client import HTTPMessage
from typing import cast
from http.server import HTTPServer, BaseHTTPRequestHandler
import email.parser
import json, argparse, sys, ssl, signal
import cbor2
from rich.console import Console
from rich.syntax import Syntax
from automations import *

import pathlib, os
parent = pathlib.Path(os.path.abspath(os.path.dirname(__file__))).parent.parent
sys.path.append(f'{parent}')
from acme.helpers.MQTTConnection import MQTTConnection, MQTTHandler
from acme.helpers.TextTools import toHex
from acme.etc.RequestUtils import serializeData
from acme.etc.DateUtils import getResourceDate
from acme.etc.Types import ContentSerializationType
from acme.etc.Constants import Constants as C

##############################################################################
#
#	HTTP Server
#

port = 9999	# Change this variable to specify another port.
messageColor = 'spring_green2'
errorColor = 'red'

failVerification = False

class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):

	def do_GET(self) -> None:
		"""	Just provide a simple web page.
		"""
		self.send_response(200)
		self.send_header('Content-type', 'text/html')
		self.end_headers()
		self.wfile.write(bytes("<html><head><title>[ACME] Notification Server</title></head><body>This server doesn't provide a web page.</body></html>","utf-8")) 


	def do_POST(self) -> None:
		"""	Handle notification.
		"""

		_responseHeaders:list = []

		# Get headers and content data
		length = int(self.headers['Content-Length'])
		contentType = self.headers['Content-Type']
		requestID = self.headers['X-M2M-RI']
		post_data = self.rfile.read(length)

		# Construct return header
		# Always acknowledge the verification requests
		self.send_response(200)
		self.send_header('X-M2M-RSC', '2000' if not failVerification else '4101')
		self.send_header('X-M2M-RI', requestID)
		_responseHeaders = self._headers_buffer	# type:ignore [attr-defined]
		self.end_headers()


		
		# Print the content data
		console.print(f'[{messageColor}]### Notification (http)')
		console.print(self.headers, highlight = False)

		#Start Automation
		automation(mqttClient, post_data)
		# Print JSON
		if contentType in [ 'application/json', 'application/vnd.onem2m-res+json' ]:
			console.print(Syntax(json.dumps(json.loads(post_data.decode('utf-8')), indent=4),
							 'json', 
							 theme='monokai',
							 line_numbers=False))
		
		# Print CBOR
		elif contentType in [ 'application/cbor', 'application/vnd.onem2m-res+cbor' ]:
			console.print('[dim]Content as Hexdump:\n')
			console.print(toHex(post_data), highlight=False)
			console.print('\n[dim]Content as JSON:\n')
			console.print(Syntax(json.dumps(cbor2.loads(post_data), indent=4),
							 'json', 
							 theme='monokai',
							 line_numbers=False))		

		# Print other binary content
		else:
			console.print(toHex(post_data), highlight=False)
		
		# Print HTTP Response
		# This looks a it more complicated but is necessary to render nicely in Jupyter
		console.print(f'[{messageColor}]### Notification Response (http)')
		console.print(email.parser.Parser(_class = HTTPMessage).parsestr(b''.join(_responseHeaders).decode('iso-8859-1')), highlight = False)


	def log_message(self, format:str, *args:int) -> None:
		if (msg := format%args).startswith('"GET'):	return	# ignore GET log messages
		console.print(f'[{messageColor} reverse]{msg}', highlight = False)


##############################################################################
#
#	MQTT Client
#

mqttNotificationTopic = [ '/oneM2M/req/id-in/+/#' ]

class MQTTClientHandler(MQTTHandler):

	def __init__(self, topic:str|list[str], enableLogging:bool) -> None:
		super().__init__()
		self.topic = topic
		self.enableLogging = enableLogging
		self.isShutdown = False


	def onConnect(self, connection:MQTTConnection) -> bool:
		try:
			connection.subscribeTopic(self.topic, self._requestCB)					# Subscribe to general requests
		except Exception as e:
			self.onError(connection, -1, f'MQTT {str(e)}')
		return True


	def onError(self, connection:MQTTConnection, rc:int=-1, msg:str=None) -> bool:
		if rc in [5]:		# authorization error
			console.print(f'[{errorColor}]MQTT authorization error')
			connection.shutdown()
			console.print(f'[{messageColor}]MQTT client shutdown')
		if rc == -1: 	# unknown. probably connection error?
			if msg:
				console.print(f'[{errorColor}]{msg}')
			connection.shutdown()
			console.print(f'[{messageColor}]MQTT client shutdown')
		# ignore all others
		return True


	def logging(self, connection:MQTTConnection, level:int, message:str) -> bool:
		if self.enableLogging:
			console.print(f'{level}: {message}')
		return True


	def onShutdown(self, connection: MQTTConnection) -> None:
		if not self.isShutdown:
			os.kill(os.getpid(), signal.SIGUSR1)
	

	def _requestCB(self, connection:MQTTConnection, topic:str, data:bytes) -> None:
		# TODO is it actually a notification? no -> do nothing, yes -> reply

		def _constructResponse(frm:str, to:str, jsn:dict) -> dict:
			responseData = 	{ 	'fr':	frm,
								'to':	to, 
								'rsc':	2000 if not failVerification else 4101,
								'ot':	getResourceDate()
							}
			if (rqi := jsn.get('rqi')):
				responseData['rqi'] = rqi
			if (rvi := jsn.get('rvi')):
				responseData['rvi'] = rvi
			return responseData


		console.print(f'[{messageColor}]### Notification (MQTT)')
		console.print(f'Topic: {topic}')
		topicArray	= topic.split('/')
		if len(topicArray) > 4 and topicArray[-4] == 'req' and topicArray[-5] == 'oneM2M':
			_frm		= topicArray[-3]
			_to			= topicArray[-2]
			encoding	= topicArray[-1]
		else:
			_frm 		= 'non-onem2m-entity'	
			_to 		= 'unknown'
			encoding	= 'json'

		# Print JSON
		responseData = None
		if encoding.upper() == 'JSON':
			console.print(Syntax(json.dumps((jsn := json.loads(data)), indent=4),
							 'json', 
							 theme='monokai',
							 line_numbers=False))
			to 	= jsn['to'] if 'to' in jsn else _to
			frm = jsn['fr'] if 'fr' in jsn else _frm
			responseData = cast(bytes, serializeData(_constructResponse(to, frm, jsn), ContentSerializationType.JSON))
			console.print(responseData)



		# Print CBOR
		elif encoding.upper() == 'CBOR':
			console.print('[dim]Content as Hexdump:\n')
			console.print(toHex(data), highlight=False)
			console.print('\n[dim]Content as JSON:\n')
			console.print(Syntax(json.dumps((jsn := cbor2.loads(data)), indent=4),
							 'json', 
							 theme='monokai',
							 line_numbers=False))		
			to 	= jsn['to'] if 'to' in jsn else to
			frm = jsn['fr'] if 'fr' in jsn else frm
			responseData = cast(bytes, serializeData(_constructResponse(to, frm, jsn), ContentSerializationType.CBOR))

		# Print other binary content
		else:
			console.print('[dim]Content as Hexdump:\n')
			console.print(toHex(data), highlight=False)
		# TODO send a response

		if responseData:
			# connection.publish(f'/oneM2M/resp{frm}/{to.lstrip("/").replace("/", ":")}/{encoding}', responseData)
			connection.publish(f'/oneM2M/resp/{_frm}/{_to}/{encoding}', responseData)


class MQTTClient(object):

	def __init__(self, args:argparse.Namespace) -> None:
		# Assume a couple of meaningful defaults here
		self.mqttConnection = MQTTConnection(address			= args.mqttAddress,
											 port				= args.mqttPort,
											 keepalive			= 60,
											 interface			= '0.0.0.0',
											 username 			= args.mqttUsername,
											 password			= args.mqttPassword,
											 useTLS				= args.mqtts,
											 caFile				= args.caFile,
											 certfile			= args.mqttCert,
											 keyfile			= args.mqttKey,
											 messageHandler 	= MQTTClientHandler	(args.mqttTopic, args.mqttLogging))
	

	def run(self) -> None:
		self.mqttConnection.run()
		console.print(f'[{messageColor}]MQTT client started. Connected to {args.mqttAddress}:{args.mqttPort}.')
	

	def shutdown(self) -> None:
		self.mqttConnection.messageHandler.isShutdown = True	# type:ignore [attr-defined]
		if self.mqttConnection:
			self.mqttConnection.shutdown()
			console.print(f'[{messageColor}]MQTT client stopped.')
	
##############################################################################

#
#	Help with exiting and terminating all the threads programmatically
#	
class ExitCommand(Exception):
	pass

def exitSignalHandler(signal, frame) -> None:	# type: ignore [no-untyped-def]
	raise ExitCommand()

def exitAll() -> None:
	os.kill(os.getpid(), signal.SIGUSR1)

signal.signal(signal.SIGUSR1, exitSignalHandler)
	
if __name__ == '__main__':
	console = Console()
	console.print(f'\n{C.textLogo} - [bold]Notification Server[/bold]\n\n')


	# parse command line argiments
	parser = argparse.ArgumentParser()

	# mutual exlcusive arguments for http
	groupHttp = parser.add_mutually_exclusive_group()
	groupHttp.add_argument('--http', action='store_false', dest='usehttps', default=None, help='run as http server (the default)')
	groupHttp.add_argument('--https', action='store_true', dest='usehttps', default=None, help='run as https server')
	parser.add_argument('--port', action='store', dest='port', default=port, type=int, help='specify the server port')
	parser.add_argument('--certfile', action='store', dest='certfile', required='--https' in sys.argv, metavar='<filename>', help='specify the certificate file for https')
	parser.add_argument('--keyfile', action='store', dest='keyfile', required='--https' in sys.argv, metavar='<filename>', 	help='specify the key file for https')

	# MQTT arguments
	parser.add_argument('--mqtt', action='store_true', dest='mqtt', default=False, help='enable MQTT for notifications')
	parser.add_argument('--mqtt-address', action='store', dest='mqttAddress', default='localhost', required='--mqtt' in sys.argv, metavar='<host>', help='MQTT broker address (default: localhost)')
	parser.add_argument('--mqtt-port', action='store', dest='mqttPort', default=1883, metavar='<port>',  help='MQTT broker port (default: 1883)')
	parser.add_argument('--mqtt-topic', action='store', dest='mqttTopic', default=mqttNotificationTopic, metavar='<topic>', nargs='+', help=f'MQTT topic list (default: {mqttNotificationTopic})')
	parser.add_argument('--mqtt-username', action='store', dest='mqttUsername', default=None, metavar='<username>',  help='MQTT username (default: None)')
	parser.add_argument('--mqtt-password', action='store', dest='mqttPassword', default=None, required='--mqttUsername' in sys.argv, metavar='<password>',  help='MQTT password (default: None)')
	parser.add_argument('--mqtt-logging', action='store_true', dest='mqttLogging', default=False, help='MQTT enable logging (default: disabled)')
	parser.add_argument('--mqtt-useTLS', action='store_true', dest='mqtts', default=False, help='Enable MQTTS (default: disabled)')
	parser.add_argument('--mqtt-caFile', action='store', metavar='<cafile>', dest='caFile', default=None, help='CA Certficiate ca.crt file (default: None)')
	parser.add_argument('--mqtt-certfile', action='store', metavar='<certfile>', dest='mqttCert', default=None, help='Certificate file *.crt for MQTTS (default: None)')
	parser.add_argument('--mqtt-keyfile', action='store', metavar='<keyfile>', dest='mqttKey', default=None, help='Key file *.key for MQTTS (default: None)')

	# Generic
	parser.add_argument('--fail-verification', action='store_true', dest='failVerification', default=False, help='Fail verification requests (default: false)')



	args = parser.parse_args()

	# Generic arguments
	failVerification = args.failVerification

	# run http(s) server
	httpd = HTTPServer(('', args.port), SimpleHTTPRequestHandler)
	
	if args.usehttps:
		# init ssl socket
		context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)					# Create a SSL Context
		context.load_cert_chain(args.certfile, args.keyfile)				# Load the certificate and private key
		httpd.socket = context.wrap_socket(httpd.socket, server_side=True)	# wrap the original http server socket as an SSL/TLS socket
	
	console.print(f'[{messageColor}]Notification server started.')
	try:
		# run mqtt client
		mqttClient = None
		if args.mqtt:
			mqttClient = MQTTClient(args)
			mqttClient.run()
		# run http server
		console.print(f'[{messageColor}]Listening for http(s) connections on port {args.port}.')
		httpd.serve_forever()
	except KeyboardInterrupt as e:
		if mqttClient:
			mqttClient.shutdown()
		# The http server is stopped implicitly
	except ExitCommand:
		pass
	except Exception:
		console.print()
		console.print_exception()
	finally:
		console.print(f'[{messageColor}]Notification server stopped.')

#
#	MQTTConnection.py
#
#	(c) 2021 by Andreas Kraft
#	License: BSD 3-Clause License. See the LICENSE file for further details.
#
#	Implementation of an MQTT Client helper class.
#

from __future__ import annotations
import ssl, time
from dataclasses import dataclass
from typing import Callable, Any, Tuple
import logging

from .BackgroundWorker import BackgroundWorkerPool, BackgroundWorker
from .TextTools import simpleMatch

import paho.mqtt.client as mqtt

@dataclass
class MQTTTopic:
	"""	Structure that represents a subscribed-to topic.
	"""
	topic:str				= None
	mid:int					= None
	isSubscribed:bool		= False
	callback:MQTTCallback	= None
	callbackArgs:dict 		= None


class MQTTHandler(object):
	"""	This base class defines the interface for an MQTT handler class. 
		The abstract methods defined here must be implemented by the implementing class.

		The implementing class acts as a handler for various callbacks when dealing with
		the MQTT handler. To receive messages a client implementation must register topics
		and the callbacks for them in the `onConnect()` method.
	"""

	def onConnect(self, connection:MQTTConnection) -> bool:
		"""	This method is called after the MQTT client connected to the MQTT broker. 
			Usually, an MQTT client should subscribe to topics and register the callback
			methods here.
		"""
		return True

	def onDisconnect(self, connection:MQTTConnection) -> bool:
		"""	This method is called after the MQTT client disconnected from the MQTT broker. 
		"""
		return True

	def onSubscribed(self, connection:MQTTConnection, topic:str) -> bool:
		"""	This method is called after the MQTT client successfully subsribed to a topic. 
		"""
		connection.subscribedCount += 1
		return True

	def onUnsubscribed(self, connection:MQTTConnection, topic:str) -> bool:
		"""	This method is called after the MQTT client successfully unsubsribed from a topic. 
		"""
		connection.subscribedCount -= 1
		return True

	def onError(self, connection:MQTTConnection, rc:int) -> bool:
		"""	This method is called when receiving an error when communicating with the MQTT broker. 
		"""
		return True

	def logging(self, connection:MQTTConnection, level:int, message:str) -> bool:
		"""	This method is called when a log message should be handled. 
		"""
		return True
	
	def onShutdown(self, connection:MQTTConnection) -> None:
		"""	This method is called after the ```connection``` was shut down.
		"""


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


class MQTTConnection(object):

	#
	#	Runtime methods
	#

	def __init__(self, address:str, port:int=None, keepalive:int=60, interface:str='0.0.0.0', 
					clientID:str=None, username:str=None, password:str=None,
					useTLS:bool=False, caFile:str=None, verifyCertificate:bool=False, certfile: str = None, keyfile: str = None,
					lowLevelLogging:bool=True,
					messageHandler:MQTTHandler=None
				) -> None:
		self.address								= address
		self.port									= port if port else 8883 if useTLS else 1883
		self.keepalive								= keepalive
		self.bindIF									= interface
		self.username:str							= username
		self.password:str 							= password
		self.useTLS:bool							= useTLS
		self.verifyCertificate						= verifyCertificate
		self.caFile									= caFile
		self.mqttsCertfile 							= certfile
		self.mqttsKeyfile 							= keyfile
		self.clientID								= clientID
		self.lowLevelLogging						= lowLevelLogging

		self.isStopped								= True
		self.isConnected							= False
		self.subscribedCount 						= 0


		self.mqttClient:mqtt.Client 				= None
		self.messageHandler:MQTTHandler				= messageHandler
		self.actor:BackgroundWorker 				= None
		self.subscribedTopics:dict[str, MQTTTopic]	= {}

	
	def shutdown(self) -> bool:
		"""	Shutting down the MQTT client.
		"""

		self.isStopped = True
		if self.mqttClient:
		# if self.mqttClient and self.isConnected:
			# Unsubscribe from all topics
			for t in list(self.subscribedTopics.values()):
				self.unsubscribeTopic(t)
			# wait a moment for all unsubscribe ACKs to arrive
			while len(self.subscribedTopics) > 0:
				time.sleep(0.1)
			# Then disconnect. The actor is stoped implicitly
			self.mqttClient.disconnect()
			self.actor = None

		self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.INFO, 'MQTT client shut down')
		return True


	def run(self) -> None:
		"""	Initialize and run the MQTT client as a BackgroundWorker/Actor.
		"""
		self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.DEBUG, f'MQTT: client name: {self.clientID}')
		self.mqttClient = mqtt.Client(client_id=self.clientID, clean_session=False if self.clientID else True)	# clean_session=False is defined by TS-0010

		# Enable SSL see: https://pypi.org/project/paho-mqtt/
		if self.useTLS:
			self.mqttClient.tls_set(ca_certs=self.caFile, certfile=self.mqttsCertfile, keyfile=self.mqttsKeyfile, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None)
			# If tls_insecure_set is set to True, it is impossible to guarantee that the host you are connecting to is not impersonating your server. This can be useful in initial server testing, but makes it possible for a malicious third party to impersonate your server through DNS spoofing, for example.
			self.mqttClient.tls_insecure_set(True)

		# Set username/password
		if self.username and self.password:
			self.mqttClient.username_pw_set(self.username, self.password)
		
		self.mqttClient.on_connect 		= self._onConnect
		self.mqttClient.on_disconnect	= self._onDisconnect
		self.mqttClient.on_log			= self._onLog
		self.mqttClient.on_subscribe	= self._onSubscribe
		self.mqttClient.on_unsubscribe	= self._onUnsubscribe
		self.mqttClient.on_message		= self._onMessage

		try:
			self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.DEBUG, f'MQTT: connecting to host:{self.address}, port:{self.port}, keepalive: {self.keepalive}, bind: {self.bindIF}')
			self.mqttClient.connect(host=self.address, port=self.port, keepalive=self.keepalive, bind_address=self.bindIF)
		except Exception as e:
			if self.messageHandler:
				self.messageHandler.logging(self.mqttClient, logging.ERROR, f'MQTT: cannot connect to broker: {e}')
				self.messageHandler.onError(self, -1)

		# Actually start the actor to run the MQTT client as a thread
		self.actor = BackgroundWorkerPool.newActor(self._mqttActor, name='MQTTClient').start()


	def _mqttActor(self) -> bool:
		"""	Backgroundworker callback to run the actuall MQTT loop.
		"""
		self.isStopped = False
		self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.INFO, 'MQTT: client started')
		while not self.isStopped:
			self.mqttClient.loop_forever()	# Will return when disconnect() is called
		if self.messageHandler:
			self.messageHandler.onShutdown(self)
		return True
	
	
	#
	#	MQTT/paho callbacks
	#

	def _onConnect(self, client:mqtt.Client, userdata:Any, flags:dict, rc:int) -> None:
		"""	Callback when the MQTT client connected to the broker.
		"""
		self.messageHandler and self.messageHandler.logging(self, logging.DEBUG, f'MQTT: Connected with result code: {rc} ({mqtt.error_string(rc)})')
		if rc == 0:
			self.isConnected = True
			self.messageHandler and self.messageHandler.onConnect(self)
		else:
			self.isConnected = False
			if self.messageHandler:
				self.messageHandler.logging(self, logging.ERROR, f'MQTT: Cannot connect to broker. Result code: {rc} ({mqtt.error_string(rc)})')
				self.messageHandler.onError(self, rc)


	def _onDisconnect(self, client:mqtt.Client, userdata:Any, rc:int) -> None:
		"""	Callback when the MQTT client disconnected from the broker.
		"""
		self.messageHandler and self.messageHandler.logging(self, logging.DEBUG, f'MQTT: Disconnected with result code: {rc} ({mqtt.error_string(rc)})')
		self.subscribedTopics.clear()
		if rc == 0:
			self.isConnected = False
			self.messageHandler and	self.messageHandler.onDisconnect(self)
		elif rc == 7:
			self.isConnected = False
			self.messageHandler.logging(self, logging.ERROR, f'MQTT: Cannot disconnect from broker. Result code: {rc} ({mqtt.error_string(rc)})')
			self.messageHandler.logging(self, logging.ERROR, f'MQTT: Did another client connected with the same ID ({self.clientID})?')
			self.messageHandler and	self.messageHandler.onDisconnect(self)
		else:
			self.isConnected = False
			if self.messageHandler:
				self.messageHandler.logging(self, logging.ERROR, f'MQTT: Cannot disconnect from broker. Result code: {rc} ({mqtt.error_string(rc)})')
				self.messageHandler.onDisconnect(self)
				self.messageHandler.onError(self, rc)


	def _onLog(self, client:mqtt.Client, userdata:Any, level:int, buf:str) -> None:
		"""	Mapping of the paho MQTT client's log to the logging system. Also handles different log-level scheme.
		"""
		self.lowLevelLogging and self.messageHandler and self.messageHandler.logging(self, mqtt.LOGGING_LEVEL[level], f'MQTT: {buf}')
	

	def _onSubscribe(self, client:mqtt.Client, userdata:Any, mid:int, granted_qos:int) -> None:
		# TODO doc, error check when not connected, not subscribed
		for t in self.subscribedTopics.values():
			if t.mid == mid:
				t.isSubscribed = True
				self.messageHandler and self.messageHandler.onSubscribed(self, t.topic)
				break
	

	def _onUnsubscribe(self, client:mqtt.Client, userdata:Any, mid:int) -> None:
		# TODO doc, error check when not connected, not subscribed
		"""	Callback when the client successfulle unsubscribed from a topic. The topic
			is also removed from the internal list.
		"""
		for t in self.subscribedTopics.values():
			if t.mid == mid:
				del self.subscribedTopics[t.topic]
				self.messageHandler and self.messageHandler.onUnsubscribed(self, t.topic)
				break


	def _onMessage(self, client:mqtt.Client, userdata:Any, message:mqtt.MQTTMessage) -> None:
		"""	Handle a received message. Forward it to the apropriate handler callback (in a Thread)
		"""
		self.lowLevelLogging and self.messageHandler and self.messageHandler.logging(self, logging.DEBUG, f'MQTT: received topic:{message.topic}, payload:{message.payload}')
		for t in self.subscribedTopics.keys():
			if simpleMatch(message.topic, t, star='#'):
				if (topic := self.subscribedTopics[t]).callback:
					# Run actual request handling in a thread
					# For some reasons mid is not initialized in the on on_message callback, so we use the timestamp for the actor name
					BackgroundWorkerPool.newActor(topic.callback, name=f'mid_{message.timestamp}').start(	connection=self,
																											topic=message.topic,
																											data=message.payload, 
																											**topic.callbackArgs)
					break	# break at first occurence


	#
	#	MQTT messaging methods
	#

	def subscribeTopic(self, topic:str|list[str], callback:MQTTCallback=None, **kwargs:Any) -> None:
		"""	Add one or more MQTT topics to subscribe to. Add the topic(s) afterwards
			to the list of subscribed-to topics.
		"""
		def _subscribe(topic:str) -> None:
			"""	Handle subscription of a single topic.
			"""
			if topic in self.subscribedTopics:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.WARNING, f'MQTT: topic already subscribed: {topic}')
				return
			if (r := self.mqttClient.subscribe(topic))[0] == 0:
				t = MQTTTopic(topic = topic, mid=r[1], callback=callback, callbackArgs=kwargs)
				self.subscribedTopics[topic] = t
			else:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.ERROR, f'MQTT: cannot subscribe: {r[0]}')

		if not self.mqttClient or not self.isConnected:
			self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.ERROR, 'MQTT: Client missing or not initialized')
			return

		# either subscribe a list of topics or a single topic
		list(map(_subscribe, topic if isinstance(topic, list) else [topic]))


	def unsubscribeTopic(self, topic:str|MQTTTopic) -> None:
		"""	Unsubscribe from a topic. `topic` is either an MQTTTopic structure with
			a previously subscribed to topic, or a topic name, in which case
			it is searched for in the list of MQTTTopics.
		"""
		if isinstance(topic, MQTTTopic):
			if topic.topic not in self.subscribedTopics:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.WARNING, f'MQTT: unknown topic: {topic.topic}')
				return
			if (r := self.mqttClient.unsubscribe(topic.topic))[0] == 0:
				topic.mid = r[1]
			else:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.ERROR, f'MQTT: cannot unsubscribe: {r[0]}')
				return

		else:	# if topic is just the name we need to subscribe to
			if topic not in self.subscribedTopics:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.WARNING, f'MQTT: unknown topic: {topic}')
				return
			t = self.subscribedTopics[topic]
			if t.isSubscribed:
				if (r := self.mqttClient.unsubscribe(t.topic))[0] == 0:
					t.mid = r[1]
				else:
					self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.ERROR, f'MQTT: cannot unsubscribe: {r[0]}')
					return
			else:
				self.messageHandler and self.messageHandler.logging(self.mqttClient, logging.WARNING, f'MQTT: topic not subscribed: {topic}')

		# topic is removed in _onUnsubscribe() callback


	def isFullySubscribed(self) -> bool:
		"""	Check whether the number managed subscriptions matches the number of
			currently subscribed-to topics.
		"""
		return self.subscribedCount == len(self.subscribedTopics)



	def publish(self, topic:str, data:bytes) -> None:
		"""	Publish the message `data` with the topic `topic` with the MQTT broker.
		"""
		self.mqttClient.publish(topic, data)



##############################################################################
#
#	Utility functions
#

def idToMQTT(id:str) -> str:
	"""	Convert a oneM2M ID to an MQTT compatible path element.
	"""
	return f'{id.lstrip("/").replace("/", ":")}'


def idToMQTTClientID(id:str, isCSE:bool=True) -> str:
	"""	Convert a oneM2M ID to an MQTT client ID.
	"""
	return f'{"C::" if isCSE else "A::"}{id.lstrip("/")}'

def mqttToId(mqttId:str, isCSE:bool=True) -> Tuple[str, bool]:
	"""	Convert an MQTT compatible path element to an ID.
	"""
	if mqttId.startswith('A:'):
		isCSE = False
	elif mqttId.startswith('C:'):
		isCSE = True
	else:
		return None, False
	return mqttId[2:].replace(':', '/'), isCSE


# Type for an MQTT Callback
MQTTCallback = Callable[[MQTTConnection, str, bytes], None]

[
    // ModuleClass: mioAirqualitySensor (aiQSr)
    {
        "type"      : "mio:aiQSr",
        "lname"     : "mioAirqualitySensor",
        "cnd"       : "org.fhtwmio.common.moduleclass.mioAirqualitySensor",
        "attributes": [
            // DataPoint: dataGenerationTime
            { 
                "sname" : "dgt",
                "lname" : "dataGenerationTime", 
                "type" : "timestamp", 
                "car" : "01" 
            }, 
            // DataPoint: co2
            { 
                "sname" : "co2", 
                "lname" : "co2Value", 
                "type" : "number",
                "car" : "01" 
            },
            // DataPoint: temperature
            { 
                "sname" : "temp", 
                "lname" : "temperatureValue", 
                "type" : "number",
                "car" : "01" 
            },
            // DataPoint: humidity
            { 
                "sname" : "hum", 
                "lname" : "humidityValue", 
                "type" : "number",
                "car" : "01" 
            }
        ]
    },

    // ModuleClass: mioColorSmartLed (coSLd)
    {
        "type"      : "mio:coSLd",
        "lname"     : "mioColorSmartLed",
        "cnd"       : "org.fhtwmio.common.moduleclass.mioColorSmartLed",
        "attributes": [
            // DataPoint: dataGenerationTime
            { 
                "sname" : "dgt",
                "lname" : "dataGenerationTime", 
                "type" : "timestamp", 
                "car" : "01" 
            },            
            // DataPoint: hue
            { 
                "sname" : "hue", 
                "lname" : "hueValue", 
                "type" : "nonNegInteger",
                "car" : "1" 
            },
            // DataPoint: saturation
            { 
                "sname" : "sat", 
                "lname" : "saturationValue", 
                "type" : "nonNegInteger",
                "car" : "1" 
            },
            // DataPoint: colorX
            { 
                "sname" : "x", 
                "lname" : "xValue", 
                "type" : "number",
                "car" : "1"
            },
            // DataPoint: colorY
            { 
                "sname" : "y", 
                "lname" : "yValue", 
                "type" : "number",
                "car" : "1" 
            },
            // DataPoint: color temperature
            { 
                "sname" : "colT", 
                "lname" : "colorTemperatureValue", 
                "type" : "nonNegInteger",
                "car" : "1" 
            }        
        ]
    },
    
    // DeviceClass: deviceSmartLed
    {
        "type"      : "mio:devSd",
        "lname"     : "mioDeviceSmartLed",
        "cnd"       : "org.fhtwmio.common.device.mioDeviceSmartLed",
        "children"  : [
            "cod:fauDn", 
            "cod:binSh",
            "cod:runSe",
            "cod:sigSh",
            "cod:brigs",
            "mio:coSLd"
        ]
    },

    // DeviceClass: deviceAirQualitySensor
    {
        "type"      : "mio:devAir",
        "lname"     : "mioDeviceAirQualitySensor",
        "cnd"       : "org.fhtwmio.common.device.mioDeviceAirQualitySensor",
        "children"  : [
            "cod:fauDn", 
            "mio:aiQSr"
        ]
    },

    // DeviceClass: deviceActuator
    {
        "type"      : "mio:devAct",
        "lname"     : "mioDeviceActuator",
        "cnd"       : "org.fhtwmio.common.device.mioDeviceActuator",
        "children"  : [
            "cod:fauDn", 
            "cod:binSh",
            "cod:sigSh"
        ]
    }
]