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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
"""Scenes class managing all scene data."""
import logging
from aiopvapi.helpers.aiorequest import AioRequest
from aiopvapi.helpers.api_base import ApiEntryPoint
from aiopvapi.helpers.constants import (
ATTR_ID,
ATTR_NAME,
ATTR_NAME_UNICODE,
ATTR_SHADE_DATA,
)
from aiopvapi.helpers.tools import base64_to_unicode
from aiopvapi.resources import shade
from aiopvapi.resources.model import PowerviewData
_LOGGER = logging.getLogger(__name__)
class Shades(ApiEntryPoint):
"""Shades entry point"""
api_endpoint = "shades"
def __init__(self, request: AioRequest) -> None:
super().__init__(request, self.api_endpoint)
def _sanitize_resources(self, resources: dict) -> dict | None:
"""Cleans up incoming shade data
:param resources: The dict with shade data to be sanitized.
:returns: Cleaned up shade dict.
"""
if self.api_version < 3:
resources = resources[ATTR_SHADE_DATA]
try:
for _shade in resources:
_name = _shade.get(ATTR_NAME)
if _name:
_shade[ATTR_NAME_UNICODE] = base64_to_unicode(_name)
return resources
except (KeyError, TypeError):
_LOGGER.debug("No shade data available")
return None
def _resource_factory(self, raw):
return shade.factory(raw, self.request)
def _loop_raw(self, raw):
if self.api_version < 3:
raw = raw[ATTR_SHADE_DATA]
for _raw in raw:
yield _raw
def _get_to_actual_data(self, raw):
if self.api_version >= 3:
return raw
return raw.get("shade")
async def get_shades(self, **kwargs) -> PowerviewData:
"""Get a list of shades.
:returns PowerviewData object
:raises PvApiError when an error occurs.
"""
resources = await self.get_resources(**kwargs)
if self.api_version < 3:
resources = resources[ATTR_SHADE_DATA]
_LOGGER.debug("Raw shades data: %s", resources)
processed = {
entry[ATTR_ID]: shade.factory(entry, self.request) for entry in resources
}
return PowerviewData(raw=resources, processed=processed)
# async def get_shade(self, shade_id: int):
# _url = '{}/{}'.format(self.api_path, shade_id)
# _raw = await self.request.get(_url)
# return shade.factory(_raw, self.request)
|
