"""RelayTV integration.""" from __future__ import annotations import asyncio import logging from pathlib import Path import time from urllib.parse import urlparse from homeassistant.components import frontend from homeassistant.components.media_source import async_resolve_media from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse, callback from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady, ServiceValidationError from homeassistant.helpers import aiohttp_client, config_validation as cv, entity_registry as er from homeassistant.helpers.event import async_track_state_change_event from homeassistant.helpers.storage import Store from homeassistant.helpers.target import TargetSelection, async_extract_referenced_entity_ids from .const import ( CONF_BASE_URL, CONF_API_TOKEN, CONF_PANEL_ENABLED, CONF_PANEL_TARGET_ENTRY_ID, CONF_RESUME_POSITIONS, CONF_SENSOR_STREAM_MAPPINGS, DATA_API, DATA_COORDINATOR, DATA_LAST_SNAPSHOT_URL, DATA_PANEL_SETTINGS, DATA_STORE, DEFAULT_PANEL_ICON, DEFAULT_PANEL_PATH, DEFAULT_PANEL_TITLE, DOMAIN, PLATFORMS, SERVICE_ANNOUNCE, SERVICE_OVERLAY, SERVICE_PLAY_NOW, SERVICE_PLAY_SYNCED, SERVICE_PLAY_TEMPORARY, SERVICE_PLAY_WITH_RESUME, SERVICE_SMART_URL, SERVICE_SNAPSHOT, SERVICE_UPLOAD_MEDIA, SERVICE_UPLOAD_MEDIA_ENQUEUE, SERVICE_UPLOAD_MEDIA_PLAY, ) from .coordinator import RelayTVCoordinator from .relaytv_api import RelayTVApi, RelayTVApiError, RelayTVAuthError from .url_utils import canonical_media_key _LOGGER = logging.getLogger(__name__) RUNTIME_STORE_KEY = f"{DOMAIN}_runtime" MAX_RESUME_POSITIONS = 500 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) def _normalize_base_url(raw: str) -> str: raw = (raw or "").strip() if not raw: return "" if "://" not in raw: raw = f"http://{raw}" parsed = urlparse(raw) if not parsed.netloc: return raw normalized = f"{parsed.scheme}://{parsed.netloc}{parsed.path}".rstrip("/") if parsed.query: normalized += f"?{parsed.query}" return normalized def _absolute_url(base_url: str, maybe_url: str | None) -> str | None: if not maybe_url: return None value = str(maybe_url).strip() if not value: return None parsed = urlparse(value) if parsed.scheme in ("http", "https"): return value base = (base_url or "").rstrip("/") tail = value.lstrip("/") if not base or not tail: return None return f"{base}/{tail}" async def async_setup(hass: HomeAssistant, config: dict) -> bool: hass.data.setdefault(DOMAIN, {}) return True async def _async_ensure_settings(hass: HomeAssistant) -> dict: domain_data = hass.data.setdefault(DOMAIN, {}) if DATA_STORE not in domain_data: domain_data[DATA_STORE] = Store(hass, 1, f"{DOMAIN}_panel_settings") if DATA_PANEL_SETTINGS not in domain_data: domain_data[DATA_PANEL_SETTINGS] = await domain_data[DATA_STORE].async_load() or { CONF_PANEL_ENABLED: True, CONF_PANEL_TARGET_ENTRY_ID: None, } return domain_data[DATA_PANEL_SETTINGS] async def _async_save_settings(hass: HomeAssistant) -> None: data = hass.data.setdefault(DOMAIN, {}) await data[DATA_STORE].async_save(data[DATA_PANEL_SETTINGS]) async def _async_load_runtime_data(hass: HomeAssistant) -> dict: data = hass.data.setdefault(DOMAIN, {}) if "runtime_store" not in data: data["runtime_store"] = Store(hass, 1, RUNTIME_STORE_KEY) if "runtime_data" not in data: data["runtime_data"] = await data["runtime_store"].async_load() or {CONF_RESUME_POSITIONS: {}} original = data["runtime_data"].get(CONF_RESUME_POSITIONS, {}) migrated = _migrate_resume_positions(original) data["runtime_data"][CONF_RESUME_POSITIONS] = migrated if migrated != original: await data["runtime_store"].async_save(data["runtime_data"]) data["runtime_data"].setdefault(CONF_RESUME_POSITIONS, {}) data.setdefault("runtime_lock", asyncio.Lock()) return data["runtime_data"] async def _async_save_runtime_data(hass: HomeAssistant) -> None: data = hass.data.setdefault(DOMAIN, {}) await data["runtime_store"].async_save(data["runtime_data"]) def _migrate_resume_positions(value: object) -> dict[str, float]: """Canonicalize, deduplicate, and cap persisted resume positions.""" if not isinstance(value, dict): return {} migrated: dict[str, float] = {} for raw_url, raw_position in value.items(): key = canonical_media_key(raw_url) if not key: continue try: position = float(raw_position) except (TypeError, ValueError): continue if position < 0: continue previous = migrated.pop(key, None) migrated[key] = max(previous, position) if previous is not None else position if len(migrated) > MAX_RESUME_POSITIONS: migrated = dict(list(migrated.items())[-MAX_RESUME_POSITIONS:]) return migrated def _get_entry_data(hass: HomeAssistant, entry_id: str) -> dict | None: return hass.data.get(DOMAIN, {}).get(entry_id) def _fallback_entry_id(hass: HomeAssistant) -> str | None: for entry_id, value in hass.data.get(DOMAIN, {}).items(): if isinstance(value, dict) and DATA_API in value: return entry_id return None def _resolve_entry_ids_for_call(hass: HomeAssistant, call: ServiceCall) -> list[str]: target = TargetSelection(call.data) registry = er.async_get(hass) entry_ids: list[str] = [] selected = async_extract_referenced_entity_ids(hass, target) entity_ids = selected.referenced | selected.indirectly_referenced for entity_id in entity_ids: reg_entry = registry.async_get(entity_id) if reg_entry and _get_entry_data(hass, reg_entry.config_entry_id): entry_ids.append(reg_entry.config_entry_id) if entry_ids: return list(dict.fromkeys(entry_ids)) if target.has_any_target: raise ServiceValidationError("The selected target does not contain a loaded RelayTV media player") panel_target = hass.data.get(DOMAIN, {}).get(DATA_PANEL_SETTINGS, {}).get(CONF_PANEL_TARGET_ENTRY_ID) if panel_target and _get_entry_data(hass, panel_target): return [panel_target] fallback = _fallback_entry_id(hass) return [fallback] if fallback else [] def _resolve_entry_id_for_call(hass: HomeAssistant, call: ServiceCall) -> str | None: entry_ids = _resolve_entry_ids_for_call(hass, call) return entry_ids[0] if entry_ids else None async def _resolve_upload_path(hass: HomeAssistant, call: ServiceCall) -> str: media = call.data.get("file") if isinstance(media, dict): media_content_id = media.get("media_content_id") if isinstance(media_content_id, str) and media_content_id: resolved = await async_resolve_media(hass, media_content_id, None) if resolved.path is None: raise ServiceValidationError("RelayTV upload requires a local media file") path = str(resolved.path) if await hass.async_add_executor_job(Path(path).is_file): return path raise ServiceValidationError(f"RelayTV upload file does not exist: {path}") file_path = str(call.data.get("file_path") or "").strip() if not file_path: raise ServiceValidationError("RelayTV upload requires either file or file_path") allowed = await hass.async_add_executor_job(hass.config.is_allowed_path, file_path) if not allowed: raise ServiceValidationError( f"RelayTV upload file path is not allowed by Home Assistant: {file_path}. " "Add the directory to allowlist_external_dirs." ) if not await hass.async_add_executor_job(Path(file_path).is_file): raise ServiceValidationError(f"RelayTV upload file does not exist: {file_path}") return file_path def _upload_title(call: ServiceCall, path: str) -> str | None: title = str(call.data.get("title") or "").strip() return title or Path(path).stem def _resolve_entries_for_entities(hass: HomeAssistant, entity_ids: list[str]) -> list[str]: registry = er.async_get(hass) result: list[str] = [] for entity_id in entity_ids: reg_entry = registry.async_get(entity_id) if reg_entry and _get_entry_data(hass, reg_entry.config_entry_id): result.append(reg_entry.config_entry_id) return list(dict.fromkeys(result)) def _register_panel(hass: HomeAssistant, *, path: str, title: str, icon: str, url: str) -> None: frontend.async_register_built_in_panel( hass, component_name="iframe", sidebar_title=title, sidebar_icon=icon, frontend_url_path=path, config={"url": url}, require_admin=False, ) def _async_unregister_panel(hass: HomeAssistant) -> None: try: frontend.async_remove_panel(hass, DEFAULT_PANEL_PATH) except Exception: _LOGGER.debug("Panel removal failed (it may not exist)", exc_info=True) async def _async_update_panel(hass: HomeAssistant) -> None: settings = await _async_ensure_settings(hass) if not settings.get(CONF_PANEL_ENABLED, True): _async_unregister_panel(hass) return target_entry_id = settings.get(CONF_PANEL_TARGET_ENTRY_ID) target = _get_entry_data(hass, target_entry_id) if target_entry_id else None if target is None: configured_ids = {entry.entry_id for entry in hass.config_entries.async_entries(DOMAIN)} if target_entry_id in configured_ids: # Preserve both the selection and the existing panel while that # entry is still loading. return fallback_id = _fallback_entry_id(hass) if not fallback_id: _async_unregister_panel(hass) return settings[CONF_PANEL_TARGET_ENTRY_ID] = fallback_id await _async_save_settings(hass) target = _get_entry_data(hass, fallback_id) target_entry_id = fallback_id if not target: _async_unregister_panel(hass) return url = target[DATA_API].base_url _async_unregister_panel(hass) _register_panel(hass, path=DEFAULT_PANEL_PATH, title=DEFAULT_PANEL_TITLE, icon=DEFAULT_PANEL_ICON, url=url) _LOGGER.info("Registered RelayTV panel to entry %s (%s)", target_entry_id, url) async def _async_set_default_sidebar_target(hass: HomeAssistant, entry: ConfigEntry) -> None: settings = await _async_ensure_settings(hass) if not settings.get(CONF_PANEL_TARGET_ENTRY_ID): settings[CONF_PANEL_TARGET_ENTRY_ID] = entry.entry_id await _async_save_settings(hass) def _entry_mappings(entry: ConfigEntry) -> list[dict]: mappings = entry.options.get(CONF_SENSOR_STREAM_MAPPINGS, []) if not isinstance(mappings, list): return [] valid: list[dict] = [] for item in mappings: if not isinstance(item, dict): continue sensor = item.get("sensor_entity_id") url = item.get("url") if isinstance(sensor, str) and isinstance(url, str) and sensor and url: valid.append({"sensor_entity_id": sensor, "url": url}) return valid def _setup_mapping_listeners(hass: HomeAssistant, entry: ConfigEntry) -> list: entry_data = _get_entry_data(hass, entry.entry_id) if not entry_data: return [] unsubscribers = [] for mapping in _entry_mappings(entry): sensor_entity_id = mapping["sensor_entity_id"] url = mapping["url"] @callback def _listener(event, sensor_entity_id=sensor_entity_id, url=url): old_state = event.data.get("old_state") new_state = event.data.get("new_state") if new_state is None: return if old_state is not None and old_state.state == "on": return if new_state.state != "on": return hass.async_create_task(entry_data[DATA_API].play_temporary(url=url)) unsubscribers.append(async_track_state_change_event(hass, [sensor_entity_id], _listener)) return unsubscribers async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: base_url = _normalize_base_url(entry.data.get(CONF_BASE_URL, "")) if not base_url: _LOGGER.error("RelayTV base URL is empty") return False await _async_ensure_settings(hass) await _async_load_runtime_data(hass) session = aiohttp_client.async_get_clientsession(hass) @callback def _start_reauth() -> None: entry.async_start_reauth_if_available(hass) api = RelayTVApi( session=session, base_url=base_url, api_token=str(entry.data.get(CONF_API_TOKEN) or ""), ) try: await api.validate() except RelayTVAuthError as err: raise ConfigEntryAuthFailed from err except RelayTVApiError as err: raise ConfigEntryNotReady(str(err)) from err api.on_auth_failure = _start_reauth coordinator = RelayTVCoordinator(hass=hass, api=api) hass.data.setdefault(DOMAIN, {})[entry.entry_id] = { DATA_API: api, DATA_COORDINATOR: coordinator, DATA_LAST_SNAPSHOT_URL: None, "mapping_unsubs": [], } @callback def _save_resume_position() -> None: status = coordinator.data if isinstance(coordinator.data, dict) else {} url = status.get("url") or (status.get("now_playing") or {}).get("url") position = status.get("position") duration = status.get("duration") key = canonical_media_key(url) if not key: return try: pos = float(position) except Exception: return try: dur = float(duration) if duration is not None else 0.0 except Exception: dur = 0.0 if pos < 60.0 or dur < 120.0: return async def _save() -> None: await _async_load_runtime_data(hass) async with hass.data[DOMAIN]["runtime_lock"]: runtime = hass.data[DOMAIN]["runtime_data"] positions = runtime[CONF_RESUME_POSITIONS] if dur and (pos / dur >= 0.98 or dur - pos <= 30.0): if positions.pop(key, None) is not None: await _async_save_runtime_data(hass) return previous = positions.get(key) if previous is not None and abs(float(previous) - pos) < 10.0: return positions.pop(key, None) positions[key] = pos while len(positions) > MAX_RESUME_POSITIONS: positions.pop(next(iter(positions))) await _async_save_runtime_data(hass) hass.async_create_task(_save()) coordinator.async_add_listener(_save_resume_position) await coordinator.async_config_entry_first_refresh() await coordinator.async_start() await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) hass.data[DOMAIN][entry.entry_id]["mapping_unsubs"] = _setup_mapping_listeners(hass, entry) await _async_set_default_sidebar_target(hass, entry) await _async_update_panel(hass) async def _handle_smart_url(call: ServiceCall): url = (call.data.get("url") or "").strip() if not url: return for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].smart_url(url) await store[DATA_COORDINATOR].async_request_refresh() async def _handle_play_now(call: ServiceCall): url = (call.data.get("url") or "").strip() if not url: return for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].play(url=url, use_ytdlp=call.data.get("use_ytdlp"), cec=call.data.get("cec")) await store[DATA_COORDINATOR].async_request_refresh() async def _handle_announce(call: ServiceCall): await _handle_play_now(call) async def _handle_play_temporary(call: ServiceCall): url = (call.data.get("url") or "").strip() if not url: return for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].play_temporary( url=url, timeout_sec=call.data.get("timeout"), volume_override=call.data.get("volume"), ) await store[DATA_COORDINATOR].async_request_refresh() async def _handle_overlay(call: ServiceCall): for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].overlay( text=call.data.get("text"), duration=call.data.get("duration"), position=call.data.get("position"), image_url=call.data.get("image_url"), ) async def _handle_play_synced(call: ServiceCall): url = (call.data.get("url") or "").strip() if not url: return delay = float(call.data.get("delay_buffer_sec") or 2) start_at = time.time() + delay targets = call.data.get("target_entities") entry_ids: list[str] = [] if isinstance(targets, str): targets = [targets] if isinstance(targets, list): entry_ids = _resolve_entries_for_entities(hass, [e for e in targets if isinstance(e, str)]) if not entry_ids: entry_ids = _resolve_entry_ids_for_call(hass, call) for entry_id in entry_ids: store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].play_at(url=url, start_at=start_at) await store[DATA_COORDINATOR].async_request_refresh() async def _handle_snapshot(call: ServiceCall): for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue data = await store[DATA_API].snapshot() or {} snapshot_url = data.get("image_url") if isinstance(data, dict) else None snapshot_url = _absolute_url(store[DATA_API].base_url, snapshot_url) if snapshot_url: store[DATA_LAST_SNAPSHOT_URL] = snapshot_url await store[DATA_COORDINATOR].async_request_refresh() async def _handle_play_with_resume(call: ServiceCall): url = (call.data.get("url") or "").strip() if not url: return runtime = await _async_load_runtime_data(hass) resume_position = runtime.get(CONF_RESUME_POSITIONS, {}).get(canonical_media_key(url)) for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue await store[DATA_API].play(url=url, use_ytdlp=call.data.get("use_ytdlp"), cec=call.data.get("cec")) if resume_position is not None: await store[DATA_API].seek_abs(float(resume_position)) await store[DATA_COORDINATOR].async_request_refresh() async def _handle_upload_media(call: ServiceCall): path = await _resolve_upload_path(hass, call) title = _upload_title(call, path) results: list[dict] = [] for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue data = await store[DATA_API].upload_media(path, title=title) if data is None: raise ServiceValidationError(f"RelayTV media upload failed for {store[DATA_API].base_url}") results.append({"entry_id": entry_id, "base_url": store[DATA_API].base_url, "response": data}) return {"results": results} async def _handle_upload_media_play(call: ServiceCall): path = await _resolve_upload_path(hass, call) title = _upload_title(call, path) results: list[dict] = [] for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue data = await store[DATA_API].upload_media_play(path, title=title) if data is None: raise ServiceValidationError(f"RelayTV media upload/play failed for {store[DATA_API].base_url}") await store[DATA_COORDINATOR].async_request_refresh() results.append({"entry_id": entry_id, "base_url": store[DATA_API].base_url, "response": data}) return {"results": results} async def _handle_upload_media_enqueue(call: ServiceCall): path = await _resolve_upload_path(hass, call) title = _upload_title(call, path) results: list[dict] = [] for entry_id in _resolve_entry_ids_for_call(hass, call): store = _get_entry_data(hass, entry_id) if not store: continue data = await store[DATA_API].upload_media_enqueue(path, title=title) if data is None: raise ServiceValidationError(f"RelayTV media upload/enqueue failed for {store[DATA_API].base_url}") await store[DATA_COORDINATOR].async_request_refresh() results.append({"entry_id": entry_id, "base_url": store[DATA_API].base_url, "response": data}) return {"results": results} for service_name, handler in ( (SERVICE_SMART_URL, _handle_smart_url), (SERVICE_PLAY_NOW, _handle_play_now), (SERVICE_ANNOUNCE, _handle_announce), (SERVICE_PLAY_TEMPORARY, _handle_play_temporary), (SERVICE_OVERLAY, _handle_overlay), (SERVICE_PLAY_SYNCED, _handle_play_synced), (SERVICE_SNAPSHOT, _handle_snapshot), (SERVICE_PLAY_WITH_RESUME, _handle_play_with_resume), ): if not hass.services.has_service(DOMAIN, service_name): hass.services.async_register(DOMAIN, service_name, handler) for service_name, handler in ( (SERVICE_UPLOAD_MEDIA, _handle_upload_media), (SERVICE_UPLOAD_MEDIA_PLAY, _handle_upload_media_play), (SERVICE_UPLOAD_MEDIA_ENQUEUE, _handle_upload_media_enqueue), ): if not hass.services.has_service(DOMAIN, service_name): hass.services.async_register( DOMAIN, service_name, handler, supports_response=SupportsResponse.OPTIONAL, ) entry.async_on_unload(entry.add_update_listener(_async_update_listener)) return True async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) entry_data = hass.data.get(DOMAIN, {}).pop(entry.entry_id, None) if entry_data: await entry_data[DATA_COORDINATOR].async_stop() for unsub in entry_data.get("mapping_unsubs", []): unsub() settings = await _async_ensure_settings(hass) if settings.get(CONF_PANEL_TARGET_ENTRY_ID) == entry.entry_id: replacement_id = _fallback_entry_id(hass) settings[CONF_PANEL_TARGET_ENTRY_ID] = replacement_id await _async_save_settings(hass) if not any(cfg_entry.entry_id in hass.data.get(DOMAIN, {}) for cfg_entry in hass.config_entries.async_entries(DOMAIN)): _async_unregister_panel(hass) for service_name in ( SERVICE_SMART_URL, SERVICE_PLAY_NOW, SERVICE_ANNOUNCE, SERVICE_PLAY_TEMPORARY, SERVICE_OVERLAY, SERVICE_PLAY_SYNCED, SERVICE_SNAPSHOT, SERVICE_PLAY_WITH_RESUME, SERVICE_UPLOAD_MEDIA, SERVICE_UPLOAD_MEDIA_PLAY, SERVICE_UPLOAD_MEDIA_ENQUEUE, ): if hass.services.has_service(DOMAIN, service_name): hass.services.async_remove(DOMAIN, service_name) return unload_ok await _async_update_panel(hass) return unload_ok async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: base_url = _normalize_base_url(entry.data.get(CONF_BASE_URL, "")) store = _get_entry_data(hass, entry.entry_id) if store and base_url: store[DATA_API].base_url = base_url store[DATA_API].api_token = str(entry.data.get(CONF_API_TOKEN) or "") await store[DATA_COORDINATOR].async_restart() for unsub in store.get("mapping_unsubs", []): unsub() store["mapping_unsubs"] = _setup_mapping_listeners(hass, entry) await _async_update_panel(hass)