Commander1024/homeassistant-config
Commander1024/homeassistant-config
1
0
Fork 0
Code Pull Requests Projects Releases Activity

Compare commits

base: Commander1024:32e9c3af4dee57904abc6ff9ed6307d2ff13e6cf
Branches Tags
Commander1024:main
..
compare: Commander1024:8cd5ae5283f5d422f9e2351f71bed85131891f5c
Branches Tags
Commander1024:main

No commits in common. "32e9c3af4dee57904abc6ff9ed6307d2ff13e6cf" and "8cd5ae5283f5d422f9e2351f71bed85131891f5c" have entirely different histories.
32e9c3af4d ... 8cd5ae5283

10 changed files with 1 additions and 478 deletions
Show Stats Expand all files Collapse all files

2
.HA_VERSION
View File

@ -1 +1 @@
2024.10.1 2024.10.0

76
custom_components/hochwasserportal/__init__.py
View File

@ -1,76 +0,0 @@
"""The Länderübergreifendes Hochwasser Portal integration."""
from __future__ import annotations
from lhpapi import HochwasserPortalAPI, LHPError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import (
CONF_ADD_UNAVAILABLE,
CONF_PEGEL_IDENTIFIER,
DOMAIN,
LOGGER,
PLATFORMS,
)
from .coordinator import HochwasserPortalCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
pegel_identifier: str = entry.data[CONF_PEGEL_IDENTIFIER]
# Initialize the API and coordinator.
try:
api = await hass.async_add_executor_job(HochwasserPortalAPI, pegel_identifier)
coordinator = HochwasserPortalCoordinator(hass, api)
except LHPError as err:
LOGGER.exception("Setup of %s failed: %s", pegel_identifier, err)
return False
# No need to refresh via the following line because api runs
# update during init automatically
# 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)
return True
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
LOGGER.debug(
"Migrating %s from version %s.%s",
config_entry.title,
config_entry.version,
config_entry.minor_version,
)
if config_entry.version > 1:
# This means the user has downgraded from a future version
return False
if config_entry.version == 1:
new = {**config_entry.data}
if config_entry.minor_version < 2:
new[CONF_ADD_UNAVAILABLE] = True # Behaviour as in 1.1
config_entry.minor_version = 2
hass.config_entries.async_update_entry(config_entry, data=new)
LOGGER.debug(
"Migration of %s to version %s.%s successful",
config_entry.title,
config_entry.version,
config_entry.minor_version,
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

62
custom_components/hochwasserportal/config_flow.py
View File

@ -1,62 +0,0 @@
"""Config flow for the hochwasserportal integration."""
from __future__ import annotations
from typing import Any
from lhpapi import HochwasserPortalAPI, LHPError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from .const import CONF_ADD_UNAVAILABLE, CONF_PEGEL_IDENTIFIER, DOMAIN, LOGGER
class HochwasserPortalConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the config flow for the hochwasserportal integration."""
VERSION = 1
MINOR_VERSION = 2
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict = {}
if user_input is not None:
pegel_identifier = user_input[CONF_PEGEL_IDENTIFIER]
# Validate pegel identifier using the API
try:
api = await self.hass.async_add_executor_job(
HochwasserPortalAPI, pegel_identifier
)
LOGGER.debug(
"%s (%s): Successfully added!",
api.ident,
api.name,
)
except LHPError as err:
LOGGER.exception("Setup of %s failed: %s", pegel_identifier, err)
errors["base"] = "invalid_identifier"
if not errors:
# Set the unique ID for this config entry.
await self.async_set_unique_id(f"{DOMAIN}_{pegel_identifier.lower()}")
self._abort_if_unique_id_configured()
return self.async_create_entry(title=f"{api.name}", data=user_input)
return self.async_show_form(
step_id="user",
errors=errors,
data_schema=vol.Schema(
{
vol.Required(CONF_PEGEL_IDENTIFIER): cv.string,
vol.Required(CONF_ADD_UNAVAILABLE, default=False): cv.boolean,
}
),
)

46
custom_components/hochwasserportal/const.py
View File

@ -1,46 +0,0 @@
"""Constants for the Länderübergreifendes Hochwasser Portal integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Final
from homeassistant.const import Platform
LOGGER = logging.getLogger(__package__)
DOMAIN: Final = "hochwasserportal"
CONF_PEGEL_IDENTIFIER: Final = "pegel_identifier"
CONF_ADD_UNAVAILABLE: Final = "add_unavailable"
ATTR_DATA_PROVIDERS: Final[dict[str, str]] = {
"BB": "LfU Brandenburg",
"BE": "SenMVKU Berlin",
"BW": "LUBW Baden-Württemberg",
"BY": "LfU Bayern",
"HB": "SUKW Bremen",
"HE": "HLNUG",
"HH": "LSBG Hamburg",
"MV": "LUNG Mecklenburg-Vorpommern",
"NI": "NLWKN",
"NW": "LANUV Nordrhein-Westfalen",
"RP": "Luf Rheinland-Pfalz",
"SH": "Luf Schleswig-Holstein",
"SL": "LUA Saarland",
"SN": "LfULG Sachsen",
"ST": "Land Sachsen-Anhalt",
"TH": "TLUBN",
}
ATTR_LAST_UPDATE: Final = "last_update"
ATTR_URL: Final = "url"
ATTR_HINT: Final = "hint"
LEVEL_SENSOR: Final = "level"
STAGE_SENSOR: Final = "stage"
FLOW_SENSOR: Final = "flow"
DEFAULT_SCAN_INTERVAL: Final = timedelta(minutes=15)
PLATFORMS: Final[list[Platform]] = [Platform.SENSOR]

32
custom_components/hochwasserportal/coordinator.py
View File

@ -1,32 +0,0 @@
"""Data coordinator for the hochwasserportal integration."""
from __future__ import annotations
from lhpapi import HochwasserPortalAPI, LHPError
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
class HochwasserPortalCoordinator(DataUpdateCoordinator[None]):
"""Custom coordinator for the hochwasserportal integration."""
def __init__(self, hass: HomeAssistant, api: HochwasserPortalAPI) -> None:
"""Initialize the hochwasserportal coordinator."""
super().__init__(
hass, LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
)
self.api = api
LOGGER.debug("%s", repr(self.api))
async def _async_update_data(self) -> None:
"""Get the latest data from the hochwasserportal API."""
try:
await self.hass.async_add_executor_job(self.api.update)
LOGGER.debug("%s", repr(self.api))
except LHPError as err:
LOGGER.exception("Update of %s failed: %s", self.api.ident, err)
return False

14
custom_components/hochwasserportal/manifest.json
View File

@ -1,14 +0,0 @@
{
"domain": "hochwasserportal",
"name": "Länderübergreifendes Hochwasser Portal",
"codeowners": ["@stephan192"],
"config_flow": true,
"dependencies": [],
"documentation": "https://github.com/stephan192/hochwasserportal",
"integration_type": "device",
"iot_class": "cloud_polling",
"issue_tracker": "https://github.com/stephan192/hochwasserportal/issues",
"loggers": ["hochwasserportal"],
"requirements": ["lhpapi==1.0.3"],
"version": "1.0.1"
}

148
custom_components/hochwasserportal/sensor.py
View File

@ -1,148 +0,0 @@
"""Platform for sensor integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from lhpapi import HochwasserPortalAPI
from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfLength
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTR_DATA_PROVIDERS,
ATTR_HINT,
ATTR_LAST_UPDATE,
ATTR_URL,
CONF_ADD_UNAVAILABLE,
DOMAIN,
FLOW_SENSOR,
LEVEL_SENSOR,
LOGGER,
STAGE_SENSOR,
)
from .coordinator import HochwasserPortalCoordinator
@dataclass(frozen=True, kw_only=True)
class HochwasserPortalSensorEntityDescription(SensorEntityDescription):
"""Describes HochwasserPortal sensor entity."""
value_fn: Callable[[HochwasserPortalAPI], int | float | None]
available_fn: Callable[[HochwasserPortalAPI], bool]
SENSOR_TYPES: tuple[HochwasserPortalSensorEntityDescription, ...] = (
HochwasserPortalSensorEntityDescription(
key=LEVEL_SENSOR,
translation_key=LEVEL_SENSOR,
icon="mdi:waves",
native_unit_of_measurement=UnitOfLength.CENTIMETERS,
device_class=None,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda api: api.level,
available_fn=lambda api: api.level is not None,
),
HochwasserPortalSensorEntityDescription(
key=STAGE_SENSOR,
translation_key=STAGE_SENSOR,
icon="mdi:waves-arrow-up",
native_unit_of_measurement=None,
device_class=None,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda api: api.stage,
available_fn=lambda api: api.stage is not None,
),
HochwasserPortalSensorEntityDescription(
key=FLOW_SENSOR,
translation_key=FLOW_SENSOR,
icon="mdi:waves-arrow-right",
native_unit_of_measurement="m³/s",
device_class=None,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda api: api.flow,
available_fn=lambda api: api.flow is not None,
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up entities from config entry."""
coordinator: HochwasserPortalCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
HochwasserPortalSensor(coordinator, entry, description)
for description in SENSOR_TYPES
if description.available_fn(coordinator.api)
or entry.data.get(CONF_ADD_UNAVAILABLE, False)
]
)
class HochwasserPortalSensor(
CoordinatorEntity[HochwasserPortalCoordinator], SensorEntity
):
"""Sensor representation."""
entity_description: HochwasserPortalSensorEntityDescription
_attr_has_entity_name = True
def __init__(
self,
coordinator: HochwasserPortalCoordinator,
entry: ConfigEntry,
description: HochwasserPortalSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.api = coordinator.api
self.entity_description = description
self._attr_unique_id = f"{entry.unique_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
name=f"{entry.title}",
configuration_url=self.api.url,
manufacturer=f"{ATTR_DATA_PROVIDERS[self.api.ident[:2]]}",
model=f"{self.api.ident}",
)
self._attr_attribution = (
f"Data provided by {ATTR_DATA_PROVIDERS[self.api.ident[:2]]}"
)
LOGGER.debug("Setting up sensor: %s", self._attr_unique_id)
@property
def native_value(self):
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.api)
@property
def extra_state_attributes(self):
"""Return the state attributes of the sensor."""
data = {}
if self.api.last_update is not None:
data[ATTR_LAST_UPDATE] = self.api.last_update
if self.api.url is not None:
data[ATTR_URL] = self.api.url
if self.api.hint is not None:
data[ATTR_HINT] = self.api.hint
if bool(data):
return data
return None
@property
def available(self) -> bool:
"""Could the device be accessed during the last update call."""
return self.entity_description.available_fn(self.api)

33
custom_components/hochwasserportal/string.json
View File

@ -1,33 +0,0 @@
{
"config": {
"step": {
"user": {
"description": "To identify the pegel, the pegel ID is required.",
"data": {
"pegel_identifier": "Pegel ID",
"add_unavailable": "Add unavailable entities anyway"
}
}
},
"error": {
"invalid_identifier": "The specified pegel identifier is invalid."
},
"abort": {
"already_configured": "Pegel is already configured.",
"invalid_identifier": "[%key:component::hochwasserportal::config::error::invalid_identifier%]"
}
},
"entity": {
"sensor": {
"level": {
"name": "Level"
},
"stage": {
"name": "Stage"
},
"flow": {
"name": "Flow"
}
}
}
}

33
custom_components/hochwasserportal/translations/de.json
View File

@ -1,33 +0,0 @@
{
"config": {
"step": {
"user": {
"description": "Um den Pegel zu identifizieren, ist die Pegel-ID erforderlich.",
"data": {
"pegel_identifier": "Pegel ID",
"add_unavailable": "Nicht verfügbare Entitäten trotzdem hinzufügen"
}
}
},
"error": {
"invalid_identifier": "Der angegebene Pegel ist ungültig."
},
"abort": {
"already_configured": "Pegel bereits konfiguriert.",
"invalid_identifier": "[%key:component::hochwasserportal::config::error::invalid_identifier%]"
}
},
"entity": {
"sensor": {
"level": {
"name": "Pegelstand"
},
"stage": {
"name": "Warnstufe"
},
"flow": {
"name": "Abfluss"
}
}
}
}

33
custom_components/hochwasserportal/translations/en.json
View File

@ -1,33 +0,0 @@
{
"config": {
"step": {
"user": {
"description": "To identify the pegel, the pegel ID is required.",
"data": {
"pegel_identifier": "Pegel ID",
"add_unavailable": "Add unavailable entities anyway"
}
}
},
"error": {
"invalid_identifier": "The specified pegel identifier is invalid."
},
"abort": {
"already_configured": "Pegel is already configured.",
"invalid_identifier": "[%key:component::hochwasserportal::config::error::invalid_identifier%]"
}
},
"entity": {
"sensor": {
"level": {
"name": "Level"
},
"stage": {
"name": "Stage"
},
"flow": {
"name": "Flow"
}
}
}
}