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
|
from datetime import datetime, timedelta
from functools import wraps
from typing import Callable
from PyViCare import Feature
VICARE_DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
# This decorator handles access to underlying JSON properties.
# If the property is not found (KeyError) or the index does not
# exists (IndexError), the requested feature is not supported by
# the device.
def isSupported(method: Callable) -> bool:
try:
result = method()
return bool(result != 'error')
except PyViCareNotSupportedFeatureError:
return False
def time_as_delta(date_time: datetime) -> timedelta:
return timedelta(
hours=date_time.hour,
minutes=date_time.minute,
seconds=date_time.second
)
def parse_time_as_delta(time_string: str) -> timedelta:
return timedelta(
hours=int(time_string[0:2]),
minutes=int(time_string[3:5])
)
class ViCareTimer:
# class is used to replace logic in unittest
def now(self) -> datetime:
return datetime.now()
def handleNotSupported(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except (KeyError, IndexError):
raise PyViCareNotSupportedFeatureError(func.__name__)
# You can remove that wrapper after the feature flag gets removed entirely.
def feature_flag_wrapper(*args, **kwargs):
try:
return wrapper(*args, **kwargs)
except PyViCareNotSupportedFeatureError:
if Feature.raise_exception_on_not_supported_device_feature:
raise
return "error"
return feature_flag_wrapper
def handleAPICommandErrors(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except (KeyError, IndexError):
raise PyViCareCommandError(func.__name__)
# You can remove that wrapper after the feature flag gets removed entirely.
def feature_flag_wrapper(*args, **kwargs):
try:
return wrapper(*args, **kwargs)
except PyViCareCommandError:
if Feature.raise_exception_on_command_failure:
raise
return "error"
return feature_flag_wrapper
class PyViCareNotSupportedFeatureError(Exception):
pass
class PyViCareInvalidConfigurationError(Exception):
def __init__(self, response):
error = response['error']
error_description = response['error_description']
msg = f'Invalid credentials. Error: {error}. Description: {error_description}. Please check your configuration: clientid and redirect uri.'
super().__init__(self, msg)
self.message = msg
class PyViCareInvalidCredentialsError(Exception):
pass
class PyViCareBrowserOAuthTimeoutReachedError(Exception):
pass
class PyViCareInvalidDataError(Exception):
pass
class PyViCareRateLimitError(Exception):
def __init__(self, response):
extended_payload = response["extendedPayload"]
name = extended_payload["name"]
requestCountLimit = extended_payload["requestCountLimit"]
limitReset = extended_payload["limitReset"]
limitResetDate = datetime.utcfromtimestamp(limitReset / 1000)
msg = f'API rate limit {name} exceeded. Max {requestCountLimit} calls in timewindow. Limit reset at {limitResetDate.isoformat()}.'
super().__init__(self, msg)
self.message = msg
self.limitResetDate = limitResetDate
class PyViCareInternalServerError(Exception):
def __init__(self, response):
statusCode = response["statusCode"]
message = response["message"]
viErrorId = response["viErrorId"]
msg = f'Request failed with status code {statusCode} and message "{message}". ViCare ErrorId: {viErrorId}'
super().__init__(self, msg)
self.message = msg
class PyViCareCommandError(Exception):
def __init__(self, response):
if isinstance(response, str):
msg = f'Command failed with message "{response}"'
else:
statusCode = response["statusCode"]
extended_payload = response["extendedPayload"]
if "reason" in extended_payload:
reason = extended_payload["reason"]
else:
reason = "Unknown"
msg = f'Command failed with status code {statusCode}. Reason given was: {reason}'
super().__init__(self, msg)
self.message = msg
|
