Source code for mdns_beacon.beacon

"""Beacon module."""
import logging
import time
from ipaddress import IPv4Address, IPv6Address, ip_address
from typing import Any, Dict, List, Optional, Union

from slugify import slugify
from typing_extensions import Literal
from zeroconf import ServiceInfo

from .base import BaseBeacon

logger = logging.getLogger(__name__)

PROTOCOL = Literal["tcp", "udp"]


[docs]class Beacon(BaseBeacon): """mDNS Beacon. Attributes: aliases: Service alias name list. addresses: IP addresses that the service runs on. port: Port that the service runs on. type_: Service type. protocol: Service protocol. ttl: TTL used for the announce of the service. weight: Weight of the service. priority: Priority of the service. properties: Dict of properties (or a bytes object with the content of the `text` field). delay_startup: Amount of time to wait before trying to start the zeroconf service (in seconds). *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. """ _SLUG_REGEX_PATTERN = r"[^-a-z0-9_.]+" _SLUG_SEPARATOR = "-" _services: Optional[List[ServiceInfo]] = None def __init__( self, aliases: Optional[List[str]] = None, addresses: Optional[List[Union[IPv4Address, IPv6Address]]] = None, port: int = 80, type_: str = "http", protocol: PROTOCOL = "tcp", ttl: int = 60, weight: int = 0, priority: int = 0, properties: Optional[Union[bytes, Dict[str, Any]]] = None, delay_startup: int = 0, *args: Any, **kwargs: Any, ) -> None: """Init a mDNS Beacon instance. Args: aliases: Service alias name list. addresses: IP addresses that the service runs on. port: Port that the service runs on. type_: Service type. protocol: Service protocol. ttl: TTL used for the announce of the service. weight: Weight of the service. priority: Priority of the service. properties: Dict of properties (or a bytes object with the content of the `text` field) of the service. delay_startup: Amount of time to wait before trying to start the zeroconf service (in seconds). *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. """ super().__init__(*args, **kwargs) self.aliases = set(aliases or []) self.addresses = set(addresses or [ip_address("127.0.0.1")]) self.port = port self.type_ = type_ self.protocol = protocol self.ttl = ttl self.weight = weight self.priority = priority self.properties = properties or b"" self.delay_startup = delay_startup def _build_service_host(self, name: str) -> str: """Build service host for a given name. Args: name: Service name. Returns: Fully qualified service host name. """ slug = slugify( name, separator=self._SLUG_SEPARATOR, regex_pattern=self._SLUG_REGEX_PATTERN ) return f"{slug}.local." def _build_service_name(self, name: str) -> str: """Build service name for a given name. Args: name: Service name. Returns: Fully qualified service name. """ return f"{name}.{self.service_type}" @property def service_type(self) -> str: """Beacon service type.""" return f"_{self.type_}._{self.protocol}.local." @property def services(self) -> List[ServiceInfo]: """Services to register on the local network.""" if not self._services: self._services = [ ServiceInfo( type_=self.service_type, name=self._build_service_name(alias), parsed_addresses=[str(addr) for addr in self.addresses], port=self.port, host_ttl=self.ttl, weight=self.weight, priority=self.priority, properties=self.properties, server=self._build_service_host(alias), ) for alias in self.aliases ] return self._services
[docs] def stop(self) -> None: """Stop Beacon. Unregister all the announced services. """ logger.info( "Unregistering %(services_len)s services", {"services_len": len(self.services)} ) for service in self.services: logger.debug("Unregistering %(service_name)s", {"service_name": service.name}) self.zeroconf.unregister_service(service) super().stop()
def _wait(self) -> None: """Wait before trying to start the zeroconf service.""" time.sleep(self.delay_startup) def _execute(self) -> None: """Register aliases on the local network.""" self._wait() logger.info("Registering %(services_len)s services", {"services_len": len(self.services)}) for service in self.services: logger.debug("Registering %(service_name)s", {"service_name": service.name}) self.zeroconf.register_service(service)