Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
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(
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
``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
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()
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()
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)
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