dev: Improved everything

Author: ping-localhost

.github/workflows/lint.yml

@@ -22,13 +22,15 @@ jobs:
         uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c # v6.0.0
         with:
           python-version: "3.13"
-          cache: "pip"
+
+      - name: Install uv
+        run: pip install uv
 
       - name: Install requirements
-        run: python3 -m pip install -r requirements.txt
+        run: uv pip install --system -r pyproject.toml
 
       - name: Lint
-        run: python3 -m ruff check .
+        run: uv run ruff check .
 
       - name: Format
-        run: python3 -m ruff format . --check
+        run: uv run ruff format . --check

custom_components/panda_status/config_flow.py

@@ -2,21 +2,17 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import voluptuous as vol
 
 from custom_components.panda_status import tools
-from homeassistant import config_entries
-from homeassistant.config_entries import OptionsFlowWithReload
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
 from homeassistant.const import CONF_NAME, CONF_URL
-from homeassistant.core import callback
 from homeassistant.helpers import selector
 
 from .const import DOMAIN, LOGGER
-from .websocket import (
-    PandaStatusWebSocket,
-    PandaStatusWebsocketCommunicationError,
-    PandaStatusWebsocketError,
-)
+from .websocket import PandaStatusWebsocketCommunicationError, PandaStatusWebsocketError
 
 OPTIONS_SCHEMA = vol.Schema(
     {
@@ -30,27 +26,23 @@
 )
 
 
-async def _test_credentials(url: str) -> dict:
-    """Validate credentials."""
-    client = PandaStatusWebSocket(url=url, session=None)
-    return await client.async_get_data()
-
-
-class PandaStatusFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+class PandaStatusFlowHandler(ConfigFlow, domain=DOMAIN):
     """Config flow for PandaStatus."""
 
     VERSION = 1
     MINOR_VERSION = 1
 
     async def async_step_user(
-        self, user_input: dict | None = None
-    ) -> config_entries.ConfigFlowResult:
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
         """Handle a flow initialized by the user."""
         errors = {}
         if user_input is not None:
             try:
                 # Try and get initial data to validate the connection
-                initial_data = await _test_credentials(url=user_input[CONF_URL])
+                initial_data = await tools.test_credentials(
+                    url=user_input.get(CONF_URL)
+                )
             except PandaStatusWebsocketCommunicationError as exception:
                 LOGGER.error(exception)
                 errors["base"] = "connection"
@@ -63,7 +55,7 @@ async def async_step_user(
 
                 # Use tools to extract printer and device name
                 printer_name = tools.get_printer_name(initial_data=initial_data)
-                device_name = user_input[CONF_NAME] or tools.get_device_name(
+                device_name = user_input.get(CONF_NAME, None) or tools.get_device_name(
                     initial_data=initial_data
                 )
 
@@ -72,7 +64,7 @@ async def async_step_user(
                     title=device_name,
                     description=f"Panda status for {printer_name}",
                     data={
-                        CONF_URL: user_input[CONF_URL],
+                        CONF_URL: user_input.get(CONF_URL),
                         CONF_NAME: device_name,
                     },
                 )
@@ -82,59 +74,3 @@ async def async_step_user(
             data_schema=OPTIONS_SCHEMA,
             errors=errors,
         )
-
-    @staticmethod
-    @callback
-    def async_get_options_flow(
-        config_entry: config_entries.ConfigEntry,
-    ) -> config_entries.OptionsFlow:
-        """Return the options flow handler for PandaStatus."""
-        return PandaStatusOptionsFlowHandler(config_entry)
-
-
-class PandaStatusOptionsFlowHandler(OptionsFlowWithReload):
-    """Options flow for PandaStatus."""
-
-    def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
-        """Initialize options flow."""
-        self.config_entry = config_entry
-
-    async def async_step_init(
-        self, user_input: dict | None = None
-    ) -> config_entries.ConfigFlowResult:
-        """Manage the options."""
-        errors = {}
-        if user_input is not None:
-            try:
-                # Try and get initial data to validate the connection
-                initial_data = await _test_credentials(url=user_input[CONF_URL])
-            except PandaStatusWebsocketCommunicationError as exception:
-                LOGGER.error(exception)
-                errors["base"] = "connection"
-            except PandaStatusWebsocketError as exception:
-                LOGGER.exception(exception)
-                errors["base"] = "unknown"
-            else:
-                # Use tools to extract printer and device name
-                printer_name = tools.get_printer_name(initial_data=initial_data)
-                device_name = user_input[CONF_NAME] or tools.get_device_name(
-                    initial_data=initial_data
-                )
-
-                # If user provided a name, use it; otherwise, use the device name
-                return self.async_create_entry(
-                    title=device_name,
-                    description=f"Panda status for {printer_name}",
-                    data={
-                        CONF_URL: user_input[CONF_URL],
-                        CONF_NAME: device_name,
-                    },
-                )
-
-        return self.async_show_form(
-            step_id="init",
-            data_schema=self.add_suggested_values_to_schema(
-                OPTIONS_SCHEMA, self.config_entry.options
-            ),
-            errors=errors,
-        )

custom_components/panda_status/system_health.py

@@ -1,15 +1,24 @@
-"""System Health platform for panda_status custom component."""
+"""System health platform for the Panda Status custom component."""
 
-from typing import TYPE_CHECKING, Any
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any, cast
 
-from custom_components.panda_status.const import DOMAIN
-from homeassistant.components import system_health
 from homeassistant.const import CONF_URL
 from homeassistant.core import HomeAssistant, callback
 
+from . import tools
+from .const import DOMAIN
+from .websocket import PandaStatusWebsocketError, PandaStatusWebsocketTimeoutError
+
 if TYPE_CHECKING:
+    from homeassistant.components import system_health
+
     from .data import PandaStatusConfigEntry
 
+_LOGGER = logging.getLogger(__name__)
+
 
 @callback
 def async_register(
@@ -21,11 +30,30 @@ def async_register(
 
 
 async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
-    """Get info for the info page."""
-    config_entry: PandaStatusConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
-
-    return {
-        "can_reach_server": system_health.async_check_can_reach_url(
-            hass, config_entry.data[CONF_URL]
-        ),
-    }
+    """Return system-health information for the Panda Status integration."""
+    # Grab the first config entry for this integration.
+    # If none are present we report nothing.
+    config_entries = hass.config_entries.async_entries(DOMAIN)
+    if not config_entries:
+        _LOGGER.warning("No %s config entries found", DOMAIN)
+        return {}
+
+    config_entry: PandaStatusConfigEntry = cast(
+        "PandaStatusConfigEntry", config_entries[0]
+    )
+
+    url = config_entry.data.get(CONF_URL)
+    if not url:
+        _LOGGER.warning("Config entry for %s does not contain a URL", DOMAIN)
+        return {"websocket_reachable": "Missing URL"}
+
+    try:
+        await tools.test_credentials(url)
+    except (TimeoutError, PandaStatusWebsocketTimeoutError):
+        data = "timeout"
+    except (Exception, PandaStatusWebsocketError):  # noqa: BLE001
+        data = "unreachable"
+    else:
+        data = "ok"
+
+    return {"websocket_reachable": data}

custom_components/panda_status/tools.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from custom_components.panda_status import PandaStatusWebSocket
+
 
 def extract_value(data: dict, dotted_key: str) -> Any:
     """Extract nested JSON value using dotted path."""
@@ -22,3 +24,9 @@ def get_printer_name(initial_data: dict) -> str:
 def get_device_name(initial_data: dict) -> str:
     """Get device name from initial data."""
     return f"{extract_value(initial_data, 'printer.name')} - Panda Status"
+
+
+async def test_credentials(url: str) -> dict:
+    """Validate credentials."""
+    client = PandaStatusWebSocket(url=url, session=None)
+    return await client.async_get_data()

custom_components/panda_status/translations/en.json

@@ -1,25 +1,40 @@
 {
-    "config": {
-        "step": {
-            "user": {
-                "description": "If you need help with the configuration have a look here: https://github.com/ping-localhost/panda_status",
-                "data": {
-                    "url": "URL",
-                    "name": "Name for the device (optional)"
-                }
-            }
-        },
-        "error": {
-            "connection": "Unable to connect to the server.",
-            "unknown": "Unknown error occurred."
-        },
-        "abort": {
-            "already_configured": "This entry is already configured."
+  "config": {
+    "step": {
+      "user": {
+        "description": "If you need help with the configuration have a look here: https://github.com/ping-localhost/panda_status",
+        "data": {
+          "url": "URL",
+          "name": "Name for the device (optional)"
         }
+      }
     },
-    "system_health": {
-        "info": {
-            "can_reach_server": "WebSocket reachable"
+    "error": {
+      "connection": "Unable to connect to the server.",
+      "unknown": "Unknown error occurred."
+    },
+    "abort": {
+      "already_configured": "This entry is already configured."
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "description": "If you need help with the configuration have a look here: https://github.com/ping-localhost/panda_status",
+        "data": {
+          "url": "URL",
+          "name": "Name for the device (optional)"
         }
+      }
+    },
+    "error": {
+      "connection": "Unable to connect to the server.",
+      "unknown": "Unknown error occurred."
+    }
+  },
+  "system_health": {
+    "info": {
+      "websocket_reachable": "WebSocket reachable"
     }
+  }
 }

custom_components/panda_status/websocket.py

@@ -7,7 +7,7 @@
 import logging
 
 from websockets.asyncio.client import ClientConnection, connect
-from websockets.exceptions import ConnectionClosed
+from websockets.exceptions import ConnectionClosed, InvalidStatus
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -16,6 +16,12 @@ class PandaStatusWebsocketError(Exception):
     """Exception to indicate a general WebSocket error."""
 
 
+class PandaStatusWebsocketTimeoutError(
+    PandaStatusWebsocketError,
+):
+    """Exception to indicate a timeout error."""
+
+
 class PandaStatusWebsocketCommunicationError(
     PandaStatusWebsocketError,
 ):
@@ -55,18 +61,18 @@ async def async_get_data(self) -> dict:
             PandaStatusWebsocketError: If JSON is invalid or connection fails.
 
         """
-        async with self._session as websocket:
-            try:
+        try:
+            async with self._session as websocket:
                 data = json.loads(await websocket.recv())
-            except TimeoutError as e:
-                msg = f"Timeout error getting data - {e}"
-                raise PandaStatusWebsocketCommunicationError(msg) from e
-            except (OSError, ConnectionClosed, TypeError) as e:
-                msg = f"Communication error - {e}"
-                raise PandaStatusWebsocketCommunicationError(msg) from e
-            except Exception as e:
-                msg = f"Unexpected error parsing data payload - {e}"
-                raise PandaStatusWebsocketError(msg) from e
+        except TimeoutError as e:
+            msg = f"Timeout error getting data - {e}"
+            raise PandaStatusWebsocketTimeoutError(msg) from e
+        except (OSError, ConnectionClosed, TypeError, InvalidStatus) as e:
+            msg = f"Communication error - {e}"
+            raise PandaStatusWebsocketCommunicationError(msg) from e
+        except Exception as e:
+            msg = f"Unexpected error parsing data payload - {e}"
+            raise PandaStatusWebsocketError(msg) from e
 
         _LOGGER.debug("Latest data received: %s", json.dumps(data))
 

pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "panda_status"
+version = "1.2.0"
+requires-python = ">=3.13.2"
+dependencies = [
+    "colorlog>=6.9.0",
+    "hassil>=3.2.0",
+    "home-assistant-intents==2025.8.29",
+    "homeassistant==2025.8.1",
+    "mutagen>=1.47.0",
+    "numpy>=2.3.2",
+    "pip>=25.2",
+    "ruff>=0.12.12",
+    "websockets>=15.0.1",
+]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.12.12"
+]

requirements.txt

@@ -4,4 +4,4 @@ set -e
 
 cd "$(dirname "$0")/.."
 
-python3 -m pip install --requirement requirements.txt
+uv venv && uv pip install -r pyproject.toml