/
/
/
1"""
2Cloud connection manager for Yandex Smart Home via yaha-cloud.ru relay.
3
4Manages a persistent WebSocket connection to the yaha-cloud.ru relay service.
5Incoming Yandex Smart Home API requests are received over WS, processed by
6the on_request callback, and the response is sent back over WS.
7
8Adapted from dext0r/yandex_smart_home cloud.py, stripped of HA dependencies.
9"""
10
11from __future__ import annotations
12
13import asyncio
14import json
15import logging
16from collections.abc import Awaitable, Callable
17from typing import TYPE_CHECKING, Any
18
19import aiohttp
20
21if TYPE_CHECKING:
22 from ya_dialogs_api import SecretStr
23
24from .constants import (
25 CLOUD_BASE_URL,
26 CLOUD_HEARTBEAT_INTERVAL,
27 CLOUD_RECONNECT_MAX,
28 CLOUD_RECONNECT_MIN,
29 CLOUD_REGISTER_URL,
30 CLOUD_WS_URL,
31)
32from .schema import CloudRequest
33
34_LOGGER = logging.getLogger(__name__)
35
36
37class CloudManager:
38 """Manages WebSocket connection to yaha-cloud.ru for Smart Home API relay."""
39
40 def __init__(
41 self,
42 session: aiohttp.ClientSession,
43 connection_token: SecretStr,
44 on_request: Callable[[CloudRequest], Awaitable[dict[str, Any]]],
45 logger: logging.Logger | None = None,
46 ) -> None:
47 """Initialize cloud relay manager."""
48 self._session = session
49 self._token = connection_token
50 self._on_request = on_request
51 self._logger = logger or _LOGGER
52 self._ws: aiohttp.ClientWebSocketResponse | None = None
53 self._running = False
54 self._reconnect_delay = CLOUD_RECONNECT_MIN
55
56 @property
57 def connected(self) -> bool:
58 """Return True if WebSocket is connected."""
59 return self._ws is not None and not self._ws.closed
60
61 async def connect(self) -> None:
62 """Start the WebSocket connection loop (runs until disconnect is called)."""
63 self._running = True
64 while self._running:
65 try:
66 await self._connect_once()
67 except asyncio.CancelledError:
68 break
69 except Exception:
70 if not self._running:
71 break # type: ignore[unreachable]
72 self._logger.exception(
73 "Cloud connection error, reconnecting in %ds", self._reconnect_delay
74 )
75 if not self._running:
76 break # type: ignore[unreachable]
77 # Backoff before reconnect (both after errors and clean disconnects)
78 await asyncio.sleep(self._reconnect_delay)
79 self._reconnect_delay = min(self._reconnect_delay * 2, CLOUD_RECONNECT_MAX)
80
81 async def disconnect(self) -> None:
82 """Stop the connection loop and close WebSocket."""
83 self._running = False
84 if self._ws and not self._ws.closed:
85 await self._ws.close()
86 self._ws = None
87 self._logger.info("Cloud relay disconnected")
88
89 async def _connect_once(self) -> None:
90 """Single WebSocket connection attempt + message loop."""
91 headers = {"Authorization": f"Bearer {self._token.get_secret()}"}
92 async with self._session.ws_connect(
93 CLOUD_WS_URL,
94 headers=headers,
95 heartbeat=CLOUD_HEARTBEAT_INTERVAL,
96 ) as ws:
97 self._ws = ws
98 self._reconnect_delay = CLOUD_RECONNECT_MIN
99 self._logger.info("Connected to cloud relay at %s", CLOUD_WS_URL)
100
101 async for msg in ws:
102 if not self._running:
103 break
104
105 if msg.type == aiohttp.WSMsgType.TEXT:
106 try:
107 data = json.loads(msg.data)
108 except json.JSONDecodeError:
109 self._logger.warning("Received invalid JSON from cloud relay: %r", msg.data)
110 continue
111 await self._handle_message(ws, data)
112 elif msg.type == aiohttp.WSMsgType.ERROR:
113 self._logger.error("WebSocket error: %s", ws.exception())
114 break
115 elif msg.type in (
116 aiohttp.WSMsgType.CLOSE,
117 aiohttp.WSMsgType.CLOSING,
118 aiohttp.WSMsgType.CLOSED,
119 ):
120 break
121
122 self._ws = None
123 self._logger.info("Cloud relay connection closed")
124
125 async def _handle_message(
126 self, ws: aiohttp.ClientWebSocketResponse, data: dict[str, Any]
127 ) -> None:
128 """Parse incoming WS message, call handler, and send response."""
129 try:
130 # message may be a JSON string or already parsed dict
131 raw_message = data.get("message")
132 if isinstance(raw_message, str) and raw_message:
133 raw_message = json.loads(raw_message)
134 request = CloudRequest(
135 request_id=data["request_id"],
136 action=data["action"],
137 message=raw_message if isinstance(raw_message, dict) else None,
138 )
139 self._logger.debug("Cloud request: action=%s", request.action)
140 response = await self._on_request(request)
141 await ws.send_json(response)
142 except Exception:
143 self._logger.exception("Error handling cloud message: %s", data)
144 # Send best-effort error response so the relay doesn't hang
145 request_id = data.get("request_id") if isinstance(data, dict) else None
146 if request_id and ws and not ws.closed:
147 try:
148 await ws.send_json(
149 {"request_id": request_id, "payload": {"error": "INTERNAL_ERROR"}}
150 )
151 except Exception:
152 self._logger.debug("Failed to send error response for %s", request_id)
153
154
155# ---------------------------------------------------------------------------
156# Cloud instance registration helpers
157# ---------------------------------------------------------------------------
158
159
160async def register_cloud_instance(
161 session: aiohttp.ClientSession,
162 platform: str | None = None,
163) -> dict[str, str]:
164 """
165 Register a new cloud instance on yaha-cloud.ru.
166
167 Returns dict with 'id', 'password', 'connection_token'.
168 No authentication is required â the relay auto-generates credentials.
169
170 For Cloud Plus mode, pass platform="yandex" so the relay can validate
171 the client_id during OAuth account linking.
172 """
173 kwargs: dict[str, Any] = {}
174 if platform:
175 kwargs["json"] = {"platform": platform}
176 async with session.post(CLOUD_REGISTER_URL, **kwargs) as resp:
177 resp.raise_for_status()
178 # yaha-cloud.ru may return text/plain content-type for JSON
179 data = await resp.json(content_type=None)
180 _LOGGER.info("Registered cloud instance: %s", data.get("id"))
181 return dict(data)
182
183
184async def get_cloud_otp(
185 session: aiohttp.ClientSession,
186 instance_id: str,
187 token: SecretStr,
188) -> str:
189 """
190 Get a one-time password for linking the instance in the Yandex app.
191
192 User enters this OTP in the Yandex Smart Home app to link their account.
193 The token parameter is the connection_token from registration.
194 """
195 url = f"{CLOUD_BASE_URL}/api/home_assistant/v1/instance/{instance_id}/otp"
196 headers = {"Authorization": f"Bearer {token.get_secret()}"}
197 async with session.post(url, headers=headers) as resp:
198 resp.raise_for_status()
199 # yaha-cloud.ru may return text/plain content-type for JSON
200 data = await resp.json(content_type=None)
201 return str(data["code"])
202