1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
"""Host client for supervisor."""
from .client import _SupervisorComponentClient
from .const import TIMEOUT_60_SECONDS
from .models.host import (
DiskUsage,
HostInfo,
HostOptions,
RebootOptions,
Service,
ServiceList,
ShutdownOptions,
)
class HostClient(_SupervisorComponentClient):
"""Handles host access in supervisor."""
async def info(self) -> HostInfo:
"""Get host info."""
result = await self._client.get("host/info")
return HostInfo.from_dict(result.data)
async def reboot(self, options: RebootOptions | None = None) -> None:
"""Reboot host."""
await self._client.post(
"host/reboot",
json=options.to_dict() if options else None,
timeout=TIMEOUT_60_SECONDS,
)
async def shutdown(self, options: ShutdownOptions | None = None) -> None:
"""Shutdown host."""
await self._client.post(
"host/shutdown", json=options.to_dict() if options else None
)
async def reload(self) -> None:
"""Reload host info cache."""
await self._client.post("host/reload")
async def set_options(self, options: HostOptions) -> None:
"""Set host options."""
await self._client.post("host/options", json=options.to_dict())
async def services(self) -> list[Service]:
"""Get list of available services on host."""
result = await self._client.get("host/services")
return ServiceList.from_dict(result.data).services
async def get_disk_usage(self, max_depth: int = 1) -> DiskUsage:
"""Get disk usage."""
result = await self._client.get(
"host/disks/default/usage",
params={"max_depth": str(max_depth)},
)
return DiskUsage.from_dict(result.data)
# Omitted for now - Log endpoints
|
