Add Pterodactyl integration (home-assistant#141197)

* Add Pterodactyl integration

* Remove translation for unavailable platform sensor, use constant for host

* Improve data descriptions

* Replace index based handling of data (list) with dict[str, PterodactylData]

* Replace CONF_HOST with CONF_URL

* Parse URL with YARL

* Set proper availability in binary sensor

* Remove storage of data within api.py

* Fix some review findings

* Use better unique ID for binary_sensor

* Fix more review findings

* Fix remaining review findings

* Add wrapper for server and util API, use underscore in unique ID

* Reuse result in config flow tests

* Patch async_setup_entry in config_flow tests

* Move patching of library APIs to the fixture mock_pterodactyl

Author: elmurato

CODEOWNERS

@@ -0,0 +1,27 @@
+"""The Pterodactyl integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import PterodactylConfigEntry, PterodactylCoordinator
+
+_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: PterodactylConfigEntry) -> bool:
+    """Set up Pterodactyl from a config entry."""
+    coordinator = PterodactylCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: PterodactylConfigEntry
+) -> bool:
+    """Unload a Pterodactyl config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)

homeassistant/components/pterodactyl/__init__.py

@@ -0,0 +1,120 @@
+"""API module of the Pterodactyl integration."""
+
+from dataclasses import dataclass
+import logging
+
+from pydactyl import PterodactylClient
+from pydactyl.exceptions import (
+    BadRequestError,
+    ClientConfigError,
+    PterodactylApiError,
+    PydactylError,
+)
+
+from homeassistant.core import HomeAssistant
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class PterodactylConfigurationError(Exception):
+    """Raised when the configuration is invalid."""
+
+
+class PterodactylConnectionError(Exception):
+    """Raised when no data can be fechted from the server."""
+
+
+@dataclass
+class PterodactylData:
+    """Data for the Pterodactyl server."""
+
+    name: str
+    uuid: str
+    identifier: str
+    state: str
+    memory_utilization: int
+    cpu_utilization: float
+    disk_utilization: int
+    network_rx_utilization: int
+    network_tx_utilization: int
+    uptime: int
+
+
+class PterodactylAPI:
+    """Wrapper for Pterodactyl's API."""
+
+    pterodactyl: PterodactylClient | None
+    identifiers: list[str]
+
+    def __init__(self, hass: HomeAssistant, host: str, api_key: str) -> None:
+        """Initialize the Pterodactyl API."""
+        self.hass = hass
+        self.host = host
+        self.api_key = api_key
+        self.pterodactyl = None
+        self.identifiers = []
+
+    async def async_init(self):
+        """Initialize the Pterodactyl API."""
+        self.pterodactyl = PterodactylClient(self.host, self.api_key)
+
+        try:
+            paginated_response = await self.hass.async_add_executor_job(
+                self.pterodactyl.client.servers.list_servers
+            )
+        except ClientConfigError as error:
+            raise PterodactylConfigurationError(error) from error
+        except (
+            PydactylError,
+            BadRequestError,
+            PterodactylApiError,
+        ) as error:
+            raise PterodactylConnectionError(error) from error
+        else:
+            game_servers = paginated_response.collect()
+            for game_server in game_servers:
+                self.identifiers.append(game_server["attributes"]["identifier"])
+
+            _LOGGER.debug("Identifiers of Pterodactyl servers: %s", self.identifiers)
+
+    def get_server_data(self, identifier: str) -> tuple[dict, dict]:
+        """Get all data from the Pterodactyl server."""
+        server = self.pterodactyl.client.servers.get_server(identifier)  # type: ignore[union-attr]
+        utilization = self.pterodactyl.client.servers.get_server_utilization(  # type: ignore[union-attr]
+            identifier
+        )
+
+        return server, utilization
+
+    async def async_get_data(self) -> dict[str, PterodactylData]:
+        """Update the data from all Pterodactyl servers."""
+        data = {}
+
+        for identifier in self.identifiers:
+            try:
+                server, utilization = await self.hass.async_add_executor_job(
+                    self.get_server_data, identifier
+                )
+            except (
+                PydactylError,
+                BadRequestError,
+                PterodactylApiError,
+            ) as error:
+                raise PterodactylConnectionError(error) from error
+            else:
+                data[identifier] = PterodactylData(
+                    name=server["name"],
+                    uuid=server["uuid"],
+                    identifier=identifier,
+                    state=utilization["current_state"],
+                    cpu_utilization=utilization["resources"]["cpu_absolute"],
+                    memory_utilization=utilization["resources"]["memory_bytes"],
+                    disk_utilization=utilization["resources"]["disk_bytes"],
+                    network_rx_utilization=utilization["resources"]["network_rx_bytes"],
+                    network_tx_utilization=utilization["resources"]["network_tx_bytes"],
+                    uptime=utilization["resources"]["uptime"],
+                )
+
+                _LOGGER.debug("%s", data[identifier])
+
+        return data

homeassistant/components/pterodactyl/api.py

@@ -0,0 +1,64 @@
+"""Binary sensor platform of the Pterodactyl integration."""
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .coordinator import PterodactylConfigEntry, PterodactylCoordinator
+from .entity import PterodactylEntity
+
+KEY_STATUS = "status"
+
+
+BINARY_SENSOR_DESCRIPTIONS = [
+    BinarySensorEntityDescription(
+        key=KEY_STATUS,
+        translation_key=KEY_STATUS,
+        device_class=BinarySensorDeviceClass.RUNNING,
+    ),
+]
+
+# Coordinator is used to centralize the data updates.
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: PterodactylConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the Pterodactyl binary sensor platform."""
+    coordinator = config_entry.runtime_data
+
+    async_add_entities(
+        PterodactylBinarySensorEntity(
+            coordinator, identifier, description, config_entry
+        )
+        for identifier in coordinator.api.identifiers
+        for description in BINARY_SENSOR_DESCRIPTIONS
+    )
+
+
+class PterodactylBinarySensorEntity(PterodactylEntity, BinarySensorEntity):
+    """Representation of a Pterodactyl binary sensor base entity."""
+
+    def __init__(
+        self,
+        coordinator: PterodactylCoordinator,
+        identifier: str,
+        description: BinarySensorEntityDescription,
+        config_entry: PterodactylConfigEntry,
+    ) -> None:
+        """Initialize binary sensor base entity."""
+        super().__init__(coordinator, identifier, config_entry)
+        self.entity_description = description
+        self._attr_unique_id = f"{self.game_server_data.uuid}_{description.key}"
+
+    @property
+    def is_on(self) -> bool:
+        """Return binary sensor state."""
+        return self.game_server_data.state == "running"

homeassistant/components/pterodactyl/binary_sensor.py

@@ -0,0 +1,62 @@
+"""Config flow for the Pterodactyl integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+from yarl import URL
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_API_KEY, CONF_URL
+
+from .api import (
+    PterodactylAPI,
+    PterodactylConfigurationError,
+    PterodactylConnectionError,
+)
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+DEFAULT_URL = "http://localhost:8080"
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_URL, default=DEFAULT_URL): str,
+        vol.Required(CONF_API_KEY): str,
+    }
+)
+
+
+class PterodactylConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Pterodactyl."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            url = URL(user_input[CONF_URL]).human_repr()
+            api_key = user_input[CONF_API_KEY]
+
+            self._async_abort_entries_match({CONF_URL: url})
+            api = PterodactylAPI(self.hass, url, api_key)
+
+            try:
+                await api.async_init()
+            except (PterodactylConfigurationError, PterodactylConnectionError):
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception occurred during config flow")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(title=url, data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/pterodactyl/config_flow.py

@@ -0,0 +1,3 @@
+"""Constants for the Pterodactyl integration."""
+
+DOMAIN = "pterodactyl"

homeassistant/components/pterodactyl/const.py

@@ -0,0 +1,66 @@
+"""Data update coordinator of the Pterodactyl integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .api import (
+    PterodactylAPI,
+    PterodactylConfigurationError,
+    PterodactylConnectionError,
+    PterodactylData,
+)
+
+SCAN_INTERVAL = timedelta(seconds=60)
+
+_LOGGER = logging.getLogger(__name__)
+
+type PterodactylConfigEntry = ConfigEntry[PterodactylCoordinator]
+
+
+class PterodactylCoordinator(DataUpdateCoordinator[dict[str, PterodactylData]]):
+    """Pterodactyl data update coordinator."""
+
+    config_entry: PterodactylConfigEntry
+    api: PterodactylAPI
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: PterodactylConfigEntry,
+    ) -> None:
+        """Initialize coordinator instance."""
+
+        super().__init__(
+            hass=hass,
+            name=config_entry.data[CONF_URL],
+            config_entry=config_entry,
+            logger=_LOGGER,
+            update_interval=SCAN_INTERVAL,
+        )
+
+    async def _async_setup(self) -> None:
+        """Set up the Pterodactyl data coordinator."""
+        self.api = PterodactylAPI(
+            hass=self.hass,
+            host=self.config_entry.data[CONF_URL],
+            api_key=self.config_entry.data[CONF_API_KEY],
+        )
+
+        try:
+            await self.api.async_init()
+        except PterodactylConfigurationError as error:
+            raise UpdateFailed(error) from error
+
+    async def _async_update_data(self) -> dict[str, PterodactylData]:
+        """Get updated data from the Pterodactyl server."""
+        try:
+            return await self.api.async_get_data()
+        except PterodactylConnectionError as error:
+            raise UpdateFailed(error) from error