lzy/sci-gui-agent-benchmark
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ver Jan4th

Browse Source 
updated interfaces for thunderbird evaluation, not tested
This commit is contained in:
David Chang
2024-01-04 22:41:57 +08:00
parent f831aa93df
commit 5fedf5b891
10 changed files with 361 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
desktop_env/controllers/setup.py
View File

@@ -6,7 +6,7 @@ import uuid
import os.path
from typing import Dict, List
from typing import Any
from typing import Any, Union
class SetupController:
@@ -145,7 +145,7 @@ class SetupController:
print("An error occurred while trying to send the request:", e)
def _tidy_desktop_setup(self, **config):
raise NotImplementedError
raise NotImplementedError()
def _open_setup(self, path: str):
# if not config:
@@ -170,3 +170,56 @@ class SetupController:
print("Failed to open file. Status code:", response.text)
except requests.exceptions.RequestException as e:
print("An error occurred while trying to send the request:", e)
def _launch_setup(self, command: List[str]):
if not command:
raise Exception("Empty comman to launch.")
payload = json.dumps({"command": command})
headers = {"Content-Type": "application/json"}
try:
response = requests.post(self.http_server + "/launch", headers=headers, data=payload)
if response.status_code == 200:
print("Command executed successfully:", response.text)
else:
print("Failed to launch application. Status code:", response.text)
except requests.exceptions.RequestException as e:
print("An error occurred while trying to send the request:", e)
def _execute_setup(self, command: List[str], stdout: str = "", stderr: str = ""):
if not command:
raise Exception("Empty comman to launch.")
payload = json.dumps({"command": command})
headers = {"Content-Type": "application/json"}
try:
response = requests.post(self.http_server + "/launch", headers=headers, data=payload)
if response.status_code == 200:
results: Dict[str, str] = response.json()
if stdout:
with open(os.path.join(self.cache_dir, stdout), "w") as f:
f.write(results["output"])
if stderr:
with open(os.path.join(self.cache_dir, stderr), "w") as f:
f.write(results["error"])
print( "Command executed successfully: {:} ->".format(" ".join(command))
, response.text
)
else:
print("Failed to launch application. Status code:", response.text)
except requests.exceptions.RequestException as e:
print("An error occurred while trying to send the request:", e)
def _act_setup(self, action_seq: List[Union[Dict[str, Any], str]]):
# TODO
raise NotImplementedError()
def _replay_setup(self, trajectory: str):
"""
Args:
trajectory (str): path to the replay trajectory file
"""
# TODO
raise NotImplementedError()

48
desktop_env/envs/desktop_env.py
View File

@@ -68,19 +68,8 @@ class DesktopEnv(gym.Env):
self.cache_dir_base: str = cache_dir
# task-aware stuffs
self.snapshot_path = task_config["snapshot"] # todo: handling the logic of snapshot directory
self.task_id: str = task_config["id"]
self.cache_dir: str = os.path.join(self.cache_dir_base, self.task_id)
os.makedirs(self.cache_dir, exist_ok=True)
self.instruction = task_config["instruction"]
self.config = task_config["config"]
self.evaluator = task_config["evaluator"]
self.metric: Metric = getattr(metrics, self.evaluator["func"])
self.result_getter: Getter = getattr(getters, "get_{:}".format(self.evaluator["result"]["type"]))
self.expected_getter: Getter = getattr(getters, "get_{:}".format(
self.evaluator["expected"]["type"])) if "expected" in self.evaluator else None
self.metric_options: Dict[str, Any] = self.evaluator.get("options", {})
# todo: handling the logic of snapshot directory
self._set_task_info(task_config)
# Initialize emulator and controller
print("Initializing...")
@@ -151,25 +140,27 @@ class DesktopEnv(gym.Env):
screenshot_image_path = self._get_screenshot()
return screenshot_image_path
def _set_task_info(self, task_config: Dict[str, Any]):
self.snapshot_path = task_config["snapshot"]
self.task_id: str = task_config["id"]
self.cache_dir: str = os.path.join(self.cache_dir_base, self.task_id)
os.makedirs(self.cache_dir, exist_ok=True)
self.instruction = task_config["instruction"]
self.config = task_config["config"]
self.evaluator = task_config["evaluator"]
self.metric: Metric = getattr(metrics, self.evaluator["func"])
self.result_getter: Getter = getattr(getters, "get_{:}".format(self.evaluator["result"]["type"]))
self.expected_getter: Getter = getattr(getters, "get_{:}".format(
self.evaluator["expected"]["type"])) if "expected" in self.evaluator else None
self.metric_options: Dict[str, Any] = self.evaluator.get("options", {})
def reset(self, task_config: Optional[Dict[str, Any]] = None, seed=None, options=None):
print("Resetting environment...")
print("Switching task...")
if task_config is not None:
self.snapshot_path = task_config["snapshot"]
self.task_id = task_config["id"]
self.cache_dir = os.path.join(self.cache_dir_base, self.task_id)
os.makedirs(self.cache_dir, exist_ok=True)
self.instruction = task_config["instruction"]
self.config = task_config["config"]
self.evaluator = task_config["evaluator"]
self.metric: Metric = getattr(metrics, self.evaluator["func"])
self.result_getter: Getter = getattr(getters, "get_{:}".format(self.evaluator["result"]["type"]))
self.expected_getter: Getter = getattr(getters, "get_{:}".format(
self.evaluator["expected"]["type"])) if "expected" in self.evaluator else None
self.metric_options = self.evaluator.get("options", {})
self._set_task_info(task_config)
self.setup_controller.reset_cache_dir(self.cache_dir)
print("Setting counters...")
@@ -228,6 +219,9 @@ class DesktopEnv(gym.Env):
"""
Evaluate whether the task is successfully completed.
"""
self.setup_controller.setup(self.evaluator["postconfig"])
result_state = self.result_getter(self, self.evaluator["result"])
expected_state = self.expected_getter(self, self.evaluator["expected"]) if "expected" in self.evaluator \
else None

2
desktop_env/evaluators/getters/__init__.py
View File

@@ -1,2 +1,2 @@
from .file import get_cloud_file, get_vm_file
from .file import get_cloud_file, get_vm_file, get_cache_file
from .misc import get_rule

11
desktop_env/evaluators/getters/file.py
View File

@@ -33,8 +33,6 @@ def get_vm_file(env, config: Dict[str, str]) -> str:
"""
_path = os.path.join(env.cache_dir, config["dest"])
if os.path.exists(_path):
return _path
file = env.controller.get_file(config["path"])
with open(_path, "wb") as f:
@@ -42,3 +40,12 @@ def get_vm_file(env, config: Dict[str, str]) -> str:
return _path
def get_cache_file(env, config: Dict[str, str]) -> str:
"""
Config:
path (str): relative path in cache dir
"""
_path = os.path.join(env.cache_dir, config["path"])
assert os.path.exists(_path)
return _path

55
desktop_env/evaluators/getters/misc.py
View File

@@ -1,8 +1,63 @@
from typing import TypeVar
import platform
import subprocess
import ctypes
import os
R = TypeVar("Rule")
def get_rule(env, config: R) -> R:
"""
Returns the rule as-is.
"""
return config["rules"]
def get_desktop_path():
username = os.getlogin() # Get the current username
if platform.system() == "Windows":
return os.path.join("C:", "Users", username, "Desktop")
elif platform.system() == "Darwin": # macOS is identified as 'Darwin'
return os.path.join("/Users", username, "Desktop")
elif platform.system() == "Linux":
return os.path.join("/home", username, "Desktop")
else:
raise Exception("Unsupported operating system")
def get_wallpaper():
def get_wallpaper_windows():
SPI_GETDESKWALLPAPER = 0x73
MAX_PATH = 260
buffer = ctypes.create_unicode_buffer(MAX_PATH)
ctypes.windll.user32.SystemParametersInfoW(SPI_GETDESKWALLPAPER, MAX_PATH, buffer, 0)
return buffer.value
def get_wallpaper_macos():
script = """
tell application "System Events" to tell every desktop to get picture
"""
process = subprocess.Popen(['osascript', '-e', script], stdout=subprocess.PIPE)
output, error = process.communicate()
if error:
print("Error:", error)
else:
return output.strip().decode('utf-8')
def get_wallpaper_linux():
try:
output = subprocess.check_output(["gsettings", "get", "org.gnome.desktop.background", "picture-uri"])
return output.decode('utf-8').strip().replace('file://', '').replace("'", "")
except Exception as e:
print("Error:", e)
return None
os_name = platform.system()
if os_name == 'Windows':
return get_wallpaper_windows()
elif os_name == 'Darwin':
return get_wallpaper_macos()
elif os_name == 'Linux':
return get_wallpaper_linux()
else:
return "Unsupported OS"

1
desktop_env/evaluators/metrics/__init__.py
View File

@@ -5,3 +5,4 @@ from .docs import compare_font_names, compare_subscript_contains, has_page_numbe
from .docs import is_first_line_centered, check_file_exists, compare_contains_image
from .pdf import check_pdf_pages
from .libreoffice import check_libre_locale
from .general import check_csv

79
desktop_env/evaluators/metrics/general.py
View File

@@ -1,57 +1,30 @@
import platform
import subprocess
import ctypes
import os
import csv
from typing import Dict, List
def _match_record(pattern: Dict[str, str], item: Dict[str, str]) -> float:
return all(k in item and item[k]==val for k, val in pattern.items())
# todo: move to getter module
def get_desktop_path():
username = os.getlogin() # Get the current username
if platform.system() == "Windows":
return os.path.join("C:", "Users", username, "Desktop")
elif platform.system() == "Darwin": # macOS is identified as 'Darwin'
return os.path.join("/Users", username, "Desktop")
elif platform.system() == "Linux":
return os.path.join("/home", username, "Desktop")
else:
raise Exception("Unsupported operating system")
def get_wallpaper():
def get_wallpaper_windows():
SPI_GETDESKWALLPAPER = 0x73
MAX_PATH = 260
buffer = ctypes.create_unicode_buffer(MAX_PATH)
ctypes.windll.user32.SystemParametersInfoW(SPI_GETDESKWALLPAPER, MAX_PATH, buffer, 0)
return buffer.value
def get_wallpaper_macos():
script = """
tell application "System Events" to tell every desktop to get picture
"""
process = subprocess.Popen(['osascript', '-e', script], stdout=subprocess.PIPE)
output, error = process.communicate()
if error:
print("Error:", error)
else:
return output.strip().decode('utf-8')
def get_wallpaper_linux():
try:
output = subprocess.check_output(["gsettings", "get", "org.gnome.desktop.background", "picture-uri"])
return output.decode('utf-8').strip().replace('file://', '').replace("'", "")
except Exception as e:
print("Error:", e)
return None
os_name = platform.system()
if os_name == 'Windows':
return get_wallpaper_windows()
elif os_name == 'Darwin':
return get_wallpaper_macos()
elif os_name == 'Linux':
return get_wallpaper_linux()
else:
return "Unsupported OS"
def check_csv(result: str, rules: Dict[str, List[Dict[str, str]]]) -> float:
"""
Args:
result (str): path to csv file
rules (Dict[str, List[Dict[str, str]]]): dict like
{
"expect": [{key: value}]
"unexpect": [{key: value}]
}
Returns:
float
"""
expect_metrics = [False] * len(rules["except"])
unexpect_metric = True
with open(result) as f:
reader = csv.DictReader(f)
for rcd in reader:
for i, r in enumerate(rules["expect"]):
expect_metrics[i] = expect_metrics[i] or _match_record(r, rcd)
unexpect_metric = unexpect_metric and all(_match_record(r, rcd) for r in rules["unexpect"])
return float(all(expect_metrics) and unexpect_metric)

21
desktop_env/server/main.py
View File

@@ -9,6 +9,7 @@ import pyautogui
# from PIL import ImageGrab, Image
from PIL import Image
from flask import Flask, request, jsonify, send_file
from typing import List
app = Flask(__name__)
@@ -16,15 +17,16 @@ pyautogui.PAUSE = 0
pyautogui.DARWIN_CATCH_UP_TIME = 0
@app.route('/setup/execute', methods=['POST'])
@app.route('/execute', methods=['POST'])
def execute_command():
data = request.json
# The 'command' key in the JSON request should contain the command to be executed.
command = data.get('command', '')
command = data.get('command', [])
# Execute the command without any safety checks.
try:
result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
return jsonify({
'status': 'success',
'output': result.stdout,
@@ -36,6 +38,21 @@ def execute_command():
'message': str(e)
}), 500
@app.route('/setup/launch', methods=["POST"])
def launch_app():
data = request.json
command: List[str] = data.get("command", [])
try:
subprocess.Popen(command)
return "{:} launched successfully".format(" ".join(command))
except Exception as e:
return jsonify( { "status": "error"
, "message": str(e)
}
)\
, 500
@app.route('/screenshot', methods=['GET'])
def capture_screen_with_cursor():