g3pylib.system.battery

 1import asyncio
 2from enum import Enum
 3from typing import Awaitable, Tuple, cast
 4
 5from g3pylib._utils import APIComponent, EndpointKind
 6from g3pylib.g3typing import URI, SignalBody
 7from g3pylib.websocket import G3WebSocketClientProtocol
 8
 9
10class BatteryState(Enum):
11    """Defines battery levels."""
12
13    FULL = "full"
14    GOOD = "good"
15    LOW = "low"
16    VERY_LOW = "verylow"
17    UNKNOWN = "unknown"
18
19
20class Battery(APIComponent):
21    def __init__(self, connection: G3WebSocketClientProtocol, api_uri: URI) -> None:
22        self._connection = connection
23        super().__init__(api_uri)
24
25    async def get_charging(self) -> bool:
26        return cast(
27            bool,
28            await self._connection.require_get(
29                self.generate_endpoint_uri(EndpointKind.PROPERTY, "charging")
30            ),
31        )
32
33    async def get_level(self) -> float:
34        return cast(
35            float,
36            await self._connection.require_get(
37                self.generate_endpoint_uri(EndpointKind.PROPERTY, "level")
38            ),
39        )
40
41    async def get_name(self) -> str:
42        return cast(
43            str,
44            await self._connection.require_get(
45                self.generate_endpoint_uri(EndpointKind.PROPERTY, "name")
46            ),
47        )
48
49    async def get_remaining_time(self) -> int:
50        return cast(
51            int,
52            await self._connection.require_get(
53                self.generate_endpoint_uri(EndpointKind.PROPERTY, "remaining-time")
54            ),
55        )
56
57    async def get_state(self) -> BatteryState:
58        return BatteryState(
59            await self._connection.require_get(
60                self.generate_endpoint_uri(EndpointKind.PROPERTY, "state")
61            ),
62        )
63
64    async def subscribe_to_state_changed(
65        self,
66    ) -> Tuple[asyncio.Queue[SignalBody], Awaitable[None]]:
67        return await self._connection.subscribe_to_signal(
68            self.generate_endpoint_uri(EndpointKind.SIGNAL, "state-changed")
69        )