Use entry runtime data on Filesize (#116962)

* Use entry runtime data on Filesize

* Fix comment

* ignore

* Another way

* Refactor
This commit is contained in:
G Johansson 2024-05-07 20:07:32 +02:00 committed by GitHub
parent 789aadcc4c
commit 5bef2d5d25
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 31 additions and 36 deletions

View file

@ -2,18 +2,39 @@
from __future__ import annotations from __future__ import annotations
import pathlib
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_FILE_PATH from homeassistant.const import CONF_FILE_PATH
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .const import PLATFORMS from .const import PLATFORMS
from .coordinator import FileSizeCoordinator from .coordinator import FileSizeCoordinator
FileSizeConfigEntry = ConfigEntry[FileSizeCoordinator]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
def _get_full_path(hass: HomeAssistant, path: str) -> pathlib.Path:
"""Check if path is valid, allowed and return full path."""
get_path = pathlib.Path(path)
if not hass.config.is_allowed_path(path):
raise ConfigEntryNotReady(f"Filepath {path} is not valid or allowed")
if not get_path.exists() or not get_path.is_file():
raise ConfigEntryNotReady(f"Can not access file {path}")
return get_path.absolute()
async def async_setup_entry(hass: HomeAssistant, entry: FileSizeConfigEntry) -> bool:
"""Set up from a config entry.""" """Set up from a config entry."""
coordinator = FileSizeCoordinator(hass, entry.data[CONF_FILE_PATH]) path = await hass.async_add_executor_job(
_get_full_path, hass, entry.data[CONF_FILE_PATH]
)
coordinator = FileSizeCoordinator(hass, path)
await coordinator.async_config_entry_first_refresh() await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True return True

View file

@ -19,7 +19,7 @@ _LOGGER = logging.getLogger(__name__)
class FileSizeCoordinator(DataUpdateCoordinator[dict[str, int | float | datetime]]): class FileSizeCoordinator(DataUpdateCoordinator[dict[str, int | float | datetime]]):
"""Filesize coordinator.""" """Filesize coordinator."""
def __init__(self, hass: HomeAssistant, unresolved_path: str) -> None: def __init__(self, hass: HomeAssistant, path: pathlib.Path) -> None:
"""Initialize filesize coordinator.""" """Initialize filesize coordinator."""
super().__init__( super().__init__(
hass, hass,
@ -28,28 +28,12 @@ class FileSizeCoordinator(DataUpdateCoordinator[dict[str, int | float | datetime
update_interval=timedelta(seconds=60), update_interval=timedelta(seconds=60),
always_update=False, always_update=False,
) )
self._unresolved_path = unresolved_path self.path: pathlib.Path = path
self._path: pathlib.Path | None = None
def _get_full_path(self) -> pathlib.Path:
"""Check if path is valid, allowed and return full path."""
path = self._unresolved_path
get_path = pathlib.Path(path)
if not self.hass.config.is_allowed_path(path):
raise UpdateFailed(f"Filepath {path} is not valid or allowed")
if not get_path.exists() or not get_path.is_file():
raise UpdateFailed(f"Can not access file {path}")
return get_path.absolute()
def _update(self) -> os.stat_result: def _update(self) -> os.stat_result:
"""Fetch file information.""" """Fetch file information."""
if not self._path:
self._path = self._get_full_path()
try: try:
return self._path.stat() return self.path.stat()
except OSError as error: except OSError as error:
raise UpdateFailed(f"Can not retrieve file statistics {error}") from error raise UpdateFailed(f"Can not retrieve file statistics {error}") from error

View file

@ -4,7 +4,6 @@ from __future__ import annotations
from datetime import datetime from datetime import datetime
import logging import logging
import pathlib
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorDeviceClass, SensorDeviceClass,
@ -12,13 +11,13 @@ from homeassistant.components.sensor import (
SensorEntityDescription, SensorEntityDescription,
SensorStateClass, SensorStateClass,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.const import EntityCategory, UnitOfInformation
from homeassistant.const import CONF_FILE_PATH, EntityCategory, UnitOfInformation
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import FileSizeConfigEntry
from .const import DOMAIN from .const import DOMAIN
from .coordinator import FileSizeCoordinator from .coordinator import FileSizeCoordinator
@ -53,20 +52,12 @@ SENSOR_TYPES = (
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,
entry: ConfigEntry, entry: FileSizeConfigEntry,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up the platform from config entry.""" """Set up the platform from config entry."""
path = entry.data[CONF_FILE_PATH]
get_path = await hass.async_add_executor_job(pathlib.Path, path)
fullpath = str(get_path.absolute())
coordinator = FileSizeCoordinator(hass, fullpath)
await coordinator.async_config_entry_first_refresh()
async_add_entities( async_add_entities(
FilesizeEntity(description, fullpath, entry.entry_id, coordinator) FilesizeEntity(description, entry.entry_id, entry.runtime_data)
for description in SENSOR_TYPES for description in SENSOR_TYPES
) )
@ -79,13 +70,12 @@ class FilesizeEntity(CoordinatorEntity[FileSizeCoordinator], SensorEntity):
def __init__( def __init__(
self, self,
description: SensorEntityDescription, description: SensorEntityDescription,
path: str,
entry_id: str, entry_id: str,
coordinator: FileSizeCoordinator, coordinator: FileSizeCoordinator,
) -> None: ) -> None:
"""Initialize the Filesize sensor.""" """Initialize the Filesize sensor."""
super().__init__(coordinator) super().__init__(coordinator)
base_name = path.split("/")[-1] base_name = str(coordinator.path.absolute()).rsplit("/", maxsplit=1)[-1]
self._attr_unique_id = ( self._attr_unique_id = (
entry_id if description.key == "file" else f"{entry_id}-{description.key}" entry_id if description.key == "file" else f"{entry_id}-{description.key}"
) )