Add outage map alerts to PECO (#69825)
This commit is contained in:
parent
c87992715f
commit
6c75eaa1bc
8 changed files with 86 additions and 33 deletions
|
@ -5,7 +5,7 @@ import asyncio
|
|||
from datetime import timedelta
|
||||
from typing import Final
|
||||
|
||||
from peco import BadJSONError, HttpError, OutageResults, PecoOutageApi
|
||||
from peco import AlertResults, BadJSONError, HttpError, OutageResults, PecoOutageApi
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
|
@ -18,6 +18,18 @@ from .const import CONF_COUNTY, DOMAIN, LOGGER, SCAN_INTERVAL
|
|||
PLATFORMS: Final = [Platform.SENSOR]
|
||||
|
||||
|
||||
class PECOCoordinatorData:
|
||||
"""Something to hold the data for PECO."""
|
||||
|
||||
outages: OutageResults
|
||||
alerts: AlertResults
|
||||
|
||||
def __init__(self, outages: OutageResults, alerts: AlertResults) -> None:
|
||||
"""Initialize the data holder for PECO."""
|
||||
self.outages = outages
|
||||
self.alerts = alerts
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up PECO Outage Counter from a config entry."""
|
||||
|
||||
|
@ -25,14 +37,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
api = PecoOutageApi()
|
||||
county: str = entry.data[CONF_COUNTY]
|
||||
|
||||
async def async_update_data() -> OutageResults:
|
||||
async def async_update_data() -> PECOCoordinatorData:
|
||||
"""Fetch data from API."""
|
||||
try:
|
||||
data: OutageResults = (
|
||||
outages: OutageResults = (
|
||||
await api.get_outage_totals(websession)
|
||||
if county == "TOTAL"
|
||||
else await api.get_outage_count(county, websession)
|
||||
)
|
||||
alerts: AlertResults = await api.get_map_alerts(websession)
|
||||
data = PECOCoordinatorData(outages, alerts)
|
||||
except HttpError as err:
|
||||
raise UpdateFailed(f"Error fetching data: {err}") from err
|
||||
except BadJSONError as err:
|
||||
|
|
|
@ -16,3 +16,4 @@ COUNTY_LIST: Final = [
|
|||
CONFIG_FLOW_COUNTIES: Final = [{county: county.capitalize()} for county in COUNTY_LIST]
|
||||
SCAN_INTERVAL: Final = 9
|
||||
CONF_COUNTY: Final = "county"
|
||||
ATTR_CONTENT: Final = "content"
|
||||
|
|
|
@ -5,5 +5,5 @@
|
|||
"documentation": "https://www.home-assistant.io/integrations/peco",
|
||||
"codeowners": ["@IceBotYT"],
|
||||
"iot_class": "cloud_polling",
|
||||
"requirements": ["peco==0.0.25"]
|
||||
"requirements": ["peco==0.0.29"]
|
||||
}
|
||||
|
|
|
@ -5,8 +5,6 @@ from collections.abc import Callable
|
|||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
from peco import OutageResults
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
|
@ -21,14 +19,16 @@ from homeassistant.helpers.update_coordinator import (
|
|||
DataUpdateCoordinator,
|
||||
)
|
||||
|
||||
from .const import CONF_COUNTY, DOMAIN
|
||||
from . import PECOCoordinatorData
|
||||
from .const import ATTR_CONTENT, CONF_COUNTY, DOMAIN
|
||||
|
||||
|
||||
@dataclass
|
||||
class PECOSensorEntityDescriptionMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
value_fn: Callable[[OutageResults], int]
|
||||
value_fn: Callable[[PECOCoordinatorData], int | str]
|
||||
attribute_fn: Callable[[PECOCoordinatorData], dict[str, str]]
|
||||
|
||||
|
||||
@dataclass
|
||||
|
@ -43,23 +43,33 @@ SENSOR_LIST: tuple[PECOSensorEntityDescription, ...] = (
|
|||
PECOSensorEntityDescription(
|
||||
key="customers_out",
|
||||
name="Customers Out",
|
||||
value_fn=lambda data: int(data.customers_out),
|
||||
value_fn=lambda data: int(data.outages.customers_out),
|
||||
attribute_fn=lambda data: {},
|
||||
),
|
||||
PECOSensorEntityDescription(
|
||||
key="percent_customers_out",
|
||||
name="Percent Customers Out",
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
value_fn=lambda data: int(data.percent_customers_out),
|
||||
value_fn=lambda data: int(data.outages.percent_customers_out),
|
||||
attribute_fn=lambda data: {},
|
||||
),
|
||||
PECOSensorEntityDescription(
|
||||
key="outage_count",
|
||||
name="Outage Count",
|
||||
value_fn=lambda data: int(data.outage_count),
|
||||
value_fn=lambda data: int(data.outages.outage_count),
|
||||
attribute_fn=lambda data: {},
|
||||
),
|
||||
PECOSensorEntityDescription(
|
||||
key="customers_served",
|
||||
name="Customers Served",
|
||||
value_fn=lambda data: int(data.customers_served),
|
||||
value_fn=lambda data: int(data.outages.customers_served),
|
||||
attribute_fn=lambda data: {},
|
||||
),
|
||||
PECOSensorEntityDescription(
|
||||
key="map_alert",
|
||||
name="Map Alert",
|
||||
value_fn=lambda data: str(data.alerts.alert_title),
|
||||
attribute_fn=lambda data: {ATTR_CONTENT: data.alerts.alert_content},
|
||||
),
|
||||
)
|
||||
|
||||
|
@ -80,26 +90,36 @@ async def async_setup_entry(
|
|||
return
|
||||
|
||||
|
||||
class PecoSensor(CoordinatorEntity[DataUpdateCoordinator[OutageResults]], SensorEntity):
|
||||
class PecoSensor(
|
||||
CoordinatorEntity[DataUpdateCoordinator[PECOCoordinatorData]], SensorEntity
|
||||
):
|
||||
"""PECO outage counter sensor."""
|
||||
|
||||
_attr_state_class = SensorStateClass.MEASUREMENT
|
||||
_attr_icon: str = "mdi:power-plug-off"
|
||||
entity_description: PECOSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
description: PECOSensorEntityDescription,
|
||||
county: str,
|
||||
coordinator: DataUpdateCoordinator[OutageResults],
|
||||
coordinator: DataUpdateCoordinator[PECOCoordinatorData],
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self._attr_name = f"{county.capitalize()} {description.name}"
|
||||
self._attr_unique_id = f"{county}-{description.key}"
|
||||
self._attr_icon = (
|
||||
"mdi:alert" if description.key == "map_alert" else "mdi:power-plug-off"
|
||||
)
|
||||
if description.key != "map_alert":
|
||||
self._attr_state_class = SensorStateClass.MEASUREMENT
|
||||
self.entity_description = description
|
||||
|
||||
@property
|
||||
def native_value(self) -> int:
|
||||
def native_value(self) -> int | str:
|
||||
"""Return the value of the sensor."""
|
||||
return self.entity_description.value_fn(self.coordinator.data)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, str]:
|
||||
"""Return state attributes for the sensor."""
|
||||
return self.entity_description.attribute_fn(self.coordinator.data)
|
||||
|
|
|
@ -1184,7 +1184,7 @@ panasonic_viera==0.3.6
|
|||
pdunehd==1.3.2
|
||||
|
||||
# homeassistant.components.peco
|
||||
peco==0.0.25
|
||||
peco==0.0.29
|
||||
|
||||
# homeassistant.components.pencom
|
||||
pencompy==0.0.3
|
||||
|
|
|
@ -789,7 +789,7 @@ panasonic_viera==0.3.6
|
|||
pdunehd==1.3.2
|
||||
|
||||
# homeassistant.components.peco
|
||||
peco==0.0.25
|
||||
peco==0.0.29
|
||||
|
||||
# homeassistant.components.aruba
|
||||
# homeassistant.components.cisco_ios
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
import asyncio
|
||||
from unittest.mock import patch
|
||||
|
||||
from peco import BadJSONError, HttpError
|
||||
from peco import AlertResults, BadJSONError, HttpError, OutageResults
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.peco.const import DOMAIN
|
||||
|
@ -23,15 +23,21 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
|
|||
|
||||
with patch(
|
||||
"peco.PecoOutageApi.get_outage_totals",
|
||||
return_value={
|
||||
"customers_out": 0,
|
||||
"percent_customers_out": 0,
|
||||
"outage_count": 0,
|
||||
"customers_served": 350394,
|
||||
},
|
||||
return_value=OutageResults(
|
||||
customers_out=0,
|
||||
percent_customers_out=0,
|
||||
outage_count=0,
|
||||
customers_served=350394,
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
with patch(
|
||||
"peco.PecoOutageApi.get_map_alerts",
|
||||
return_value=AlertResults(
|
||||
alert_content="Testing 1234", alert_title="Testing 4321"
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.data[DOMAIN]
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Test the PECO Outage Counter sensors."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from peco import OutageResults
|
||||
from peco import AlertResults, OutageResults
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.peco.const import DOMAIN
|
||||
|
@ -42,8 +42,14 @@ async def test_sensor_available(
|
|||
customers_served=789,
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
with patch(
|
||||
"peco.PecoOutageApi.get_map_alerts",
|
||||
return_value=AlertResults(
|
||||
alert_content="Testing 1234", alert_title="Testing 4321"
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.data[DOMAIN]
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
|
@ -69,8 +75,14 @@ async def test_sensor_available(
|
|||
customers_served=789,
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
with patch(
|
||||
"peco.PecoOutageApi.get_map_alerts",
|
||||
return_value=AlertResults(
|
||||
alert_content="Testing 1234", alert_title="Testing 4321"
|
||||
),
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(entries) == 2
|
||||
|
|
Loading…
Add table
Reference in a new issue