jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 00:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Exchange co2signal package with aioelectricitymaps (#101955)

Browse Source
This commit is contained in:
Jan-Philipp Benecke 2023-11-13 20:48:33 +01:00 committed by GitHub
parent 2557e41ec0
commit 685537e475
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 135 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -187,6 +187,7 @@ omit =
homeassistant/components/control4/director_utils.py
homeassistant/components/control4/light.py
homeassistant/components/coolmaster/coordinator.py
homeassistant/components/co2signal/coordinator.py
homeassistant/components/cppm_tracker/device_tracker.py
homeassistant/components/crownstone/__init__.py
homeassistant/components/crownstone/devices.py

10
homeassistant/components/co2signal/__init__.py
View File

@ -1,9 +1,12 @@
"""The CO2 Signal integration."""
from __future__ import annotations
from aioelectricitymaps import ElectricityMaps
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import CO2SignalCoordinator
@ -13,7 +16,10 @@ PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up CO2 Signal from a config entry."""
coordinator = CO2SignalCoordinator(hass, entry)
session = async_get_clientsession(hass)
coordinator = CO2SignalCoordinator(
hass, ElectricityMaps(token=entry.data[CONF_API_KEY], session=session)
)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator

32
homeassistant/components/co2signal/config_flow.py
View File

@ -3,11 +3,14 @@ from __future__ import annotations
from typing import Any
from aioelectricitymaps import ElectricityMaps
from aioelectricitymaps.exceptions import ElectricityMapsError, InvalidToken
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.selector import (
SelectSelector,
@ -16,8 +19,7 @@ from homeassistant.helpers.selector import (
)
from .const import CONF_COUNTRY_CODE, DOMAIN
from .coordinator import get_data
from .exceptions import APIRatelimitExceeded, InvalidAuth
from .helpers import fetch_latest_carbon_intensity
from .util import get_extra_name
TYPE_USE_HOME = "use_home_location"
@ -117,19 +119,19 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Validate data and show form if it is invalid."""
errors: dict[str, str] = {}
try:
await self.hass.async_add_executor_job(get_data, self.hass, data)
except InvalidAuth:
errors["base"] = "invalid_auth"
except APIRatelimitExceeded:
errors["base"] = "api_ratelimit"
except Exception: # pylint: disable=broad-except
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=get_extra_name(data) or "CO2 Signal",
data=data,
)
session = async_get_clientsession(self.hass)
async with ElectricityMaps(token=data[CONF_API_KEY], session=session) as em:
try:
await fetch_latest_carbon_intensity(self.hass, em, data)
except InvalidToken:
errors["base"] = "invalid_auth"
except ElectricityMapsError:
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=get_extra_name(data) or "CO2 Signal",
data=data,
)
return self.async_show_form(
step_id=step_id,

88
homeassistant/components/co2signal/coordinator.py
View File

@ -1,94 +1,50 @@
"""DataUpdateCoordinator for the co2signal integration."""
from __future__ import annotations
from collections.abc import Mapping
from datetime import timedelta
import logging
from typing import Any, cast
import CO2Signal
from requests.exceptions import JSONDecodeError
from aioelectricitymaps import ElectricityMaps
from aioelectricitymaps.exceptions import ElectricityMapsError, InvalidToken
from aioelectricitymaps.models import CarbonIntensityResponse
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_COUNTRY_CODE, DOMAIN
from .exceptions import APIRatelimitExceeded, CO2Error, InvalidAuth, UnknownError
from .models import CO2SignalResponse
from .const import DOMAIN
from .helpers import fetch_latest_carbon_intensity
_LOGGER = logging.getLogger(__name__)
class CO2SignalCoordinator(DataUpdateCoordinator[CO2SignalResponse]):
class CO2SignalCoordinator(DataUpdateCoordinator[CarbonIntensityResponse]):
"""Data update coordinator."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, client: ElectricityMaps) -> None:
"""Initialize the coordinator."""
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=timedelta(minutes=15)
)
self._entry = entry
self.client = client
@property
def entry_id(self) -> str:
"""Return entry ID."""
return self._entry.entry_id
return self.config_entry.entry_id
async def _async_update_data(self) -> CO2SignalResponse:
async def _async_update_data(self) -> CarbonIntensityResponse:
"""Fetch the latest data from the source."""
try:
data = await self.hass.async_add_executor_job(
get_data, self.hass, self._entry.data
)
except InvalidAuth as err:
raise ConfigEntryAuthFailed from err
except CO2Error as err:
raise UpdateFailed(str(err)) from err
return data
def get_data(hass: HomeAssistant, config: Mapping[str, Any]) -> CO2SignalResponse:
"""Get data from the API."""
if CONF_COUNTRY_CODE in config:
latitude = None
longitude = None
else:
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
try:
data = CO2Signal.get_latest(
config[CONF_API_KEY],
config.get(CONF_COUNTRY_CODE),
latitude,
longitude,
wait=False,
)
except JSONDecodeError as err:
# raise occasional occurring json decoding errors as CO2Error so the data update coordinator retries it
raise CO2Error from err
except ValueError as err:
err_str = str(err)
if "Invalid authentication credentials" in err_str:
raise InvalidAuth from err
if "API rate limit exceeded." in err_str:
raise APIRatelimitExceeded from err
_LOGGER.exception("Unexpected exception")
raise UnknownError from err
if "error" in data:
raise UnknownError(data["error"])
if data.get("status") != "ok":
_LOGGER.exception("Unexpected response: %s", data)
raise UnknownError
return cast(CO2SignalResponse, data)
async with self.client as em:
try:
return await fetch_latest_carbon_intensity(
self.hass, em, self.config_entry.data
)
except InvalidToken as err:
raise ConfigEntryError from err
except ElectricityMapsError as err:
raise UpdateFailed(str(err)) from err

3
homeassistant/components/co2signal/diagnostics.py
View File

@ -1,6 +1,7 @@
"""Diagnostics support for CO2Signal."""
from __future__ import annotations
from dataclasses import asdict
from typing import Any
from homeassistant.components.diagnostics import async_redact_data
@ -22,5 +23,5 @@ async def async_get_config_entry_diagnostics(
return {
"config_entry": async_redact_data(config_entry.as_dict(), TO_REDACT),
"data": coordinator.data,
"data": asdict(coordinator.data),
}

18
homeassistant/components/co2signal/exceptions.py
View File

@ -1,18 +0,0 @@
"""Exceptions to the co2signal integration."""
from homeassistant.exceptions import HomeAssistantError
class CO2Error(HomeAssistantError):
"""Base error."""
class InvalidAuth(CO2Error):
"""Raised when invalid authentication credentials are provided."""
class APIRatelimitExceeded(CO2Error):
"""Raised when the API rate limit is exceeded."""
class UnknownError(CO2Error):
"""Raised when an unknown error occurs."""

28
homeassistant/components/co2signal/helpers.py Normal file
View File

@ -0,0 +1,28 @@
"""Helper functions for the CO2 Signal integration."""
from types import MappingProxyType
from typing import Any
from aioelectricitymaps import ElectricityMaps
from aioelectricitymaps.models import CarbonIntensityResponse
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from .const import CONF_COUNTRY_CODE
async def fetch_latest_carbon_intensity(
hass: HomeAssistant,
em: ElectricityMaps,
config: dict[str, Any] | MappingProxyType[str, Any],
) -> CarbonIntensityResponse:
"""Fetch the latest carbon intensity based on country code or location coordinates."""
if CONF_COUNTRY_CODE in config:
return await em.latest_carbon_intensity_by_country_code(
code=config[CONF_COUNTRY_CODE]
)
return await em.latest_carbon_intensity_by_coordinates(
lat=config.get(CONF_LATITUDE, hass.config.latitude),
lon=config.get(CONF_LONGITUDE, hass.config.longitude),
)

4
homeassistant/components/co2signal/manifest.json
View File

@ -6,6 +6,6 @@
"documentation": "https://www.home-assistant.io/integrations/co2signal",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["CO2Signal"],
"requirements": ["CO2Signal==0.4.2"]
"loggers": ["aioelectricitymaps"],
"requirements": ["aioelectricitymaps==0.1.5"]
}

24
homeassistant/components/co2signal/models.py
View File

@ -1,24 +0,0 @@
"""Models to the co2signal integration."""
from typing import TypedDict
class CO2SignalData(TypedDict):
"""Data field."""
carbonIntensity: float
fossilFuelPercentage: float
class CO2SignalUnit(TypedDict):
"""Unit field."""
carbonIntensity: str
class CO2SignalResponse(TypedDict):
"""API response."""
status: str
countryCode: str
data: CO2SignalData
units: CO2SignalUnit

47
homeassistant/components/co2signal/sensor.py
View File

@ -1,9 +1,11 @@
"""Support for the CO2signal platform."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
from typing import cast
from aioelectricitymaps.models import CarbonIntensityResponse
from homeassistant.components.sensor import (
SensorEntity,
@ -24,11 +26,21 @@ SCAN_INTERVAL = timedelta(minutes=3)
@dataclass
class CO2SensorEntityDescription(SensorEntityDescription):
class ElectricityMapsMixin:
"""Mixin for value and unit_of_measurement_fn function."""
value_fn: Callable[[CarbonIntensityResponse], float | None]
@dataclass
class CO2SensorEntityDescription(SensorEntityDescription, ElectricityMapsMixin):
"""Provide a description of a CO2 sensor."""
# For backwards compat, allow description to override unique ID key to use
unique_id: str | None = None
unit_of_measurement_fn: Callable[
[CarbonIntensityResponse], str | None
] | None = None
SENSORS = (
@ -36,12 +48,14 @@ SENSORS = (
key="carbonIntensity",
translation_key="carbon_intensity",
unique_id="co2intensity",
# No unit, it's extracted from response.
value_fn=lambda response: response.data.carbon_intensity,
unit_of_measurement_fn=lambda response: response.units.carbon_intensity,
),
CO2SensorEntityDescription(
key="fossilFuelPercentage",
translation_key="fossil_fuel_percentage",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda response: response.data.fossil_fuel_percentage,
),
)
@ -51,7 +65,9 @@ async def async_setup_entry(
) -> None:
"""Set up the CO2signal sensor."""
coordinator: CO2SignalCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(CO2Sensor(coordinator, description) for description in SENSORS)
async_add_entities(
[CO2Sensor(coordinator, description) for description in SENSORS], False
)
class CO2Sensor(CoordinatorEntity[CO2SignalCoordinator], SensorEntity):
@ -71,7 +87,7 @@ class CO2Sensor(CoordinatorEntity[CO2SignalCoordinator], SensorEntity):
self.entity_description = description
self._attr_extra_state_attributes = {
"country_code": coordinator.data["countryCode"],
"country_code": coordinator.data.country_code,
}
self._attr_device_info = DeviceInfo(
configuration_url="https://www.electricitymaps.com/",
@ -84,26 +100,15 @@ class CO2Sensor(CoordinatorEntity[CO2SignalCoordinator], SensorEntity):
f"{coordinator.entry_id}_{description.unique_id or description.key}"
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return (
super().available
and self.entity_description.key in self.coordinator.data["data"]
)
@property
def native_value(self) -> float | None:
"""Return sensor state."""
if (value := self.coordinator.data["data"][self.entity_description.key]) is None: # type: ignore[literal-required]
return None
return round(value, 2)
return self.entity_description.value_fn(self.coordinator.data)
@property
def native_unit_of_measurement(self) -> str | None:
"""Return the unit of measurement."""
if self.entity_description.native_unit_of_measurement:
return self.entity_description.native_unit_of_measurement
return cast(
str, self.coordinator.data["units"].get(self.entity_description.key)
)
if self.entity_description.unit_of_measurement_fn:
return self.entity_description.unit_of_measurement_fn(self.coordinator.data)
return self.entity_description.native_unit_of_measurement

6
requirements_all.txt
View File

@ -21,9 +21,6 @@ Ambiclimate==0.2.1
# homeassistant.components.blinksticklight
BlinkStick==1.2.0
# homeassistant.components.co2signal
CO2Signal==0.4.2
# homeassistant.components.doorbird
DoorBirdPy==2.1.0
@ -235,6 +232,9 @@ aioeagle==1.1.0
# homeassistant.components.ecowitt
aioecowitt==2023.5.0
# homeassistant.components.co2signal
aioelectricitymaps==0.1.5
# homeassistant.components.emonitor
aioemonitor==1.0.5

6
requirements_test_all.txt
View File

@ -18,9 +18,6 @@ Adax-local==0.1.5
# homeassistant.components.ambiclimate
Ambiclimate==0.2.1
# homeassistant.components.co2signal
CO2Signal==0.4.2
# homeassistant.components.doorbird
DoorBirdPy==2.1.0
@ -214,6 +211,9 @@ aioeagle==1.1.0
# homeassistant.components.ecowitt
aioecowitt==2023.5.0
# homeassistant.components.co2signal
aioelectricitymaps==0.1.5
# homeassistant.components.emonitor
aioemonitor==1.0.5

8
tests/components/co2signal/snapshots/test_diagnostics.ambr
View File

@ -19,14 +19,14 @@
'version': 1,
}),
'data': dict({
'countryCode': 'FR',
'country_code': 'FR',
'data': dict({
'carbonIntensity': 45.98623190095805,
'fossilFuelPercentage': 5.461182741937103,
'carbon_intensity': 45.98623190095805,
'fossil_fuel_percentage': 5.461182741937103,
}),
'status': 'ok',
'units': dict({
'carbonIntensity': 'gCO2eq/kWh',
'carbon_intensity': 'gCO2eq/kWh',
}),
}),
})

37
tests/components/co2signal/test_config_flow.py
View File

@ -1,7 +1,11 @@
"""Test the CO2 Signal config flow."""
from json import JSONDecodeError
from unittest.mock import Mock, patch
from unittest.mock import patch
from aioelectricitymaps.exceptions import (
ElectricityMapsDecodeError,
ElectricityMapsError,
InvalidToken,
)
import pytest
from homeassistant import config_entries
@ -22,7 +26,7 @@ async def test_form_home(hass: HomeAssistant) -> None:
assert result["errors"] is None
with patch(
"CO2Signal.get_latest",
"homeassistant.components.co2signal.config_flow.ElectricityMaps._get",
return_value=VALID_PAYLOAD,
), patch(
"homeassistant.components.co2signal.async_setup_entry",
@ -64,7 +68,7 @@ async def test_form_coordinates(hass: HomeAssistant) -> None:
assert result2["type"] == FlowResultType.FORM
with patch(
"CO2Signal.get_latest",
"homeassistant.components.co2signal.config_flow.ElectricityMaps._get",
return_value=VALID_PAYLOAD,
), patch(
"homeassistant.components.co2signal.async_setup_entry",
@ -108,7 +112,7 @@ async def test_form_country(hass: HomeAssistant) -> None:
assert result2["type"] == FlowResultType.FORM
with patch(
"CO2Signal.get_latest",
"homeassistant.components.co2signal.config_flow.ElectricityMaps._get",
return_value=VALID_PAYLOAD,
), patch(
"homeassistant.components.co2signal.async_setup_entry",
@ -135,27 +139,16 @@ async def test_form_country(hass: HomeAssistant) -> None:
("side_effect", "err_code"),
[
(
ValueError("Invalid authentication credentials"),
InvalidToken,
"invalid_auth",
),
(
ValueError("API rate limit exceeded."),
"api_ratelimit",
),
(ValueError("Something else"), "unknown"),
(JSONDecodeError(msg="boom", doc="", pos=1), "unknown"),
(Exception("Boom"), "unknown"),
(Mock(return_value={"error": "boom"}), "unknown"),
(Mock(return_value={"status": "error"}), "unknown"),
(ElectricityMapsError("Something else"), "unknown"),
(ElectricityMapsDecodeError("Boom"), "unknown"),
],
ids=[
"invalid auth",
"rate limit exceeded",
"unknown value error",
"generic error",
"json decode error",
"unknown error",
"error in json dict",
"status error",
],
)
async def test_form_error_handling(hass: HomeAssistant, side_effect, err_code) -> None:
@ -165,7 +158,7 @@ async def test_form_error_handling(hass: HomeAssistant, side_effect, err_code) -
)
with patch(
"CO2Signal.get_latest",
"homeassistant.components.co2signal.config_flow.ElectricityMaps._get",
side_effect=side_effect,
):
result = await hass.config_entries.flow.async_configure(
@ -180,7 +173,7 @@ async def test_form_error_handling(hass: HomeAssistant, side_effect, err_code) -
assert result["errors"] == {"base": err_code}
with patch(
"CO2Signal.get_latest",
"homeassistant.components.co2signal.config_flow.ElectricityMaps._get",
return_value=VALID_PAYLOAD,
):
result = await hass.config_entries.flow.async_configure(

5
tests/components/co2signal/test_diagnostics.py
View File

@ -27,7 +27,10 @@ async def test_entry_diagnostics(
entry_id="904a74160aa6f335526706bee85dfb83",
)
config_entry.add_to_hass(hass)
with patch("CO2Signal.get_latest", return_value=VALID_PAYLOAD):
with patch(
"homeassistant.components.co2signal.coordinator.ElectricityMaps._get",
return_value=VALID_PAYLOAD,
):
assert await async_setup_component(hass, DOMAIN, {})
result = await get_diagnostics_for_config_entry(hass, hass_client, config_entry)