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

Fix bugs of a11y tree when the tag is empty

Browse Source
This commit is contained in:
Timothyxxx
2024-01-23 14:37:54 +08:00
parent 7e61ab7e53
commit 303c89411e
1 changed files with 48 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
desktop_env/server/main.py
View File

@@ -14,7 +14,7 @@ import pyautogui
import requests import requests
from PIL import Image from PIL import Image
from Xlib import display, X from Xlib import display, X
from flask import Flask, request, jsonify, send_file, abort #, send_from_directory from flask import Flask, request, jsonify, send_file, abort # , send_from_directory
from lxml.etree import _Element from lxml.etree import _Element
from pyatspi import Accessible, StateType from pyatspi import Accessible, StateType
from pyatspi import Action as ATAction from pyatspi import Action as ATAction
@@ -33,6 +33,7 @@ logger = app.logger
recording_process = None # fixme: this is a temporary solution for recording, need to be changed to support multiple-process recording_process = None # fixme: this is a temporary solution for recording, need to be changed to support multiple-process
recording_path = "/tmp/recording.mp4" recording_path = "/tmp/recording.mp4"
@app.route('/setup/execute', methods=['POST']) @app.route('/setup/execute', methods=['POST'])
@app.route('/execute', methods=['POST']) @app.route('/execute', methods=['POST'])
def execute_command(): def execute_command():
@@ -147,7 +148,7 @@ def get_terminal_output():
terminals: List[_Element] = desktop_xml.xpath(xpath, namespaces=_accessibility_ns_map) terminals: List[_Element] = desktop_xml.xpath(xpath, namespaces=_accessibility_ns_map)
output = terminals[0].text.rstrip() if len(terminals) == 1 else None output = terminals[0].text.rstrip() if len(terminals) == 1 else None
else: # windows and macos platform is not implemented currently else: # windows and macos platform is not implemented currently
#raise NotImplementedError # raise NotImplementedError
return "Currently not implemented for platform {:}.".format(platform.platform()), 500 return "Currently not implemented for platform {:}.".format(platform.platform()), 500
return jsonify({"output": output, "status": "success"}) return jsonify({"output": output, "status": "success"})
except: except:
@@ -172,11 +173,10 @@ def _create_atspi_node(node: Accessible) -> _Element:
states: List[StateType] = node.getState().get_states() states: List[StateType] = node.getState().get_states()
for st in states: for st in states:
state_name: str = StateType._enum_lookup[st] state_name: str = StateType._enum_lookup[st]
attribute_dict["{{{:}}}{:}" \ if len(state_name.split("_", maxsplit=1)[1].lower()) == 0:
.format(_accessibility_ns_map["st"] continue
, state_name.split("_", maxsplit=1)[1].lower() attribute_dict[
) "{{{:}}}{:}".format(_accessibility_ns_map["st"], state_name.split("_", maxsplit=1)[1].lower())] = "true"
] = "true"
# }}} States # # }}} States #
# Attributes {{{ # # Attributes {{{ #
@@ -185,11 +185,9 @@ def _create_atspi_node(node: Accessible) -> _Element:
attribute_name: str attribute_name: str
attribute_value: str attribute_value: str
attribute_name, attribute_value = attrbt.split(":", maxsplit=1) attribute_name, attribute_value = attrbt.split(":", maxsplit=1)
attribute_dict["{{{:}}}{:}" \ if len(attribute_name) == 0:
.format(_accessibility_ns_map["attr"] continue
, attribute_name attribute_dict["{{{:}}}{:}".format(_accessibility_ns_map["attr"], attribute_name)] = attribute_value
)
] = attribute_value
# }}} Attributes # # }}} Attributes #
# Component {{{ # # Component {{{ #
@@ -220,11 +218,9 @@ def _create_atspi_node(node: Accessible) -> _Element:
attribute_name: str attribute_name: str
attribute_value: str attribute_value: str
attribute_name, attribute_value = attrbt.split(":", maxsplit=1) attribute_name, attribute_value = attrbt.split(":", maxsplit=1)
attribute_dict["{{{:}}}{:}" \ if len(attribute_name) == 0:
.format(_accessibility_ns_map["docattr"] continue
, attribute_name attribute_dict["{{{:}}}{:}".format(_accessibility_ns_map["docattr"], attribute_name)] = attribute_value
)
] = attribute_value
# }}} Document # # }}} Document #
# Text {{{ # # Text {{{ #
@@ -277,12 +273,18 @@ def _create_atspi_node(node: Accessible) -> _Element:
, action_name , action_name
) )
] = action.getKeyBinding(i) ] = action.getKeyBinding(i)
# }}} Action # # }}} Action #
xml_node = lxml.etree.Element(node.getRoleName().replace(" ", "-") if node.getRoleName().strip() == "":
, attrib=attribute_dict node_role_name = "unknown"
, nsmap=_accessibility_ns_map else:
) node_role_name = node.getRoleName().replace(" ", "-")
xml_node = lxml.etree.Element(
node_role_name,
attrib=attribute_dict,
nsmap=_accessibility_ns_map
)
if "text" in locals() and len(text) > 0: if "text" in locals() and len(text) > 0:
xml_node.text = text xml_node.text = text
for ch in node: for ch in node:
@@ -589,7 +591,7 @@ def activate_window():
window_name = data.get('window_name', None) window_name = data.get('window_name', None)
if not window_name: if not window_name:
return "window_name required", 400 return "window_name required", 400
strict: bool = data.get("strict", False) # compare case-sensitively and match the whole string strict: bool = data.get("strict", False) # compare case-sensitively and match the whole string
by_class_name: bool = data.get("by_class", False) by_class_name: bool = data.get("by_class", False)
os_name = platform.system() os_name = platform.system()
@@ -601,11 +603,11 @@ def activate_window():
windows: List[gw.Window] = gw.getWindowsWithTitle(window_name) windows: List[gw.Window] = gw.getWindowsWithTitle(window_name)
window: Optional[gw.Window] = None window: Optional[gw.Window] = None
if len(windows)==0: if len(windows) == 0:
return "Window {:} not found (empty results)".format(window_name), 404 return "Window {:} not found (empty results)".format(window_name), 404
elif strict: elif strict:
for wnd in windows: for wnd in windows:
if wnd.title==wnd: if wnd.title == wnd:
window = wnd window = wnd
if window is None: if window is None:
return "Window {:} not found (strict mode).".format(window_name), 404 return "Window {:} not found (strict mode).".format(window_name), 404
@@ -621,11 +623,11 @@ def activate_window():
windows = gw.getWindowsWithTitle(window_name) windows = gw.getWindowsWithTitle(window_name)
window: Optional[gw.Window] = None window: Optional[gw.Window] = None
if len(windows)==0: if len(windows) == 0:
return "Window {:} not found (empty results)".format(window_name), 404 return "Window {:} not found (empty results)".format(window_name), 404
elif strict: elif strict:
for wnd in windows: for wnd in windows:
if wnd.title==wnd: if wnd.title == wnd:
window = wnd window = wnd
if window is None: if window is None:
return "Window {:} not found (strict mode).".format(window_name), 404 return "Window {:} not found (strict mode).".format(window_name), 404
@@ -638,26 +640,27 @@ def activate_window():
elif os_name == 'Linux': elif os_name == 'Linux':
# Attempt to activate VS Code window using wmctrl # Attempt to activate VS Code window using wmctrl
subprocess.run( [ "wmctrl" subprocess.run(["wmctrl"
, "-{:}{:}a".format( "x" if by_class_name else "" , "-{:}{:}a".format("x" if by_class_name else ""
, "F" if strict else "" , "F" if strict else ""
) )
, window_name , window_name
] ]
) )
else: else:
return f"Operating system {os_name} not supported.", 400 return f"Operating system {os_name} not supported.", 400
return "Window activated successfully", 200 return "Window activated successfully", 200
@app.route("/setup/close_window", methods=["POST"]) @app.route("/setup/close_window", methods=["POST"])
def close_window(): def close_window():
data = request.json data = request.json
if "window_name" not in data: if "window_name" not in data:
return "window_name required", 400 return "window_name required", 400
window_name: str = data["window_name"] window_name: str = data["window_name"]
strict: bool = data.get("strict", False) # compare case-sensitively and match the whole string strict: bool = data.get("strict", False) # compare case-sensitively and match the whole string
by_class_name: bool = data.get("by_class", False) by_class_name: bool = data.get("by_class", False)
os_name: str = platform.system() os_name: str = platform.system()
@@ -669,11 +672,11 @@ def close_window():
windows: List[gw.Window] = gw.getWindowsWithTitle(window_name) windows: List[gw.Window] = gw.getWindowsWithTitle(window_name)
window: Optional[gw.Window] = None window: Optional[gw.Window] = None
if len(windows)==0: if len(windows) == 0:
return "Window {:} not found (empty results)".format(window_name), 404 return "Window {:} not found (empty results)".format(window_name), 404
elif strict: elif strict:
for wnd in windows: for wnd in windows:
if wnd.title==wnd: if wnd.title == wnd:
window = wnd window = wnd
if window is None: if window is None:
return "Window {:} not found (strict mode).".format(window_name), 404 return "Window {:} not found (strict mode).".format(window_name), 404
@@ -681,14 +684,14 @@ def close_window():
window = windows[0] window = windows[0]
window.close() window.close()
elif os_name == "Linux": elif os_name == "Linux":
subprocess.run( [ "wmctrl" subprocess.run(["wmctrl"
, "-{:}{:}c".format( "x" if by_class_name else "" , "-{:}{:}c".format("x" if by_class_name else ""
, "F" if strict else "" , "F" if strict else ""
) )
, window_name , window_name
] ]
) )
elif os_name=="Darwin": elif os_name == "Darwin":
import pygetwindow as gw import pygetwindow as gw
return "Currently not supported on macOS.", 500 return "Currently not supported on macOS.", 500
else: else:
@@ -696,6 +699,7 @@ def close_window():
return "Window closed successfully.", 200 return "Window closed successfully.", 200
@app.route('/start_recording', methods=['POST']) @app.route('/start_recording', methods=['POST'])
def start_recording(): def start_recording():
global recording_process global recording_process
@@ -722,7 +726,7 @@ def end_recording():
recording_process.terminate() recording_process.terminate()
recording_process.wait() recording_process.wait()
#return_code = recording_process.returncode # return_code = recording_process.returncode
output, error = recording_process.communicate() output, error = recording_process.communicate()
recording_process = None recording_process = None