95aac1d809
- https://gitea.bitfire.work → https://gitea.bitfire.work in allen Dateien - Vorausgefülltes Default-Passwort aus config_flow.py entfernt - DEFAULT_MOUNT auf /stream korrigiert (war noch /stream.mp3) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
"""Config Flow für die Busch-Radio Spotify Bridge Integration.
|
|
|
|
Führt den Nutzer durch die Einrichtung in der HA-Oberfläche.
|
|
Fragt nach:
|
|
- Hostname/IP des HA-Hosts (normalerweise 'localhost' wenn Add-on läuft)
|
|
- Port (Standard: 8000)
|
|
- Stream-Mount-Pfad (Standard: /stream.mp3)
|
|
- Icecast Admin-Passwort (identisch mit 'icecast_password' in der Add-on Konfig)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import aiohttp
|
|
import voluptuous as vol
|
|
from homeassistant import config_entries
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
from . import CONF_HOST, CONF_MOUNT, CONF_PASSWORD, CONF_PORT, DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
DEFAULT_HOST = "localhost"
|
|
DEFAULT_PORT = 8000
|
|
DEFAULT_MOUNT = "/stream"
|
|
|
|
|
|
async def _test_icecast_connection(
|
|
hass: HomeAssistant, host: str, port: int, password: str
|
|
) -> str | None:
|
|
"""Verbindung zu Icecast testen.
|
|
|
|
Gibt None zurück wenn alles ok, sonst einen Fehlercode.
|
|
"""
|
|
session = async_get_clientsession(hass)
|
|
url = f"http://{host}:{port}/admin/stats"
|
|
auth = aiohttp.BasicAuth("admin", password)
|
|
|
|
try:
|
|
async with session.get(
|
|
url,
|
|
auth=auth,
|
|
timeout=aiohttp.ClientTimeout(total=5),
|
|
) as resp:
|
|
if resp.status == 200:
|
|
return None
|
|
if resp.status == 401:
|
|
return "invalid_auth"
|
|
return "cannot_connect"
|
|
except aiohttp.ClientConnectorError:
|
|
return "cannot_connect"
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
class BuschRadioSpotifyConfigFlow(
|
|
config_entries.ConfigFlow, domain=DOMAIN
|
|
):
|
|
"""Config Flow für Busch-Radio Spotify Bridge."""
|
|
|
|
VERSION = 1
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict | None = None
|
|
) -> config_entries.FlowResult:
|
|
"""Ersten Einrichtungsschritt anzeigen."""
|
|
errors: dict[str, str] = {}
|
|
|
|
if user_input is not None:
|
|
error = await _test_icecast_connection(
|
|
self.hass,
|
|
user_input[CONF_HOST],
|
|
user_input[CONF_PORT],
|
|
user_input[CONF_PASSWORD],
|
|
)
|
|
|
|
if error is None:
|
|
# Doppelte Einrichtung verhindern
|
|
await self.async_set_unique_id(
|
|
f"{user_input[CONF_HOST]}:{user_input[CONF_PORT]}"
|
|
)
|
|
self._abort_if_unique_id_configured()
|
|
|
|
return self.async_create_entry(
|
|
title=f"Busch-Radio ({user_input[CONF_HOST]}:{user_input[CONF_PORT]})",
|
|
data=user_input,
|
|
)
|
|
else:
|
|
errors["base"] = error
|
|
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(CONF_HOST, default=DEFAULT_HOST): str,
|
|
vol.Required(CONF_PORT, default=DEFAULT_PORT): vol.All(
|
|
vol.Coerce(int), vol.Range(min=1, max=65535)
|
|
),
|
|
vol.Required(CONF_MOUNT, default=DEFAULT_MOUNT): str,
|
|
vol.Required(CONF_PASSWORD): str,
|
|
}
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=schema,
|
|
errors=errors,
|
|
)
|