generated from ludeeus/integration_blueprint
-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
175 additions
and
132 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,87 +1,87 @@ | ||
| """Binary sensor platform for Healthchecksio.""" | ||
|
|
||
| try: | ||
| from homeassistant.components.binary_sensor import BinarySensorEntity | ||
| except ImportError: | ||
| from homeassistant.components.binary_sensor import ( | ||
| BinarySensorDevice as BinarySensorEntity, | ||
| ) | ||
| from __future__ import annotations | ||
|
|
||
| from typing import TYPE_CHECKING, Any | ||
|
|
||
| from homeassistant.components.binary_sensor import ( | ||
| BinarySensorDeviceClass, | ||
| BinarySensorEntity, | ||
| ) | ||
| from homeassistant.helpers.update_coordinator import CoordinatorEntity | ||
|
|
||
| from custom_components.healthchecksio import LOGGER | ||
|
|
||
| from .const import ATTRIBUTION, BINARY_SENSOR_DEVICE_CLASS, DOMAIN, DOMAIN_DATA | ||
| from .const import ATTRIBUTION, DOMAIN | ||
|
|
||
| if TYPE_CHECKING: | ||
| from homeassistant.core import HomeAssistant | ||
|
|
||
| from .coordinator import HealthchecksioDataUpdateCoordinator | ||
|
|
||
|
|
||
| async def async_setup_entry(hass, config_entry, async_add_devices): | ||
| """Setup sensor platform.""" | ||
| # Send update "signal" to the component | ||
| await hass.data[DOMAIN_DATA]["client"].update_data() | ||
| checks = [] | ||
| for check in hass.data[DOMAIN_DATA].get("data", {}).get("checks", []): | ||
| check_data = { | ||
| "name": check.get("name"), | ||
| "last_ping": check.get("last_ping"), | ||
| "status": check.get("status"), | ||
| "ping_url": check.get("ping_url"), | ||
| } | ||
| checks.append(HealthchecksioBinarySensor(hass, check_data, config_entry)) | ||
| async_add_devices(checks, True) | ||
| coordinator: HealthchecksioDataUpdateCoordinator = config_entry.runtime_data | ||
| async_add_devices( | ||
| HealthchecksioBinarySensor( | ||
| hass=hass, | ||
| ping_url=check.get("ping_url"), | ||
| coordinator=coordinator, | ||
| ) | ||
| for check in coordinator.data.get("checks", []) | ||
| ) | ||
|
|
||
|
|
||
| class HealthchecksioBinarySensor(BinarySensorEntity): | ||
| class HealthchecksioBinarySensor(BinarySensorEntity, CoordinatorEntity): | ||
| """Healthchecksio binary_sensor class.""" | ||
|
|
||
| def __init__(self, hass, check_data, config_entry): | ||
| self.hass = hass | ||
| self.attr = {} | ||
| self.config_entry = config_entry | ||
| self._status = None | ||
| self.check_data = check_data | ||
| self.check = {} | ||
|
|
||
| async def async_update(self): | ||
| """Update the binary_sensor.""" | ||
| # Send update "signal" to the component | ||
| await self.hass.data[DOMAIN_DATA]["client"].update_data() | ||
|
|
||
| # Check the data and update the value. | ||
| for check in self.hass.data[DOMAIN_DATA].get("data", {}).get("checks", []): | ||
| if self.unique_id == check.get("ping_url").split("/")[-1]: | ||
| self.check = check | ||
| break | ||
| self._status = self.check.get("status") != "down" | ||
|
|
||
| # Set/update attributes | ||
| self.attr["attribution"] = ATTRIBUTION | ||
| self.attr["last_ping"] = self.check.get("last_ping") | ||
| coordinator: HealthchecksioDataUpdateCoordinator | ||
|
|
||
| @property | ||
| def unique_id(self): | ||
| """Return a unique ID to use for this binary_sensor.""" | ||
| return self.check_data.get("ping_url", "").split("/")[-1] | ||
| _attr_attribution = ATTRIBUTION | ||
| _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY | ||
|
|
||
| @property | ||
| def device_info(self): | ||
| return { | ||
| "identifiers": {(DOMAIN, self.config_entry.entry_id)}, | ||
| def __init__( | ||
| self, | ||
| *, | ||
| hass: HomeAssistant, | ||
| coordinator: HealthchecksioDataUpdateCoordinator, | ||
| ping_url: str, | ||
| ): | ||
| super().__init__(coordinator) | ||
| self.hass = hass | ||
| self._ping_url = ping_url | ||
|
|
||
| self._attr_unique_id = ping_url.split("/")[-1] | ||
| self._attr_device_info = { | ||
| "identifiers": {(DOMAIN, self.coordinator.config_entry.entry_id)}, | ||
| "name": "Healthchecks.io", | ||
| "manufacturer": "SIA Monkey See Monkey Do", | ||
| } | ||
|
|
||
| def get_check(self) -> dict[str, Any]: | ||
| """Get check data.""" | ||
| for check in self.coordinator.data.get("checks", []): | ||
| if self._ping_url == check.get("ping_url"): | ||
| return check | ||
| return {} | ||
|
|
||
| @property | ||
| def name(self): | ||
| """Return the name of the binary_sensor.""" | ||
| return self.check.get("name") | ||
|
|
||
| @property | ||
| def device_class(self): | ||
| """Return the class of this binary_sensor.""" | ||
| return BINARY_SENSOR_DEVICE_CLASS | ||
| check = self.get_check() | ||
| return check.get("name") | ||
|
|
||
| @property | ||
| def is_on(self): | ||
| """Return true if the binary_sensor is on.""" | ||
| return self._status | ||
| check = self.get_check() | ||
| LOGGER.debug("Check: %s", check) | ||
| return check.get("status") != "down" | ||
|
|
||
| @property | ||
| def extra_state_attributes(self): | ||
| """Return the state attributes.""" | ||
| return self.attr | ||
| check = self.get_check() | ||
| return {"last_ping": check.get("last_ping")} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| """Add coordinator for Healthchecks.io integration.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import asyncio | ||
| from logging import getLogger | ||
| from typing import TYPE_CHECKING, Any | ||
|
|
||
| from aiohttp import ClientSession, ClientTimeout | ||
| from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed | ||
|
|
||
| from .const import DOMAIN, MIN_TIME_BETWEEN_UPDATES | ||
|
|
||
| if TYPE_CHECKING: | ||
| from homeassistant.config_entries import ConfigEntry | ||
| from homeassistant.core import HomeAssistant | ||
|
|
||
| LOGGER = getLogger(__name__) | ||
|
|
||
|
|
||
| class HealthchecksioDataUpdateCoordinator(DataUpdateCoordinator): | ||
| """Class to manage fetching Healthchecks.io data.""" | ||
|
|
||
| config_entry: ConfigEntry | ||
|
|
||
| def __init__( | ||
| self, | ||
| hass: HomeAssistant, | ||
| *, | ||
| api_key: str, | ||
| session: ClientSession, | ||
| self_hosted: bool = False, | ||
| check_id: str | None = None, | ||
| site_root: str | None = None, | ||
| ping_endpoint: str | None = None, | ||
| ): | ||
| """Initialize.""" | ||
| super().__init__( | ||
| hass=hass, | ||
| logger=LOGGER, | ||
| name=DOMAIN, | ||
| update_interval=MIN_TIME_BETWEEN_UPDATES, | ||
| ) | ||
| self._api_key = api_key | ||
| self._site_root = site_root | ||
| self._session = session | ||
| self._self_hosted = self_hosted | ||
| self._check_id = check_id | ||
| self._ping_endpoint = ping_endpoint | ||
|
|
||
| async def _async_update_data(self) -> dict[str, Any]: | ||
| """Update data.""" | ||
| try: | ||
| data = await self._session.get( | ||
| f"{self._site_root}/api/v1/checks/", | ||
| headers={"X-Api-Key": self._api_key}, | ||
| timeout=ClientTimeout(total=10), | ||
| ) | ||
| except Exception as error: | ||
| raise UpdateFailed(error) from error | ||
|
|
||
| check_url = ( | ||
| f"https://hc-ping.com/{self._check_id}" | ||
| if not self._self_hosted | ||
| else f"{self._site_root}/{self._ping_endpoint}/{self._check_id}" | ||
| ) | ||
| if self._self_hosted: | ||
| await asyncio.sleep(1) # needed for self-hosted instances | ||
|
|
||
| try: | ||
| await self._session.get( | ||
| check_url, | ||
| timeout=ClientTimeout(total=10), | ||
| ) | ||
| except Exception: | ||
| LOGGER.exception("Could not ping") | ||
|
|
||
| return await data.json() |