File: scene_members.py

package info (click to toggle)
python-aiopvapi 3.1.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 684 kB
  • sloc: python: 3,123; xml: 850; makefile: 5
file content (84 lines) | stat: -rw-r--r-- 2,561 bytes parent folder | download
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
"""Scenes class managing all scene data."""

import logging

from aiopvapi.helpers.api_base import ApiEntryPoint
from aiopvapi.helpers.aiorequest import AioRequest
from aiopvapi.helpers.constants import (
    ATTR_SCENE_ID,
    ATTR_SHADE_ID,
    ATTR_POSITIONS,
    SCENE_MEMBER_DATA,
)
from aiopvapi.resources.scene_member import ATTR_SCENE_MEMBER, SceneMember

from aiopvapi.resources.model import PowerviewData

_LOGGER = logging.getLogger("__name__")


class SceneMembers(ApiEntryPoint):
    """A scene member is a device, like a shade, being a member
    of a specific scene."""

    api_endpoint = "scenemembers"

    def __init__(self, request: AioRequest) -> None:
        super().__init__(request, self.api_endpoint)

    async def create_scene_member(self, shade_position, scene_id, shade_id):
        """Adds a shade to an existing scene"""

        data = {
            ATTR_SCENE_MEMBER: {
                ATTR_POSITIONS: shade_position,
                ATTR_SCENE_ID: scene_id,
                ATTR_SHADE_ID: shade_id,
            }
        }
        return await self.request.post(self.base_path, data=data)

    def _resource_factory(self, raw):
        return SceneMember(raw, self.request)

    def _loop_raw(self, raw):
        if self.api_version < 3:
            raw = raw[SCENE_MEMBER_DATA]

        for _raw in raw:
            yield _raw

    def _get_to_actual_data(self, raw):
        if self.api_version >= 3:
            return raw
        return raw.get(SCENE_MEMBER_DATA)

    async def get_scene_members_old(self, scene_id):
        """Return all scene members for a particular Scene ID."""

        return await self.get_instances(sceneId=scene_id)

    async def delete_shade_from_scene(self, shade_id, scene_id):
        """Delete a shade from a scene."""
        return await self.request.delete(
            self.base_path,
            params={ATTR_SCENE_ID: scene_id, ATTR_SHADE_ID: shade_id},
        )

    async def get_scene_members(self, **kwargs) -> PowerviewData:
        """Get a list of scene members.

        :raises PvApiError when an error occurs.
        """
        resources = await self.get_resources(**kwargs)
        if self.api_version < 3:
            resources = resources[SCENE_MEMBER_DATA]

        _LOGGER.debug("Raw scene_member data: %s", resources)
        
        # return array of scenes attached to a shade
        processed = {
            entry["shadeId"]: SceneMember(entry, self.request) for entry in resources
        }

        return PowerviewData(raw=resources, processed=processed)