delete old files
This commit is contained in:
@@ -1,41 +0,0 @@
|
|||||||
from enum import Enum
|
|
||||||
|
|
||||||
import numpy as np
|
|
||||||
from PIL import Image
|
|
||||||
|
|
||||||
from simulator import EmulatorSimulator, MouseClick
|
|
||||||
|
|
||||||
|
|
||||||
class Action(Enum):
|
|
||||||
CLICK = "click"
|
|
||||||
MOUSE_DOWN = "mousedown"
|
|
||||||
MOUSE_UP = "mouseup"
|
|
||||||
MOUSE_MOVE = "mousemove"
|
|
||||||
KEY = "key"
|
|
||||||
TYPE = "type"
|
|
||||||
|
|
||||||
|
|
||||||
class Controller:
|
|
||||||
def __init__(self, vm_name: str, username: str, password: str, host: str) -> None:
|
|
||||||
self.simulator = EmulatorSimulator(vm_name=vm_name, username=username,
|
|
||||||
password=password, host=host)
|
|
||||||
|
|
||||||
def get_state(self) -> np.ndarray:
|
|
||||||
image_path = self.simulator.get_screenshot()
|
|
||||||
with Image.open(image_path) as img:
|
|
||||||
return np.array(img)
|
|
||||||
|
|
||||||
def step(self, action: Action, **action_args) -> None:
|
|
||||||
if action == Action.CLICK:
|
|
||||||
print(action_args)
|
|
||||||
self.simulator.click(**action_args)
|
|
||||||
elif action == Action.MOUSE_DOWN:
|
|
||||||
self.simulator.mousedown(**action_args)
|
|
||||||
elif action == Action.MOUSE_UP:
|
|
||||||
self.simulator.mouseup(**action_args)
|
|
||||||
elif action == Action.MOUSE_MOVE:
|
|
||||||
self.simulator.mouse_move(**action_args)
|
|
||||||
elif action == Action.KEY:
|
|
||||||
self.simulator.key(**action_args)
|
|
||||||
elif action == Action.TYPE:
|
|
||||||
self.simulator.type(**action_args)
|
|
||||||
72
simulator.py
72
simulator.py
@@ -1,72 +0,0 @@
|
|||||||
from enum import Enum
|
|
||||||
|
|
||||||
import subprocess
|
|
||||||
from fabric import Connection
|
|
||||||
|
|
||||||
|
|
||||||
class MouseClick(Enum):
|
|
||||||
LEFT = 1
|
|
||||||
MIDDLE = 2
|
|
||||||
RIGHT = 3
|
|
||||||
WHEEL_UP = 4
|
|
||||||
WHEEL_DOWN = 5
|
|
||||||
|
|
||||||
|
|
||||||
class EmulatorSimulator:
|
|
||||||
def __init__(self, vm_name: str, username: str, password: str, host:str, snapshot_name: str = "snapshot"):
|
|
||||||
self.vm_name = vm_name
|
|
||||||
self.username = username
|
|
||||||
self.password = password
|
|
||||||
self.host = host
|
|
||||||
self.snapshot_name = snapshot_name
|
|
||||||
self.ssh_connection = Connection(host=self.host, user=self.username, connect_kwargs={"password": password})
|
|
||||||
self._execute_command(["VBoxManage", "startvm", self.vm_name])
|
|
||||||
|
|
||||||
def _execute_command(self, command: list[str]) -> None:
|
|
||||||
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
||||||
stdout, stderr = process.communicate()
|
|
||||||
if process.returncode != 0:
|
|
||||||
print(f"Error executing command: {command}")
|
|
||||||
print(stderr.decode())
|
|
||||||
return None
|
|
||||||
else:
|
|
||||||
return stdout.decode()
|
|
||||||
|
|
||||||
def _execute_xdotool_command(self, command: list[str]) -> None:
|
|
||||||
result = self.ssh_connection.run(f"DISPLAY=:0 xdotool {command}", hide=True)
|
|
||||||
return result.stdout.strip()
|
|
||||||
|
|
||||||
def get_logs(self) -> str:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def load_state(self) -> None:
|
|
||||||
self._execute_command(["VBoxManage", "snapshot", self.vm_name, "restore", self.snapshot_name])
|
|
||||||
|
|
||||||
def save_state(self) -> None:
|
|
||||||
self._execute_command(["VBoxManage", "snapshot", self.vm_name, "take", self.snapshot_name])
|
|
||||||
|
|
||||||
def click(self, click: MouseClick) -> None:
|
|
||||||
self._execute_xdotool_command(f"click {click.value}")
|
|
||||||
|
|
||||||
def mousedown(self, click: MouseClick) -> None:
|
|
||||||
self._execute_xdotool_command(f"mousedown {click}")
|
|
||||||
|
|
||||||
def mouseup(self, click: MouseClick) -> None:
|
|
||||||
self._execute_xdotool_command(f"mouseup {click}")
|
|
||||||
|
|
||||||
def mouse_move(self, x: int, y: int) -> None:
|
|
||||||
self._execute_xdotool_command(f"mousemove {x} {y}")
|
|
||||||
|
|
||||||
def key(self, key: str) -> None:
|
|
||||||
self._execute_xdotool_command(f"key {key}")
|
|
||||||
|
|
||||||
def type(self, text: str) -> None:
|
|
||||||
self._execute_xdotool_command(f"type {text}")
|
|
||||||
|
|
||||||
def get_screenshot(self) -> str:
|
|
||||||
image_path = "./screenshot.png"
|
|
||||||
self._execute_command(["VBoxManage", "controlvm", self.vm_name, "screenshotpng", image_path])
|
|
||||||
return image_path
|
|
||||||
|
|
||||||
def close(self) -> None:
|
|
||||||
pass
|
|
||||||
Reference in New Issue
Block a user