-
Notifications
You must be signed in to change notification settings - Fork 43
/
async_connection.py
184 lines (148 loc) · 6.13 KB
/
async_connection.py
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
"""
SamsungTVWS - Samsung Smart TV WS API wrapper
Copyright (C) 2019 Xchwarze
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.
This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor,
Boston, MA 02110-1335 USA
"""
import asyncio
import contextlib
import json
import logging
import ssl
from types import TracebackType
from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Union
from websockets.client import WebSocketClientProtocol, connect
from websockets.exceptions import ConnectionClosed
from . import connection, exceptions, helper
from .command import SamsungTVCommand, SamsungTVSleepCommand
from .event import (
IGNORE_EVENTS_AT_STARTUP,
MS_CHANNEL_CONNECT_EVENT,
MS_CHANNEL_UNAUTHORIZED,
)
_LOGGING = logging.getLogger(__name__)
class SamsungTVWSAsyncConnection(connection.SamsungTVWSBaseConnection):
connection: Optional[WebSocketClientProtocol]
_recv_loop: Optional["asyncio.Task[None]"]
async def __aenter__(self) -> "SamsungTVWSAsyncConnection":
return self
async def __aexit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> None:
await self.close()
async def open(self) -> WebSocketClientProtocol:
if self.connection:
# someone else already created a new connection
return self.connection
url = self._format_websocket_url(self.endpoint)
_LOGGING.debug("WS url %s", url)
connect_kwargs: Dict[str, Any] = {}
if self._is_ssl_connection():
ssl_context = ssl.SSLContext()
ssl_context.verify_mode = ssl.CERT_NONE
connect_kwargs["ssl"] = ssl_context
connection = await connect(url, open_timeout=self.timeout, **connect_kwargs)
event: Optional[str] = None
while event is None or event in IGNORE_EVENTS_AT_STARTUP:
data = await connection.recv()
response = helper.process_api_response(data)
event = response.get("event", "*")
assert event
self._websocket_event(event, response)
if event == MS_CHANNEL_UNAUTHORIZED:
await self.close()
raise exceptions.UnauthorizedError(response)
if event != MS_CHANNEL_CONNECT_EVENT:
# Unexpected event received during connection routine
await self.close()
raise exceptions.ConnectionFailure(response)
self._check_for_token(response)
self.connection = connection
return connection
async def start_listening(
self, callback: Optional[Callable[[str, Any], Optional[Awaitable[None]]]] = None
) -> None:
"""Open, and start listening."""
if self.connection:
raise exceptions.ConnectionFailure("Connection already exists")
self.connection = await self.open()
self._recv_loop = asyncio.ensure_future(
self._do_start_listening(callback, self.connection)
)
async def _do_start_listening(
self,
callback: Optional[Callable[[str, Any], Optional[Awaitable[None]]]],
connection: WebSocketClientProtocol,
) -> None:
"""Do start listening."""
with contextlib.suppress(ConnectionClosed):
while True:
data = await connection.recv()
response = helper.process_api_response(data)
event = response.get("event", "*")
self._websocket_event(event, response)
if callback:
awaitable = callback(event, response)
if awaitable:
await awaitable
async def close(self) -> None:
if self.connection:
await self.connection.close()
if self._recv_loop:
await self._recv_loop
self.connection = None
_LOGGING.debug("Connection closed.")
async def send_commands(
self,
commands: Sequence[Union[SamsungTVCommand, Dict[str, Any]]],
key_press_delay: Optional[float] = None,
) -> None:
if self.connection is None:
self.connection = await self.open()
delay = self.key_press_delay if key_press_delay is None else key_press_delay
for command in commands:
await self._send_command(self.connection, command, delay)
async def send_command(
self,
command: Union[List[SamsungTVCommand], SamsungTVCommand, Dict[str, Any]],
key_press_delay: Optional[float] = None,
) -> None:
if isinstance(command, list):
_LOGGING.warn(
"Using send_command to send multiple commands is deprecated, "
"please use send_commands."
)
await self.send_commands(command, key_press_delay)
return
await self.send_commands([command], key_press_delay)
@staticmethod
async def _send_command(
connection: WebSocketClientProtocol,
command: Union[SamsungTVCommand, Dict[str, Any]],
delay: float,
) -> None:
if isinstance(command, SamsungTVSleepCommand):
await asyncio.sleep(command.delay)
return
if isinstance(command, SamsungTVCommand):
payload = command.get_payload()
else:
payload = json.dumps(command)
_LOGGING.debug("SamsungTVWS websocket command: %s", payload)
await connection.send(payload)
await asyncio.sleep(delay)
def is_alive(self) -> bool:
return self.connection is not None and not self.connection.closed