How to use the zeroconf.ServiceBrowser function in zeroconf

To help you get started, we’ve selected a few zeroconf examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Ultimaker / Cura / plugins / UM3NetworkPrinting / NetworkPrinterOutputDevicePlugin.py View on Github external
def startDiscovery(self):
        self.stop()
        if self._browser:
            self._browser.cancel()
            self._browser = None
            self._old_printers = [printer_name for printer_name in self._printers]
            self._printers = {}
            self.printerListChanged.emit()
        # After network switching, one must make a new instance of Zeroconf
        # On windows, the instance creation is very fast (unnoticable). Other platforms?
        self._zero_conf = Zeroconf()
        self._browser = ServiceBrowser(self._zero_conf, u'_ultimaker._tcp.local.', [self._appendServiceChangedRequest])

        # Look for manual instances from preference
        for address in self._manual_instances:
            if address:
                self.addManualPrinter(address)
github home-assistant / home-assistant-cli / homeassistant_cli / config.py View on Github external
def _locate_ha() -> Optional[str]:

    _zeroconf = zeroconf.Zeroconf()
    listener = _ZeroconfListener()
    zeroconf.ServiceBrowser(_zeroconf, "_home-assistant._tcp.local.", listener)
    try:
        import time

        retries = 0
        while not listener.services and retries < 5:
            _LOGGING.info(
                "Trying to locate Home Assistant on local network..."
            )
            time.sleep(0.5)
            retries = retries + 1
    finally:
        _zeroconf.close()

    if listener.services:
        if len(listener.services) > 1:
            _LOGGING.warning(
github guirem / plugin-googlecast / resources / pychromecast / pychromecast / discovery.py View on Github external
Start discovering chromecasts on the network.

    This method will start discovering chromecasts on a separate thread. When
    a chromecast is discovered, the callback will be called with the
    discovered chromecast's zeroconf name. This is the dictionary key to find
    the chromecast metadata in listener.services.

    This method returns the CastListener object and the zeroconf ServiceBrowser
    object. The CastListener object will contain information for the discovered
    chromecasts. To stop discovery, call the stop_discovery method with the
    ServiceBrowser object.
    """
    listener = CastListener(add_callback, remove_callback)
    service_browser = False
    try:
        service_browser = zeroconf.ServiceBrowser(
            zeroconf.Zeroconf(), "_googlecast._tcp.local.", listener
        )
    except (
        zeroconf.BadTypeInNameException,
        NotImplementedError,
        OSError,
        socket.error,
        zeroconf.NonUniqueNameException,
    ):
        pass

    return listener, service_browser
github pklaus / ds1054z / ds1054z / discovery.py View on Github external
``zc_type``, and ``zc_info``.
    :rtype: list of dict
    """
    zc = Zeroconf()

    def ds1000z_filter(result):
        check_results = [
          re.match(b'DS1\d\d\dZ', result['zc_info'].properties[b'Model']),
          re.match(b'RIGOL TECHNOLOGIES', result['zc_info'].properties[b'Manufacturer']),
        ]
        if not all(check_results):
            return False
        return True

    listener = Listener(filter_func=ds1000z_filter, cast_service_info=DS1000ZServiceInfo)
    browser = ServiceBrowser(zc, '_scpi-raw._tcp.local.', listener=listener)

    start = clock()
    while True:
        # Because multithreading sucks.
        et = clock() - start # elapsed time
        if len(listener.results) and et >= if_any_return_after:
            break
        if et >= timeout:
            break
        time.sleep(0.005)

    zc.close()

    return listener.results
github masmu / pulseaudio-dlna / pulseaudio_dlna / plugins / chromecast / mdns.py View on Github external
def run(self, ttl=None):
        if self.host:
            self.zeroconf = zeroconf.Zeroconf(interfaces=[self.host])
        else:
            self.zeroconf = zeroconf.Zeroconf()
        zeroconf.ServiceBrowser(self.zeroconf, self.domain, MDNSHandler(self))

        if ttl:
            GObject.timeout_add(ttl * 1000, self.shutdown)

        self.__running = True
        self.__mainloop = GObject.MainLoop()
        context = self.__mainloop.get_context()
        try:
            while self.__running:
                if context.pending():
                    context.iteration(True)
                else:
                    time.sleep(0.01)
        except KeyboardInterrupt:
            pass
        self.zeroconf.close()
github janten / dpt-rp1-py / dptrp1 / dptrp1.py View on Github external
def find(self, id, timeout=30):
        from zeroconf import ServiceBrowser, Zeroconf

        if not self.quiet:
            print("Discovering Digital Paper for {} seconds…".format(timeout))
        sys.stdout.flush()
        self.id = id
        zc = Zeroconf()
        self.lock.acquire()
        ServiceBrowser(zc, "_digitalpaper._tcp.local.", self)
        wait = self.lock.acquire(timeout=timeout) or (self.addr is not None)
        zc.close()
        if not wait:
            print("Failed".format(timeout))
            return None
        else:
            if not self.quiet:
                print("Found digital paper at", self.addr)
                print("To skip the discovery process (and this message), call:")
                print()
                print(
                    "    {} --addr {} {}".format(
                        sys.argv[0], self.addr, " ".join(sys.argv[1:])
                    )
                )
                print()
github google / cloudprint_logocert / _zconf.py View on Github external
def __init__(self, logger, wifi_interfaces=[]):
    """Initialization requires a logger.

    Args:
      logger: initialized logger object.
      if_addr: string, interface address for Zeroconf, None means
               all interfaces.
    """
    self.logger = logger
    self.l = _Listener(logger)
    if not wifi_interfaces:
      self.z = Zeroconf()
    else:
      self.z = Zeroconf(wifi_interfaces)
    self.sb = ServiceBrowser(zc=self.z, type_='_privet._tcp.local.',
                             listener=self.l)
github google / cloudprint_logocert / _mdns.py View on Github external
proto: string, type of traffic to listen for.
    Returns:
      boolean, True = browser activated, False = errors detected.
    """
    protocols = {'http': '_http._tcp.local.',
                 'ipp': '_ipp._tcp.local.',
                 'mdns': '_mdns._udp.local.',
                 'printer': '_printer._tcp.local.',
                 'privet': '_privet._tcp.local.',
                }

    if proto not in protocols:
      self.logger.error('Error starting listener, %s protocol unkown', proto)
      return False

    self.sb = ServiceBrowser(self.zeroconf, protocols[proto], self.listener)
    self.logger.info('Browsing for %s services...', proto)
    return True