Implement DroidCast screenshot backend

adb_vision/screenshot.py

@@ -0,0 +1,71 @@
+import asyncio
+import base64
+import requests
+from typing import Callable, Coroutine
+
+# Define AdbRunFn type for type hinting
+# async def _adb_run(*args, timeout) -> bytes
+AdbRunFn = Callable[..., Coroutine[None, None, bytes]]
+
+async def _capture_droidcast(*, adb_run: AdbRunFn, serial: str, adb_exe: str) -> str:
+    """Return base64-encoded PNG screenshot via DroidCast HTTP stream."""
+    url = "http://localhost:53516/screenshot?format=png"
+
+    async def fetch_screenshot(raise_for_status=False):
+        try:
+            # Using asyncio.to_thread for the synchronous requests call
+            response = await asyncio.to_thread(requests.get, url, timeout=5.0)
+            if response.status_code == 200:
+                return base64.b64encode(response.content).decode("ascii")
+            if raise_for_status:
+                response.raise_for_status()
+        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
+            if raise_for_status:
+                raise
+        except requests.exceptions.HTTPError:
+            if raise_for_status:
+                raise
+        return None
+
+    # Try fetching first (idempotency)
+    result = await fetch_screenshot()
+    if result:
+        return result
+
+    # Setup DroidCast if not running or not responding correctly
+    apk_path = "alas_wrapped/bin/DroidCast/DroidCast-debug-1.2.3.apk"
+
+    # 1. Install APK
+    await adb_run("install", "-r", apk_path, timeout=30.0)
+
+    # 2. Start service
+    try:
+        await adb_run(
+            "shell", "am", "start-foreground-service",
+            "-n", "com.rayworks.droidcast/.Main",
+            "-a", "android.intent.action.MAIN",
+            timeout=10.0
+        )
+    except Exception:
+        # Fallback to app_process method
+        await adb_run(
+            "shell", "CLASSPATH=/data/local/tmp/DroidCast.dex",
+            "app_process", "/", "com.rayworks.droidcast.Main",
+            timeout=5.0
+        )
+
+    # 3. Forward port
+    await adb_run("forward", "tcp:53516", "tcp:53516", timeout=5.0)
+
+    # 4. Fetch again
+    for i in range(3):
+        try:
+            # On the last attempt, we want to know why it failed
+            result = await fetch_screenshot(raise_for_status=(i == 2))
+            if result:
+                return result
+        except Exception as e:
+            raise RuntimeError(f"DroidCast HTTP failure: {e}")
+        await asyncio.sleep(1.0)
+
+    raise RuntimeError("Failed to capture screenshot via DroidCast: service did not respond with 200 OK")

adb_vision/test_server.py

@@ -0,0 +1,124 @@
+import pytest
+import unittest.mock as mock
+import base64
+import io
+import requests
+from adb_vision.screenshot import _capture_droidcast
+
+def _fake_png() -> bytes:
+    """Return a minimal valid PNG (1x1 transparent) for use in tests."""
+    return (
+        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00"
+        b"\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00"
+        b"\x01\x00\x00\x05\x00\x01\r\n\x2d\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
+    )
+
+@pytest.mark.asyncio
+async def test_capture_droidcast_happy_path():
+    """Happy path: first fetch fails, then install/start/forward/fetch succeeds."""
+    adb_run = mock.AsyncMock()
+    fake_png_data = _fake_png()
+    fake_png_b64 = base64.b64encode(fake_png_data).decode("ascii")
+
+    # Mock requests.get to fail once, then succeed
+    mock_response = mock.Mock()
+    mock_response.status_code = 200
+    mock_response.content = fake_png_data
+
+    with mock.patch("requests.get", side_effect=[
+        requests.exceptions.ConnectionError(), # Initial check for idempotency
+        mock_response                          # After setup
+    ]) as m_get:
+        result = await _capture_droidcast(
+            adb_run=adb_run,
+            serial="emulator-5554",
+            adb_exe="adb"
+        )
+
+        assert result == fake_png_b64
+        assert m_get.call_count == 2
+
+        # Verify ADB calls
+        assert adb_run.call_count >= 3
+        adb_run.assert_any_call("install", "-r", "alas_wrapped/bin/DroidCast/DroidCast-debug-1.2.3.apk", timeout=30.0)
+        adb_run.assert_any_call("forward", "tcp:53516", "tcp:53516", timeout=5.0)
+
+@pytest.mark.asyncio
+async def test_capture_droidcast_idempotent():
+    """Idempotency: first fetch succeeds, skipping all ADB setup calls."""
+    adb_run = mock.AsyncMock()
+    fake_png_data = _fake_png()
+    fake_png_b64 = base64.b64encode(fake_png_data).decode("ascii")
+
+    mock_response = mock.Mock()
+    mock_response.status_code = 200
+    mock_response.content = fake_png_data
+
+    with mock.patch("requests.get", return_value=mock_response) as m_get:
+        result = await _capture_droidcast(
+            adb_run=adb_run,
+            serial="emulator-5554",
+            adb_exe="adb"
+        )
+
+        assert result == fake_png_b64
+        assert m_get.call_count == 1
+        assert adb_run.call_count == 0
+
+@pytest.mark.asyncio
+async def test_capture_droidcast_http_failure():
+    """Failure: HTTP returns 404 after setup -> raises RuntimeError."""
+    adb_run = mock.AsyncMock()
+
+    mock_response_404 = mock.Mock()
+    mock_response_404.status_code = 404
+    # mock_response_404.raise_for_status will be called on last retry
+    mock_response_404.raise_for_status.side_effect = requests.exceptions.HTTPError("404 Client Error")
+
+    with mock.patch("requests.get", side_effect=[
+        requests.exceptions.ConnectionError(), # Initial check
+        mock_response_404,                     # After setup
+        mock_response_404,
+        mock_response_404
+    ]) as m_get:
+        with pytest.raises(RuntimeError, match="DroidCast HTTP failure: 404 Client Error"):
+            await _capture_droidcast(
+                adb_run=adb_run,
+                serial="emulator-5554",
+                adb_exe="adb"
+            )
+
+        assert m_get.call_count == 4 # 1 init + 3 retries
+
+@pytest.mark.asyncio
+async def test_capture_droidcast_fallback_start():
+    """Verify fallback to app_process if am start-foreground-service fails."""
+    adb_run = mock.AsyncMock(side_effect=[
+        None, # install
+        RuntimeError("am failed"), # am start
+        None, # app_process
+        None  # forward
+    ])
+    fake_png_data = _fake_png()
+    fake_png_b64 = base64.b64encode(fake_png_data).decode("ascii")
+
+    mock_response = mock.Mock()
+    mock_response.status_code = 200
+    mock_response.content = fake_png_data
+
+    with mock.patch("requests.get", side_effect=[
+        requests.exceptions.ConnectionError(),
+        mock_response
+    ]):
+        result = await _capture_droidcast(
+            adb_run=adb_run,
+            serial="emulator-5554",
+            adb_exe="adb"
+        )
+
+        assert result == fake_png_b64
+        adb_run.assert_any_call(
+            "shell", "CLASSPATH=/data/local/tmp/DroidCast.dex",
+            "app_process", "/", "com.rayworks.droidcast.Main",
+            timeout=5.0
+        )