File: automation.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 (233 lines) | stat: -rw-r--r-- 7,896 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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
"""Scene class managing all scenes."""

from aiopvapi.helpers.aiorequest import AioRequest, PvApiMaintenance
from aiopvapi.helpers.api_base import ApiResource
from aiopvapi.helpers.tools import get_base_path, join_path
from aiopvapi.helpers.constants import (
    ATTR_SCENE_ID,
    ATTR_SCHEDULED_EVENT,
    ATTR_ID,
    FUNCTION_SCHEDULE,
)
from aiopvapi.resources.scene import Scene

import logging

_LOGGER = logging.getLogger(__name__)


class Automation(ApiResource):
    """Powerview Automation class."""

    def __init__(self, raw_data: dict, request: AioRequest) -> None:
        self.api_endpoint = "scheduledevents"
        if request.api_version >= 3:
            self.api_endpoint = "automations"
        super().__init__(request, self.api_endpoint, raw_data)
        self._name = None
        self._room_id = None
        self._scene: Scene = None

    def is_supported(self, function: str) -> bool:
        """Return if api supports this function."""
        if self.api_version >= 3:
            return False
        else:
            if function in FUNCTION_SCHEDULE:
                return True
        return False

    @property
    def enabled(self) -> bool:
        """Return the automation state."""
        return self._raw_data.get("enabled")

    @property
    def id(self) -> int:
        return self._raw_data.get(ATTR_ID)

    @property
    def name(self) -> str:
        if self._name is not None:
            return self._name
        return self._raw_data.get(ATTR_SCENE_ID)

    @property
    def scene_id(self) -> str:
        """Return the scene id of the automation."""
        return self._raw_data.get(ATTR_SCENE_ID)

    @property
    def room_id(self) -> int | None:
        """Return the room id of the automation."""
        return self._room_id

    def convert_to_12_hour(self, hour: int):
        """Convert 24 hour time to 12 hour"""
        if hour < 0 or hour > 24:
            _LOGGER.error("%s is not a valid 24 hour time", hour)
            return 0
        if hour == 0:
            return 12
        if 0 < hour <= 12:
            return hour
        return hour - 12

    def format_time(self, hour: int, minute: int):
        """Convert hour and minute to friendly text format"""
        meridiem = "AM" if hour < 12 else "PM"
        hour = hour % 12 if hour % 12 != 0 else 12

        if minute >= 60:
            hour += minute // 60
            minute %= 60

        hour = self.convert_to_12_hour(hour)
        return f"{hour}:{str(abs(minute)).zfill(2)} {meridiem}"

    def get_execution_time(self):
        """Return a friendly string, in the same format the hub
        does incicating when the time the schedule will execute.
        """
        # {'id': 437, 'type': 14, 'enabled': False, 'days': 127, 'hour': 1, 'min': 0, 'bleId': 2, 'sceneId': 220, 'errorShd_Ids': []}
        # {'enabled': True, 'sceneId': 14067, 'daySunday': False, 'dayMonday': True, 'dayTuesday': True, 'dayWednesday': True, 'dayThursday': True, 'dayFriday': True, 'daySaturday': False, 'eventType': 0, 'hour': 7, 'minute': 0, 'id': 38971}

        if self.api_version >= 3:
            # 2 = Before sunrise, 10 = After sunrise
            # 6 = Before sunset, 14 = After sunset
            sunrise = [2, 10]
            valid_events = [2, 6, 10, 14]
        else:
            # - Sunrise = 1, Sunset = 2
            # before and after are caluclated by hour/minute
            sunrise = [1]
            valid_events = [1, 2]

        attr_type = "type" if self.api_version >= 3 else "eventType"
        attr_hour = "hour"
        attr_minute = "min" if self.api_version >= 3 else "minute"

        event_type = self.raw_data.get(attr_type)
        hour = self.raw_data.get(attr_hour)
        minute = self.raw_data.get(attr_minute)

        # event type 0 represents clock based for all generations
        if event_type == 0:
            return self.format_time(hour, minute)

        if event_type in valid_events:
            when = "Sunrise" if event_type in sunrise else "Sunset"

            if hour == 0 and minute == 0:
                return f"At {when}"

            if self.api_version >= 3:
                before_after = "Before" if event_type in [2, 6] else "After"
            else:
                before_after = "Before" if minute < 0 else "After"
                hour = abs(minute) // 60
                minute = abs(minute) % 60
                # hour = floor(abs(minute) / 60)
                # minute = abs(minute) - (hour * 60)
            return f"{hour}h {minute}m {before_after} {when}"
            # return f"{abs(hour)}h {abs(minute)}m {before_after} {when}"

        return f"Unknown Event {event_type}"

    def get_execution_days(self):
        """Return a friendly string, in the same format the hub
        does incicating when the days the schedule will execute.
        """

        if self.api_version >= 3:
            day_mapping = {
                0x40: "Sun",
                0x01: "Mon",
                0x02: "Tue",
                0x04: "Wed",
                0x08: "Thu",
                0x10: "Fri",
                0x20: "Sat",
            }

            enabled_days = [
                day for bit, day in day_mapping.items() if self.raw_data["days"] & bit
            ]

        else:
            day_mapping = {
                "daySunday": "Sun",
                "dayMonday": "Mon",
                "dayTuesday": "Tue",
                "dayWednesday": "Wed",
                "dayThursday": "Thu",
                "dayFriday": "Fri",
                "daySaturday": "Sat",
            }

            enabled_days = [
                day_mapping[key]
                for key, value in self.raw_data.items()
                if key in day_mapping and value
            ]

        if len(enabled_days) == len(day_mapping):
            output = "Every Day"
        elif set(enabled_days) == {"Sat", "Sun"}:
            output = "Weekends"
        elif set(enabled_days) == {"Mon", "Tue", "Wed", "Thu", "Fri"}:
            output = "Weekdays"
        else:
            output = ", ".join(enabled_days)
        return output

    @property
    def details(self) -> dict[str, str]:
        """Return the specifics of the automation."""
        details = {
            "ID": self.id,
            "Time": self.get_execution_time(),
            "Days": self.get_execution_days(),
        }

        _LOGGER.debug(
            "Automation: %s (Enabled: %s), %s, %s",
            self.name,
            self.enabled,
            details.get("Time"),
            details.get("Days"),
        )
        return details

    async def fetch_associated_scene_data(self) -> None:
        """Update the automation with friendly scene info."""
        scene_url = join_path(
            get_base_path(self.request.hub_ip, self.api_path),
            "scenes",
            str(self.scene_id),
        )
        self._scene: Scene = Scene(
            await self.request.get(scene_url),
            self.request,
        )
        self._name = self._scene.name
        self._room_id = self._scene.room_id

    async def set_state(self, state: bool) -> None:
        """Update the automation enabled status."""
        resource_path = join_path(self.base_path, str(self.id))
        data = self.raw_data
        data["enabled"] = state
        if self.api_version <= 2:
            data = {"scheduledEvent": data}
        await self.request.put(resource_path, data)

    async def refresh(self):
        """Query the hub and for updated automation information."""
        try:
            raw_data = await self.request.get(self._resource_path)
            # Gen <= 2 API has raw data under shade key.  Gen >= 3 API this is flattened.
            self._raw_data = raw_data.get(ATTR_SCHEDULED_EVENT, raw_data)
        except PvApiMaintenance:
            _LOGGER.debug("Hub undergoing maintenance. Please try again")
        return