platypush/platypush/plugins/switch/wemo/__init__.py

import contextlib
import ipaddress
from typing import List, Optional

from platypush.plugins import action
from platypush.plugins.switch import SwitchPlugin
from platypush.utils.workers import Workers

from .lib import WemoRunner
from .scanner import Scanner


class SwitchWemoPlugin(SwitchPlugin):
    """
    Plugin to control a Belkin WeMo smart switches
    (https://www.belkin.com/us/Products/home-automation/c/wemo-home-automation/)
    """

    _default_port = 49153

    def __init__(
        self,
        devices=None,
        netmask: Optional[str] = None,
        port: int = _default_port,
        **kwargs
    ):
        """
        :param devices: List of IP addresses or name->address map containing the WeMo Switch devices to control.
            This plugin previously used ouimeaux for auto-discovery but it's been dropped because
            1. too slow 2. too heavy 3. auto-discovery failed too often.
        :type devices: list or dict

        :param netmask: Alternatively to a list of static IP->name pairs,  you can specify the network mask where
            the devices should be scanned (e.g. '192.168.1.0/24')

        :param port: Port where the WeMo devices are expected to expose the RPC/XML over HTTP service (default: 49153)
        """

        super().__init__(**kwargs)
        self.port = port
        self.netmask = netmask
        self._devices = {}
        self._init_devices(devices)

    def _init_devices(self, devices):
        if devices:
            self._devices.update(
                devices
                if isinstance(devices, dict)
                else {addr: addr for addr in devices}
            )
        else:
            self._devices = {}

        self._addresses = set(self._devices.values())

    @property
    def switches(self) -> List[dict]:
        """
        Get the list of available devices
        :returns: The list of devices.

            .. code-block:: json

                [
                    {
                        "ip": "192.168.1.123",
                        "name": "Switch 1",
                        "on": true
                    },
                    {
                        "ip": "192.168.1.124",
                        "name": "Switch 2",
                        "on": false
                    }
                ]

        """

        return [
            self.status(device).output  # type: ignore
            for device in self._devices.values()
        ]

    def _get_address(self, device: str) -> str:
        if device not in self._addresses:
            with contextlib.suppress(KeyError):
                return self._devices[device]

        return device

    @action
    def status(self, device: Optional[str] = None, *_, **__):
        devices = {device: device} if device else self._devices.copy()

        ret = [
            {
                "id": addr,
                "ip": addr,
                "name": name if name != addr else WemoRunner.get_name(addr),
                "on": WemoRunner.get_state(addr),
            }
            for (name, addr) in devices.items()
        ]

        self.publish_entities(ret)  # type: ignore
        return ret[0] if device else ret

    def transform_entities(self, devices: List[dict]):
        from platypush.entities.switches import Switch

        return super().transform_entities(  # type: ignore
            [
                Switch(
                    id=dev["id"],
                    name=dev["name"],
                    state=dev["on"],
                    data={
                        "ip": dev["ip"],
                    },
                )
                for dev in (devices or [])
            ]
        )

    @action
    def on(self, device: str, **_):
        """
        Turn a switch on

        :param device: Device name or address
        """
        device = self._get_address(device)
        WemoRunner.on(device)
        return self.status(device)

    @action
    def off(self, device: str, **_):
        """
        Turn a switch off

        :param device: Device name or address
        """
        device = self._get_address(device)
        WemoRunner.off(device)
        return self.status(device)

    @action
    def toggle(self, device: str, *_, **__):
        """
        Toggle a device on/off state

        :param device: Device name or address
        """
        device = self._get_address(device)
        WemoRunner.toggle(device)
        return self.status(device)

    @action
    def get_state(self, device: str):
        """
        Get the on state of a device (True/False)

        :param device: Device name or address
        """
        device = self._get_address(device)
        return WemoRunner.get_state(device)

    @action
    def get_name(self, device: str):
        """
        Get the friendly name of a device

        :param device: Device name or address
        """
        device = self._get_address(device)
        return WemoRunner.get_name(device)

    @action
    def scan(self, netmask: Optional[str] = None):
        netmask = netmask or self.netmask
        assert netmask, "Scan not supported: No netmask specified"

        workers = Workers(10, Scanner, port=self.port)
        with workers:
            for addr in ipaddress.IPv4Network(netmask):
                workers.put(addr.exploded)

        devices = {dev.name: dev.addr for dev in workers.responses}

        self._init_devices(devices)
        return self.status()


# vim:sw=4:ts=4:et: