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

ver Jan11th

Browse Source 
finally set up a simple task, or which should be simple
This commit is contained in:
David Chang
2024-01-11 20:03:33 +08:00
parent 5e2a03720d
commit 27eaf2f5d5
9 changed files with 272 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
desktop_env/controllers/setup.py
View File

@@ -6,12 +6,13 @@ import uuid
import os.path
from typing import Dict, List
from typing import Any, Union
from typing import Any, Union, Optional
import logging
logger = logging.getLogger("desktopenv.setup")
import traceback
import time
class SetupController:
def __init__(self, http_server: str, cache_dir: str):
@@ -45,6 +46,8 @@ class SetupController:
assert hasattr(self, setup_function)
getattr(self, setup_function)(**parameters)
logger.info("SETUP: %s(%s)", setup_function, str(parameters))
# self._download_setup(config)
# self._change_wallpaper(config)
# self._tidy_desktop(config) todo: implement this
@@ -219,35 +222,60 @@ class SetupController:
except requests.exceptions.RequestException as e:
logger.error("An error occurred while trying to send the request: %s", e)
def _execute_setup(self, command: List[str], stdout: str = "", stderr: str = ""):
def _execute_setup( self, command: List[str]
, stdout: str = "", stderr: str = ""
, shell: bool = False, until: Optional[Dict[str, Any]] = None):
if not command:
raise Exception("Empty comman to launch.")
payload = json.dumps({"command": command})
until: Dict[str, Any] = until or {}
terminates: bool = False
nb_failings = 0
payload = json.dumps({"command": command, "shell": shell})
headers = {"Content-Type": "application/json"}
try:
response = requests.post(self.http_server_setup_root + "/execute", 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"])
logger.info( "Command executed successfully: %s -> %s"
, " ".join(command)
, response.text
)
else:
logger.error("Failed to launch application. Status code: %s", response.text)
except requests.exceptions.RequestException as e:
logger.error("An error occurred while trying to send the request: %s", e)
traceback.print_exc()
while not terminates:
try:
response = requests.post(self.http_server_setup_root + "/execute", 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"])
logger.info( "Command executed successfully: %s -> %s"
, " ".join(command)
, response.text
)
else:
logger.error("Failed to launch application. Status code: %s", response.text)
results = None
nb_failings += 1
except requests.exceptions.RequestException as e:
logger.error("An error occurred while trying to send the request: %s", e)
traceback.print_exc()
def _command_setup(self, command: List[str], stdout: str = "", stderr: str = ""):
self._execute_setup(command, stdout, stderr)
results = None
nb_failings += 1
if len(until)==0:
terminates = True
elif results is not None:
terminates = "returncode" in until and results["returncode"]==until["returncode"]\
or "stdout" in until and until["stdout"] in results["output"]\
or "stderr" in until and until["stderr"] in results["error"]
terminates = terminates or nb_failings>=5
if not terminates:
time.sleep(0.3)
def _command_setup(self, command: List[str], **kwargs):
self._execute_setup(command, **kwargs)
def _sleep_setup(self, seconds: float):
time.sleep(seconds)
def _act_setup(self, action_seq: List[Union[Dict[str, Any], str]]):
# TODO