Storage space usage API

.gitignore

@@ -100,3 +100,6 @@ ENV/
 # mypy
 /.mypy_cache/*
 /.dmypy.json
+
+# Mac
+.DS_Store

supervisor/api/__init__.py

@@ -198,6 +198,7 @@ def _register_host(self) -> None:
                 web.post("/host/reload", api_host.reload),
                 web.post("/host/options", api_host.options),
                 web.get("/host/services", api_host.services),
+                web.get("/host/disks/default/usage", api_host.disk_usage),
             ]
         )
 

supervisor/api/const.py

@@ -49,6 +49,7 @@
 ATTR_LOCAL_ONLY = "local_only"
 ATTR_LOCATION_ATTRIBUTES = "location_attributes"
 ATTR_LOCATIONS = "locations"
+ATTR_MAX_DEPTH = "max_depth"
 ATTR_MDNS = "mdns"
 ATTR_MODEL = "model"
 ATTR_MOUNTS = "mounts"

supervisor/api/host.py

@@ -51,6 +51,7 @@
     ATTR_FORCE,
     ATTR_IDENTIFIERS,
     ATTR_LLMNR_HOSTNAME,
+    ATTR_MAX_DEPTH,
     ATTR_STARTUP_TIME,
     ATTR_USE_NTP,
     ATTR_VIRTUALIZATION,
@@ -289,3 +290,49 @@ async def advanced_logs(
     ) -> web.StreamResponse:
         """Return systemd-journald logs. Wrapped as standard API handler."""
         return await self.advanced_logs_handler(request, identifier, follow)
+
+    @api_process
+    async def disk_usage(self, request: web.Request) -> dict:
+        """Return a breakdown of storage usage for the system."""
+
+        max_depth = request.query.get(ATTR_MAX_DEPTH, 1)
+        try:
+            max_depth = int(max_depth)
+        except ValueError:
+            max_depth = 1
+
+        disk = self.sys_hardware.disk
+
+        total, used, _ = await self.sys_run_in_executor(
+            disk.disk_usage, self.sys_config.path_supervisor
+        )
+
+        known_paths = await self.sys_run_in_executor(
+            disk.get_dir_sizes,
+            {
+                "addons_data": self.sys_config.path_addons_data,
+                "addons_config": self.sys_config.path_addon_configs,
+                "media": self.sys_config.path_media,
+                "share": self.sys_config.path_share,
+                "backup": self.sys_config.path_backup,
+                "ssl": self.sys_config.path_ssl,
+                "homeassistant": self.sys_config.path_homeassistant,
+            },
+            max_depth,
+        )
+        return {
+            # this can be the disk/partition ID in the future
+            "id": "root",
+            "label": "Root",
+            "total_bytes": total,
+            "used_bytes": used,
+            "children": [
+                {
+                    "id": "system",
+                    "label": "System",
+                    "used_bytes": used
+                    - sum(path["used_bytes"] for path in known_paths),
+                },
+                *known_paths,
+            ],
+        }

supervisor/hardware/disk.py

@@ -1,8 +1,12 @@
 """Read disk hardware info from system."""
 
+import errno
 import logging
 from pathlib import Path
 import shutil
+from typing import Any
+
+from supervisor.resolution.const import UnhealthyReason
 
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import DBusError, DBusObjectError, HardwareNotFound
@@ -54,25 +58,100 @@ def get_disk_total_space(self, path: str | Path) -> float:
 
         Must be run in executor.
         """
-        total, _, _ = shutil.disk_usage(path)
+        total, _, _ = self.disk_usage(path)
         return round(total / (1024.0**3), 1)
 
     def get_disk_used_space(self, path: str | Path) -> float:
         """Return used space (GiB) on disk for path.
 
         Must be run in executor.
         """
-        _, used, _ = shutil.disk_usage(path)
+        _, used, _ = self.disk_usage(path)
         return round(used / (1024.0**3), 1)
 
     def get_disk_free_space(self, path: str | Path) -> float:
         """Return free space (GiB) on disk for path.
 
         Must be run in executor.
         """
-        _, _, free = shutil.disk_usage(path)
+        _, _, free = self.disk_usage(path)
         return round(free / (1024.0**3), 1)
 
+    def disk_usage(self, path: str | Path) -> tuple[int, int, int]:
+        """Return (total, used, free) in bytes for path.
+
+        Must be run in executor.
+        """
+        return shutil.disk_usage(path)
+
+    def get_dir_structure_sizes(self, path: Path, max_depth: int = 1) -> dict[str, Any]:
+        """Return a recursive dict of subdirectories and their sizes, only if size > 0.
+
+        Excludes external mounts and symlinks to avoid counting files on other filesystems
+        or following symlinks that could lead to infinite loops or incorrect sizes.
+        """
+
+        size = 0
+        if not path.exists():
+            return {"used_bytes": size}
+
+        children: list[dict[str, Any]] = []
+        root_device = path.stat().st_dev
+
+        for child in path.iterdir():
+            if not child.is_dir():
+                size += child.stat(follow_symlinks=False).st_size
+                continue
+
+            # Skip symlinks to avoid infinite loops
+            if child.is_symlink():
+                continue
+
+            try:
+                # Skip if not on same device (external mount)
+                if child.stat().st_dev != root_device:
+                    continue
+            except OSError as err:
+                if err.errno == errno.EBADMSG:
+                    self.sys_resolution.add_unhealthy_reason(
+                        UnhealthyReason.OSERROR_BAD_MESSAGE
+                    )
+                    break
+                continue
+
+            child_result = self.get_dir_structure_sizes(child, max_depth - 1)
+            if child_result["used_bytes"] > 0:
+                size += child_result["used_bytes"]
+                if max_depth > 1:
+                    children.append(
+                        {
+                            "id": child.name,
+                            "label": child.name,
+                            **child_result,
+                        }
+                    )
+
+        if children:
+            return {"used_bytes": size, "children": children}
+
+        return {"used_bytes": size}
+
+    def get_dir_sizes(
+        self, request: dict[str, Path], max_depth: int = 1
+    ) -> list[dict[str, Any]]:
+        """Accept a dictionary of `name: Path` and return a dictionary with `name: <size>`.
+
+        Must be run in executor.
+        """
+        return [
+            {
+                "id": name,
+                "label": name,
+                **self.get_dir_structure_sizes(path, max_depth),
+            }
+            for name, path in request.items()
+        ]
+
     def _get_mountinfo(self, path: str) -> list[str] | None:
         mountinfo = _MOUNTINFO.read_text(encoding="utf-8")
         for line in mountinfo.splitlines():