/usr/lib/python3/dist-packages/pychromecast/discovery.py is in python3-pychromecast 0.8.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Discovers Chromecasts on the network using mDNS/zeroconf."""
from uuid import UUID
import six
from zeroconf import ServiceBrowser, Zeroconf
DISCOVER_TIMEOUT = 5
class CastListener(object):
"""Zeroconf Cast Services collection."""
def __init__(self, callback=None):
self.services = {}
self.callback = callback
@property
def count(self):
"""Number of discovered cast services."""
return len(self.services)
@property
def devices(self):
"""List of tuples (ip, host) for each discovered device."""
return list(self.services.values())
# pylint: disable=unused-argument
def remove_service(self, zconf, typ, name):
""" Remove a service from the collection. """
self.services.pop(name, None)
def add_service(self, zconf, typ, name):
""" Add a service to the collection. """
service = None
tries = 0
while service is None and tries < 4:
try:
service = zconf.get_service_info(typ, name)
except IOError:
# If the zerconf fails to receive the necesarry data we abort
# adding the service
break
tries += 1
if not service:
return
def get_value(key):
"""Retrieve value and decode for Python 2/3."""
value = service.properties.get(key.encode('utf-8'))
if value is None or isinstance(value, six.text_type):
return value
return value.decode('utf-8')
ips = zconf.cache.entries_with_name(service.server.lower())
host = repr(ips[0]) if ips else service.server
model_name = get_value('md')
uuid = get_value('id')
friendly_name = get_value('fn')
if uuid:
uuid = UUID(uuid)
self.services[name] = (host, service.port, uuid, model_name,
friendly_name)
if self.callback:
self.callback(name)
def start_discovery(callback=None):
"""
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(callback)
return listener, \
ServiceBrowser(Zeroconf(), "_googlecast._tcp.local.", listener)
def stop_discovery(browser):
"""Stop the chromecast discovery thread."""
browser.zc.close()
def discover_chromecasts(max_devices=None, timeout=DISCOVER_TIMEOUT):
""" Discover chromecasts on the network. """
from threading import Event
try:
# pylint: disable=unused-argument
def callback(name):
"""Called when zeroconf has discovered a new chromecast."""
if max_devices is not None and listener.count >= max_devices:
discover_complete.set()
discover_complete = Event()
listener, browser = start_discovery(callback)
# Wait for the timeout or the maximum number of devices
discover_complete.wait(timeout)
return listener.devices
finally:
stop_discovery(browser)
|