Use HassKey in hardware (#144337)

This commit is contained in:
epenet 2025-05-09 10:04:11 +02:00 committed by GitHub
parent 19b1dc8d65
commit eab1d5717f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 37 additions and 15 deletions

View File

@ -7,14 +7,15 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from . import websocket_api
from .const import DOMAIN
from .const import DATA_HARDWARE, DOMAIN
from .models import HardwareData
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Hardware."""
hass.data[DOMAIN] = {}
hass.data[DATA_HARDWARE] = HardwareData()
await websocket_api.async_setup(hass)

View File

@ -1,3 +1,14 @@
"""Constants for the Hardware integration."""
from __future__ import annotations
from typing import TYPE_CHECKING
from homeassistant.util.hass_dict import HassKey
if TYPE_CHECKING:
from .models import HardwareData
DOMAIN = "hardware"
DATA_HARDWARE: HassKey[HardwareData] = HassKey(DOMAIN)

View File

@ -8,13 +8,15 @@ from homeassistant.helpers.integration_platform import (
async_process_integration_platforms,
)
from .const import DOMAIN
from .const import DATA_HARDWARE, DOMAIN
from .models import HardwareProtocol
async def async_process_hardware_platforms(hass: HomeAssistant) -> None:
async def async_process_hardware_platforms(
hass: HomeAssistant,
) -> None:
"""Start processing hardware platforms."""
hass.data[DOMAIN]["hardware_platform"] = {}
hass.data[DATA_HARDWARE].hardware_platform = {}
await async_process_integration_platforms(
hass, DOMAIN, _register_hardware_platform, wait_for_platforms=True
@ -30,4 +32,4 @@ def _register_hardware_platform(
return
if not hasattr(platform, "async_info"):
raise HomeAssistantError(f"Invalid hardware platform {platform}")
hass.data[DOMAIN]["hardware_platform"][integration_domain] = platform
hass.data[DATA_HARDWARE].hardware_platform[integration_domain] = platform

View File

@ -3,10 +3,21 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
from typing import TYPE_CHECKING, Protocol
from homeassistant.core import HomeAssistant, callback
if TYPE_CHECKING:
from .websocket_api import SystemStatus
@dataclass
class HardwareData:
"""Hardware data."""
hardware_platform: dict[str, HardwareProtocol] = None # type: ignore[assignment]
system_status: SystemStatus = None # type: ignore[assignment]
@dataclass(slots=True)
class BoardInfo:

View File

@ -16,9 +16,8 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .const import DATA_HARDWARE
from .hardware import async_process_hardware_platforms
from .models import HardwareProtocol
@dataclass(slots=True)
@ -34,7 +33,7 @@ async def async_setup(hass: HomeAssistant) -> None:
"""Set up the hardware websocket API."""
websocket_api.async_register_command(hass, ws_info)
websocket_api.async_register_command(hass, ws_subscribe_system_status)
hass.data[DOMAIN]["system_status"] = SystemStatus(
hass.data[DATA_HARDWARE].system_status = SystemStatus(
ha_psutil=await hass.async_add_executor_job(ha_psutil.PsutilWrapper),
remove_periodic_timer=None,
subscribers=set(),
@ -53,12 +52,10 @@ async def ws_info(
"""Return hardware info."""
hardware_info = []
if "hardware_platform" not in hass.data[DOMAIN]:
if hass.data[DATA_HARDWARE].hardware_platform is None:
await async_process_hardware_platforms(hass)
hardware_platform: dict[str, HardwareProtocol] = hass.data[DOMAIN][
"hardware_platform"
]
hardware_platform = hass.data[DATA_HARDWARE].hardware_platform
for platform in hardware_platform.values():
if hasattr(platform, "async_info"):
with contextlib.suppress(HomeAssistantError):
@ -78,7 +75,7 @@ def ws_subscribe_system_status(
) -> None:
"""Subscribe to system status updates."""
system_status: SystemStatus = hass.data[DOMAIN]["system_status"]
system_status = hass.data[DATA_HARDWARE].system_status
@callback
def async_update_status(now: datetime) -> None: