[DOC] Receiving signal ?

[Kochise]

Hi.

The documentation is not clear and the example is not functional.

How set a signal receiver without having to handle the "introspection" ?

I'd like to have def cb_on_tested(testid) called :

#!/usr/bin/env python3

# author: d.koch
# coding: utf-8
# naming: pep-0008
# typing: pep-0484
# docstring: pep-0257
# indentation: tabulation

""" dbus_interface.py
	DBus interface
"""

#  --- IMPORT ---

# Standard libraries (installed with python)

import sys
sys.path.insert(0, "..")

import logging
import os

# . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .

# External libraries (installed with pip, conda, setup.py, ...)

import asyncio
import json

# . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .

# Included libraries (this module, local files)

# https://pypi.org/project/dbus-next/

try:
	from dbus_next import BusType
	from dbus_next import Message
	from dbus_next import MessageType
	from dbus_next import Variant

	from dbus_next.aio import MessageBus
	#from dbus_next.aio.message_bus import MessageBus

	from dbus_next.service import ServiceInterface

	from dbus_next.service import dbus_property
	from dbus_next.service import method
	from dbus_next.service import signal
except ImportError:
	print(f'dbus-next not installed')

#  --- GLOBAL ---

logging.basicConfig(level = logging.DEBUG)

G_LOG = logging.getLogger("dbus_interface")
G_LOG.info(f'Starting...')

# Local settings (might be present in other files yet with different values)

# https://develop.kde.org/docs/features/d-bus/introduction_to_dbus/
# https://telepathy.freedesktop.org/doc/book/sect.basics.dbus.html
# https://dbus-cxx.github.io/local-and-remote.html
# https://dbus.freedesktop.org/doc/dbus-api-design.html

# DBus topology :
#  Service : org.me.project ("well-known bus name", or :1.xx)
#    Object : /zb/... (object path)
#      Interface : org.freedesktop.DBus.Introspectable
#        Method : Introspect (return a XML)
#      Interface : org.me.project.test
#        Methods :
#        Properties (R/W) :
#        Signals :

G_BUS_TEST		        = 'org.me.project'
G_PATH_TEST			    = '/test'
G_INTERFACE_TEST		= 'org.me.project.test'
G_SIGNAL_TESTED			= 'Tested'

#  --- CLASS ---

# https://python-dbus-next.readthedocs.io/en/latest/type-system/index.html

class TestInterface(ServiceInterface):
	""" Test interface
		Receive :
			Signals
				Tested
					Checked signal
		Send :
			Signals
				Tested	: {testid}
					Signal to be checked
	"""
	def __init__(self, i_oStrName):
		super().__init__(i_oStrName)

	if True:
		def cb_on_tested(testid):
			# NEVER CALLED <----------------
			G_LOG.info(f'Received message (1): {testid}')

		def on_tested(self, cb_on_tested):
			pass

	@signal(name = G_SIGNAL_TESTED)
	def signal_tested(self) -> 's':
		return '00000001'

# - - - MAIN - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

async def main():
	l_eBusType = BusType.SESSION

	l_oBus = await MessageBus(bus_type = l_eBusType).connect()
	#negotiate_unix_fd = True

	l_oTestInterface = TestInterface(G_INTERFACE_TEST)

	l_oBus.export(G_PATH_TEST, l_oTestInterface)

	# Work on BusType.SESSION
	await l_oBus.request_name(G_BUS_TEST)

	print(f'Waiting...')

	# FIRST TRY
	if True:
		l_oTestInterface.signal_tested()

	# SECOND TRY
	if True:
		def new_cb_on_tested(testid):
			G_LOG.info(f'Received message (2): {testid}')

		l_oTestInterface.on_tested(new_cb_on_tested)
		l_oTestInterface.signal_tested()

	# THIRD TRY
	if False:
		l_oIntrospect = await l_oBus.introspect(G_INTERFACE_TEST, G_PATH_TEST)
		l_oProxyObject = l_oBus.get_proxy_object(G_INTERFACE_TEST, G_PATH_TEST, l_oIntrospect)
		l_oInterface = l_oProxyObject.get_interface(G_INTERFACE_TEST)

	await l_oBus.wait_for_disconnect()

#  --- MAIN ---

def __main__():
	""" Basic self test (debugging)
	"""
	l_oLoop = asyncio.get_event_loop()
	l_oLoop.set_debug(1)

	if True:
		try:
			l_oLoop.run_until_complete(main())
			#asyncio.run(main())
		except KeyboardInterrupt:
			print(f'')
			G_LOG.info(f'Process interrupted...')
			pass
		finally:
			l_oLoop.stop()
			l_oLoop.close()
	else:
		pass

if __name__ == "__main__":
	""" Routine selector
	"""
	__main__()

Regards