add support /features, /current_activity

Author: codeskyblue

uiautodev/app.py

@@ -9,7 +9,7 @@
 import platform
 import signal
 from pathlib import Path
-from typing import List
+from typing import Dict, List
 
 import adbutils
 import uvicorn
@@ -25,6 +25,7 @@
 from uiautodev.model import Node
 from uiautodev.provider import AndroidProvider, HarmonyProvider, IOSProvider, MockProvider
 from uiautodev.remote.scrcpy import ScrcpyServer
+from uiautodev.router.android import router as android_device_router
 from uiautodev.router.device import make_router
 from uiautodev.router.xml import router as xml_router
 from uiautodev.utils.envutils import Environment
@@ -68,7 +69,23 @@ def enable_logger_to_console():
     app.include_router(harmony_router, prefix="/api/harmony", tags=["harmony"])
 
 app.include_router(xml_router, prefix="/api/xml", tags=["xml"])
-
+app.include_router(android_device_router, prefix="/api/android", tags=["android"])
+
+
+@app.get('/api/{platform}/features')
+def get_features(platform: str) -> Dict[str, bool]:
+    """Get features supported by the specified platform"""
+    features = {}
+    # 获取所有带有指定平台tag的路由
+    for route in app.routes:
+        if hasattr(route, 'tags') and platform in route.tags:
+            if route.path.startswith(f"/api/{platform}/{{serial}}/"):
+                # 提取特性名称
+                parts = route.path.split('/')
+                feature_name = parts[-1]
+                if not feature_name.startswith('{'):
+                    features[feature_name] = True
+    return features
 
 class InfoResponse(BaseModel):
     version: str

uiautodev/driver/android.py

@@ -31,6 +31,15 @@ def __init__(self, serial: str):
     @cached_property
     def ud(self) -> u2.Device:
         return u2.connect_usb(self.serial)
+
+    def get_current_activity(self) -> str:
+        ret = self.adb_device.shell2(["dumpsys", "activity", "activities"], rstrip=True, timeout=5)
+        # 使用正则查找包含前台 activity 的行
+        match = re.search(r"mResumedActivity:.*? ([\w\.]+\/[\w\.]+)", ret.output)
+        if match:
+            return match.group(1)  # 返回包名/类名，例如 com.example/.MainActivity
+        else:
+            return ""
 
     def screenshot(self, id: int) -> Image.Image:
         if id > 0:

uiautodev/router/android.py

@@ -0,0 +1,42 @@
+# prefix for /api/android/{serial}/shell
+
+import logging
+from typing import Dict, Optional
+
+from fastapi import APIRouter, Request, Response
+from pydantic import BaseModel
+
+from uiautodev.driver.android import AndroidDriver
+from uiautodev.model import ShellResponse
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+class AndroidShellPayload(BaseModel):
+    command: str
+    
+@router.post("/{serial}/shell")
+def shell(serial: str, payload: AndroidShellPayload) -> ShellResponse:
+    """Run a shell command on an Android device"""
+    try:
+        driver = AndroidDriver(serial)
+        return driver.shell(payload.command)
+    except NotImplementedError as e:
+        return Response(content="shell not implemented", media_type="text/plain", status_code=501)
+    except Exception as e:
+        logger.exception("shell failed")
+        return ShellResponse(output="", error=str(e))
+
+
+@router.get("/{serial}/current_activity")
+async def get_current_activity(serial: str) -> Response:
+    """Get the current activity of the Android device"""
+    try:
+        driver = AndroidDriver(serial)
+        activity = driver.get_current_activity()
+        return Response(content=activity, media_type="text/plain")
+    except Exception as e:
+        logger.exception("get_current_activity failed")
+        return Response(content="", media_type="text/plain")

uiautodev/router/device.py

@@ -6,9 +6,9 @@
 
 import io
 import logging
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
-from fastapi import APIRouter, Response
+from fastapi import APIRouter, Query, Request, Response
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
@@ -19,9 +19,6 @@
 
 logger = logging.getLogger(__name__)
 
-class AndroidShellPayload(BaseModel):
-    command: str
-
 
 def make_router(provider: BaseProvider) -> APIRouter:
     router = APIRouter()
@@ -37,18 +34,6 @@ def _list() -> List[DeviceInfo]:
             logger.exception("list_devices failed")
             return Response(content=str(e), media_type="text/plain", status_code=500)
 
-    @router.post("/{serial}/shell")
-    def android_shell(serial: str, payload: AndroidShellPayload) -> ShellResponse:
-        """Run a shell command on an Android device"""
-        try:
-            driver = provider.get_device_driver(serial)
-            return driver.shell(payload.command)
-        except NotImplementedError as e:
-            return Response(content="shell not implemented", media_type="text/plain", status_code=501)
-        except Exception as e:
-            logger.exception("shell failed")
-            return ShellResponse(output="", error=str(e))
-
     @router.get(
         "/{serial}/screenshot/{id}",
         responses={200: {"content": {"image/jpeg": {}}}},