File: discovery.py

package info (click to toggle)
pychromecast 2.4.0-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 332 kB
  • sloc: python: 1,751; makefile: 6; sh: 3
file content (130 lines) | stat: -rw-r--r-- 4,051 bytes parent folder | download
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
"""Discovers Chromecasts on the network using mDNS/zeroconf."""
import socket
from uuid import UUID

import 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 to UTF-8."""
            value = service.properties.get(key.encode('utf-8'))

            if value is None or isinstance(value, str):
                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)
    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


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
    browser = False
    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
    except Exception:  # pylint: disable=broad-except
        raise
    finally:
        if browser is not False:
            stop_discovery(browser)