Source code for mdns_beacon.base
"""Base mDNS Beacon module."""
import asyncio
import logging
from abc import ABC, abstractmethod
from typing import Optional
from zeroconf import IPVersion, Zeroconf
logger = logging.getLogger(__name__)
[docs]class BaseBeacon(ABC):
"""mDNS Beacon base class.
Note:
Derived beacons must override the `_execute` method.
Attributes:
ip_version: IP protocol version to use.
"""
_zeroconf: Optional[Zeroconf] = None
def __init__(self, ip_version: Optional[IPVersion] = None) -> None:
"""Init a mDNS Beacon instance.
Args:
ip_version: IP protocol version to use.
"""
self.ip_version = ip_version
@property
def zeroconf(self) -> Zeroconf:
"""Zeroconf instance."""
if not self._zeroconf:
self._zeroconf = Zeroconf(ip_version=self.ip_version)
return self._zeroconf
[docs] def stop(self) -> None:
"""Stop Beacon.
Ends zeroconf background threads, and prevent zeroconf instance from
servicing further queries.
"""
logger.debug("Stoping zeroconf")
self.zeroconf.close()
self._zeroconf = None
@abstractmethod
def _execute(self) -> None:
"""Execute Beacon work.
Method that derived beacons must override.
"""
[docs] def run_forever(self) -> None:
"""Run beacon forever."""
self._execute()
logger.debug("Starting forever loop")
loop = asyncio.get_event_loop()
try:
loop.run_forever()
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()