Implement 'get stream subscription'

pymammotion/http/dataclass/stream_subscription_response.py

@@ -0,0 +1,27 @@
+from dataclasses import dataclass
+from typing import List, Optional
+
+from mashumaro.config import BaseConfig
+from mashumaro.mixins.orjson import DataClassORJSONMixin
+
+@dataclass
+class Camera(DataClassORJSONMixin):
+    cameraId: int
+    token: str
+
+@dataclass
+class Data(DataClassORJSONMixin):
+    appid: str
+    cameras: List[Camera]
+    channelName: str
+    token: str
+    uid: int
+
+@dataclass
+class StreamSubscriptionResponse(DataClassORJSONMixin):
+    code: int
+    msg: str
+    data: Optional[Data] = None
+
+    class Config(BaseConfig):
+        omit_default = True

pymammotion/http/http.py

@@ -6,6 +6,7 @@
 from mashumaro import DataClassDictMixin
 from mashumaro.mixins.orjson import DataClassORJSONMixin
 
+from pymammotion.http.dataclass.stream_subscription_response import StreamSubscriptionResponse
 from pymammotion.aliyun.dataclass.connect_response import Device
 from pymammotion.const import (
     MAMMOTION_CLIENT_ID,
@@ -47,12 +48,14 @@ class LoginResponseData(DataClassORJSONMixin):
 
 
 class MammotionHTTP:
+    _headers = dict()
+
     def __init__(self, response: Response):
-        self._headers = dict()
         self.login_info = LoginResponseData.from_dict(response.data) if response.data else None
         self._headers["Authorization"] = f"Bearer {self.login_info.access_token}" if response.data else None
         self.msg = response.msg
         self.code = response.code
+
 
     @classmethod
     async def login(cls, session: ClientSession, username: str, password: str) -> Response[LoginResponseData]:
@@ -71,7 +74,31 @@ async def login(cls, session: ClientSession, username: str, password: str) -> Re
                 response = Response.from_dict(data)
                 # TODO catch errors from mismatch user / password elsewhere
                 # Assuming the data format matches the expected structure
+                cls._session = session
                 return response
+
+    @classmethod
+    async def get_stream_subscription(cls, iot_id: str) -> Response[StreamSubscriptionResponse]:
+        """Get agora.io data for view camera stream"""
+        print(cls._headers["Authorization"])
+        async with ClientSession('https://domestic.mammotion.com') as session:
+            async with session.post(
+                "/device-server/v1/stream/subscription",
+                json={
+                    "deviceId" : iot_id
+                },
+                headers={
+                    "Authorization": f"{cls._headers["Authorization"]}",
+                    "Content-Type": "application/json"  # Se necessario
+                }
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    response = StreamSubscriptionResponse.from_dict(data)
+                    # TODO catch errors from mismatch like token expire etc
+                    # Assuming the data format matches the expected structure
+                    return response
+
 
 
 async def connect_http(username: str, password: str) -> MammotionHTTP:

pymammotion/mammotion/devices/mammotion.py

@@ -13,7 +13,7 @@
 from collections import deque
 from enum import Enum
 from functools import cache
-from typing import Any, Callable, Optional, cast, Awaitable
+from typing import Any, Callable, Optional, cast, Awaitable, Tuple
 from uuid import UUID
 
 import betterproto
@@ -37,7 +37,7 @@
 from pymammotion.data.model.device import MowingDevice
 from pymammotion.data.mqtt.event import ThingEventMessage
 from pymammotion.data.state_manager import StateManager
-from pymammotion.http.http import connect_http
+from pymammotion.http.http import connect_http, MammotionHTTP
 from pymammotion.mammotion.commands.mammotion_command import MammotionCommand
 from pymammotion.mqtt import MammotionMQTT
 from pymammotion.mqtt.mammotion_future import MammotionFuture
@@ -200,9 +200,9 @@ async def create_devices(ble_device: BLEDevice,
     mammotion = Mammotion(ble_device, preference)
 
     if cloud_credentials:
-        cloud_client = await Mammotion.login(cloud_credentials.account_id or cloud_credentials.email,
+        cloud_client, mammotion_http = await Mammotion.login(cloud_credentials.account_id or cloud_credentials.email,
                                              cloud_credentials.password)
-        await mammotion.initiate_cloud_connection(cloud_client)
+        await mammotion.initiate_cloud_connection(cloud_client, mammotion_http)
 
     return mammotion
 
@@ -214,7 +214,7 @@ class Mammotion(object):
     devices = MammotionDevices()
     cloud_client: CloudIOTGateway | None = None
     mqtt: MammotionMQTT | None = None
-
+    mammotion_http_client: MammotionHTTP | None = None
 
 
     def __init__(
@@ -229,12 +229,13 @@ def __init__(
         if preference:
             self._preference = preference
 
-    async def initiate_cloud_connection(self, cloud_client: CloudIOTGateway) -> None:
+    async def initiate_cloud_connection(self, cloud_client: CloudIOTGateway, mammotion_http: MammotionHTTP) -> None:
         if self.mqtt is not None:
             if self.mqtt.is_connected:
                 return
 
         self.cloud_client = cloud_client
+        self.mammotion_http_client = mammotion_http
         self.mqtt = MammotionMQTT(region_id=cloud_client._region_response.data.regionId,
                                         product_key=cloud_client._aep_response.data.productKey,
                                         device_name=cloud_client._aep_response.data.deviceName,
@@ -257,7 +258,7 @@ def set_disconnect_strategy(self, disconnect: bool):
                 ble_device.set_disconnect_strategy(disconnect)
 
     @staticmethod
-    async def login(account: str, password: str) -> CloudIOTGateway:
+    async def login(account: str, password: str) -> Tuple[CloudIOTGateway, MammotionHTTP]:
         """Login to mammotion cloud."""
         cloud_client = CloudIOTGateway()
         async with ClientSession(MAMMOTION_DOMAIN) as session:
@@ -266,15 +267,14 @@ async def login(account: str, password: str) -> CloudIOTGateway:
             _LOGGER.debug("CountryCode: " + country_code)
             _LOGGER.debug("AuthCode: " + mammotion_http.login_info.authorization_code)
             loop = asyncio.get_running_loop()
-            cloud_client.set_http(mammotion_http)
             await loop.run_in_executor(None, cloud_client.get_region, country_code, mammotion_http.login_info.authorization_code)
             await cloud_client.connect()
             await cloud_client.login_by_oauth(country_code, mammotion_http.login_info.authorization_code)
             await loop.run_in_executor(None, cloud_client.aep_handle)
             await loop.run_in_executor(None, cloud_client.session_by_auth_code)
 
             await loop.run_in_executor(None, cloud_client.list_binding_by_account)
-            return cloud_client
+            return cloud_client, mammotion_http
 
 
     def get_device_by_name(self, name: str) -> MammotionMixedDeviceManager:
@@ -322,6 +322,13 @@ def mower(self, name: str):
         device = self.get_device_by_name(name)
         if device:
             return device.mower_state()
+
+    async def get_stream_subsctiption(self, name: str):
+        device = self.get_device_by_name(name)
+        if self._preference is ConnectionPreference.WIFI:
+            if self.mammotion_http_client is not None and device.has_cloud():
+                _stream_response = await self.mammotion_http_client.get_stream_subscription(device.cloud().iot_id)
+                _LOGGER.debug(_stream_response)
 
 def has_field(message: betterproto.Message) -> bool:
     """Check if the message has any fields serialized on wire."""
@@ -816,6 +823,7 @@ async def _execute_command_locked(self, key: str, command: bytes) -> bytes:
         except asyncio.TimeoutError:
             timeout_expired = True
             notify_msg = b''
+            self._notify_future.set_result(notify_msg)
         finally:
             if not timeout_expired:
                 timeout_handle.cancel()
@@ -1219,7 +1227,4 @@ async def _handle_mqtt_message(self, topic: str, payload: dict) -> None:
 
     def _disconnect(self):
         """Disconnect the MQTT client."""
-        self._mqtt_client.disconnect()
-
-
-
+        self._mqtt_client.disconnect()

tests/login_and_get_stream_token.py

@@ -0,0 +1,50 @@
+import asyncio
+import logging
+import os
+
+from aiohttp import ClientSession
+import traceback
+
+from pymammotion import MammotionHTTP
+from pymammotion.aliyun.cloud_gateway import CloudIOTGateway
+from pymammotion.const import MAMMOTION_DOMAIN
+from pymammotion.http.http import connect_http
+from pymammotion.mammotion.commands.mammotion_command import MammotionCommand
+from pymammotion.mqtt.mammotion_mqtt import MammotionMQTT, logger
+from pymammotion.mammotion.devices.mammotion import MammotionBaseCloudDevice
+from pymammotion.data.model.account import Credentials
+from pymammotion.mammotion.devices.mammotion import create_devices, ConnectionPreference, Mammotion
+
+logger = logging.getLogger(__name__)
+
+
+async def run():
+    EMAIL = os.environ.get('EMAIL')
+    PASSWORD = os.environ.get('PASSWORD')
+    DEVICE_NAME = "Luba-VSXXXXXX"
+
+    try:
+        credentials = Credentials(
+        email=EMAIL,
+        password=PASSWORD
+        )   
+        _mammotion = await create_devices(ble_device=None, cloud_credentials=credentials, preference=ConnectionPreference.WIFI)
+
+
+
+        await _mammotion.get_stream_subsctiption(DEVICE_NAME)
+
+        return _mammotion
+    except Exception as ex:
+        logger.error(f"{ex}")
+        logger.error(traceback.format_exc())
+        return None
+
+
+if __name__ ==  '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    logger.getChild("paho").setLevel(logging.WARNING)
+    event_loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(event_loop)
+    cloud_client = event_loop.run_until_complete(run())
+    event_loop.run_forever()