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))

print(f"Info from zeroconf.get_service_info: {info!r}")

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))

addresses = [f"{addr}:{cast(int, info.port)}" for addr in info.parsed_addresses()]

print(f" Name: {name}")

print(f" Addresses: {', '.join(addresses)}")

print(f" Weight: {info.weight}, priority: {info.priority}")

print(f" Server: {info.server}")

if info.properties:

print(" Properties are:")

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)

print(f"\nBrowsing {ALL_SERVICES} service(s), press Ctrl-C to exit...\n")

kwargs = {

"handlers": [async_on_service_state_change],

"question_type": DNSQuestionType.QU,

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))

print(f"Info from zeroconf.get_service_info: {info!r}")

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))

addresses = [f"{addr}:{cast(int, info.port)}" for addr in info.parsed_scoped_addresses()]

print(f" Name: {name}")

print(f" Addresses: {', '.join(addresses)}")

print(f" Weight: {info.weight}, priority: {info.priority}")

print(f" Server: {info.server}")

if info.properties:

print(" Properties are:")

await AsyncZeroconfServiceTypes.async_find(aiozc=self.aiozc, ip_version=ip_version)

)

print("\nBrowsing %s service(s), press Ctrl-C to exit...\n" % services)

print(f"\nBrowsing {services} service(s), press Ctrl-C to exit...\n")

self.aiobrowser = AsyncServiceBrowser(

self.aiozc.zeroconf, services, handlers=[async_on_service_state_change]

)

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))

print(f"Info for {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))

addresses = [f"{addr}:{cast(int, info.port)}" for addr in info.parsed_addresses()]

print(f" Addresses: {', '.join(addresses)}")

print(f" Weight: {info.weight}, priority: {info.priority}")

print(f" Server: {info.server}")

if info.properties:

print(" Properties are:")

if state_change is ServiceStateChange.Added:

info = zeroconf.get_service_info(service_type, name)

print("Info from zeroconf.get_service_info: %r" % (info))

print(f"Info from zeroconf.get_service_info: {info!r}")

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))

addresses = [f"{addr}:{cast(int, info.port)}" for addr in info.parsed_scoped_addresses()]

print(f" Addresses: {', '.join(addresses)}")

print(f" Weight: {info.weight}, priority: {info.priority}")

print(f" Server: {info.server}")

if info.properties:

print(" Properties are:")

if args.find:

services = list(ZeroconfServiceTypes.find(zc=zeroconf))

print("\nBrowsing %d service(s), press Ctrl-C to exit...\n" % len(services))

print(f"\nBrowsing {len(services)} service(s), press Ctrl-C to exit...\n")

browser = ServiceBrowser(zeroconf, services, handlers=[on_service_state_change])

try:

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(f" Getting ZOE service: {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.")

ignore = [

"S101", # use of assert

"S104", # S104 Possible binding to all interfaces

"UP031", # UP031 use f-strings -- too many to fix right now

]

select = [

"B", # flake8-bugbear

self.get_class_(self.class_),

"-unique" if self.unique else "",

self.name,

"=%s" % cast(Any, other) if other is not None else "",

f"={cast(Any, other)}" if other is not None else "",

)

return self._num_authorities > 0

def __repr__(self) -> str:

return "<DNSIncoming:{%s}>" % ", ".join(

[

"id=%s" % self.id,

"flags=%s" % self.flags,

"truncated=%s" % self.truncated,

"n_q=%s" % self._num_questions,

"n_ans=%s" % self._num_answers,

"n_auth=%s" % self._num_authorities,

"n_add=%s" % self._num_additionals,

"questions=%s" % self._questions,

"answers=%s" % self.answers(),

]

return "<DNSIncoming:{}>".format(

", ".join(

[

f"id={self.id}",

f"flags={self.flags}",

f"truncated={self.truncated}",

f"n_q={self._num_questions}",

f"n_ans={self._num_answers}",

f"n_auth={self._num_authorities}",

f"n_add={self._num_additionals}",

f"questions={self._questions}",

f"answers={self.answers()}",

]

)

)

def _read_header(self) -> None:

self.allow_long = True

def __repr__(self) -> str:

return "<DNSOutgoing:{%s}>" % ", ".join(

[

"multicast=%s" % self.multicast,

"flags=%s" % self.flags,

"questions=%s" % self.questions,

"answers=%s" % self.answers,

"authorities=%s" % self.authorities,

"additionals=%s" % self.additionals,

]

return "<DNSOutgoing:{}>".format(

", ".join(

[

f"multicast={self.multicast}",

f"flags={self.flags}",

f"questions={self.questions}",

f"answers={self.answers}",

f"authorities={self.authorities}",

f"additionals={self.additionals}",

]

)

)

def add_question(self, record: DNSQuestion) -> None:

cast(int, adapter.index),

)

raise RuntimeError("No adapter found for IP address %s" % ip)

raise RuntimeError(f"No adapter found for IP address {ip}")

def interface_index_to_ip6_address(adapters: List[Any], index: int) -> Tuple[str, int, int]:

if isinstance(adapter_ip.ip, tuple):

return cast(Tuple[str, int, int], adapter_ip.ip)

raise RuntimeError("No adapter found for index %s" % index)

raise RuntimeError(f"No adapter found for index {index}")

def ip6_addresses_to_indexes(

result.extend(get_all_addresses())

if not result:

raise RuntimeError(

"No interfaces to listen on, check that any interfaces have IP version %s" % ip_version

f"No interfaces to listen on, check that any interfaces have IP version {ip_version}"

)

elif isinstance(choice, list):

# First, take IPv4 addresses.

result = [i for i in choice if isinstance(i, str) and ipaddress.ip_address(i).version == 4]

# Unlike IP_ADD_MEMBERSHIP, IPV6_JOIN_GROUP requires interface indexes.

result += ip6_addresses_to_indexes(choice)

else:

raise TypeError("choice must be a list or InterfaceChoice, got %r" % choice)

raise TypeError(f"choice must be a list or InterfaceChoice, got {choice!r}")

return result