From 92c3fae892dd88fb83fe9ee06917b7fec23d9595 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 12 Jun 2021 16:08:09 -1000 Subject: [PATCH] Move ZeroconfServiceTypes to zeroconf.services.types --- zeroconf/__init__.py | 60 +--------------------------- zeroconf/services/types.py | 82 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 59 deletions(-) create mode 100644 zeroconf/services/types.py diff --git a/zeroconf/__init__.py b/zeroconf/__init__.py index 625f7f9d..8242c4e1 100644 --- a/zeroconf/__init__.py +++ b/zeroconf/__init__.py @@ -21,9 +21,6 @@ """ import sys -import time -from typing import Optional, Union -from typing import Set, Tuple # noqa # used in type hints from .const import ( # noqa # import needed for backwards compat _BROWSER_BACKOFF_LIMIT, @@ -112,6 +109,7 @@ ServiceStateChange, ) from .services.registry import ServiceRegistry # noqa # import needed for backwards compat +from .services.types import ZeroconfServiceTypes # noqa # import needed for backwards compat from .utils.name import service_type_name # noqa # import needed for backwards compat from .utils.net import ( # noqa # import needed for backwards compat add_multicast_member, @@ -155,59 +153,3 @@ If you need support for Python 3.5 please use version 0.28.0 ''' ) - - -# implementation classes - - -class ZeroconfServiceTypes(ServiceListener): - """ - Return all of the advertised services on any local networks - """ - - def __init__(self) -> None: - """Keep track of found services in a set.""" - self.found_services = set() # type: Set[str] - - def add_service(self, zc: 'Zeroconf', type_: str, name: str) -> None: - """Service added.""" - self.found_services.add(name) - - def update_service(self, zc: 'Zeroconf', type_: str, name: str) -> None: - """Service updated.""" - - def remove_service(self, zc: 'Zeroconf', type_: str, name: str) -> None: - """Service removed.""" - - @classmethod - def find( - cls, - zc: Optional['Zeroconf'] = None, - timeout: Union[int, float] = 5, - interfaces: InterfacesType = InterfaceChoice.All, - ip_version: Optional[IPVersion] = None, - ) -> Tuple[str, ...]: - """ - Return all of the advertised services on any local networks. - - :param zc: Zeroconf() instance. Pass in if already have an - instance running or if non-default interfaces are needed - :param timeout: seconds to wait for any responses - :param interfaces: interfaces to listen on. - :param ip_version: IP protocol version to use. - :return: tuple of service type strings - """ - local_zc = zc or Zeroconf(interfaces=interfaces, ip_version=ip_version) - listener = cls() - browser = ServiceBrowser(local_zc, _SERVICE_TYPE_ENUMERATION_NAME, listener=listener) - - # wait for responses - time.sleep(timeout) - - browser.cancel() - - # close down anything we opened - if zc is None: - local_zc.close() - - return tuple(sorted(listener.found_services)) diff --git a/zeroconf/services/types.py b/zeroconf/services/types.py new file mode 100644 index 00000000..e27defff --- /dev/null +++ b/zeroconf/services/types.py @@ -0,0 +1,82 @@ +""" Multicast DNS Service Discovery for Python, v0.14-wmcbrine + Copyright 2003 Paul Scott-Murphy, 2014 William McBrine + + This module provides a framework for the use of DNS Service Discovery + using IP multicast. + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License as published by the Free Software Foundation; either + version 2.1 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public + License along with this library; if not, write to the Free Software + Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 + USA +""" + +import time +from typing import Optional, Set, Tuple, Union + +from ..const import _SERVICE_TYPE_ENUMERATION_NAME +from ..core import Zeroconf +from ..services import ServiceBrowser, ServiceListener +from ..utils.net import IPVersion, InterfaceChoice, InterfacesType + + +class ZeroconfServiceTypes(ServiceListener): + """ + Return all of the advertised services on any local networks + """ + + def __init__(self) -> None: + """Keep track of found services in a set.""" + self.found_services: Set[str] = set() + + def add_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Service added.""" + self.found_services.add(name) + + def update_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Service updated.""" + + def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None: + """Service removed.""" + + @classmethod + def find( + cls, + zc: Optional[Zeroconf] = None, + timeout: Union[int, float] = 5, + interfaces: InterfacesType = InterfaceChoice.All, + ip_version: Optional[IPVersion] = None, + ) -> Tuple[str, ...]: + """ + Return all of the advertised services on any local networks. + + :param zc: Zeroconf() instance. Pass in if already have an + instance running or if non-default interfaces are needed + :param timeout: seconds to wait for any responses + :param interfaces: interfaces to listen on. + :param ip_version: IP protocol version to use. + :return: tuple of service type strings + """ + local_zc = zc or Zeroconf(interfaces=interfaces, ip_version=ip_version) + listener = cls() + browser = ServiceBrowser(local_zc, _SERVICE_TYPE_ENUMERATION_NAME, listener=listener) + + # wait for responses + time.sleep(timeout) + + browser.cancel() + + # close down anything we opened + if zc is None: + local_zc.close() + + return tuple(sorted(listener.found_services))