| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -5,7 +5,8 @@ | |
|
|
||
| from ..credentials import Credentials | ||
| from ..emeterstatus import EmeterStatus | ||
| from ..smartdevice import DeviceType | ||
| from ..modules import Emeter | ||
| from ..smartdevice import DeviceType, requires_update | ||
| from .tapodevice import TapoDevice | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
| Expand All | @@ -24,6 +25,15 @@ def __init__( | |
| ) -> None: | ||
| super().__init__(host, port=port, credentials=credentials, timeout=timeout) | ||
| self._device_type = DeviceType.Plug | ||
| self.modules: Dict[str, Any] = {} | ||
| self.emeter_type = "emeter" | ||
| self.modules["emeter"] = Emeter(self, self.emeter_type) | ||
|
Comment thread
Copy link
Copy Markdown
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityI think we need to rethink the whole modules system for "tapo" devices separately from the legacy protocol, but I suppose this is fine for now.
Sorry, something went wrong.
All reactions
|
||
|
|
||
| @property # type: ignore | ||
| @requires_update | ||
| def has_emeter(self) -> bool: | ||
| """Return that the plug has an emeter.""" | ||
| return True | ||
|
|
||
| async def update(self, update_children: bool = True): | ||
| """Call the device endpoint and update the device data.""" | ||
| Expand Down Expand Up | @@ -52,17 +62,24 @@ def state_information(self) -> Dict[str, Any]: | |
| @property | ||
| def emeter_realtime(self) -> EmeterStatus: | ||
| """Get the emeter status.""" | ||
| return EmeterStatus({"power_mw": self._energy.get("current_power")}) | ||
| return EmeterStatus( | ||
| { | ||
| "power_mw": self._energy.get("current_power"), | ||
| "total": self._convert_energy_data( | ||
| self._energy.get("today_energy"), 1 / 1000 | ||
| ), | ||
| } | ||
| ) | ||
|
|
||
| @property | ||
| def emeter_today(self) -> Optional[float]: | ||
| """Get the emeter value for today.""" | ||
| return None | ||
| return self._convert_energy_data(self._energy.get("today_energy"), 1 / 1000) | ||
|
|
||
| @property | ||
| def emeter_this_month(self) -> Optional[float]: | ||
| """Get the emeter value for this month.""" | ||
| return None | ||
| return self._convert_energy_data(self._energy.get("month_energy"), 1 / 1000) | ||
|
|
||
| @property | ||
| def on_since(self) -> Optional[datetime]: | ||
| Expand All | @@ -71,3 +88,7 @@ def on_since(self) -> Optional[datetime]: | |
| return None | ||
| on_time = cast(float, self._info.get("on_time")) | ||
| return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time) | ||
|
|
||
| def _convert_energy_data(self, data, scale) -> Optional[float]: | ||
| """Return adjusted emeter information.""" | ||
| return data if not data else data * scale | ||
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.