From ce1127ddac3b419b594a7db78d4de5e89cde26eb Mon Sep 17 00:00:00 2001 From: jeff-tcs <160525204+jeff-tcs@users.noreply.github.com> Date: Thu, 7 Mar 2024 16:53:59 -0500 Subject: [PATCH] add zero conf examples --- async_apple_scanner.py | 119 +++++++++++++++++++++ async_browser.py | 106 +++++++++++++++++++ async_registration.py | 74 +++++++++++++ async_service_info_request.py | 101 ++++++++++++++++++ browser.py | 82 +++++++++++++++ example_node_implementation-jeff.py | 157 ++++++++++++++++++++++++++++ registration.py | 53 ++++++++++ resolver.py | 24 +++++ self_test.py | 48 +++++++++ 9 files changed, 764 insertions(+) create mode 100644 async_apple_scanner.py create mode 100644 async_browser.py create mode 100644 async_registration.py create mode 100644 async_service_info_request.py create mode 100755 browser.py create mode 100644 example_node_implementation-jeff.py create mode 100755 registration.py create mode 100755 resolver.py create mode 100755 self_test.py diff --git a/async_apple_scanner.py b/async_apple_scanner.py new file mode 100644 index 0000000..ff558f8 --- /dev/null +++ b/async_apple_scanner.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 + +""" Scan for apple devices. """ + +import argparse +import asyncio +import logging +from typing import Any, Optional, cast + +from zeroconf import DNSQuestionType, IPVersion, ServiceStateChange, Zeroconf +from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf + +HOMESHARING_SERVICE: str = "_appletv-v2._tcp.local." +DEVICE_SERVICE: str = "_touch-able._tcp.local." +MEDIAREMOTE_SERVICE: str = "_mediaremotetv._tcp.local." +AIRPLAY_SERVICE: str = "_airplay._tcp.local." +COMPANION_SERVICE: str = "_companion-link._tcp.local." +RAOP_SERVICE: str = "_raop._tcp.local." +AIRPORT_ADMIN_SERVICE: str = "_airport._tcp.local." +DEVICE_INFO_SERVICE: str = "_device-info._tcp.local." + +ALL_SERVICES = [ + HOMESHARING_SERVICE, + DEVICE_SERVICE, + MEDIAREMOTE_SERVICE, + AIRPLAY_SERVICE, + COMPANION_SERVICE, + RAOP_SERVICE, + AIRPORT_ADMIN_SERVICE, + DEVICE_INFO_SERVICE, +] + +log = logging.getLogger(__name__) + + +def async_on_service_state_change( + zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange +) -> None: + print(f"Service {name} of type {service_type} state changed: {state_change}") + if state_change is not ServiceStateChange.Added: + return + base_name = name[: -len(service_type) - 1] + device_name = f"{base_name}.{DEVICE_INFO_SERVICE}" + asyncio.ensure_future(_async_show_service_info(zeroconf, service_type, name)) + # Also probe for device info + asyncio.ensure_future(_async_show_service_info(zeroconf, DEVICE_INFO_SERVICE, device_name)) + + +async def _async_show_service_info(zeroconf: Zeroconf, service_type: str, name: str) -> None: + info = AsyncServiceInfo(service_type, name) + await info.async_request(zeroconf, 3000, question_type=DNSQuestionType.QU) + print("Info from zeroconf.get_service_info: %r" % (info)) + if info: + addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_addresses()] + print(" Name: %s" % name) + print(" Addresses: %s" % ", ".join(addresses)) + print(" Weight: %d, priority: %d" % (info.weight, info.priority)) + print(f" Server: {info.server}") + if info.properties: + print(" Properties are:") + for key, value in info.properties.items(): + print(f" {key!r}: {value!r}") + else: + print(" No properties") + else: + print(" No info") + print('\n') + + +class AsyncAppleScanner: + def __init__(self, args: Any) -> None: + self.args = args + self.aiobrowser: Optional[AsyncServiceBrowser] = None + self.aiozc: Optional[AsyncZeroconf] = None + + async def async_run(self) -> None: + self.aiozc = AsyncZeroconf(ip_version=ip_version) + await self.aiozc.zeroconf.async_wait_for_start() + print("\nBrowsing %s service(s), press Ctrl-C to exit...\n" % ALL_SERVICES) + kwargs = {'handlers': [async_on_service_state_change], 'question_type': DNSQuestionType.QU} + if self.args.target: + kwargs["addr"] = self.args.target + self.aiobrowser = AsyncServiceBrowser(self.aiozc.zeroconf, ALL_SERVICES, **kwargs) # type: ignore + while True: + await asyncio.sleep(1) + + async def async_close(self) -> None: + assert self.aiozc is not None + assert self.aiobrowser is not None + await self.aiobrowser.async_cancel() + await self.aiozc.async_close() + + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--target', help='Unicast target') + version_group.add_argument('--v6', action='store_true') + version_group.add_argument('--v6-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6: + ip_version = IPVersion.All + elif args.v6_only: + ip_version = IPVersion.V6Only + else: + ip_version = IPVersion.V4Only + + loop = asyncio.get_event_loop() + runner = AsyncAppleScanner(args) + try: + loop.run_until_complete(runner.async_run()) + except KeyboardInterrupt: + loop.run_until_complete(runner.async_close()) diff --git a/async_browser.py b/async_browser.py new file mode 100644 index 0000000..f7fb715 --- /dev/null +++ b/async_browser.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 + +""" Example of browsing for a service. + +The default is HTTP and HAP; use --find to search for all available services in the network +""" + +import argparse +import asyncio +import logging +from typing import Any, Optional, cast + +from zeroconf import IPVersion, ServiceStateChange, Zeroconf +from zeroconf.asyncio import ( + AsyncServiceBrowser, + AsyncServiceInfo, + AsyncZeroconf, + AsyncZeroconfServiceTypes, +) + + +def async_on_service_state_change( + zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange +) -> None: + print(f"Service {name} of type {service_type} state changed: {state_change}") + if state_change is not ServiceStateChange.Added: + return + asyncio.ensure_future(async_display_service_info(zeroconf, service_type, name)) + + +async def async_display_service_info(zeroconf: Zeroconf, service_type: str, name: str) -> None: + info = AsyncServiceInfo(service_type, name) + await info.async_request(zeroconf, 3000) + print("Info from zeroconf.get_service_info: %r" % (info)) + if info: + addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_scoped_addresses()] + print(" Name: %s" % name) + print(" Addresses: %s" % ", ".join(addresses)) + print(" Weight: %d, priority: %d" % (info.weight, info.priority)) + print(f" Server: {info.server}") + if info.properties: + print(" Properties are:") + for key, value in info.properties.items(): + print(f" {key!r}: {value!r}") + else: + print(" No properties") + else: + print(" No info") + print('\n') + + +class AsyncRunner: + def __init__(self, args: Any) -> None: + self.args = args + self.aiobrowser: Optional[AsyncServiceBrowser] = None + self.aiozc: Optional[AsyncZeroconf] = None + + async def async_run(self) -> None: + self.aiozc = AsyncZeroconf(ip_version=ip_version) + + services = ["_http._tcp.local.", "_hap._tcp.local."] + if self.args.find: + services = list( + await AsyncZeroconfServiceTypes.async_find(aiozc=self.aiozc, ip_version=ip_version) + ) + + print("\nBrowsing %s service(s), press Ctrl-C to exit...\n" % services) + self.aiobrowser = AsyncServiceBrowser( + self.aiozc.zeroconf, services, handlers=[async_on_service_state_change] + ) + while True: + await asyncio.sleep(1) + + async def async_close(self) -> None: + assert self.aiozc is not None + assert self.aiobrowser is not None + await self.aiobrowser.async_cancel() + await self.aiozc.async_close() + + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + parser.add_argument('--find', action='store_true', help='Browse all available services') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--v6', action='store_true') + version_group.add_argument('--v6-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6: + ip_version = IPVersion.All + elif args.v6_only: + ip_version = IPVersion.V6Only + else: + ip_version = IPVersion.V4Only + + loop = asyncio.get_event_loop() + runner = AsyncRunner(args) + try: + loop.run_until_complete(runner.async_run()) + except KeyboardInterrupt: + loop.run_until_complete(runner.async_close()) diff --git a/async_registration.py b/async_registration.py new file mode 100644 index 0000000..c3aab32 --- /dev/null +++ b/async_registration.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Example of announcing 250 services (in this case, a fake HTTP server).""" + +import argparse +import asyncio +import logging +import socket +from typing import List, Optional + +from zeroconf import IPVersion +from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf + + +class AsyncRunner: + def __init__(self, ip_version: IPVersion) -> None: + self.ip_version = ip_version + self.aiozc: Optional[AsyncZeroconf] = None + + async def register_services(self, infos: List[AsyncServiceInfo]) -> None: + self.aiozc = AsyncZeroconf(ip_version=self.ip_version) + tasks = [self.aiozc.async_register_service(info) for info in infos] + background_tasks = await asyncio.gather(*tasks) + await asyncio.gather(*background_tasks) + print("Finished registration, press Ctrl-C to exit...") + while True: + await asyncio.sleep(1) + + async def unregister_services(self, infos: List[AsyncServiceInfo]) -> None: + assert self.aiozc is not None + tasks = [self.aiozc.async_unregister_service(info) for info in infos] + background_tasks = await asyncio.gather(*tasks) + await asyncio.gather(*background_tasks) + await self.aiozc.async_close() + + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--v6', action='store_true') + version_group.add_argument('--v6-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6: + ip_version = IPVersion.All + elif args.v6_only: + ip_version = IPVersion.V6Only + else: + ip_version = IPVersion.V4Only + + infos = [] + for i in range(250): + infos.append( + AsyncServiceInfo( + "_http._tcp.local.", + f"Paul's Test Web Site {i}._http._tcp.local.", + addresses=[socket.inet_aton("127.0.0.1")], + port=80, + properties={'path': '/~paulsm/'}, + server=f"zcdemohost-{i}.local.", + ) + ) + + print("Registration of 250 services...") + loop = asyncio.get_event_loop() + runner = AsyncRunner(ip_version) + try: + loop.run_until_complete(runner.register_services(infos)) + except KeyboardInterrupt: + loop.run_until_complete(runner.unregister_services(infos)) diff --git a/async_service_info_request.py b/async_service_info_request.py new file mode 100644 index 0000000..5bb2476 --- /dev/null +++ b/async_service_info_request.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Example of perodic dump of homekit services. + +This example is useful when a user wants an ondemand +list of HomeKit devices on the network. + +""" + +import argparse +import asyncio +import logging +from typing import Any, List, Optional, cast + +from zeroconf import IPVersion, ServiceBrowser, ServiceStateChange, Zeroconf +from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf + +HAP_TYPE = "_hap._tcp.local." + + +async def async_watch_services(aiozc: AsyncZeroconf) -> None: + zeroconf = aiozc.zeroconf + while True: + await asyncio.sleep(5) + infos: List[AsyncServiceInfo] = [] + for name in zeroconf.cache.names(): + if not name.endswith(HAP_TYPE): + continue + infos.append(AsyncServiceInfo(HAP_TYPE, name)) + tasks = [info.async_request(aiozc.zeroconf, 3000) for info in infos] + await asyncio.gather(*tasks) + for info in infos: + print("Info for %s" % (info.name)) + if info: + addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_addresses()] + print(" Addresses: %s" % ", ".join(addresses)) + print(" Weight: %d, priority: %d" % (info.weight, info.priority)) + print(f" Server: {info.server}") + if info.properties: + print(" Properties are:") + for key, value in info.properties.items(): + print(f" {key!r}: {value!r}") + else: + print(" No properties") + else: + print(" No info") + print('\n') + + +class AsyncRunner: + def __init__(self, args: Any) -> None: + self.args = args + self.threaded_browser: Optional[ServiceBrowser] = None + self.aiozc: Optional[AsyncZeroconf] = None + + async def async_run(self) -> None: + self.aiozc = AsyncZeroconf(ip_version=ip_version) + assert self.aiozc is not None + + def on_service_state_change( + zeroconf: Zeroconf, service_type: str, state_change: ServiceStateChange, name: str + ) -> None: + """Dummy handler.""" + + self.threaded_browser = ServiceBrowser( + self.aiozc.zeroconf, [HAP_TYPE], handlers=[on_service_state_change] + ) + await async_watch_services(self.aiozc) + + async def async_close(self) -> None: + assert self.aiozc is not None + assert self.threaded_browser is not None + self.threaded_browser.cancel() + await self.aiozc.async_close() + + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--v6', action='store_true') + version_group.add_argument('--v6-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6: + ip_version = IPVersion.All + elif args.v6_only: + ip_version = IPVersion.V6Only + else: + ip_version = IPVersion.V4Only + + print(f"Services with {HAP_TYPE} will be shown every 5s, press Ctrl-C to exit...") + loop = asyncio.get_event_loop() + runner = AsyncRunner(args) + try: + loop.run_until_complete(runner.async_run()) + except KeyboardInterrupt: + loop.run_until_complete(runner.async_close()) diff --git a/browser.py b/browser.py new file mode 100755 index 0000000..237de01 --- /dev/null +++ b/browser.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 + +""" Example of browsing for a service. + +The default is HTTP and HAP; use --find to search for all available services in the network +""" + +import argparse +import logging +from time import sleep +from typing import cast + +from zeroconf import ( + IPVersion, + ServiceBrowser, + ServiceStateChange, + Zeroconf, + ZeroconfServiceTypes, +) + + +def on_service_state_change( + zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange +) -> None: + print(f"Service {name} of type {service_type} state changed: {state_change}") + + if state_change is ServiceStateChange.Added: + info = zeroconf.get_service_info(service_type, name) + print("Info from zeroconf.get_service_info: %r" % (info)) + + if info: + addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_scoped_addresses()] + print(" Addresses: %s" % ", ".join(addresses)) + print(" Weight: %d, priority: %d" % (info.weight, info.priority)) + print(f" Server: {info.server}") + if info.properties: + print(" Properties are:") + for key, value in info.properties.items(): + print(f" {key!r}: {value!r}") + else: + print(" No properties") + else: + print(" No info") + print('\n') + + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + parser.add_argument('--find', action='store_true', help='Browse all available services') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--v6-only', action='store_true') + version_group.add_argument('--v4-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6_only: + ip_version = IPVersion.V6Only + elif args.v4_only: + ip_version = IPVersion.V4Only + else: + ip_version = IPVersion.All + + zeroconf = Zeroconf(ip_version=ip_version) + + services = ["_http._tcp.local.", "_hap._tcp.local.", "_esphomelib._tcp.local.", "_airplay._tcp.local."] + if args.find: + services = list(ZeroconfServiceTypes.find(zc=zeroconf)) + + print("\nBrowsing %d service(s), press Ctrl-C to exit...\n" % len(services)) + browser = ServiceBrowser(zeroconf, services, handlers=[on_service_state_change]) + + try: + while True: + sleep(0.1) + except KeyboardInterrupt: + pass + finally: + zeroconf.close() diff --git a/example_node_implementation-jeff.py b/example_node_implementation-jeff.py new file mode 100644 index 0000000..39cd2bd --- /dev/null +++ b/example_node_implementation-jeff.py @@ -0,0 +1,157 @@ +''' +Demo of using the datagram service to send and receive a datagram + +Usage: +python3 example_node_implementation.py [host|host:port] + +Options: +host|host:port (optional) Set the address (or using a colon, + the address and port). Defaults to a hard-coded test + address and port. +''' +from openlcb.tcpsocket import TcpSocket + +from canbus.canphysicallayergridconnect import CanPhysicalLayerGridConnect +# from canbus.canframe import CanFrame +from canbus.canlink import CanLink +# from openlcb.controlframe import ControlFrame +from openlcb.nodeid import NodeID +from openlcb.datagramservice import ( + # DatagramWriteMemo, + # DatagramReadMemo, + DatagramService, +) +from openlcb.memoryservice import ( + # MemoryReadMemo, + # MemoryWriteMemo, + MemoryService, +) +from openlcb.localnodeprocessor import LocalNodeProcessor +from openlcb.pip import PIP +from openlcb.snip import SNIP +from openlcb.node import Node + +# specify connection information +host = "192.168.16.212" +port = 12021 +localNodeID = "05.01.01.01.03.01" +farNodeID = "09.00.99.03.00.35" + +# region same code as other examples + + +def usage(): + print(__doc__, file=sys.stderr) + + +if __name__ == "__main__": + # global host # only necessary if this is moved to a main/other function + import sys + if len(sys.argv) == 2: + host = sys.argv[1] + parts = host.split(":") + if len(parts) == 2: + host = parts[0] + try: + port = int(parts[1]) + except ValueError: + usage() + print("Error: Port {} is not an integer.".format(parts[1]), + file=sys.stderr) + sys.exit(1) + elif len(parts) > 2: + usage() + print("Error: blank, address or address:port format was expected.") + sys.exit(1) + elif len(sys.argv) > 2: + usage() + print("Error: blank, address or address:port format was expected.") + sys.exit(1) + +# endregion same code as other examples + +s = TcpSocket() +s.connect(host, port) + +print("RR, SR are raw socket interface receive and send;" + " RL, SL are link interface; RM, SM are message interface") + + +def sendToSocket(string): + print(" SR: {}".format(string)) + s.send(string) + + +def printFrame(frame): + print(" RL: {}".format(frame)) + + +canPhysicalLayerGridConnect = CanPhysicalLayerGridConnect(sendToSocket) +canPhysicalLayerGridConnect.registerFrameReceivedListener(printFrame) + + +def printMessage(message): + print("RM: {} from {}".format(message, message.source)) + + +canLink = CanLink(NodeID(localNodeID)) +canLink.linkPhysicalLayer(canPhysicalLayerGridConnect) +canLink.registerMessageReceivedListener(printMessage) + +datagramService = DatagramService(canLink) +canLink.registerMessageReceivedListener(datagramService.process) + + +def printDatagram(memo): + """create a call-back to print datagram contents when received + + Args: + memo (_type_): _description_ + + Returns: + bool: Always False (True would mean we sent a reply to the datagram, + but let the MemoryService do that). + """ + print("Datagram receive call back: {}".format(memo.data)) + return False + + +datagramService.registerDatagramReceivedListener(printDatagram) + +memoryService = MemoryService(datagramService) + + +# createcallbacks to get results of memory read +def memoryReadSuccess(memo): + print("successful memory read: {}".format(memo.data)) + + +def memoryReadFail(memo): + print("memory read failed: {}".format(memo.data)) + + +# create a node and connect it update +# This is a very minimal node, which just takes part in the low-level common +# protocols +localNode = Node( + NodeID(localNodeID), + SNIP("PythonOlcbNode", "example_node_implementation", + "0.1", "0.2", "User Name Here", "User Description Here"), + set([PIP.SIMPLE_NODE_IDENTIFICATION_PROTOCOL, PIP.DATAGRAM_PROTOCOL]) +) + +localNodeProcessor = LocalNodeProcessor(canLink, localNode) +canLink.registerMessageReceivedListener(localNodeProcessor.process) + +####################### + +# have the socket layer report up to bring the link layer up and get an alias +print(" SL : link up") +canPhysicalLayerGridConnect.physicalLayerUp() + +# process resulting activity +while True: + input = s.receive() + print(" RR: "+input) + # pass to link processor + canPhysicalLayerGridConnect.receiveString(input) diff --git a/registration.py b/registration.py new file mode 100755 index 0000000..65c2219 --- /dev/null +++ b/registration.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 + +""" Example of announcing a service (in this case, a fake HTTP server) """ + +import argparse +import logging +import socket +from time import sleep + +from zeroconf import IPVersion, ServiceInfo, Zeroconf + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + + parser = argparse.ArgumentParser() + parser.add_argument('--debug', action='store_true') + version_group = parser.add_mutually_exclusive_group() + version_group.add_argument('--v6', action='store_true') + version_group.add_argument('--v6-only', action='store_true') + args = parser.parse_args() + + if args.debug: + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + if args.v6: + ip_version = IPVersion.All + elif args.v6_only: + ip_version = IPVersion.V6Only + else: + ip_version = IPVersion.V4Only + + desc = {'path': '/~paulsm/'} + + info = ServiceInfo( + "_http._tcp.local.", + "Paul's Test Web Site._http._tcp.local.", + addresses=[socket.inet_aton("127.0.0.1")], + port=80, + properties=desc, + server="ash-2.local.", + ) + + zeroconf = Zeroconf(ip_version=ip_version) + print("Registration of a service, press Ctrl-C to exit...") + zeroconf.register_service(info) + try: + while True: + sleep(0.1) + except KeyboardInterrupt: + pass + finally: + print("Unregistering...") + zeroconf.unregister_service(info) + zeroconf.close() diff --git a/resolver.py b/resolver.py new file mode 100755 index 0000000..6a550fc --- /dev/null +++ b/resolver.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 + +""" Example of resolving a service with a known name """ + +import logging +import sys + +from zeroconf import Zeroconf + +TYPE = '_test._tcp.local.' +NAME = 'My Service Name' + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + if len(sys.argv) > 1: + assert sys.argv[1:] == ['--debug'] + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + + zeroconf = Zeroconf() + + try: + print(zeroconf.get_service_info(TYPE, NAME + '.' + TYPE)) + finally: + zeroconf.close() diff --git a/self_test.py b/self_test.py new file mode 100755 index 0000000..2178629 --- /dev/null +++ b/self_test.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 + +import logging +import socket +import sys + +from zeroconf import ServiceInfo, Zeroconf, __version__ + +if __name__ == '__main__': + logging.basicConfig(level=logging.DEBUG) + if len(sys.argv) > 1: + assert sys.argv[1:] == ['--debug'] + logging.getLogger('zeroconf').setLevel(logging.DEBUG) + + # Test a few module features, including service registration, service + # query (for Zoe), and service unregistration. + print(f"Multicast DNS Service Discovery for Python, version {__version__}") + r = Zeroconf() + print("1. Testing registration of a service...") + desc = {'version': '0.10', 'a': 'test value', 'b': 'another value'} + addresses = [socket.inet_aton("127.0.0.1")] + expected = {'127.0.0.1'} + if socket.has_ipv6: + addresses.append(socket.inet_pton(socket.AF_INET6, '::1')) + expected.add('::1') + info = ServiceInfo( + "_http._tcp.local.", + "My Service Name._http._tcp.local.", + addresses=addresses, + port=1234, + properties=desc, + ) + print(" Registering service...") + r.register_service(info) + print(" Registration done.") + print("2. Testing query of service information...") + print(" Getting ZOE service: %s" % (r.get_service_info("_http._tcp.local.", "ZOE._http._tcp.local."))) + print(" Query done.") + print("3. Testing query of own service...") + queried_info = r.get_service_info("_http._tcp.local.", "My Service Name._http._tcp.local.") + assert queried_info + assert set(queried_info.parsed_addresses()) == expected + print(f" Getting self: {queried_info}") + print(" Query done.") + print("4. Testing unregister of service information...") + r.unregister_service(info) + print(" Unregister done.") + r.close()