Browse avahi services through DBUS using PyQt5 and QtDbus

pyqt5-avahi-sample.py

#!/usr/bin/python3

import sys
import logging
import signal

from PyQt5.QtCore import QObject, QVariant, pyqtSlot, pyqtSignal
from PyQt5.QtWidgets import QApplication
from PyQt5.QtDBus import QDBus, QDBusConnection, QDBusInterface, QDBusMessage

from models import Server

logger = logging.getLogger(__name__)
signal.signal(signal.SIGINT, signal.SIG_DFL)


class Service(QObject):

    def __init__(
        self,
        interface,
        protocol,
        name,
        stype,
        domain,
        host=None,
        aprotocol=None,
        address=None,
        port=None,
        txt=None
    ):
        super(Service, self).__init__()
        self.interface = interface
        self.protocol = protocol
        self.name = name
        self.stype = stype
        self.domain = domain
        self.host = host
        self.aprotocol = aprotocol
        self.address = address
        self.port = port
        self.txt = txt


    def __str__(self):
        return '{s.name} ({s.stype}.{s.domain})'.format(s=self)

    def __eq__(self, other):
        return self.__dict__ == other.__dict__


class Discoverer(QObject):

    # Use those signals to get notified of changes in subscribed services
    # Emitted when initial scanning of avahi services is done
    initialized = pyqtSignal()
    # Emitted when a new service for our watched type is found
    added = pyqtSignal(Service)
    removed = pyqtSignal(Service)

    def __init__(self, parent, service, interface=-1, protocol=-1, domain='local'):
        super(Discoverer, self).__init__(parent)
        self.protocol = protocol
        self.bus = QDBusConnection.systemBus()
        self.bus.registerObject('/', self)
        self.server = QDBusInterface(
            'org.freedesktop.Avahi',
            '/',
            'org.freedesktop.Avahi.Server',
            self.bus
        )
        flags = QVariant(0)
        flags.convert(QVariant.UInt)
        browser_path = self.server.call(
            'ServiceBrowserNew',
            interface,
            self.protocol,
            service,
            domain,
            flags
        )
        logger.debug('New ServiceBrowser: {}'.format(browser_path.arguments()))
        self.bus.connect(
            'org.freedesktop.Avahi',
            browser_path.arguments()[0],
            'org.freedesktop.Avahi.ServiceBrowser',
            'ItemNew',
            self.onItemNew
        )
        self.bus.connect(
            'org.freedesktop.Avahi',
            browser_path.arguments()[0],
            'org.freedesktop.Avahi.ServiceBrowser',
            'ItemRemove',
            self.onItemRemove
        )
        self.bus.connect(
            'org.freedesktop.Avahi',
            browser_path.arguments()[0],
            'org.freedesktop.Avahi.ServiceBrowser',
            'AllForNow',
            self.onAllForNow
        )

    @pyqtSlot(QDBusMessage)
    def onItemNew(self, msg):
        logger.debug('Avahi service discovered: {}'.format(msg.arguments()))
        flags = QVariant(0)
        flags.convert(QVariant.UInt)
        resolved = self.server.callWithArgumentList(
            QDBus.AutoDetect,
            'ResolveService',
            [
                *msg.arguments()[:5],
                self.protocol,
                flags
            ]
        ).arguments()
        logger.debug('Avahi service resolved: {}'.format(resolved))
        service = Service(*resolved[:10])
        self.added.emit(service)

    @pyqtSlot(QDBusMessage)
    def onItemRemove(self, msg):
        arguments = msg.arguments()
        logger.debug('Avahi service removed: {}'.format(arguments))
        service = Service(*arguments[:5])
        self.removed.emit(service)

    @pyqtSlot(QDBusMessage)
    def onAllForNow(self, msg):
        logger.debug('Avahi emitted all signals for discovered peers')
        self.initialized.emit()


# Main Function
if __name__ == '__main__':

    logging.basicConfig(level=logging.DEBUG)

    # Create main app
    app = QApplication(sys.argv)

    # Create avahi discoverer
    d = Discoverer(app, '_workstation._tcp')

    # Execute the application and exit
    sys.exit(app.exec_())