Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move rainforest_eagle coordinator to separate module (#117556)

pull/117558/head
epenet 2024-05-16 12:57:20 +02:00 committed by GitHub
parent 32a9cb4b14
commit 388132cfc8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 142 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/rainforest_eagle/__init__.py
View File

@ -6,15 +6,15 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import data
from .const import DOMAIN
from .coordinator import EagleDataCoordinator
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Rainforest Eagle from a config entry."""
coordinator = data.EagleDataCoordinator(hass, entry)
coordinator = EagleDataCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

8
homeassistant/components/rainforest_eagle/config_flow.py
View File

@ -10,8 +10,8 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_TYPE
from . import data
from .const import CONF_CLOUD_ID, CONF_HARDWARE_ADDRESS, CONF_INSTALL_CODE, DOMAIN
from .data import CannotConnect, InvalidAuth, async_get_type
_LOGGER = logging.getLogger(__name__)
@ -49,15 +49,15 @@ class RainforestEagleConfigFlow(ConfigFlow, domain=DOMAIN):
errors = {}
try:
eagle_type, hardware_address = await data.async_get_type(
eagle_type, hardware_address = await async_get_type(
self.hass,
user_input[CONF_CLOUD_ID],
user_input[CONF_INSTALL_CODE],
user_input[CONF_HOST],
)
except data.CannotConnect:
except CannotConnect:
errors["base"] = "cannot_connect"
except data.InvalidAuth:
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")

131
homeassistant/components/rainforest_eagle/coordinator.py Normal file
View File

@ -0,0 +1,131 @@
"""Rainforest data."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import aioeagle
from eagle100 import Eagle as Eagle100Reader
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_CLOUD_ID,
CONF_HARDWARE_ADDRESS,
CONF_INSTALL_CODE,
TYPE_EAGLE_100,
)
from .data import UPDATE_100_ERRORS
_LOGGER = logging.getLogger(__name__)
class EagleDataCoordinator(DataUpdateCoordinator):
"""Get the latest data from the Eagle device."""
eagle100_reader: Eagle100Reader | None = None
eagle200_meter: aioeagle.ElectricMeter | None = None
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the data object."""
self.entry = entry
if self.type == TYPE_EAGLE_100:
self.model = "EAGLE-100"
update_method = self._async_update_data_100
else:
self.model = "EAGLE-200"
update_method = self._async_update_data_200
super().__init__(
hass,
_LOGGER,
name=entry.data[CONF_CLOUD_ID],
update_interval=timedelta(seconds=30),
update_method=update_method,
)
@property
def cloud_id(self):
"""Return the cloud ID."""
return self.entry.data[CONF_CLOUD_ID]
@property
def type(self):
"""Return entry type."""
return self.entry.data[CONF_TYPE]
@property
def hardware_address(self):
"""Return hardware address of meter."""
return self.entry.data[CONF_HARDWARE_ADDRESS]
@property
def is_connected(self):
"""Return if the hub is connected to the electric meter."""
if self.eagle200_meter:
return self.eagle200_meter.is_connected
return True
async def _async_update_data_200(self):
"""Get the latest data from the Eagle-200 device."""
if (eagle200_meter := self.eagle200_meter) is None:
hub = aioeagle.EagleHub(
aiohttp_client.async_get_clientsession(self.hass),
self.cloud_id,
self.entry.data[CONF_INSTALL_CODE],
host=self.entry.data[CONF_HOST],
)
eagle200_meter = aioeagle.ElectricMeter.create_instance(
hub, self.hardware_address
)
is_connected = True
else:
is_connected = eagle200_meter.is_connected
async with asyncio.timeout(30):
data = await eagle200_meter.get_device_query()
if self.eagle200_meter is None:
self.eagle200_meter = eagle200_meter
elif is_connected and not eagle200_meter.is_connected:
_LOGGER.warning("Lost connection with electricity meter")
_LOGGER.debug("API data: %s", data)
return {var["Name"]: var["Value"] for var in data.values()}
async def _async_update_data_100(self):
"""Get the latest data from the Eagle-100 device."""
try:
data = await self.hass.async_add_executor_job(self._fetch_data_100)
except UPDATE_100_ERRORS as error:
raise UpdateFailed from error
_LOGGER.debug("API data: %s", data)
return data
def _fetch_data_100(self):
"""Fetch and return the four sensor values in a dict."""
if self.eagle100_reader is None:
self.eagle100_reader = Eagle100Reader(
self.cloud_id,
self.entry.data[CONF_INSTALL_CODE],
self.entry.data[CONF_HOST],
)
out = {}
resp = self.eagle100_reader.get_instantaneous_demand()["InstantaneousDemand"]
out["zigbee:InstantaneousDemand"] = resp["Demand"]
resp = self.eagle100_reader.get_current_summation()["CurrentSummation"]
out["zigbee:CurrentSummationDelivered"] = resp["SummationDelivered"]
out["zigbee:CurrentSummationReceived"] = resp["SummationReceived"]
return out

118
homeassistant/components/rainforest_eagle/data.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
import aioeagle
@ -11,20 +10,10 @@ import aiohttp
from eagle100 import Eagle as Eagle100Reader
from requests.exceptions import ConnectionError as ConnectError, HTTPError, Timeout
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
CONF_CLOUD_ID,
CONF_HARDWARE_ADDRESS,
CONF_INSTALL_CODE,
TYPE_EAGLE_100,
TYPE_EAGLE_200,
)
from .const import TYPE_EAGLE_100, TYPE_EAGLE_200
_LOGGER = logging.getLogger(__name__)
@ -86,108 +75,3 @@ async def async_get_type(hass, cloud_id, install_code, host):
return TYPE_EAGLE_100, None
return None, None
class EagleDataCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Get the latest data from the Eagle device."""
eagle100_reader: Eagle100Reader | None = None
eagle200_meter: aioeagle.ElectricMeter | None = None
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the data object."""
self.entry = entry
if self.type == TYPE_EAGLE_100:
self.model = "EAGLE-100"
update_method = self._async_update_data_100
else:
self.model = "EAGLE-200"
update_method = self._async_update_data_200
super().__init__(
hass,
_LOGGER,
name=entry.data[CONF_CLOUD_ID],
update_interval=timedelta(seconds=30),
update_method=update_method,
)
@property
def cloud_id(self):
"""Return the cloud ID."""
return self.entry.data[CONF_CLOUD_ID]
@property
def type(self):
"""Return entry type."""
return self.entry.data[CONF_TYPE]
@property
def hardware_address(self):
"""Return hardware address of meter."""
return self.entry.data[CONF_HARDWARE_ADDRESS]
@property
def is_connected(self):
"""Return if the hub is connected to the electric meter."""
if self.eagle200_meter:
return self.eagle200_meter.is_connected
return True
async def _async_update_data_200(self):
"""Get the latest data from the Eagle-200 device."""
if (eagle200_meter := self.eagle200_meter) is None:
hub = aioeagle.EagleHub(
aiohttp_client.async_get_clientsession(self.hass),
self.cloud_id,
self.entry.data[CONF_INSTALL_CODE],
host=self.entry.data[CONF_HOST],
)
eagle200_meter = aioeagle.ElectricMeter.create_instance(
hub, self.hardware_address
)
is_connected = True
else:
is_connected = eagle200_meter.is_connected
async with asyncio.timeout(30):
data = await eagle200_meter.get_device_query()
if self.eagle200_meter is None:
self.eagle200_meter = eagle200_meter
elif is_connected and not eagle200_meter.is_connected:
_LOGGER.warning("Lost connection with electricity meter")
_LOGGER.debug("API data: %s", data)
return {var["Name"]: var["Value"] for var in data.values()}
async def _async_update_data_100(self):
"""Get the latest data from the Eagle-100 device."""
try:
data = await self.hass.async_add_executor_job(self._fetch_data_100)
except UPDATE_100_ERRORS as error:
raise UpdateFailed from error
_LOGGER.debug("API data: %s", data)
return data
def _fetch_data_100(self):
"""Fetch and return the four sensor values in a dict."""
if self.eagle100_reader is None:
self.eagle100_reader = Eagle100Reader(
self.cloud_id,
self.entry.data[CONF_INSTALL_CODE],
self.entry.data[CONF_HOST],
)
out = {}
resp = self.eagle100_reader.get_instantaneous_demand()["InstantaneousDemand"]
out["zigbee:InstantaneousDemand"] = resp["Demand"]
resp = self.eagle100_reader.get_current_summation()["CurrentSummation"]
out["zigbee:CurrentSummationDelivered"] = resp["SummationDelivered"]
out["zigbee:CurrentSummationReceived"] = resp["SummationReceived"]
return out

2
homeassistant/components/rainforest_eagle/diagnostics.py
View File

@ -9,7 +9,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import CONF_CLOUD_ID, CONF_INSTALL_CODE, DOMAIN
from .data import EagleDataCoordinator
from .coordinator import EagleDataCoordinator
TO_REDACT = {CONF_CLOUD_ID, CONF_INSTALL_CODE}

2
homeassistant/components/rainforest_eagle/sensor.py
View File

@ -17,7 +17,7 @@ from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .data import EagleDataCoordinator
from .coordinator import EagleDataCoordinator
SENSORS = (
SensorEntityDescription(

2
tests/components/rainforest_eagle/conftest.py
View File

@ -66,7 +66,7 @@ async def setup_rainforest_100(hass):
},
).add_to_hass(hass)
with patch(
"homeassistant.components.rainforest_eagle.data.Eagle100Reader",
"homeassistant.components.rainforest_eagle.coordinator.Eagle100Reader",
return_value=Mock(
get_instantaneous_demand=Mock(
return_value={"InstantaneousDemand": {"Demand": "1.152000"}}

2
tests/components/rainforest_eagle/test_config_flow.py
View File

@ -27,7 +27,7 @@ async def test_form(hass: HomeAssistant) -> None:
with (
patch(
"homeassistant.components.rainforest_eagle.data.async_get_type",
"homeassistant.components.rainforest_eagle.config_flow.async_get_type",
return_value=(TYPE_EAGLE_200, "mock-hw"),
),
patch(