Psion-fanatic/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Added Zeroconf integration [closes #105]

Browse source
This commit is contained in:
Fabio Manganiello 2020-08-14 00:34:13 +02:00
parent bc3c01254e
commit 7e8bef40cd
19 changed files with 261 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/source/backends.rst
View file

@ -11,6 +11,11 @@ Backends
platypush/backend/assistant.rst
platypush/backend/assistant.google.rst
platypush/backend/assistant.snowboy.rst
platypush/backend/bluetooth.rst
platypush/backend/bluetooth.fileserver.rst
platypush/backend/bluetooth.pushserver.rst
platypush/backend/bluetooth.scanner.rst
platypush/backend/bluetooth.scanner.ble.rst
platypush/backend/button.flic.rst
platypush/backend/camera.pi.rst
platypush/backend/chat.telegram.rst
@ -48,6 +53,7 @@ Backends
platypush/backend/sensor.distance.vl53l1x.rst
platypush/backend/sensor.envirophat.rst
platypush/backend/sensor.ir.zeroborg.rst
platypush/backend/sensor.leap.rst
platypush/backend/sensor.ltr559.rst
platypush/backend/sensor.mcp3008.rst
platypush/backend/sensor.motion.pwm3901.rst

1
docs/source/conf.py
View file

@ -253,6 +253,7 @@ autodoc_mock_imports = ['googlesamples.assistant.grpc.audio_helpers',
'samsungtvws',
'paramiko',
'luma',
'zeroconf',
]
sys.path.insert(0, os.path.abspath('../..'))

1
docs/source/events.rst
View file

@ -61,5 +61,6 @@ Events
platypush/events/web.widget.rst
platypush/events/wiimote.rst
platypush/events/zeroborg.rst
platypush/events/zeroconf.rst
platypush/events/zigbee.mqtt.rst
platypush/events/zwave.rst

5
docs/source/platypush/events/zeroconf.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.message.event.zeroconf``
====================================
.. automodule:: platypush.message.event.zeroconf
:members:

5
docs/source/platypush/plugins/zeroconf.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.plugins.zeroconf``
==============================
.. automodule:: platypush.plugins.zeroconf
:members:

4
docs/source/plugins.rst
View file

@ -8,6 +8,7 @@ Plugins
platypush/plugins/adafruit.io.rst
platypush/plugins/alarm.rst
platypush/plugins/arduino.rst
platypush/plugins/assistant.rst
platypush/plugins/assistant.echo.rst
platypush/plugins/assistant.google.rst
@ -85,6 +86,7 @@ Plugins
platypush/plugins/music.mpd.rst
platypush/plugins/music.snapcast.rst
platypush/plugins/nmap.rst
platypush/plugins/otp.rst
platypush/plugins/pihole.rst
platypush/plugins/ping.rst
platypush/plugins/printer.cups.rst
@ -97,6 +99,7 @@ Plugins
platypush/plugins/sound.rst
platypush/plugins/ssh.rst
platypush/plugins/stt.rst
platypush/plugins/stt.deepspeech.rst
platypush/plugins/stt.picovoice.hotword.rst
platypush/plugins/stt.picovoice.speech.rst
platypush/plugins/switch.rst
@ -122,5 +125,6 @@ Plugins
platypush/plugins/weather.darksky.rst
platypush/plugins/websocket.rst
platypush/plugins/wiimote.rst
platypush/plugins/zeroconf.rst
platypush/plugins/zigbee.mqtt.rst
platypush/plugins/zwave.rst

64
platypush/backend/__init__.py
View file

@ -4,6 +4,8 @@
"""
import logging
import re
import socket
import threading
import time
@ -13,9 +15,11 @@ from typing import Optional
from platypush.bus import Bus
from platypush.config import Config
from platypush.context import get_backend
from platypush.message.event.zeroconf import ZeroconfServiceAddedEvent, ZeroconfServiceRemovedEvent
from platypush.utils import set_timeout, clear_timeout, \
get_redis_queue_name_by_message, set_thread_name
from platypush import __version__
from platypush.event import EventGenerator
from platypush.message import Message
from platypush.message.event import Event, StopEvent
@ -62,6 +66,8 @@ class Backend(Thread, EventGenerator):
self._stop_event = threading.Event()
self._kwargs = kwargs
self.logger = logging.getLogger(self.__class__.__name__)
self.zeroconf = None
self.zeroconf_info = None
# Internal-only, we set the request context on a backend if that
# backend is intended to react for a response to a specific request
@ -257,7 +263,7 @@ class Backend(Thread, EventGenerator):
def on_stop(self):
""" Callback invoked when the process stops """
pass
self.unregister_service()
def stop(self):
""" Stops the backend thread by sending a STOP event on its bus """
@ -304,5 +310,61 @@ class Backend(Thread, EventGenerator):
except Exception as e:
self.logger.error('Error while processing response to {}: {}'.format(msg, str(e)))
@staticmethod
def _get_ip() -> str:
"""
Get the IP address of the machine.
"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(('8.8.8.8', 80))
addr = s.getsockname()[0]
s.close()
return addr
def register_service(self, port: Optional[int] = None, name: Optional[str] = None, udp: bool = False):
"""
Initialize the Zeroconf service configuration for this backend.
"""
try:
from zeroconf import ServiceInfo, Zeroconf
except ImportError:
self.logger.warning('zeroconf package not available, service discovery will be disabled.')
return
self.zeroconf = Zeroconf()
srv_desc = {
'name': 'Platypush',
'vendor': 'Platypush',
'version': __version__,
}
name = name or re.sub(r'Backend$', '', self.__class__.__name__).lower()
srv_type = '_platypush-{name}._{proto}.local.'.format(name=name, proto='udp' if udp else 'tcp')
srv_name = '{host}.{type}'.format(host=self.device_id, type=srv_type)
if port:
srv_port = port
else:
srv_port = self.port if hasattr(self, 'port') else None
self.zeroconf_info = ServiceInfo(srv_type, srv_name, socket.inet_aton(self._get_ip()),
srv_port, 0, 0, srv_desc)
self.zeroconf.register_service(self.zeroconf_info)
self.bus.post(ZeroconfServiceAddedEvent(service_type=srv_type, service_name=srv_name))
def unregister_service(self):
"""
Unregister the Zeroconf service configuration if available.
"""
if self.zeroconf and self.zeroconf_info:
self.zeroconf.unregister_service(self.zeroconf_info)
self.zeroconf.close()
self.bus.post(ZeroconfServiceRemovedEvent(service_type=self.zeroconf_info.type,
service_name=self.zeroconf_info.name))
self.zeroconf_info = None
self.zeroconf = None
# vim:sw=4:ts=4:et:

1
platypush/backend/assistant/snowboy/__init__.py
View file

@ -165,6 +165,7 @@ class AssistantSnowboyBackend(AssistantBackend):
return callback
def on_stop(self):
super().on_stop()
if self.detector:
self.detector.terminate()
self.detector = None

2
platypush/backend/http/__init__.py
View file

@ -198,6 +198,7 @@ class HttpBackend(Backend):
def on_stop(self):
""" On backend stop """
super().on_stop()
self.logger.info('Received STOP event on HttpBackend')
if self.server_proc:
@ -312,6 +313,7 @@ class HttpBackend(Backend):
def run(self):
super().run()
self.register_service(port=self.port)
if not self.disable_websocket:
self.logger.info('Initializing websocket interface')

1
platypush/backend/kafka/__init__.py
View file

@ -68,6 +68,7 @@ class KafkaBackend(Backend):
server=self.server)
def on_stop(self):
super().on_stop()
try:
if self.producer:
self.producer.flush()

2
platypush/backend/nodered/__init__.py
View file

@ -30,12 +30,14 @@ class NoderedBackend(Backend):
self._server = None
def on_stop(self):
super().on_stop()
if self._server:
self._server.terminate()
self._server = None
def run(self):
super().run()
self.register_service(port=self.port, name='node')
self._server = subprocess.Popen([sys.executable, '-m', 'pynodered.server',
'--port', str(self.port), self._runner_path])

1
platypush/backend/pushbullet/__init__.py
View file

@ -126,6 +126,7 @@ class PushbulletBackend(Backend):
self.listener.close()
def on_stop(self):
super().on_stop()
return self.close()
def run(self):

1
platypush/backend/sensor/__init__.py
View file

@ -166,6 +166,7 @@ class SensorBackend(Backend):
return ret
def on_stop(self):
super().on_stop()
if not self.plugin:
return

1
platypush/backend/tcp.py
View file

@ -94,6 +94,7 @@ class TcpBackend(Backend):
def run(self):
super().run()
self.register_service(port=self.port)
serv_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serv_sock.bind((self.bind_address, self.port))

20
platypush/backend/websocket.py
View file

@ -1,7 +1,4 @@
import asyncio
import os
import ssl
import threading
import websockets
from platypush.backend import Backend
@ -66,7 +63,7 @@ class WebsocketBackend(Backend):
ssl_capath=ssl_capath) \
if ssl_cert else None
def send_message(self, msg):
def send_message(self, msg, **kwargs):
websocket = get_plugin('websocket')
websocket_args = {}
@ -78,7 +75,6 @@ class WebsocketBackend(Backend):
websocket.send(url=url, msg=msg, **websocket_args)
def notify_web_clients(self, event):
""" Notify all the connected web clients (over websocket) of a new event """
async def send_event(websocket):
@ -90,21 +86,22 @@ class WebsocketBackend(Backend):
loop = get_or_create_event_loop()
active_websockets = self.active_websockets.copy()
for websocket in active_websockets:
for ws in active_websockets:
try:
loop.run_until_complete(send_event(websocket))
loop.run_until_complete(send_event(ws))
except websockets.exceptions.ConnectionClosed:
self.logger.info('Client connection lost')
self.active_websockets.remove(websocket)
self.active_websockets.remove(ws)
def run(self):
super().run()
self.register_service(port=self.port, name='ws')
# noinspection PyUnusedLocal
async def serve_client(websocket, path):
self.active_websockets.add(websocket)
self.logger.debug('New websocket connection from {}'.
format(websocket.remote_address[0]))
format(websocket.remote_address[0]))
try:
while True:
@ -116,7 +113,7 @@ class WebsocketBackend(Backend):
msg = Message.build(msg)
self.logger.info('Received message from {}: {}'.
format(websocket.remote_address[0], msg))
format(websocket.remote_address[0], msg))
self.on_message(msg)
@ -154,4 +151,3 @@ class WebsocketBackend(Backend):
# vim:sw=4:ts=4:et:

48
platypush/message/event/zeroconf.py Normal file
View file

@ -0,0 +1,48 @@
import enum
from platypush.message.event import Event
class ZeroconfEventType(enum.Enum):
ADD = 'add'
UPDATE = 'update'
REMOVE = 'remove'
class ZeroconfEvent(Event):
def __init__(self, service_event: ZeroconfEventType, service_type: str, service_name: str, *args, **kwargs):
super().__init__(*args, service_event=service_event.value, service_type=service_type,
service_name=service_name, **kwargs)
self.service_type = service_type
self.service_name = service_name
class ZeroconfServiceAddedEvent(ZeroconfEvent):
"""
Event triggered when a service is added or discovered.
"""
def __init__(self, *args, **kwargs):
kwargs['service_event'] = ZeroconfEventType.ADD
super().__init__(*args, **kwargs)
class ZeroconfServiceUpdatedEvent(ZeroconfEvent):
"""
Event triggered when a service is updated.
"""
def __init__(self, *args, **kwargs):
kwargs['service_event'] = ZeroconfEventType.UPDATE
super().__init__(*args, **kwargs)
class ZeroconfServiceRemovedEvent(ZeroconfEvent):
"""
Event triggered when a service is removed.
"""
def __init__(self, *args, **kwargs):
kwargs['service_event'] = ZeroconfEventType.REMOVE
super().__init__(*args, **kwargs)
# vim:sw=4:ts=4:et:

107
platypush/plugins/zeroconf.py Normal file
View file

@ -0,0 +1,107 @@
import queue
import time
from typing import List, Dict
import zeroconf
from zeroconf import Zeroconf, ServiceBrowser
from platypush.context import get_bus
from platypush.message.event.zeroconf import ZeroconfServiceAddedEvent, ZeroconfServiceRemovedEvent, \
ZeroconfServiceUpdatedEvent, ZeroconfEvent
from platypush.plugins import Plugin, action
class ZeroconfListener(zeroconf.ServiceListener):
def __init__(self, evt_queue: queue.Queue):
super().__init__()
self.evt_queue = evt_queue
# noinspection PyUnusedLocal
def add_service(self, zc: Zeroconf, type_: str, name: str):
self.evt_queue.put(ZeroconfServiceAddedEvent(service_type=type_, service_name=name))
# noinspection PyUnusedLocal
def remove_service(self, zc: Zeroconf, type_: str, name: str):
self.evt_queue.put(ZeroconfServiceRemovedEvent(service_type=type_, service_name=name))
# noinspection PyUnusedLocal
def update_service(self, zc: Zeroconf, type_: str, name: str):
self.evt_queue.put(ZeroconfServiceUpdatedEvent(service_type=type_, service_name=name))
class ZeroconfPlugin(Plugin):
"""
Plugin for Zeroconf services discovery.
Requires:
* **zeroconf** (``pip install zeroconf``)
"""
@action
def get_services(self, timeout: int = 5) -> List[str]:
"""
Get the full list of services found on the network.
:param timeout: Discovery timeout in seconds (default: 5).
:return: List of the services as strings.
"""
return list(zeroconf.ZeroconfServiceTypes.find(timeout=timeout))
@action
def discover_service(self, service: str, timeout: int = 5) -> Dict[str, List[str]]:
"""
Find all the services matching the specified type.
:param service: Service type (e.g. ``_http._tcp.local.``).
:param timeout: Browser timeout in seconds (default: 5).
:return: A ``service_type -> [service_names]`` mapping. Example::
{
"_platypush-http._tcp.local.": [
"host1._platypush-http._tcp.local.",
"host2._platypush-http._tcp.local."
],
...
}
"""
evt_queue = queue.Queue()
zc = Zeroconf()
listener = ZeroconfListener(evt_queue=evt_queue)
browser = ServiceBrowser(zc, service, listener)
discovery_start = time.time()
services = {}
try:
while time.time() - discovery_start < timeout:
to = discovery_start + timeout - time.time()
try:
evt: ZeroconfEvent = evt_queue.get(block=True, timeout=to)
if isinstance(evt, ZeroconfServiceAddedEvent) or isinstance(evt, ZeroconfServiceUpdatedEvent):
if evt.service_type not in services:
services[evt.service_type] = set()
services[evt.service_type].add(evt.service_name)
elif isinstance(evt, ZeroconfServiceRemovedEvent):
if evt.service_type in services:
if evt.service_name in services[evt.service_type]:
services[evt.service_type].remove(evt.service_name)
if not services[evt.service_type]:
del services[evt.service_type]
get_bus().post(evt)
except queue.Empty:
if not services:
self.logger.warning('No such service discovered: {}'.format(service))
finally:
browser.cancel()
zc.close()
return {
type_: list(names)
for type_, names in services.items()
}
# vim:sw=4:ts=4:et:

3
requirements.txt
View file

@ -35,6 +35,9 @@ sqlalchemy
# Support for multi-users and password authentication
bcrypt
# Support for Zeroconf/Bonjour
zeroconf
# RSS feeds support
# feedparser

1
setup.py
View file

@ -161,6 +161,7 @@ setup(
'websockets',
'websocket-client',
'wheel',
'zeroconf',
],
extras_require={