fix camera feed for yuka to just one camera for now

Author: mikey0000

pymammotion/agora/agora_api.py

@@ -14,10 +14,10 @@
 import logging
 from random import randint
 import time
+from types import TracebackType
+from typing import Any
 
 import aiohttp
-from types import TracebackType
-from typing import Optional, Type
 
 # Service IDs for API requests (what you send in the request)
 SERVICE_IDS = {
@@ -112,7 +112,7 @@ class AgoraResponse:
     detail: dict
     flag: int
     opid: int
-    responses: dict = None  # Multi-flag responses: {flag: response_dict}
+    responses: dict | None = None  # Multi-flag responses: {flag: response_dict}
 
     @classmethod
     def from_api_response(cls, response_data: dict) -> "AgoraResponse":
@@ -507,7 +507,9 @@ async def __aenter__(self):
         """Context manager entry."""
         return self
 
-    async def __aexit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]) -> None:
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
         """Context manager exit - close session if we created it."""
         if self._own_session and self.session:
             await self.session.close()
@@ -795,7 +797,7 @@ async def _call_endpoint(
         domain: str,
         request_payload: dict,
         proxy_server: str | None = None,
-    ) -> dict:
+    ) -> Any:
         """Call a single Agora endpoint.
 
         Args:
@@ -826,5 +828,4 @@ async def _call_endpoint(
             if resp.status != 200:
                 raise Exception(f"HTTP {resp.status}: {await resp.text()}")
 
-            response_data = await resp.json()
-            return response_data
+            return await resp.json()

pymammotion/http/http.py

@@ -290,40 +290,17 @@ async def net_rtk_enable(self, device_id: str) -> Response:
             return Response.from_dict(data)
 
     @refresh_token_decorator
-    async def get_stream_subscription(self, iot_id: str) -> Response[StreamSubscriptionResponse]:
-        """Fetches stream subscription data from agora.io for a given IoT device."""
-        resp = await self._session.post(
-            f"{MAMMOTION_API_DOMAIN}/device-server/v1/stream/subscription",
-            json={"deviceId": iot_id},
-            headers={
-                **self._headers,
-                "Authorization": f"Bearer {self.login_info.access_token}",
-                "Content-Type": "application/json",
-                "User-Agent": "okhttp/4.9.3",
-            },
-        )
-        data = await resp.json()
-        # TODO catch errors from mismatch like token expire etc
-        # Assuming the data format matches the expected structure
-        response = Response[StreamSubscriptionResponse].from_dict(data)
-        await self.handle_expiry(response)
-        if response.code != 0:
-            return response
-        response.data = StreamSubscriptionResponse.from_dict(data.get("data", {}))
-        return response
-
-    @refresh_token_decorator
-    async def get_stream_subscription_mini_or_x_series(
-        self, iot_id: str, is_yuka: bool
-    ) -> Response[StreamSubscriptionResponse]:
+    async def get_stream_subscription(self, iot_id: str, is_yuka: bool) -> Response[StreamSubscriptionResponse]:
         # Prepare the payload with cameraStates based on is_yuka flag
         """Fetches stream subscription data for a given IoT device."""
 
         payload = {"deviceId": iot_id, "mode": 0, "cameraStates": []}
 
         # Add appropriate cameraStates based on the is_yuka flag
+        # yukas have two cameras you could view [{"cameraState": 1}, {"cameraState": 0}, {"cameraState": 1}]
+        # but its not useful so ignore this and only subscribe to the front one.
         if is_yuka:
-            payload["cameraStates"] = [{"cameraState": 1}, {"cameraState": 0}, {"cameraState": 1}]
+            payload["cameraStates"] = [{"cameraState": 1}, {"cameraState": 0}, {"cameraState": 0}]
         else:
             payload["cameraStates"] = [{"cameraState": 1}, {"cameraState": 0}, {"cameraState": 0}]
 

pymammotion/mammotion/devices/mammotion.py

@@ -542,7 +542,7 @@ async def start_schedule_sync(self, name: str) -> None:
     async def get_stream_subscription(self, name: str, iot_id: str) -> Response[StreamSubscriptionResponse] | Any:
         """Get stream subscription."""
         device = self.get_device_by_name(name)
-        _stream_response = await device.mammotion_http.get_stream_subscription_mini_or_x_series(
+        _stream_response = await device.mammotion_http.get_stream_subscription(
             iot_id, DeviceType.is_yuka(name) and not DeviceType.is_yuka_mini(name)
         )
         _LOGGER.debug(_stream_response)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "pymammotion"
-version = "0.5.72"
+version = "0.5.73"
 description = ""
 authors = [
     { name = "Michael Arthur", email = "michael@jumblesoft.co.nz" },
@@ -68,7 +68,7 @@ include = ["pymammotion"]
 include = ["pymammotion"]
 
 [tool.bumpver]
-current_version = "0.5.72"
+current_version = "0.5.73"
 version_pattern = "MAJOR.MINOR.PATCH"
 commit_message  = "Bump version {old_version} -> {new_version}"
 commit          = true