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
60
61
62
63
64
65
66
67
68
69
70
71
72
|
"""Room class managing all room data."""
import logging
from aiopvapi.helpers.api_base import ApiEntryPoint
from aiopvapi.helpers.aiorequest import AioRequest
from aiopvapi.helpers.constants import (
ATTR_ID,
ATTR_NAME,
ATTR_COLOR_ID,
ATTR_ICON_ID,
ATTR_ROOM,
ATTR_ROOM_DATA,
)
from aiopvapi.helpers.tools import unicode_to_base64
from aiopvapi.resources.room import Room
from aiopvapi.resources.model import PowerviewData
_LOGGER = logging.getLogger(__name__)
class Rooms(ApiEntryPoint):
"""Rooms entry point"""
api_endpoint = "rooms"
def __init__(self, request: AioRequest) -> None:
super().__init__(request, self.api_endpoint)
async def create_room(self, name, color_id=0, icon_id=0):
"""Create a room on the hub"""
name = unicode_to_base64(name)
data = {
ATTR_ROOM: {
ATTR_NAME: name,
ATTR_COLOR_ID: color_id,
ATTR_ICON_ID: icon_id,
}
}
return await self.request.post(self.base_path, data=data)
def _resource_factory(self, raw):
return Room(raw, self.request)
def _loop_raw(self, raw):
if self.api_version < 3:
raw = raw[ATTR_ROOM_DATA]
for _raw in raw:
yield _raw
def _get_to_actual_data(self, raw):
if self.api_version >= 3:
return raw
return raw.get("room")
async def get_rooms(self, **kwargs) -> PowerviewData:
"""Get a list of rooms.
:returns PowerviewData object
:raises PvApiError when an error occurs.
"""
resources = await self.get_resources(**kwargs)
if self.api_version < 3:
resources = resources[ATTR_ROOM_DATA]
_LOGGER.debug("Raw rooms data: %s", resources)
processed = {entry[ATTR_ID]: Room(entry, self.request) for entry in resources}
return PowerviewData(raw=resources, processed=processed)
|
