Improve emeterstatus API, move into own module (#205)

Adds the following properties to EmeterStatus for saner API:
* voltage (in V)
* power (in W)
* current (in A)
* total (in kWh)
This commit is contained in:
Teemu R 2021-09-23 17:58:19 +02:00 committed by GitHub
parent 47a1405bd2
commit 36c412a9c2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 88 additions and 53 deletions

View File

@ -14,10 +14,11 @@ to be handled by the user of the library.
from importlib_metadata import version # type: ignore
from kasa.discover import Discover
from kasa.emeterstatus import EmeterStatus
from kasa.exceptions import SmartDeviceException
from kasa.protocol import TPLinkSmartHomeProtocol
from kasa.smartbulb import SmartBulb
from kasa.smartdevice import DeviceType, EmeterStatus, SmartDevice
from kasa.smartdevice import DeviceType, SmartDevice
from kasa.smartdimmer import SmartDimmer
from kasa.smartlightstrip import SmartLightStrip
from kasa.smartplug import SmartPlug

82
kasa/emeterstatus.py Normal file
View File

@ -0,0 +1,82 @@
"""Module for emeter container."""
import logging
from typing import Optional
_LOGGER = logging.getLogger(__name__)
class EmeterStatus(dict):
"""Container for converting different representations of emeter data.
Newer FW/HW versions postfix the variable names with the used units,
where-as the olders do not have this feature.
This class automatically converts between these two to allow
backwards and forwards compatibility.
"""
@property
def voltage(self) -> Optional[float]:
"""Return voltage in V."""
try:
return self["voltage"]
except ValueError:
return None
@property
def power(self) -> Optional[float]:
"""Return power in W."""
try:
return self["power"]
except ValueError:
return None
@property
def current(self) -> Optional[float]:
"""Return current in A."""
try:
return self["current"]
except ValueError:
return None
@property
def total(self) -> Optional[float]:
"""Return total in kWh."""
try:
return self["total"]
except ValueError:
return None
def __repr__(self):
return f"<EmeterStatus power={self.power} voltage={self.voltage} current={self.current} total={self.total}>"
def __getitem__(self, item):
valid_keys = [
"voltage_mv",
"power_mw",
"current_ma",
"energy_wh",
"total_wh",
"voltage",
"power",
"current",
"total",
"energy",
]
# 1. if requested data is available, return it
if item in super().keys():
return super().__getitem__(item)
# otherwise decide how to convert it
else:
if item not in valid_keys:
raise KeyError(item)
if "_" in item: # upscale
return super().__getitem__(item[: item.find("_")]) * 1000
else: # downscale
for i in super().keys():
if i.startswith(item):
return self.__getitem__(i) / 1000
_LOGGER.debug(f"Unable to find value for '{item}'")
return None

View File

@ -3,12 +3,7 @@ import logging
import re
from typing import Any, Dict, NamedTuple, cast
from kasa.smartdevice import (
DeviceType,
SmartDevice,
SmartDeviceException,
requires_update,
)
from .smartdevice import DeviceType, SmartDevice, SmartDeviceException, requires_update
class ColorTempRange(NamedTuple):

View File

@ -19,6 +19,7 @@ from datetime import datetime, timedelta
from enum import Enum, auto
from typing import Any, Dict, List, Optional
from .emeterstatus import EmeterStatus
from .exceptions import SmartDeviceException
from .protocol import TPLinkSmartHomeProtocol
@ -50,48 +51,6 @@ class WifiNetwork:
rssi: Optional[int] = None
class EmeterStatus(dict):
"""Container for converting different representations of emeter data.
Newer FW/HW versions postfix the variable names with the used units,
where-as the olders do not have this feature.
This class automatically converts between these two to allow
backwards and forwards compatibility.
"""
def __getitem__(self, item):
valid_keys = [
"voltage_mv",
"power_mw",
"current_ma",
"energy_wh",
"total_wh",
"voltage",
"power",
"current",
"total",
"energy",
]
# 1. if requested data is available, return it
if item in super().keys():
return super().__getitem__(item)
# otherwise decide how to convert it
else:
if item not in valid_keys:
raise KeyError(item)
if "_" in item: # upscale
return super().__getitem__(item[: item.find("_")]) * 1000
else: # downscale
for i in super().keys():
if i.startswith(item):
return self.__getitem__(i) / 1000
_LOGGER.debug(f"Unable to find value for '{item}'")
return None
def requires_update(f):
"""Indicate that `update` should be called before accessing this method.""" # noqa: D202
if inspect.iscoroutinefunction(f):
@ -202,7 +161,7 @@ class SmartDevice:
>>> dev.has_emeter
True
>>> dev.emeter_realtime
{'current': 0.015342, 'err_code': 0, 'power': 0.983971, 'total': 32.448, 'voltage': 235.595234}
<EmeterStatus power=0.983971 voltage=235.595234 current=0.015342 total=32.448>
>>> dev.emeter_today
>>> dev.emeter_this_month

View File

@ -1,6 +1,6 @@
import pytest
from kasa import SmartDeviceException
from kasa import EmeterStatus, SmartDeviceException
from .conftest import has_emeter, no_emeter, pytestmark
from .newfakes import CURRENT_CONSUMPTION_SCHEMA
@ -121,8 +121,6 @@ async def test_current_consumption(dev):
async def test_emeterstatus_missing_current():
"""KL125 does not report 'current' for emeter."""
from kasa import EmeterStatus
regular = EmeterStatus(
{"err_code": 0, "power_mw": 0, "total_wh": 13, "current_ma": 123}
)