[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