platypush/backend/http/__init__.py

@@ -6,14 +6,14 @@ from multiprocessing import Process
 
 try:
     from websockets.exceptions import ConnectionClosed
-    from websockets import serve as websocket_serve  # type: ignore
+    from websockets import serve as websocket_serve
 except ImportError:
-    from websockets import ConnectionClosed, serve as websocket_serve  # type: ignore
+    from websockets import ConnectionClosed, serve as websocket_serve
 
 from platypush.backend import Backend
 from platypush.backend.http.app import application
 from platypush.context import get_or_create_event_loop
-from platypush.utils import get_ssl_server_context
+from platypush.utils import get_ssl_server_context, set_thread_name
 
 
 class HttpBackend(Backend):
@@ -186,7 +186,7 @@ class HttpBackend(Backend):
         maps=None,
         run_externally=False,
         uwsgi_args=None,
-        **kwargs,
+        **kwargs
     ):
         """
         :param port: Listen port for the web server (default: 8008)
@@ -283,13 +283,15 @@ class HttpBackend(Backend):
                 '--enable-threads',
             ]
 
-        protocol = 'https' if ssl_cert else 'http'
-        self.local_base_url = f'{protocol}://localhost:{self.port}'
+        self.local_base_url = '{proto}://localhost:{port}'.format(
+            proto=('https' if ssl_cert else 'http'), port=self.port
+        )
+
         self._websocket_lock_timeout = 10
         self._websocket_lock = threading.RLock()
         self._websocket_locks = {}
 
-    def send_message(self, msg, *_, **__):
+    def send_message(self, msg, **kwargs):
         self.logger.warning('Use cURL or any HTTP client to query the HTTP backend')
 
     def on_stop(self):
@@ -349,7 +351,9 @@ class HttpBackend(Backend):
             timeout=self._websocket_lock_timeout
         )
         if not acquire_ok:
-            raise TimeoutError(f'Websocket on address {addr} not ready to receive data')
+            raise TimeoutError(
+                'Websocket on address {} not ready to receive data'.format(addr)
+            )
 
     def _release_websocket_lock(self, ws):
         try:
@@ -364,7 +368,7 @@ class HttpBackend(Backend):
                 self._websocket_locks[addr].release()
         except Exception as e:
             self.logger.warning(
-                'Unhandled exception while releasing websocket lock: %s', e
+                'Unhandled exception while releasing websocket lock: {}'.format(str(e))
             )
         finally:
             self._websocket_lock.release()
@@ -377,7 +381,7 @@ class HttpBackend(Backend):
                 self._acquire_websocket_lock(ws)
                 await ws.send(str(event))
             except Exception as e:
-                self.logger.warning('Error on websocket send_event: %s', e)
+                self.logger.warning('Error on websocket send_event: {}'.format(e))
             finally:
                 self._release_websocket_lock(ws)
 
@@ -389,7 +393,7 @@ class HttpBackend(Backend):
                 loop.run_until_complete(send_event(_ws))
             except ConnectionClosed:
                 self.logger.warning(
-                    'Websocket client %s connection lost', _ws.remote_address
+                    'Websocket client {} connection lost'.format(_ws.remote_address)
                 )
                 self.active_websockets.remove(_ws)
                 if _ws.remote_address in self._websocket_locks:
@@ -397,6 +401,7 @@ class HttpBackend(Backend):
 
     def websocket(self):
         """Websocket main server"""
+        set_thread_name('WebsocketServer')
 
         async def register_websocket(websocket, path):
             address = (
@@ -406,14 +411,16 @@ class HttpBackend(Backend):
             )
 
             self.logger.info(
-                'New websocket connection from %s on path %s', address, path
+                'New websocket connection from {} on path {}'.format(address, path)
             )
             self.active_websockets.add(websocket)
 
             try:
                 await websocket.recv()
             except ConnectionClosed:
-                self.logger.info('Websocket client %s closed connection', address)
+                self.logger.info(
+                    'Websocket client {} closed connection'.format(address)
+                )
                 self.active_websockets.remove(websocket)
                 if address in self._websocket_locks:
                     del self._websocket_locks[address]
@@ -428,14 +435,14 @@ class HttpBackend(Backend):
                 register_websocket,
                 self.bind_address,
                 self.websocket_port,
-                **websocket_args,
+                **websocket_args
             )
         )
         self._websocket_loop.run_forever()
 
     def _start_web_server(self):
         def proc():
-            self.logger.info('Starting local web server on port %s', self.port)
+            self.logger.info('Starting local web server on port {}'.format(self.port))
             kwargs = {
                 'host': self.bind_address,
                 'port': self.port,
@@ -463,10 +470,7 @@ class HttpBackend(Backend):
 
         if not self.disable_websocket:
             self.logger.info('Initializing websocket interface')
-            self.websocket_thread = threading.Thread(
-                target=self.websocket,
-                name='WebsocketServer',
-            )
+            self.websocket_thread = threading.Thread(target=self.websocket)
             self.websocket_thread.start()
 
         if not self.run_externally:
@@ -477,13 +481,13 @@ class HttpBackend(Backend):
             self.server_proc.join()
         elif self.uwsgi_args:
             uwsgi_cmd = ['uwsgi'] + self.uwsgi_args
-            self.logger.info('Starting uWSGI with arguments %s', uwsgi_cmd)
+            self.logger.info('Starting uWSGI with arguments {}'.format(uwsgi_cmd))
             self.server_proc = subprocess.Popen(uwsgi_cmd)
         else:
             self.logger.info(
                 'The web server is configured to be launched externally but '
-                'no uwsgi_args were provided. Make sure that you run another external service'
-                'for the web server (e.g. nginx)'
+                + 'no uwsgi_args were provided. Make sure that you run another external service'
+                + 'for the webserver (e.g. nginx)'
             )
 
         self._service_registry_thread = threading.Thread(target=self._register_service)

platypush/backend/mqtt/__init__.py

@@ -13,6 +13,7 @@ from platypush.message import Message
 from platypush.message.event.mqtt import MQTTMessageEvent
 from platypush.message.request import Request
 from platypush.plugins.mqtt import MqttPlugin as MQTTPlugin
+from platypush.utils import set_thread_name
 
 
 class MqttClient(mqtt.Client, threading.Thread):
@@ -38,7 +39,6 @@ class MqttClient(mqtt.Client, threading.Thread):
         mqtt.Client.__init__(self, *args, client_id=client_id, **kwargs)
         threading.Thread.__init__(self)
 
-        self.name = f'MQTTClient:{client_id}'
         self.host = host
         self.port = port
         self.topics = set(topics or [])
@@ -393,6 +393,7 @@ class MqttBackend(Backend):
         def handler(_, __, msg):
             # noinspection PyShadowingNames
             def response_thread(msg):
+                set_thread_name('MQTTProcessor')
                 response = self.get_message_response(msg)
                 if not response:
                     return
@@ -427,9 +428,7 @@ class MqttBackend(Backend):
 
             if isinstance(msg, Request):
                 threading.Thread(
-                    target=response_thread,
-                    name='MQTTProcessorResponseThread',
-                    args=(msg,),
+                    target=response_thread, name='MQTTProcessor', args=(msg,)
                 ).start()
 
         return handler

platypush/entities/_managers/switches.py

@@ -35,7 +35,7 @@ class MultiLevelSwitchEntityManager(EntityManager, ABC):
 
     @abstractmethod
     def set_value(  # pylint: disable=redefined-builtin
-        self, *_, property=None, data=None, **__
+        self, *entities, property=None, data=None, **__
     ):
         """Set a value"""
         raise NotImplementedError()

platypush/plugins/__init__.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import threading
+import time
 
 from abc import ABC, abstractmethod
 from functools import wraps
@@ -10,7 +11,7 @@ from platypush.bus import Bus
 from platypush.common import ExtensionWithManifest
 from platypush.event import EventGenerator
 from platypush.message.response import Response
-from platypush.utils import get_decorators, get_plugin_name_by_class
+from platypush.utils import get_decorators, get_plugin_name_by_class, set_thread_name
 
 PLUGIN_STOP_TIMEOUT = 5  # Plugin stop timeout in seconds
 
@@ -106,22 +107,25 @@ class RunnablePlugin(Plugin):
         return self._should_stop.wait(timeout=timeout)
 
     def start(self):
-        self._thread = threading.Thread(
-            target=self._runner, name=self.__class__.__name__
-        )
+        set_thread_name(self.__class__.__name__)
+        self._thread = threading.Thread(target=self._runner)
         self._thread.start()
 
     def stop(self):
         self._should_stop.set()
         if self._thread and self._thread.is_alive():
-            self.logger.info('Waiting for the plugin to stop')
+            self.logger.info('Waiting for %s to stop', self.__class__.__name__)
             try:
                 if self._thread:
                     self._thread.join(timeout=self._stop_timeout)
                     if self._thread and self._thread.is_alive():
                         self.logger.warning(
-                            'Timeout (seconds=%s) on exit for the plugin',
+                            'Timeout (seconds={%s}) on exit for the plugin %s',
                             self._stop_timeout,
+                            (
+                                get_plugin_name_by_class(self.__class__)
+                                or self.__class__.__name__
+                            ),
                         )
             except Exception as e:
                 self.logger.warning('Could not join thread on stop: %s', e)
@@ -138,7 +142,7 @@ class RunnablePlugin(Plugin):
                 self.logger.exception(e)
 
             if self.poll_interval:
-                self.wait_stop(self.poll_interval)
+                time.sleep(self.poll_interval)
 
         self._thread = None
 
@@ -152,27 +156,18 @@ class AsyncRunnablePlugin(RunnablePlugin, ABC):
         super().__init__(*args, **kwargs)
 
         self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._loop_runner: Optional[threading.Thread] = None
         self._task: Optional[asyncio.Task] = None
 
     @property
     def _should_start_runner(self):
-        """
-        This property is used to determine if the runner and the event loop
-        should be started for this plugin.
-        """
         return True
 
     @abstractmethod
     async def listen(self):
-        """
-        Main body of the async plugin. When it's called, the event loop should
-        already be running and available over `self._loop`.
-        """
+        pass
 
     async def _listen(self):
-        """
-        Wrapper for :meth:`.listen` that catches any exceptions and logs them.
-        """
         try:
             await self.listen()
         except KeyboardInterrupt:
@@ -184,37 +179,41 @@ class AsyncRunnablePlugin(RunnablePlugin, ABC):
             ):
                 raise e
 
-    def _run_listener(self):
+    def _start_listener(self):
+        set_thread_name(self.__class__.__name__ + ':listener')
         self._loop = asyncio.new_event_loop()
         asyncio.set_event_loop(self._loop)
 
         self._task = self._loop.create_task(self._listen())
         if hasattr(self._task, 'set_name'):
             self._task.set_name(self.__class__.__name__ + '.listen')
-        try:
-            self._loop.run_until_complete(self._task)
-        except Exception as e:
-            self.logger.info('The loop has terminated with an error: %s', e)
-
-        self._task.cancel()
+        self._loop.run_forever()
 
     def main(self):
-        if self.should_stop():
-            self.logger.info('The plugin is already scheduled to stop')
+        if self.should_stop() or (self._loop_runner and self._loop_runner.is_alive()):
+            self.logger.info('The main loop is already being run/stopped')
             return
 
-        self._loop = asyncio.new_event_loop()
-
         if self._should_start_runner:
-            self._run_listener()
+            self._loop_runner = threading.Thread(target=self._start_listener)
+            self._loop_runner.start()
 
         self.wait_stop()
 
     def stop(self):
+        if self._task and self._loop and not self._task.done():
+            self._loop.call_soon_threadsafe(self._task.cancel)
+
         if self._loop and self._loop.is_running():
             self._loop.call_soon_threadsafe(self._loop.stop)
             self._loop = None
 
+        if self._loop_runner and self._loop_runner.is_alive():
+            try:
+                self._loop_runner.join(timeout=self._stop_timeout)
+            finally:
+                self._loop_runner = None
+
         super().stop()
 
 

platypush/plugins/ntfy/__init__.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
 import os
-from typing import Any, Callable, Dict, Optional, Collection, Mapping
+from typing import Optional, Collection, Mapping
 
 import requests
 import websockets
@@ -53,12 +53,12 @@ class NtfyPlugin(AsyncRunnablePlugin):
         reconnect_wait_secs_max = 60
 
         while not self.should_stop():
-            self.logger.debug('Connecting to %s', url)
+            self.logger.debug(f'Connecting to {url}')
 
             try:
-                async with websockets.connect(url) as ws:  # pylint: disable=no-member
+                async with websockets.connect(url) as ws:  # type: ignore
                     reconnect_wait_secs = 1
-                    self.logger.info('Connected to %s', url)
+                    self.logger.info(f'Connected to {url}')
                     async for msg in ws:
                         try:
                             msg = json.loads(msg)
@@ -122,7 +122,7 @@ class NtfyPlugin(AsyncRunnablePlugin):
         tags: Optional[Collection[str]] = None,
         schedule: Optional[str] = None,
     ):
-        r"""
+        """
         Send a message/notification to a topic.
 
         :param topic: Topic where the message will be delivered.
@@ -148,36 +148,36 @@ class NtfyPlugin(AsyncRunnablePlugin):
                 - ``broadcast``: Send an `Android broadcast <https://developer.android.com/guide/components/broadcasts>`
                       intent upon action selection (only available on Android).
 
-        Actions example:
+            Example:
 
-        .. code-block:: json
+            .. code-block:: json
 
-            [
-                {
-                    "action": "view",
-                    "label": "Open portal",
-                    "url": "https://home.nest.com/",
-                    "clear": true
-                },
-                {
-                    "action": "http",
-                    "label": "Turn down",
-                    "url": "https://api.nest.com/",
-                    "method": "PUT",
-                    "headers": {
-                        "Authorization": "Bearer abcdef..."
+                [
+                    {
+                        "action": "view",
+                        "label": "Open portal",
+                        "url": "https://home.nest.com/",
+                        "clear": true
                     },
-                    "body": "{\\"temperature\\": 65}"
-                },
-                {
-                    "action": "broadcast",
-                    "label": "Take picture",
-                    "intent": "com.myapp.TAKE_PICTURE_INTENT",
-                    "extras": {
-                        "camera": "front"
+                    {
+                        "action": "http",
+                        "label": "Turn down",
+                        "url": "https://api.nest.com/",
+                        "method": "PUT",
+                        "headers": {
+                            "Authorization": "Bearer abcdef..."
+                        },
+                        "body": "{\\"temperature\\": 65}"
+                    },
+                    {
+                        "action": "broadcast",
+                        "label": "Take picture",
+                        "intent": "com.myapp.TAKE_PICTURE_INTENT",
+                        "extras": {
+                            "camera": "front"
+                        }
                     }
-                }
-            ]
+                ]
 
         :param email: Forward the notification as an email to the specified
             address.
@@ -196,9 +196,9 @@ class NtfyPlugin(AsyncRunnablePlugin):
                   ``tomorrow, 3pm``)
 
         """
-        method: Callable[..., requests.Response] = requests.post
+        method = requests.post
         url = server_url or self._server_url
-        args: Dict[str, Any] = {}
+        args = {}
         if username and password:
             args['auth'] = (username, password)
 
@@ -247,8 +247,8 @@ class NtfyPlugin(AsyncRunnablePlugin):
             }
 
         rs = method(url, **args)
-        assert rs.ok, f'Could not send message to {topic}: ' + rs.json().get(
-            'error', f'HTTP error: {rs.status_code}'
+        assert rs.ok, 'Could not send message to {}: {}'.format(
+            topic, rs.json().get('error', f'HTTP error: {rs.status_code}')
         )
 
         return rs.json()

platypush/plugins/smartthings/__init__.py

@@ -859,9 +859,8 @@ class SmartthingsPlugin(
         return self.status(device)
 
     @action
-    # pylint: disable=redefined-builtin,arguments-differ
-    def set_value(
-        self, device: str, *_, property: Optional[str] = None, data=None, **kwargs
+    def set_value(  # pylint: disable=arguments-differ,redefined-builtin
+        self, device: str, property: Optional[str] = None, data=None, **kwargs
     ):
         """
         Set the value of a device. It is compatible with the generic

platypush/plugins/switch/__init__.py

@@ -0,0 +1,72 @@
+from abc import ABC, abstractmethod
+from typing import List, Union
+
+from platypush.plugins import Plugin, action
+
+
+# TODO REMOVE ME
+class SwitchPlugin(Plugin, ABC):
+    """
+    Abstract class for interacting with switch devices
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @action
+    @abstractmethod
+    def on(self, device, *args, **kwargs):
+        """Turn the device on"""
+        raise NotImplementedError()
+
+    @action
+    @abstractmethod
+    def off(self, device, *args, **kwargs):
+        """Turn the device off"""
+        raise NotImplementedError()
+
+    @action
+    @abstractmethod
+    def toggle(self, device, *args, **kwargs):
+        """Toggle the device status (on/off)"""
+        raise NotImplementedError()
+
+    @action
+    def switch_status(self, device=None) -> Union[dict, List[dict]]:
+        """
+        Get the status of a specified device or of all the configured devices (default).
+
+        :param device: Filter by device name or ID.
+        :return: .. schema:: switch.SwitchStatusSchema(many=True)
+        """
+        devices = self.switches
+        if device:
+            devices = [
+                d
+                for d in self.switches
+                if d.get('id') == device or d.get('name') == device
+            ]
+            return devices[0] if devices else []
+
+        return devices
+
+    @action
+    def status(self, device=None, *_, **__) -> Union[dict, List[dict]]:
+        """
+        Get the status of all the devices, or filter by device name or ID (alias for :meth:`.switch_status`).
+
+        :param device: Filter by device name or ID.
+        :return: .. schema:: switch.SwitchStatusSchema(many=True)
+        """
+        return self.switch_status(device)
+
+    @property
+    @abstractmethod
+    def switches(self) -> List[dict]:
+        """
+        :return: .. schema:: switch.SwitchStatusSchema(many=True)
+        """
+        raise NotImplementedError()
+
+
+# vim:sw=4:ts=4:et:

platypush/plugins/switchbot/__init__.py

@@ -1105,7 +1105,7 @@ class SwitchbotPlugin(
 
     @action
     # pylint: disable=redefined-builtin,arguments-differ
-    def set_value(self, device: str, *_, property=None, data=None, **__):
+    def set_value(self, device: str, property=None, data=None, **__):
         entity = self._to_entity(device, property)
         assert entity, f'No such entity: "{device}"'
 

platypush/plugins/switchbot/bluetooth/__init__.py

@@ -1,16 +1,16 @@
 import enum
 import time
-from typing import Any, Collection, Dict, List, Optional
+from typing import List, Optional
 
-from platypush.entities import EnumSwitchEntityManager
-from platypush.entities.switches import EnumSwitch
 from platypush.message.response.bluetooth import BluetoothScanResponse
 from platypush.plugins import action
 from platypush.plugins.bluetooth.ble import BluetoothBlePlugin
+from platypush.plugins.switch import SwitchPlugin
 
 
-# pylint: disable=too-many-ancestors
-class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
+class SwitchbotBluetoothPlugin(  # lgtm [py/missing-call-to-init]
+    SwitchPlugin, BluetoothBlePlugin
+):
     """
     Plugin to interact with a Switchbot (https://www.switch-bot.com/) device and
     programmatically control switches over a Bluetooth interface.
@@ -91,7 +91,7 @@ class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
                 raise e
             time.sleep(5)
 
-        return self.status(device)
+        return self.status(device)  # type: ignore
 
     @action
     def press(self, device):
@@ -127,30 +127,6 @@ class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
         """
         return self._run(device, self.Command.OFF)
 
-    @action
-    # pylint: disable=arguments-differ
-    def set_value(self, device: str, data: str, *_, **__):
-        """
-        Entity-compatible ``set_value`` method to send a command to a device.
-
-        :param device: Device name or address
-        :param data: Command to send. Possible values are:
-
-            - ``on``: Press the button and remain in the pressed state.
-            - ``off``: Release a previously pressed button.
-            - ``press``: Press and release the button.
-
-        """
-        if data == 'on':
-            return self.on(device)
-        if data == 'off':
-            return self.off(device)
-        if data == 'press':
-            return self.press(device)
-
-        self.logger.warning('Unknown command for SwitchBot "%s": "%s"', device, data)
-        return None
-
     @action
     def scan(
         self, interface: Optional[str] = None, duration: int = 10
@@ -162,16 +138,14 @@ class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
         :param duration: Scan duration in seconds
         """
 
-        compatible_devices: Dict[str, Any] = {}
-        devices = (
-            super().scan(interface=interface, duration=duration).devices  # type: ignore
-        )
+        devices = super().scan(interface=interface, duration=duration).devices
+        compatible_devices = {}
 
         for dev in devices:
             try:
                 characteristics = [
                     chrc
-                    for chrc in self.discover_characteristics(  # type: ignore
+                    for chrc in self.discover_characteristics(
                         dev['addr'],
                         channel_type='random',
                         wait=False,
@@ -183,14 +157,14 @@ class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
                 if characteristics:
                     compatible_devices[dev['addr']] = None
             except Exception as e:
-                self.logger.warning('Device scan error: %s', e)
+                self.logger.warning('Device scan error', e)
 
-        self.publish_entities(compatible_devices)
+        self.publish_entities(compatible_devices)  # type: ignore
         return BluetoothScanResponse(devices=compatible_devices)
 
-    @action
-    def status(self, *_, **__) -> List[dict]:
-        self.publish_entities(self.configured_devices)
+    @property
+    def switches(self) -> List[dict]:
+        self.publish_entities(self.configured_devices)  # type: ignore
         return [
             {
                 'address': addr,
@@ -201,17 +175,20 @@ class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager):
             for addr, name in self.configured_devices.items()
         ]
 
-    def transform_entities(self, entities: Collection[dict]) -> Collection[EnumSwitch]:
-        return [
-            EnumSwitch(
-                id=addr,
-                name=name,
-                value='on',
-                values=['on', 'off', 'press'],
-                is_write_only=True,
-            )
-            for addr, name in entities
-        ]
+    def transform_entities(self, devices: dict):
+        from platypush.entities.switches import Switch
+
+        return super().transform_entities(  # type: ignore
+            [
+                Switch(
+                    id=addr,
+                    name=name,
+                    state=False,
+                    is_write_only=True,
+                )
+                for addr, name in devices.items()
+            ]
+        )
 
 
 # vim:sw=4:ts=4:et:

platypush/plugins/zigbee/mqtt/__init__.py

@@ -1072,9 +1072,8 @@ class ZigbeeMqttPlugin(
         return properties
 
     @action
-    # pylint: disable=redefined-builtin,arguments-differ
-    def set_value(
-        self, device: str, *_, property: Optional[str] = None, data=None, **kwargs
+    def set_value(  # pylint: disable=redefined-builtin,arguments-differ
+        self, device: str, property: Optional[str] = None, data=None, **kwargs
     ):
         """
         Entity-compatible way of setting a value on a node.