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
|
from __future__ import annotations
import datetime
import random
import ssl
from functools import cache
from typing import TYPE_CHECKING, Union
from .activity import (
ACTION_BRIDGE_OFFLINE,
ACTION_BRIDGE_ONLINE,
ACTIVITY_ACTION_STATES,
ACTIVITY_MOVING_STATES,
MOVING_STATES,
BridgeOperationActivity,
DoorbellImageCaptureActivity,
DoorbellMotionActivity,
DoorOperationActivity,
LockOperationActivity,
)
from .const import CONFIGURATION_URLS, Brand
from .lock import LockDetail
LockActivityTypes = Union[
LockOperationActivity, DoorOperationActivity, BridgeOperationActivity
]
DoorbellActivityTypes = Union[
DoorbellImageCaptureActivity, DoorbellMotionActivity, BridgeOperationActivity
]
if TYPE_CHECKING:
from .doorbell import DoorbellDetail
def get_latest_activity(
activity1: LockActivityTypes | None, activity2: LockActivityTypes | None
) -> LockActivityTypes | None:
"""Return the latest activity."""
return (
activity2
if (
not activity1
or (
activity2
and activity2.action not in ACTIVITY_MOVING_STATES
and activity1.activity_start_time <= activity2.activity_start_time
)
)
else activity1
)
def update_lock_detail_from_activity(
lock_detail: LockDetail,
activity: LockActivityTypes,
) -> bool:
"""Update the LockDetail from an activity."""
activity_end_time_utc = as_utc_from_local(activity.activity_end_time)
if activity.device_id != lock_detail.device_id:
raise ValueError
if isinstance(activity, LockOperationActivity):
if lock_detail.lock_status_datetime and (
lock_detail.lock_status_datetime > activity_end_time_utc
or (
lock_detail.lock_status_datetime == activity_end_time_utc
and lock_detail.lock_status not in MOVING_STATES
)
):
return False
lock_detail.lock_status = ACTIVITY_ACTION_STATES[activity.action]
lock_detail.lock_status_datetime = activity_end_time_utc
elif isinstance(activity, DoorOperationActivity):
if (
lock_detail.door_state_datetime
and lock_detail.door_state_datetime >= activity_end_time_utc
):
return False
lock_detail.door_state = ACTIVITY_ACTION_STATES[activity.action]
lock_detail.door_state_datetime = activity_end_time_utc
elif isinstance(activity, BridgeOperationActivity):
if activity.action == ACTION_BRIDGE_ONLINE:
lock_detail.set_online(True)
elif activity.action == ACTION_BRIDGE_OFFLINE:
lock_detail.set_online(False)
else:
raise ValueError
return True
def update_doorbell_image_from_activity(
doorbell_detail: DoorbellDetail, activity: DoorbellActivityTypes
) -> bool:
"""Update the DoorDetail from an activity with a new image."""
if activity.device_id != doorbell_detail.device_id:
raise ValueError
if isinstance(activity, (DoorbellImageCaptureActivity, DoorbellMotionActivity)):
if activity.image_created_at_datetime is None:
return False
if (
doorbell_detail.image_created_at_datetime is None
or doorbell_detail.image_created_at_datetime
< activity.image_created_at_datetime
):
doorbell_detail.image_url = activity.image_url
doorbell_detail.image_created_at_datetime = (
activity.image_created_at_datetime
)
doorbell_detail.content_token = (
activity.content_token or doorbell_detail.content_token
)
else:
return False
else:
raise ValueError
return True
def as_utc_from_local(dtime: datetime.datetime) -> datetime.datetime:
"""Converts the datetime returned from an activity to UTC."""
return dtime.astimezone(tz=datetime.timezone.utc)
def get_configuration_url(brand: Brand) -> str:
"""Return the configuration URL for the brand."""
return CONFIGURATION_URLS[brand]
@cache
def get_ssl_context() -> ssl.SSLContext:
"""Return an SSL context for cloudflare."""
context = ssl.create_default_context()
ciphers = [cipher["name"] for cipher in context.get_ciphers()]
default_ciphers = ciphers[:3]
backup_ciphers = ciphers[3:]
random.shuffle(backup_ciphers)
context.set_ciphers(":".join((*default_ciphers, *backup_ciphers)))
return context
|
