pir.py

"""Support for communicating with myStrom PIRs."""
import aiohttp
from yarl import URL
from typing import Any, Dict, Iterable, List, Optional, Union

from . import _request as request

URI_PIR = URL("api/v1/")


class MyStromPir:
    """A class for a myStrom PIR."""

    def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None:
        """Initialize the switch."""
        self._close_session = False
        self._host = host
        self._session = session
        self._intensity = None
        self._day = None
        self._light_raw = None
        self._sensors = None
        self._temperature_measured = None
        self._temperature_compensated = None
        self._temperature_compensation = None
        self._temperature_raw = None
        self._motion = None
        self._settings = None
        self._pir = None

        self._actions = None
        self.uri = URL.build(scheme="http", host=self._host).join(URI_PIR)

    async def get_settings(self) -> None:
        """Get the current settings from the PIR."""
        url = URL(self.uri).join(URL("settings"))
        response = await request(self, uri=url)
        self._settings = response

    async def get_actions(self) -> None:
        """Get the current action settings from the PIR."""
        url = URL(self.uri).join(URL("action"))
        response = await request(self, uri=url)
        self._actions = response

    async def get_pir(self) -> None:
        """Get the current PIR settings."""
        url = URL(self.uri).join(URL("settings/pir"))
        response = await request(self, uri=url)
        self._pir = response

    async def get_sensors_state(self) -> None:
        """Get the state of the sensors from the PIR."""
        url = URL(self.uri).join(URL("sensors"))
        response = await request(self, uri=url)
        # The return data has the be re-written as the temperature is not rounded
        self._sensors = {
            "motion": response["motion"],
            "light": response["light"],
            "temperature": round(response["temperature"], 2),
        }

    async def get_temperatures(self) -> None:
        """Get the temperatures from the PIR."""
        # There is a different URL for the temp endpoint
        url = URL.build(scheme="http", host=self._host) / "temp"
        response = await request(self, uri=url)
        self._temperature_raw = response
        self._temperature_measured = round(response["measured"], 2)
        self._temperature_compensated = round(response["compensated"], 2)
        self._temperature_compensation = round(response["compensation"], 3)

    async def get_motion(self) -> None:
        """Get the state of the motion sensor from the PIR."""
        url = URL(self.uri).join(URL("motion"))
        response = await request(self, uri=url)
        self._motion = response["motion"]

    async def get_light(self) -> None:
        """Get the state of the light sensor from the PIR."""
        url = URL(self.uri).join(URL("light"))
        response = await request(self, uri=url)
        self._intensity = response["intensity"]
        self._day = response["day"]
        self._light_raw = response["raw"]

    @property
    def settings(self) -> Optional[dict]:
        """Return current settings."""
        return self._settings

    @property
    def actions(self) -> Optional[dict]:
        """Return current action settings."""
        return self._actions

    @property
    def pir(self) -> Optional[dict]:
        """Return current PIR settings."""
        return self._pir

    @property
    def sensors(self) -> Optional[dict]:
        """Return current sensor values."""
        return self._sensors

    @property
    def temperature_measured(self) -> Optional[str]:
        """Return current measured temperature."""
        return self._temperature_measured

    @property
    def temperature_compensated(self) -> Optional[str]:
        """Return current compensated temperature."""
        return self._temperature_compensated

    @property
    def temperature_compensation(self) -> Optional[str]:
        """Return current temperature compensation."""
        return self._temperature_compensation

    @property
    def temperature_raw(self) -> Optional[dict]:
        """Return current raw temperature values."""
        return self._temperature_raw

    @property
    def motion(self) -> Optional[str]:
        """Return the state of the motion sensor."""
        return self._motion

    @property
    def intensity(self) -> Optional[str]:
        """Return the intensity reported by the light sensor."""
        return self._intensity

    @property
    def day(self) -> Optional[str]:
        """Return the information based on the thresholds set."""
        return self._day

    @property
    def light_raw(self) -> Optional[str]:
        """Return the raw data from the ADC."""
        return {
            "visible": self._light_raw["adc0"],
            "infrared": self._light_raw["adc1"],
        }

    async def close(self) -> None:
        """Close an open client session."""
        if self._session and self._close_session:
            await self._session.close()

    async def __aenter__(self) -> "MyStromPir":
        """Async enter."""
        return self

    async def __aexit__(self, *exc_info) -> None:
        """Async exit."""
        await self.close()