device.py

"""Base MELCloud device."""
import asyncio
from abc import ABC, abstractmethod
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Dict, List, Optional

from pymelcloud.client import Client
from pymelcloud.const import (
    DEVICE_TYPE_LOOKUP,
    DEVICE_TYPE_UNKNOWN,
    UNIT_TEMP_CELSIUS,
    UNIT_TEMP_FAHRENHEIT,
    ACCESS_LEVEL,
)

PROPERTY_POWER = "power"

EFFECTIVE_FLAGS = "EffectiveFlags"
HAS_PENDING_COMMAND = "HasPendingCommand"


class Device(ABC):
    """MELCloud base device representation."""

    def __init__(
        self,
        device_conf: Dict[str, Any],
        client: Client,
        set_debounce=timedelta(seconds=1),
    ):
        """Initialize a device."""
        self.device_id = device_conf.get("DeviceID")
        self.building_id = device_conf.get("BuildingID")
        self.mac = device_conf.get("MacAddress")
        self.serial = device_conf.get("SerialNumber")
        self.access_level = device_conf.get("AccessLevel")

        self._use_fahrenheit = False
        if client.account is not None:
            self._use_fahrenheit = client.account.get("UseFahrenheit", False)

        self._device_conf = device_conf
        self._state = None
        self._device_units = None
        self._energy_report = None
        self._client = client

        self._set_debounce = set_debounce
        self._set_event = asyncio.Event()
        self._write_task: Optional[asyncio.Future[None]] = None
        self._pending_writes: Dict[str, Any] = {}

    def get_device_prop(self, name: str) -> Optional[Any]:
        """Access device properties while shortcutting the nested device access."""
        device = self._device_conf.get("Device", {})
        return device.get(name)

    def get_state_prop(self, name: str) -> Optional[Any]:
        """Access state prop without None check."""
        if self._state is None:
            return None
        return self._state.get(name)

    def round_temperature(self, temperature: float) -> float:
        """Round a temperature to the nearest temperature increment."""
        return float(
            Decimal(str(temperature / self.temperature_increment))
            .quantize(Decimal('1'), rounding=ROUND_HALF_UP)
        ) * self.temperature_increment

    @abstractmethod
    def apply_write(self, state: Dict[str, Any], key: str, value: Any):
        """Apply writes to state object.

        Used for property validation, do not modify device state.
        """
        pass

    async def update(self):
        """Fetch state of the device from MELCloud.

        List of device_confs is also updated.

        Please, rate limit calls to this method. Polling every 60 seconds should be
        enough to catch all events at the rate they are coming in to MELCloud with the
        exception of changes performed through MELCloud directly.
        """
        await self._client.update_confs()
        self._device_conf = next(
            c
            for c in self._client.device_confs
            if c.get("DeviceID") == self.device_id
            and c.get("BuildingID") == self.building_id
        )
        self._state = await self._client.fetch_device_state(self)
        self._energy_report = await self._client.fetch_energy_report(self)

        if self._device_units is None and self.access_level != ACCESS_LEVEL.get(
            "GUEST"
        ):
            self._device_units = await self._client.fetch_device_units(self)

    async def set(self, properties: Dict[str, Any]):
        """Schedule property write to MELCloud."""
        if self._write_task is not None:
            self._write_task.cancel()

        for k, value in properties.items():
            if k == PROPERTY_POWER:
                continue
            self.apply_write({}, k, value)

        self._pending_writes.update(properties)

        self._write_task = asyncio.ensure_future(self._write())
        await self._set_event.wait()

    async def _write(self):
        await asyncio.sleep(self._set_debounce.total_seconds())
        new_state = self._state.copy()

        for k, value in self._pending_writes.items():
            if k == PROPERTY_POWER:
                new_state["Power"] = value
                new_state[EFFECTIVE_FLAGS] = new_state.get(EFFECTIVE_FLAGS, 0) | 0x01
            else:
                self.apply_write(new_state, k, value)

        if new_state[EFFECTIVE_FLAGS] != 0:
            new_state.update({HAS_PENDING_COMMAND: True})

        self._pending_writes = {}
        self._state = await self._client.set_device_state(new_state)
        self._set_event.set()
        self._set_event.clear()

    @property
    def name(self) -> str:
        """Return device name."""
        return self._device_conf["DeviceName"]

    @property
    def device_type(self) -> str:
        """Return type of the device."""
        return DEVICE_TYPE_LOOKUP.get(
            self._device_conf.get("Device", {}).get("DeviceType", -1),
            DEVICE_TYPE_UNKNOWN,
        )

    @property
    def units(self) -> Optional[List[dict]]:
        """Return device model info."""
        if self._device_units is None:
            return None

        infos: List[dict] = []
        for unit in self._device_units:
            infos.append(
                {
                    "model_number": unit.get("ModelNumber"),
                    "model": unit.get("Model"),
                    "serial_number": unit.get("SerialNumber"),
                }
            )
        return infos

    @property
    def temp_unit(self) -> str:
        """Return temperature unit used by the device."""
        if self._use_fahrenheit:
            return UNIT_TEMP_FAHRENHEIT
        return UNIT_TEMP_CELSIUS

    @property
    def temperature_increment(self) -> float:
        """Return temperature increment."""
        return self._device_conf.get("Device", {}).get("TemperatureIncrement", 0.5)

    @property
    def last_seen(self) -> Optional[datetime]:
        """Return timestamp of the last communication from device to MELCloud.

        The timestamp is in UTC.
        """
        if self._state is None:
            return None
        return datetime.strptime(
            self._state.get("LastCommunication"), "%Y-%m-%dT%H:%M:%S.%f"
        ).replace(tzinfo=timezone.utc)

    @property
    def power(self) -> Optional[bool]:
        """Return power on / standby state of the device."""
        if self._state is None:
            return None
        return self._state.get("Power")

    @property
    def daily_energy_consumed(self) -> Optional[float]:
        """Return daily energy consumption for the current day in kWh.

        The value resets at midnight MELCloud time. The logic here is a bit iffy and
        fragmented between Device and Client. Here's how it goes:
          - Client requests a 5 day report. Today, 2 days from the past and 2 days from
            the past.
          - MELCloud, with its clock potentially set to a different timezone than the
            client, returns a report containing data from a couple of days from the
            past and from the current day in MELCloud time.
          - Device sums up the date from the last day bucket in the report.

        TLDR: Request some days from the past and some days from the future -> receive
        the latest day bucket.
        """
        if self._energy_report is None:
            return None

        consumption = 0

        for mode in ['Heating', 'Cooling', 'Auto', 'Dry', 'Fan', 'Other']:
            previous_reports = self._energy_report.get(mode, [0.0])
            if previous_reports:
                last_report = previous_reports[-1]
            else:
                last_report = 0.0
            consumption += last_report

        return consumption

    @property
    def wifi_signal(self) -> Optional[int]:
        """Return wifi signal in dBm (negative value)."""
        if self._device_conf is None:
            return None
        return self._device_conf.get("Device", {}).get("WifiSignalStrength", None)

    @property
    def has_error(self) -> bool:
        """Return True if the device has error state."""
        if self._state is None:
            return False
        return self._state.get("HasError", False)

    @property
    def error_code(self) -> Optional[str]:
        """Return error_code.
        This is a property that probably should be checked if "has_error" = true
        Till now I have a fixed code = 8000 and never have error on the units
        """
        if self._state is None:
            return None
        return self._state.get("ErrorCode", None)