Break out sensors for filesize (#68702)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/filesize/sensor.py

@@ -1,7 +1,7 @@
 """Sensor for monitoring the size of a file."""
 from __future__ import annotations
 
-import datetime
+from datetime import datetime, timedelta
 import logging
 import os
 import pathlib
@@ -10,14 +10,25 @@ import voluptuous as vol
 
 from homeassistant.components.sensor import (
     PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
+    SensorDeviceClass,
     SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
 )
 from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
-from homeassistant.const import CONF_FILE_PATH, DATA_MEGABYTES
+from homeassistant.const import CONF_FILE_PATH, DATA_BYTES, DATA_MEGABYTES
 from homeassistant.core import HomeAssistant
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.device_registry import DeviceEntryType
+from homeassistant.helpers.entity import DeviceInfo, EntityCategory
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+    UpdateFailed,
+)
+import homeassistant.util.dt as dt_util
 
 from .const import CONF_FILE_PATHS, DOMAIN
 
@@ -25,6 +36,34 @@ _LOGGER = logging.getLogger(__name__)
 
 ICON = "mdi:file"
 
+SENSOR_TYPES = (
+    SensorEntityDescription(
+        key="file",
+        icon=ICON,
+        name="Size",
+        native_unit_of_measurement=DATA_MEGABYTES,
+        state_class=SensorStateClass.MEASUREMENT,
+    ),
+    SensorEntityDescription(
+        key="bytes",
+        entity_registry_enabled_default=False,
+        icon=ICON,
+        name="Size bytes",
+        native_unit_of_measurement=DATA_BYTES,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+    SensorEntityDescription(
+        key="last_updated",
+        entity_registry_enabled_default=False,
+        icon=ICON,
+        name="Last Updated",
+        device_class=SensorDeviceClass.TIMESTAMP,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+)
+
 PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
     {vol.Required(CONF_FILE_PATHS): vol.All(cv.ensure_list, [cv.isfile])}
 )
@@ -65,36 +104,82 @@ async def async_setup_entry(
     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()
+
     if get_path.exists() and get_path.is_file():
-        async_add_entities([FilesizeEntity(fullpath, entry.entry_id)], True)
+        async_add_entities(
+            [
+                FilesizeEntity(description, fullpath, entry.entry_id, coordinator)
+                for description in SENSOR_TYPES
+            ]
+        )
 
 
-class FilesizeEntity(SensorEntity):
+class FileSizeCoordinator(DataUpdateCoordinator):
-    """Encapsulates file size information."""
+    """Filesize coordinator."""
 
-    _attr_native_unit_of_measurement = DATA_MEGABYTES
+    def __init__(self, hass: HomeAssistant, path: str) -> None:
-    _attr_icon = ICON
+        """Initialize filesize coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=60),
+        )
+        self._path = path
 
-    def __init__(self, path: str, entry_id: str) -> None:
+    async def _async_update_data(self) -> dict[str, float | int | datetime]:
-        """Initialize the data object."""
+        """Fetch file information."""
-        self._path = path  # Need to check its a valid path
-        self._attr_name = path.split("/")[-1]
-        self._attr_unique_id = entry_id
-
-    def update(self) -> None:
-        """Update the sensor."""
         try:
             statinfo = os.stat(self._path)
         except OSError as error:
-            _LOGGER.error("Can not retrieve file statistics %s", error)
+            raise UpdateFailed(f"Can not retrieve file statistics {error}") from error
-            self._attr_native_value = None
-            return
 
         size = statinfo.st_size
-        last_updated = datetime.datetime.fromtimestamp(statinfo.st_mtime).isoformat()
+        last_updated = datetime.fromtimestamp(statinfo.st_mtime).replace(
-        self._attr_native_value = round(size / 1e6, 2) if size else None
+            tzinfo=dt_util.UTC
-        self._attr_extra_state_attributes = {
+        )
-            "path": self._path,
+        _LOGGER.debug("size %s, last updated %s", size, last_updated)
-            "last_updated": last_updated,
+        data: dict[str, int | float | datetime] = {
+            "file": round(size / 1e6, 2),
             "bytes": size,
+            "last_updated": last_updated,
         }
+
+        return data
+
+
+class FilesizeEntity(CoordinatorEntity[FileSizeCoordinator], SensorEntity):
+    """Encapsulates file size information."""
+
+    entity_description: SensorEntityDescription
+
+    def __init__(
+        self,
+        description: SensorEntityDescription,
+        path: str,
+        entry_id: str,
+        coordinator: FileSizeCoordinator,
+    ) -> None:
+        """Initialize the data object."""
+        super().__init__(coordinator)
+        base_name = path.split("/")[-1]
+        self._attr_name = f"{base_name} {description.name}"
+        self._attr_unique_id = (
+            entry_id if description.key == "file" else f"{entry_id}-{description.key}"
+        )
+        self.entity_description = description
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={(DOMAIN, entry_id)},
+            name=base_name,
+        )
+
+    @property
+    def native_value(self) -> float | int | datetime:
+        """Return the value of the sensor."""
+        value: float | int | datetime = self.coordinator.data[
+            self.entity_description.key
+        ]
+        return value

tests/components/filesize/test_sensor.py

@@ -1,9 +1,11 @@
 """The tests for the filesize sensor."""
 import os
 
-from homeassistant.const import CONF_FILE_PATH, STATE_UNKNOWN
+from homeassistant.components.filesize.const import DOMAIN
+from homeassistant.const import CONF_FILE_PATH, STATE_UNAVAILABLE
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_component import async_update_entity
+from homeassistant.setup import async_setup_component
 
 from . import TEST_FILE, TEST_FILE_NAME, create_file
 
@@ -38,19 +40,18 @@ async def test_valid_path(
     await hass.config_entries.async_setup(mock_config_entry.entry_id)
     await hass.async_block_till_done()
 
-    state = hass.states.get("sensor.file_txt")
+    state = hass.states.get("sensor.file_txt_size")
     assert state
     assert state.state == "0.0"
-    assert state.attributes.get("bytes") == 4
 
     await hass.async_add_executor_job(os.remove, testfile)
 
 
-async def test_state_unknown(
+async def test_state_unavailable(
     hass: HomeAssistant, tmpdir: str, mock_config_entry: MockConfigEntry
 ) -> None:
     """Verify we handle state unavailable."""
-    testfile = f"{tmpdir}/file"
+    testfile = f"{tmpdir}/file.txt"
     create_file(testfile)
     hass.config.allowlist_external_dirs = {tmpdir}
     mock_config_entry.add_to_hass(hass)
@@ -61,12 +62,32 @@ async def test_state_unknown(
     await hass.config_entries.async_setup(mock_config_entry.entry_id)
     await hass.async_block_till_done()
 
-    state = hass.states.get("sensor.file")
+    state = hass.states.get("sensor.file_txt_size")
     assert state
     assert state.state == "0.0"
 
     await hass.async_add_executor_job(os.remove, testfile)
-    await async_update_entity(hass, "sensor.file")
+    await async_update_entity(hass, "sensor.file_txt_size")
 
-    state = hass.states.get("sensor.file")
+    state = hass.states.get("sensor.file_txt_size")
-    assert state.state == STATE_UNKNOWN
+    assert state.state == STATE_UNAVAILABLE
+
+
+async def test_import_query(hass: HomeAssistant, tmpdir: str) -> None:
+    """Test import from yaml."""
+    testfile = f"{tmpdir}/file.txt"
+    create_file(testfile)
+    hass.config.allowlist_external_dirs = {tmpdir}
+    config = {
+        "sensor": {
+            "platform": "filesize",
+            "file_paths": [testfile],
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+    await hass.async_block_till_done()
+
+    assert hass.config_entries.async_entries(DOMAIN)
+    data = hass.config_entries.async_entries(DOMAIN)[0].data
+    assert data[CONF_FILE_PATH] == testfile