Merge pull request #2 from AlessandroFlati/develop

First working demo

Author: AlessandroFlati

BlueStacksAgent/__init__.py

@@ -0,0 +1,4 @@
+# This file makes BlueStacksAgent a Python package.
+from .agents import *
+from .actuators import *
+from .bluestacks_agent import BlueStacksAgent

BlueStacksAgent/actuators/__init__.py

@@ -0,0 +1,2 @@
+# Expose actuator base class for easy import
+from .base import BaseActuator

BlueStacksAgent/actuators/base.py

@@ -0,0 +1,18 @@
+import abc
+
+import numpy as np
+
+
+class BaseActuator(abc.ABC):
+    """
+    Abstract class for actuators that work with BlueStacksAgent.
+    The process method must be implemented by any subclass to process frames.
+    """
+
+    @abc.abstractmethod
+    def process(self, frames: np.ndarray):
+        """
+        Callback function to process a captured frame.
+        :param frames: The buffer of captured frames (i.e., a 3D NumPy array, where the first dimension
+                       represents the frame index, and the remaining dimensions represent the frame data).
+        """

BlueStacksAgent/agents/__init__.py

@@ -0,0 +1,3 @@
+# Expose agents for easy import
+from .base import BaseAgent
+from .scrcpy_agent import ScrcpyAgent

BlueStacksAgent/agents/base.py

@@ -0,0 +1,139 @@
+import abc
+import os
+from collections import deque
+from typing import Tuple
+
+from adbutils import adb, AdbDevice
+
+
+class BaseAgent(abc.ABC):
+    """
+    Abstract Base Class for BlueStacks screen capture agents.
+    This class defines common configuration and the interface for all capture methods.
+    """
+
+    def __init__(self,
+                 adb_path: str = "adb",
+                 adb_port: int = 5555,
+                 resolution: Tuple[int, int] = None,
+                 bitrate: int = 8000000,
+                 max_fps: int = 30,
+                 queue_size: int = 3):
+        """
+        :param resolution: Tuple (width, height) for the desired resolution, or None for native
+        :param bitrate: Bitrate for video encoding (default 8Mbps)
+        :param max_fps: Maximum frames per second (default 30)
+        :param queue_size: Number of frames to buffer (default 3)
+        """
+        self.adb_path = adb_path
+        self.adb_port = adb_port
+        self.resolution = resolution
+        self.bitrate = bitrate
+        self.max_fps = max_fps
+        self.queue_size = queue_size
+
+        # Check if the resolution is valid
+        if resolution is not None and not self._is_resolution_valid():
+            raise ValueError(f"Invalid resolution: {self.resolution}")
+
+        # Check if the bitrate is valid
+        if not self._is_bitrate_valid():
+            raise ValueError(f"Invalid bitrate: {self.bitrate}")
+
+        # Check if the max_fps is valid
+        if not self._is_max_fps_valid():
+            raise ValueError(f"Invalid max_fps: {self.max_fps}")
+
+        # Check if the queue_size is valid
+        if not self._is_queue_size_valid():
+            raise ValueError(f"Invalid queue_size: {self.queue_size}")
+
+        # Create the frame buffer
+        self.frame_buffer = deque(maxlen=self.queue_size)
+
+        # Connect to ADB
+        self._connect_adb()
+        self.adb_device: AdbDevice = adb.device(serial=f"emulator-{self.adb_port}")
+
+        self._is_streaming = False
+
+    @abc.abstractmethod
+    def _start_stream(self):
+        """
+        Start the screen capture stream internally, to be handled from the concrete instances.
+        """
+
+    def start_stream(self):
+        """
+        Start the screen capture stream.
+        """
+        self._start_stream()
+        self._is_streaming = True
+
+    @abc.abstractmethod
+    def _stop_stream(self):
+        """
+        Stop the screen capture stream internally, to be handled from the concrete instances.
+        """
+
+    def stop_stream(self):
+        """
+        Stop the screen capture stream.
+        """
+        self._is_streaming = False
+        self._stop_stream()
+
+    def is_streaming(self):
+        """
+        Return whether the agent is currently streaming.
+        """
+        return self._is_streaming
+
+    def _is_resolution_valid(self):
+        """
+        Check if the resolution is valid.
+        """
+        # Check if the resolution is a tuple of two integers
+        return isinstance(self.resolution, tuple) and len(self.resolution) == 2 and all(
+            isinstance(x, int) for x in self.resolution)
+
+    def _is_bitrate_valid(self):
+        """
+        Check if the bitrate is valid.
+        """
+        # Check if the bitrate is a positive integer
+        return isinstance(self.bitrate, int) and self.bitrate > 0
+
+    def _is_max_fps_valid(self):
+        """
+        Check if the max_fps is valid.
+        """
+        # Check if the max_fps is a positive integer
+        return isinstance(self.max_fps, int) and self.max_fps > 0
+
+    def _is_queue_size_valid(self):
+        """
+        Check if the queue_size is valid.
+        """
+        # Check if the queue_size is a positive integer
+        return isinstance(self.queue_size, int) and self.queue_size > 0
+
+    def _insert_in_frame_buffer(self, frame):
+        """
+        Insert a frame into the frame buffer.
+        """
+        self.frame_buffer.append(frame)
+
+    def _connect_adb(self):
+        """
+        Connect to ADB.
+        """
+        # Check if ADB is installed
+        if not self._is_adb_installed():
+            raise FileNotFoundError("ADB not found. Please install ADB and add it to the system PATH.")
+
+    def _is_adb_installed(self):
+        """
+        Check if ADB is installed.
+        """
+        return os.system(f"{self.adb_path} --version") == 0

BlueStacksAgent/agents/mediaprojection_agent.py

@@ -0,0 +1,28 @@
+from .base import BaseAgent
+
+
+class MediaProjectionAgent(BaseAgent):
+    """
+    Concrete implementation of BaseAgent using MediaProjection.
+    """
+
+    def __init__(self,
+                 adb_path: str = "adb",
+                 resolution: tuple[int, int] = None,
+                 bitrate: int = 8000000,
+                 max_fps: int = 30,
+                 queue_size: int = 3):
+        super().__init__(adb_path, resolution, bitrate, max_fps, queue_size)
+        raise NotImplementedError("MediaProjectionAgent is not yet implemented.")
+
+    def _start_stream(self):
+        """
+        Start the MediaProjection stream.
+        """
+        raise NotImplementedError("MediaProjectionAgent is not yet implemented.")
+
+    def _stop_stream(self):
+        """
+        Stop the minicap stream.
+        """
+        raise NotImplementedError("MediaProjectionAgent is not yet implemented.")

BlueStacksAgent/agents/minicap_agent.py

@@ -0,0 +1,28 @@
+from .base import BaseAgent
+
+
+class MinicapAgent(BaseAgent):
+    """
+    Concrete implementation of BaseAgent using minicap.
+    """
+
+    def __init__(self,
+                 adb_path: str = "adb",
+                 resolution: tuple[int, int] = None,
+                 bitrate: int = 8000000,
+                 max_fps: int = 30,
+                 queue_size: int = 3):
+        super().__init__(adb_path, resolution, bitrate, max_fps, queue_size)
+        raise NotImplementedError("MinicapAgent is not yet implemented.")
+
+    def _start_stream(self):
+        """
+        Start the minicap stream.
+        """
+        raise NotImplementedError("MinicapAgent is not yet implemented.")
+
+    def _stop_stream(self):
+        """
+        Stop the minicap stream.
+        """
+        raise NotImplementedError("MinicapAgent is not yet implemented.")

BlueStacksAgent/agents/scrcpy_agent.py

@@ -0,0 +1,45 @@
+import threading
+import time
+from .base import BaseAgent
+from scrcpy import Client
+
+class ScrcpyAgent(BaseAgent):
+    """
+    Concrete implementation of BaseAgent using scrcpy.
+    """
+
+    def __init__(self,
+                 adb_path: str = "adb",
+                 adb_serial: int = 5555,
+                 resolution: tuple[int, int] = None,
+                 bitrate: int = 8000000,
+                 max_fps: int = 30,
+                 queue_size: int = 3):
+        super().__init__(adb_path, adb_serial, resolution, bitrate, max_fps, queue_size)
+        self.client = None
+        self.thread = None
+
+    def _stream_loop(self):
+        # Instantiate the scrcpy client with desired parameters.
+        # TODO: (Additional scrcpy options can be added as needed.)
+        self.client = Client(device=self.adb_device, bitrate=self.bitrate, max_fps=self.max_fps)
+        self.client.add_listener("frame", self._insert_in_frame_buffer)
+        self.client.start(threaded=True)
+        # Continue streaming until signaled to stop.
+        while self._is_streaming:
+            time.sleep(0.1)
+        self.client.stop()
+
+    def _start_stream(self):
+        """
+        Start the scrcpy stream.
+        """
+        self.thread = threading.Thread(target=self._stream_loop, daemon=True)
+        self.thread.start()
+
+    def _stop_stream(self):
+        """
+        Stop the scrcpy stream.
+        """
+        if self.thread:
+            self.thread.join()

BlueStacksAgent/bluestacks_agent.py

@@ -0,0 +1,98 @@
+import time
+from collections import deque
+from threading import Thread
+
+import numpy as np
+
+from BlueStacksAgent.actuators.base import BaseActuator
+from BlueStacksAgent.agents.base import BaseAgent
+
+
+class BlueStacksAgent:
+    """
+    BlueStacksAgent ties together a screen capture agent (e.g., scrcpy, minicap, or mediaprojection)
+    and an actuator that processes captured frames.
+
+    This class instantiates the proper capture agent based on a provided stream type and registers
+    an actuator to process each frame.
+    """
+
+    def __init__(self, stream_agent: BaseAgent = None, actuator: BaseActuator = None, **kwargs):
+        """
+        :param stream_agent: An instance of a subclass of BaseAgent that implements start_stream(callback).
+        :param actuator: An instance of a subclass of BaseActuator that implements on_frame(frame).
+        :param kwargs: Additional keyword arguments passed to the underlying capture agent's constructor.
+        """
+        if stream_agent is None:
+            raise ValueError("stream_agent must be provided.")
+        if actuator is None:
+            raise ValueError("actuator must be provided.")
+
+        self.stream_agent: BaseAgent = stream_agent
+        self.actuator: BaseActuator = actuator
+        self._error = None
+        self.thread = None
+        self.error_thread = None
+        self.is_processing = False
+
+    def _process_loop(self):
+        """
+        Continuously processes frames from the stream agent.
+        :return:
+        """
+        while self.is_processing:
+            self._error = None
+            if self.stream_agent.frame_buffer:
+                frames = self._convert_to_3d_array(self.stream_agent.frame_buffer)
+                try:
+                    if frames.shape[0] > 0:
+                        self.actuator.process(frames)
+                except Exception as e:
+                    self._error = e
+                    break
+            else:
+                time.sleep(0.1)
+
+    def _error_loop(self):
+        """
+        Continuously checks for errors in the processing loop.
+        :return:
+        """
+        while self.is_processing:
+            if self._error:
+                print(f"Error in processing loop: {self._error}")
+                self.thread.join()
+                break
+            time.sleep(0.1)
+
+    def start(self):
+        """
+        Start the capture stream and processing loop.
+        """
+        self.is_processing = True
+        self.thread = Thread(target=self._process_loop, daemon=True)
+        self.thread.start()
+        self.error_thread = Thread(target=self._error_loop, daemon=True)
+        self.error_thread.start()
+        self.stream_agent.start_stream()
+
+    def stop(self):
+        """
+        Stop the capture stream.
+        """
+        self.is_processing = False
+        self.stream_agent.stop_stream()
+        if self.thread:
+            self.thread.join()
+        if self.error_thread:
+            self.error_thread.join()
+
+    @staticmethod
+    def _convert_to_3d_array(frame_buffer: deque) -> np.ndarray:
+        """
+        Convert the frame buffer to a 3D NumPy array.
+        :param frame_buffer: A deque of frames.
+        :return: A 3D NumPy array of frames.
+        """
+        frame_buffer_copy = frame_buffer.copy()
+        return np.array([f for f in frame_buffer_copy if f is not None])