FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

chore: brand new async examples, fresh readme example and DNSText repr by bdraco · Pull Request #1873 · python-zeroconf/python-zeroconf · GitHub

Repository navigation

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (3) .rst  (1) All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
18 changes: 18 additions & 0 deletions README.rst
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,24 @@ Here's how to discover services:

.. code-block:: python

from zeroconf import ServiceBrowser, ServiceListener, Zeroconf


class EventLogger(ServiceListener):
def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:
info = zc.get_service_info(type_, name)
print(f"discovered {name}: {info}")

def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None:
print(f"lost {name}")

def update_service(self, zc: Zeroconf, type_: str, name: str) -> None:
print(f"refreshed {name}")


with Zeroconf() as zc:
ServiceBrowser(zc, "_http._tcp.local.", EventLogger())
input("browsing, press enter to stop\n")

.. note::

Expand Down
79 changes: 79 additions & 0 deletions examples/async_apple_scanner.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
#!/usr/bin/env python

"""Scan for Apple devices and dump the device info records they advertise."""

from __future__ import annotations

import argparse
import asyncio
import contextlib
import logging
from typing import Any

from zeroconf import DNSQuestionType, IPVersion, ServiceStateChange, Zeroconf
from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf

HAP_TYPE = "_hap._tcp.local."
DEVICE_INFO_TYPE = "_device-info._tcp.local."

log = logging.getLogger(__name__)

_background_tasks: set[asyncio.Task] = set()


def on_service_state_change(
zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
) -> None:
print(f"{state_change.name}: {name}")
if state_change is not ServiceStateChange.Added:
return
task = asyncio.create_task(dump_service_info(zeroconf, service_type, name))
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)


async def dump_service_info(zeroconf: Zeroconf, service_type: str, name: str) -> None:
info = AsyncServiceInfo(service_type, name)
if not await info.async_request(zeroconf, 3000, question_type=DNSQuestionType.QU):
print(f" {name}: no response")
return
show_info(info)
device_name = f"{info.server.rstrip('.')}.{DEVICE_INFO_TYPE}" if info.server else None
if device_name:
device_info = AsyncServiceInfo(DEVICE_INFO_TYPE, device_name)
if await device_info.async_request(zeroconf, 3000, question_type=DNSQuestionType.QU):
show_info(device_info)


def show_info(info: AsyncServiceInfo) -> None:
print(f" {info.name}")
print(f" reachable at: {', '.join(info.parsed_scoped_addresses())}")
print(f" host {info.server} port {info.port} (priority {info.priority}, weight {info.weight})")
for key, value in info.decoded_properties.items():
if not key:
continue
print(f" txt {key} = {value}")


async def main(args: Any) -> None:
ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All
aiozc = AsyncZeroconf(ip_version=ip_version)
browser = AsyncServiceBrowser(aiozc.zeroconf, HAP_TYPE, handlers=[on_service_state_change])
print(f"scanning for {HAP_TYPE}, press ctrl-c to exit")
try:
await asyncio.Event().wait()
finally:
await browser.async_cancel()
await aiozc.async_close()


if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--debug", action="store_true", help="enable debug logging")
parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
args = parser.parse_args()

logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)

with contextlib.suppress(KeyboardInterrupt):
asyncio.run(main(args))
64 changes: 64 additions & 0 deletions examples/async_service_info_request.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
#!/usr/bin/env python

"""Resolve every HTTP service currently on the network, on a fixed cadence."""

from __future__ import annotations

import argparse
import asyncio
import contextlib
import logging

from zeroconf import IPVersion, Zeroconf
from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf

HTTP_TYPE = "_http._tcp.local."
RESOLVE_INTERVAL = 5


async def resolve_all(zeroconf: Zeroconf, names: set[str]) -> None:
infos = [AsyncServiceInfo(HTTP_TYPE, name) for name in sorted(names)]
await asyncio.gather(*(info.async_request(zeroconf, 3000) for info in infos))
for info in infos:
print(f" {info.name}")
if not info.server:
print(" (unresolved)")
continue
print(f" reachable at: {', '.join(info.parsed_scoped_addresses())}")
print(f" host {info.server} port {info.port} (priority {info.priority}, weight {info.weight})")
for key, value in info.decoded_properties.items():
if not key:
continue
print(f" txt {key} = {value}")


async def main(ip_version: IPVersion) -> None:
seen: set[str] = set()
aiozc = AsyncZeroconf(ip_version=ip_version)
browser = AsyncServiceBrowser(
aiozc.zeroconf,
HTTP_TYPE,
handlers=[lambda zeroconf, service_type, name, state_change: seen.add(name)],
)
print(f"collecting {HTTP_TYPE} services, resolving every {RESOLVE_INTERVAL}s; ctrl-c to exit")
try:
while True:
await asyncio.sleep(RESOLVE_INTERVAL)
print(f"resolving {len(seen)} known services:")
await resolve_all(aiozc.zeroconf, seen)
finally:
await browser.async_cancel()
await aiozc.async_close()


if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--debug", action="store_true", help="enable debug logging")
parser.add_argument("--v6-only", action="store_true", help="use IPv6 only")
args = parser.parse_args()

logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO)
ip_version = IPVersion.V6Only if args.v6_only else IPVersion.All

with contextlib.suppress(KeyboardInterrupt):
asyncio.run(main(ip_version))
5 changes: 5 additions & 0 deletions src/zeroconf/_dns.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -541,6 +541,11 @@ def __hash__(self) -> int:
"""Hash to compare like DNSText."""
return self._hash

def __repr__(self) -> str:
if len(self.text) > 16:
return self.to_string(f"{len(self.text)} bytes")
return self.to_string(self.text)

def write(self, out: DNSOutgoing) -> None:
out.write_string(self.text)

Expand Down
Loading

Back | FazBrowse Home | New Git URL