repairs.py

Go to the documentation of this file.

"""Repairs implementation for supervisor integration."""
 
from __future__ import annotations
 
from collections.abc import Callable, Coroutine
from types import MethodType
from typing import Any
 
from aiohasupervisor import SupervisorError
from aiohasupervisor.models import ContextType
import voluptuous as vol
 
from homeassistant.components.repairs import RepairsFlow
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
 
from . import get_addons_info, get_issues_info
from .const import (
    ISSUE_KEY_ADDON_BOOT_FAIL,
    ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED,
    ISSUE_KEY_SYSTEM_DOCKER_CONFIG,
    PLACEHOLDER_KEY_ADDON,
    PLACEHOLDER_KEY_COMPONENTS,
    PLACEHOLDER_KEY_REFERENCE,
)
from .handler import get_supervisor_client
from .issues import Issue, Suggestion
 
HELP_URLS = {
    "help_url": "https://www.home-assistant.io/help/",
    "community_url": "https://community.home-assistant.io/",
}
 
SUGGESTION_CONFIRMATION_REQUIRED = {
    "addon_execute_remove",
    "system_adopt_data_disk",
    "system_execute_reboot",
}
 
 
EXTRA_PLACEHOLDERS = {
    "issue_mount_mount_failed": {
        "storage_url": "/config/storage",
    },
    ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED: HELP_URLS,
}
 
 
class SupervisorIssueRepairFlow(RepairsFlow):
    """Handler for an issue fixing flow."""
 
    _data: dict[str, Any] | None = None
    _issue: Issue | None = None
 
    def __init__(self, hass: HomeAssistant, issue_id: str) -> None:
        """Initialize repair flow."""
        self._issue_id_issue_id = issue_id
        self._supervisor_client_supervisor_client = get_supervisor_client(hass)
        super().__init__()
 
    @property
    def issue(self) -> Issue | None:
        """Get associated issue."""
        supervisor_issues = get_issues_info(self.hass)
        if not self._issue_issue and supervisor_issues:
            self._issue_issue = supervisor_issues.get_issue(self._issue_id_issue_id)
 
        return self._issue_issue
 
    @property
    def description_placeholders(self) -> dict[str, str] | None:
        """Get description placeholders for steps."""
        placeholders = {}
        if self.issueissue:
            placeholders = EXTRA_PLACEHOLDERS.get(self.issueissue.key, {})
            if self.issueissue.reference:
                placeholders |= {PLACEHOLDER_KEY_REFERENCE: self.issueissue.reference}
 
        return placeholders or None
 
    def _async_form_for_suggestion(self, suggestion: Suggestion) -> FlowResult:
        """Return form for suggestion."""
        return self.async_show_form(
            step_id=suggestion.key,
            data_schema=vol.Schema({}),
            description_placeholders=self.description_placeholdersdescription_placeholders,
            last_step=True,
        )
 
    async def async_step_init(self, _: None = None) -> FlowResult:
        """Handle the first step of a fix flow."""
        # Out of sync with supervisor, issue is resolved or not fixable. Remove it
        if not self.issueissue or not self.issueissue.suggestions:
            return self.async_create_entry(data={})
 
        # All suggestions have the same logic: Apply them in supervisor,
        # optionally with a confirmation step. Generating the required handler for each
        # allows for shared logic but screens can still be translated per step id.
        for suggestion in self.issueissue.suggestions:
            setattr(
                self,
                f"async_step_{suggestion.key}",
                MethodType(self._async_step_async_step(suggestion), self),
            )
 
        if len(self.issueissue.suggestions) > 1:
            return await self.async_step_fix_menuasync_step_fix_menu()
 
        # Always show a form for one suggestion to explain to user what's happening
        return self._async_form_for_suggestion_async_form_for_suggestion(self.issueissue.suggestions[0])
 
    async def async_step_fix_menu(self, _: None = None) -> FlowResult:
        """Show the fix menu."""
        assert self.issueissue
 
        return self.async_show_menu(
            step_id="fix_menu",
            menu_options=[suggestion.key for suggestion in self.issueissue.suggestions],
            description_placeholders=self.description_placeholdersdescription_placeholders,
        )
 
    async def _async_step_apply_suggestion(
        self, suggestion: Suggestion, confirmed: bool = False
    ) -> FlowResult:
        """Handle applying a suggestion as a flow step. Optionally request confirmation."""
        if not confirmed and suggestion.key in SUGGESTION_CONFIRMATION_REQUIRED:
            return self._async_form_for_suggestion_async_form_for_suggestion(suggestion)
 
        try:
            await self._supervisor_client_supervisor_client.resolution.apply_suggestion(suggestion.uuid)
        except SupervisorError:
            return self.async_abort(reason="apply_suggestion_fail")
 
        return self.async_create_entry(data={})
 
    @staticmethod
    def _async_step(
        suggestion: Suggestion,
    ) -> Callable[
        [SupervisorIssueRepairFlow, dict[str, str] | None],
        Coroutine[Any, Any, FlowResult],
    ]:
        """Generate a step handler for a suggestion."""
 
        async def _async_step(
            self: SupervisorIssueRepairFlow, user_input: dict[str, str] | None = None
        ) -> FlowResult:
            """Handle a flow step for a suggestion."""
            return await self._async_step_apply_suggestion_async_step_apply_suggestion(
                suggestion, confirmed=user_input is not None
            )
 
        return _async_step
 
 
class DockerConfigIssueRepairFlow(SupervisorIssueRepairFlow):
    """Handler for docker config issue fixing flow."""
 
    @property
    def description_placeholders(self) -> dict[str, str] | None:
        """Get description placeholders for steps."""
        placeholders = {PLACEHOLDER_KEY_COMPONENTS: ""}
        supervisor_issues = get_issues_info(self.hass)
        if supervisor_issues and self.issueissue:
            addons = get_addons_info(self.hass) or {}
            components: list[str] = []
            for issue in supervisor_issues.issues:
                if issue.key == self.issueissue.key or issue.type != self.issueissue.type:
                    continue
 
                if issue.context == ContextType.CORE:
                    components.insert(0, "Home Assistant")
                elif issue.context == ContextType.ADDON:
                    components.append(
                        next(
                            (
                                info["name"]
                                for slug, info in addons.items()
                                if slug == issue.reference
                            ),
                            issue.reference or "",
                        )
                    )
 
            placeholders[PLACEHOLDER_KEY_COMPONENTS] = "\n- ".join(components)
 
        return placeholders
 
 
class AddonIssueRepairFlow(SupervisorIssueRepairFlow):
    """Handler for addon issue fixing flows."""
 
    @property
    def description_placeholders(self) -> dict[str, str] | None:
        """Get description placeholders for steps."""
        placeholders: dict[str, str] = super().description_placeholders or {}
        if self.issueissue and self.issueissue.reference:
            addons = get_addons_info(self.hass)
            if addons and self.issueissue.reference in addons:
                placeholders[PLACEHOLDER_KEY_ADDON] = addons[self.issueissue.reference][
                    "name"
                ]
            else:
                placeholders[PLACEHOLDER_KEY_ADDON] = self.issueissue.reference
 
        return placeholders or None
 
 
async def async_create_fix_flow(
    hass: HomeAssistant,
    issue_id: str,
    data: dict[str, str | int | float | None] | None,
) -> RepairsFlow:
    """Create flow."""
    supervisor_issues = get_issues_info(hass)
    issue = supervisor_issues and supervisor_issues.get_issue(issue_id)
    if issue and issue.key == ISSUE_KEY_SYSTEM_DOCKER_CONFIG:
        return DockerConfigIssueRepairFlow(hass, issue_id)
    if issue and issue.key in {
        ISSUE_KEY_ADDON_DETACHED_ADDON_REMOVED,
        ISSUE_KEY_ADDON_BOOT_FAIL,
    }:
        return AddonIssueRepairFlow(hass, issue_id)
 
    return SupervisorIssueRepairFlow(hass, issue_id)