{text_obs}\n"
+
+ if render_screenshot:
+ # image observation
+ img_obs = observation["image"]
+ image = Image.fromarray(img_obs)
+ byte_io = io.BytesIO()
+ image.save(byte_io, format="PNG")
+ byte_io.seek(0)
+ image_bytes = base64.b64encode(byte_io.read())
+ image_str = image_bytes.decode("utf-8")
+ new_content += f"
\n"
+
+ # meta data
+ new_content += f"[\W\w]*?', '', ctx)
+ ctx = re.sub('[\W\w]*?', '', ctx)
+ ctx = '' if ctx is None else re.sub(r'\s+', ' ', ctx).strip()
+ dom_tree = html.fromstring(ctx.encode('utf-8'))
+ match = re.search(' html.HtmlElement:
+ node = tree.xpath('//*')[0]
+ while True:
+ parent = node.getparent()
+ if parent is None:
+ break
+ node = parent
+ return node
+
+ def get_node_by_bid(self, tree: html.HtmlElement, bid: str) -> html.HtmlElement:
+ nodes = tree.xpath(f'//*[@{self.id_attr}="{bid}"]')
+ if len(nodes) == 0:
+ return None
+ return nodes[0]
+
+ def id_label_converter(self, label: str) -> str:
+ return self.bids2label.get(label, '')
+
+ def id_xpath_converter(self, label: str) -> str:
+ return self.bids2xpath.get(label, '')
+
+ def mark_id(self) -> None:
+ root = self.get_root(self.dom_tree)
+ _, i2xpath, used_labels = get_xpath_top_down(root, self.id_attr, self.label_attr)
+ self.used_labels = used_labels
+ self.bids2xpath = i2xpath
+
+ def parse(self, root: html.HtmlElement, keep: list[str], obs: list[str], parent_chain: bool=False, get_new_label: bool=False) -> dict[str]:
+ def get_text(str: str) -> str:
+ return '' if str is None else str.strip()[:500]
+
+ def check_attr(attr: str, node: html.HtmlElement) -> bool:
+ tag = node.tag
+ if (
+ ( attr == 'role' and node.attrib.get(attr, '') in ['presentation', 'none', 'link'] )
+ or ( attr == 'type' and node.attrib.get(attr, '') == 'hidden' )
+ or ( attr == 'aria-hidden' and node.attrib.get(attr, '') == 'true')
+ # or ( attr == 'value' and tag in ['option'] )
+ ):
+ return False
+ return True
+
+ def is_visible(node: html.HtmlElement, bid: str) -> bool:
+ if self.dataset == 'mind2web':
+ bound = node.attrib.get('bounding_box_rect', None)
+ self.rect[bid] = rect2tuple(bound)
+
+ if self.dataset == 'pipeline':
+ bound = node.attrib.get('data-bbox', None)
+ self.rect[bid] = rect2tuple(bound)
+
+ if node.attrib.get('aria-hidden', 'false') == 'true':
+ return False
+
+ if not self.use_position:
+ return True
+
+ rect = self.rect.get(bid, None)
+ if rect is None:
+ return False
+
+ if self.window_size is None:
+ return True
+
+ # get window size
+ wx, wy, ww, wh = self.window_size
+ wx, wy = 0, 0
+ x, y, w, h = rect
+ if x + w < wx or x > wx + ww or y + h < wy or y > wy + wh:
+ return False
+
+ return True
+
+ def _dfs(node: html.HtmlElement, keep: list[str]=[], obs: list[str]=[],
+ parent_chain: bool=False, get_new_label: bool=False, par_keep: bool=False) -> (str, dict[str]):
+ # basic information
+ bid = node.attrib.get(self.id_attr, '')
+ tag = node.tag
+ label = node.attrib.get(self.label_attr, '')
+
+ # element which is keeped equivalent to visible
+ visible = is_visible(node, bid)
+ in_keep_list = bid in keep
+ in_obs_list = (bid in obs or len(label) > 0) and visible
+ par_keep = par_keep and tag == "option"
+ keep_element = in_keep_list or in_obs_list or visible or par_keep
+
+ if label:
+ keep_element = True
+
+ # mark label
+ bids2label, labeled_elems = {}, []
+ have_label = False
+ if in_keep_list or in_obs_list:
+ if label is None or len(label) == 0 or get_new_label:
+ label = self.identifier.generate()
+ node.attrib[self.label_attr] = label
+ bids2label[bid] = label
+ bids2label[label] = bid
+ have_label = True
+
+ # get text or alt_text of current element
+ text = get_text(node.text)
+
+ classes = {}
+ # keep attributes if needed
+ keep_all_attrs = len(self.keep_attrs) == 0
+ keep_attrs = node.attrib.keys() if keep_all_attrs else self.keep_attrs
+
+ # traverse attributes
+ for attr in keep_attrs:
+ if attr not in node.attrib or not check_attr(attr, node):
+ continue
+ if attr in [self.id_attr, self.label_attr]:
+ continue
+ val = get_text(node.attrib[attr])
+ if len(val) > 0 or keep_all_attrs:
+ classes[attr] = val
+
+ have_text = len(text) > 0 or len(classes) - (1 if 'data-bbox' in classes else 0) > 0
+ par_keep = keep_element and tag == 'select'
+
+ parts = []
+ clickable_count = 0
+ children = node.getchildren()
+ for child in children:
+ cres, cmsg = _dfs(child, keep, obs, parent_chain, get_new_label, par_keep)
+ clickable_count += 1 if cmsg.get('have_clickable', False) else 0
+ bids2label.update(cmsg.get('bids2label', {}))
+ labeled_elems.extend(cmsg.get('label_element', []))
+ if len(cres) != 0:
+ parts.append(cres)
+
+ dom = self.prompt.subtree_constructor(parts)
+
+ # remove if all children are text
+ keep_as_all_text = (dom.count('<') == dom.count(' 0
+ if keep_as_all_text:
+ matches = re.findall(r']+) >', dom)
+ dom = self.prompt.subtree_constructor(matches)
+
+ keep_element = keep_element and (clickable_count > 1 or have_text or have_label or keep_as_all_text)
+ keep_as_parent = len(dom) > 0 and parent_chain
+ if in_keep_list or keep_element or keep_as_parent:
+ dom = self.prompt.prompt_constructor(tag, label, text, dom, classes)
+
+ if have_label:
+ labeled_elems.append(bid)
+
+ control_msg = {
+ 'have_clickable': bool(clickable_count or have_text),
+ 'bids2label': bids2label,
+ 'label_element': labeled_elems,
+ }
+
+ return dom, control_msg
+
+ dom, cmsg = _dfs(root, keep, obs, parent_chain, get_new_label)
+ return dom, cmsg
+
+ def parse_tree(self) -> dict[str]:
+ # start from here
+ stt = time.time()
+ root = self.get_root(self.dom_tree)
+ dom, cmsg = self.parse(root, self.keep, self.obs, self.parent_chain, self.get_new_label)
+ self.bids2label = cmsg.get('bids2label', {})
+ self.keep = list(set(self.keep + cmsg.get('label_element', [])))
+
+ obj = {
+ 'html': dom,
+ 'parse_time': time.time() - stt
+ }
+
+ return obj
+
+ # From mind2web, https://github.com/OSU-NLP-Group/Mind2Web/blob/main/src/data_utils/dom_utils.py
+ def get_keep_elements(self, tree: html.HtmlElement, keep: list[str], max_depth: int, max_children: int,
+ max_sibling: int, dfs_count: int=1, keep_parent: bool=False) -> list[str]:
+ def get_anscendants(node: html.HtmlElement, max_depth: int, current_depth: int=0) -> list[str]:
+ if current_depth > max_depth:
+ return []
+
+ anscendants = []
+ parent = node.getparent()
+ if parent is not None:
+ anscendants.append(parent)
+ anscendants.extend(get_anscendants(parent, max_depth, current_depth + 1))
+
+ return anscendants
+
+ def get_descendants(node: html.HtmlElement, max_depth: int, current_depth: int=0) -> list[str]:
+ if current_depth > max_depth:
+ return []
+
+ descendants = []
+ for child in node:
+ descendants.append(child)
+ descendants.extend(get_descendants(child, max_depth, current_depth + 1))
+
+ return descendants
+
+ to_keep = set(copy.deepcopy(keep))
+ nodes_to_keep = set()
+
+ for _ in range(max(1, dfs_count)):
+ for bid in to_keep:
+ candidate_node = self.get_node_by_bid(tree, bid)
+ if candidate_node is None:
+ continue
+
+ nodes_to_keep.add(candidate_node.attrib[self.id_attr])
+ # get all ancestors or with max depth
+ nodes_to_keep.update([x.attrib.get(self.id_attr, '') for x in get_anscendants(candidate_node, max_depth)])
+
+ # get descendants with max depth
+ nodes_to_keep.update([x.attrib.get(self.id_attr, '') for x in get_descendants(candidate_node, max_depth)][:max_children])
+ # get siblings within range
+ parent = candidate_node.getparent()
+ if parent is None:
+ continue
+
+ siblings = [x for x in parent.getchildren() if x.tag != 'text']
+ if candidate_node not in siblings:
+ continue
+
+ idx_in_sibling = siblings.index(candidate_node)
+ nodes_to_keep.update([x.attrib.get(self.id_attr, '')
+ for x in siblings[max(0, idx_in_sibling - max_sibling) : idx_in_sibling + max_sibling + 1]])
+
+ max_children = int(max_children * 0.5)
+ max_depth = int(max_depth * 0.5)
+ max_sibling = int(max_sibling * 0.7)
+
+ to_keep = copy.deepcopy(nodes_to_keep)
+
+ if keep_parent:
+ for bid in keep:
+ candidate_node = self.get_node_by_bid(tree, bid)
+ if candidate_node is None:
+ continue
+ nodes_to_keep.update([x.attrib.get(self.id_attr, '') for x in candidate_node.xpath("ancestor::*")])
+
+ return list(nodes_to_keep)
+
+ def prune(self, tree: html.HtmlElement, nodes_to_keep: list[str]) -> html.HtmlElement:
+ # remove nodes not in nodes_to_keep
+ for node in tree.xpath('//*')[::-1]:
+ if node.tag != 'text':
+ is_keep = node.attrib.get(self.id_attr, '') in nodes_to_keep
+ is_candidate = node.attrib.get(self.id_attr, '') in self.keep
+ else:
+ is_keep = (node.getparent().attrib.get(self.id_attr, '') in nodes_to_keep)
+ is_candidate = (node.getparent().attrib.get(self.id_attr, '') in self.keep)
+
+ if not is_keep and node.getparent() is not None:
+ # insert all children into parent
+ for child in node.getchildren():
+ node.addprevious(child)
+ node.getparent().remove(node)
+ else:
+ # if not is_candidate or node.tag == 'text':
+ # node.attrib.pop(self.id_attr, None)
+ if (
+ len(node.attrib) == 0
+ and not any([x.tag == 'text' for x in node.getchildren()])
+ and node.getparent() is not None
+ and node.tag != "text"
+ and len(node.getchildren()) <= 1
+ ):
+ # insert all children into parent
+ for child in node.getchildren():
+ node.addprevious(child)
+ node.getparent().remove(node)
+
+ return tree
+
+ def prune_tree(self, dfs_count: int=1, max_depth: int=3, max_children: int=30,
+ max_sibling: int=3, keep_parent: bool=False) -> None:
+ # clone the tree
+ new_tree = copy.deepcopy(self.dom_tree)
+ nodes_to_keep = self.get_keep_elements(new_tree, self.keep, max_depth, max_children, max_sibling, dfs_count, keep_parent)
+ new_tree = self.prune(new_tree, nodes_to_keep)
+
+ self.dom_tree = new_tree
+
+ def get_segment(self, bid: str) -> str:
+ # clone the tree
+ new_tree = copy.deepcopy(self.dom_tree)
+ nodes_to_keep = self.get_keep_elements(new_tree, [bid], 0, 2, 1)
+ new_tree = self.prune(new_tree, nodes_to_keep)
+ dom, _ = self.parse(new_tree, self.keep, [], False)
+ return dom
+
+ def get_rect_data(self, bids: list[str]) -> list[dict[str]]:
+ res = []
+ for bid in bids:
+ label = self.bids2label.get(bid, '')
+ rect = self.rect.get(bid, None)
+ res.append({
+ 'bid': bid,
+ 'label': label,
+ 'rect': rect
+ })
+ return res
+
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/identifier.py b/VAB-WebArena-Lite/new/html_tools/identifier.py
new file mode 100755
index 0000000..793ebed
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/identifier.py
@@ -0,0 +1,64 @@
+import secrets
+
+class IdentifierTool:
+ def __init__(self, method: str='order', existing_labels: dict[str]={}) -> None:
+ self.methods = {
+ 'order': self.get_identifier_in_order,
+ 'random': self.get_random_identifier,
+ }
+
+ if method is None:
+ method = 'order'
+
+ self.func = self.methods.get(method, None)
+ self.name = method
+ if self.func is None:
+ raise ValueError(f'Invalid method for identifier: {method}')
+
+ self.reset(existing_labels)
+
+ def reset(self, exists: dict[str]={}) -> None:
+ self.identifier = -1
+ self.exists = {} if exists is None else exists
+
+ def get_identifier_in_order(self) -> str:
+ def id2str(id: int) -> str:
+ if id < 26:
+ return chr(id + 65)
+ id -= 26
+ c0 = id // 676
+ c1 = (id // 26) % 26
+ c2 = id % 26
+ label = f'{chr(c1 + 65)}{chr(c2 + 65)}'
+ return label if c0 == 0 else f'{chr(c0 + 64)}{label}'
+
+ self.identifier += 1
+ label = id2str(self.identifier)
+
+ while label in self.exists:
+ self.identifier += 1
+ label = id2str(self.identifier)
+
+ self.exists[label] = True
+ return label
+
+ def get_random_identifier(self) -> str:
+ secret_generator = secrets.SystemRandom()
+
+ def get_random_label(n: int=2) -> str:
+ tmp = ''
+ for _ in range(n):
+ tmp += chr(secret_generator.randint(65, 90))
+ return tmp
+
+ wc = 3 if len(self.exists) > 280 else 2
+
+ label = get_random_label(wc)
+ while label in self.exists:
+ label = get_random_label(wc)
+
+ self.exists[label] = True
+ return label
+
+ def generate(self):
+ return self.func()
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/prompt.py b/VAB-WebArena-Lite/new/html_tools/prompt.py
new file mode 100755
index 0000000..38d6b94
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/prompt.py
@@ -0,0 +1,97 @@
+from .configs import prompts
+
+class HtmlPrompt:
+ def __init__(self, prompt: str='') -> None:
+ prompt = self.extract(prompt, 'xml')
+ if prompt not in prompts:
+ raise Exception('Unknown prompt: ' + prompt)
+
+ constructors = {
+ 'refine': self.normal_prompt_constructor,
+ 'xml': self.normal_prompt_constructor,
+ 'new_data': self.new_data_prompt_constructor,
+ }
+
+ self.name = prompt
+ self.prompt = prompts[prompt]
+ self.constructor = constructors[prompt]
+
+ @staticmethod
+ def extract(data, default=''):
+ return data if data is not None else default
+
+ def subtree_constructor(self, subtree: list[str]=[]) -> str:
+ return self.prompt['subtree_splitter'].join(subtree)
+
+ def normal_prompt_constructor(self, tag: str='', label: str='', content: str='', subtree_str: str='', class_dict: dict[str]={}) -> str:
+ def add_prefix(data, prefix):
+ return prefix + data if len(data) > 0 else ''
+
+ tag = self.extract(tag)
+ label = self.extract(label)
+ content = self.extract(content)
+ subtree_str = self.extract(subtree_str, '')
+ class_dict = self.extract(class_dict, {})
+
+ label_str = ''
+ if len(label) > 0:
+ label_str = self.prompt['label'].format(label=label)
+
+ classes = []
+ values = set()
+ for key, val in class_dict.items():
+ if val in values:
+ continue
+ values.add(val)
+ classes.append(self.prompt['attr'].format(key=key, attr=val))
+ classes_str = self.prompt['attr_splitter'].join(classes)
+
+ content_splitter = ' ' if len(classes_str) == 0 else self.prompt['attr_splitter']
+ classes_str = add_prefix(classes_str, ' ')
+ content_str = add_prefix(content, content_splitter)
+ subtree_str = add_prefix(subtree_str, ' ')
+
+ return self.prompt['dom'].format(tag=tag, label=label_str, attr=classes_str, content=content_str, subtree=subtree_str)
+
+ def new_data_prompt_constructor(self, tag: str='', label: str='', content: str='', subtree_str: str='', class_dict: dict[str]={}) -> str:
+ def add_prefix(data, prefix):
+ return prefix + data if len(data) > 0 else ''
+
+ tag = self.extract(tag)
+ label = self.extract(label)
+ content = self.extract(content)
+ subtree_str = self.extract(subtree_str, '')
+ class_dict = self.extract(class_dict, {})
+
+ label_str = ''
+ if len(label) > 0:
+ label_str = self.prompt['label'].format(label=label)
+
+ classes = []
+ values = set()
+
+ message = []
+ for key, val in class_dict.items():
+ if val == '':
+ message.append(key)
+ continue
+ if val in values:
+ continue
+ values.add(val)
+ classes.append(self.prompt['attr'].format(key=key, attr=val))
+
+ if len(message) > 0:
+ message_str = ' '.join(message)
+ classes.append(self.prompt['attr'].format(key='message', attr=message_str))
+
+ classes_str = self.prompt['attr_splitter'].join(classes)
+
+ content_splitter = ' ' if len(classes_str) == 0 else self.prompt['attr_splitter']
+ classes_str = add_prefix(classes_str, ' ')
+ content_str = add_prefix(content, content_splitter)
+ subtree_str = add_prefix(subtree_str, ' ')
+
+ return self.prompt['dom'].format(tag=tag, label=label_str, attr=classes_str, content=content_str, subtree=subtree_str)
+
+ def prompt_constructor(self, tag: str='', label: str='', content: str='', subtree_str: str='', class_dict: dict[str]={}) -> str:
+ return self.constructor(tag, label, content, subtree_str, class_dict)
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/__init__.py b/VAB-WebArena-Lite/new/html_tools/scripts/__init__.py
new file mode 100755
index 0000000..0d36c9a
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/__init__.py
@@ -0,0 +1,43 @@
+import os
+from pathlib import Path
+rootdir = Path(__file__).parent
+
+with open(os.path.join(rootdir,'prepare.js'), 'r') as f:
+ prepare_script = f.read()
+
+with open(os.path.join(rootdir, 'clickable_checker.js'), 'r') as f:
+ clickable_checker_script = f.read()
+
+with open(os.path.join(rootdir, 'label.js'), 'r') as f:
+ label_script = f.read()
+
+with open(os.path.join(rootdir, 'element_info.js'), 'r') as f:
+ element_info_script = f.read()
+
+# draw label on page
+with open(os.path.join(rootdir, 'label_marker.js'), 'r') as f:
+ label_marker_script = f.read()
+
+# remove label draw on page
+remove_label_mark_script = """
+ () => {
+ document.querySelectorAll(".our-dom-marker").forEach(item => {
+ document.body.removeChild(item);
+ });
+ }
+"""
+
+remove_id_script = """
+ () => {
+ Array.from(document.getElementsByClassName('possible-clickable-element')).forEach((element) => {
+ element.classList.remove('possible-clickable-element');
+ element.removeAttribute('data-value');
+ element.removeAttribute('data-text');
+ element.removeAttribute('data-label');
+ element.removeAttribute('data-bbox');
+ element.removeAttribute('data-status');
+ element.removeAttribute('data-backend-node-id');
+ element.removeAttribute('data-label-id');
+ });
+ }
+"""
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/clickable_checker.js b/VAB-WebArena-Lite/new/html_tools/scripts/clickable_checker.js
new file mode 100755
index 0000000..42267f0
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/clickable_checker.js
@@ -0,0 +1,148 @@
+() => {
+ var items = Array.prototype.slice.call(
+ document.querySelectorAll('*')
+ ).map(function(element) {
+ var vw = Math.max(document.documentElement.clientWidth || 0, window.innerWidth || 0);
+ var vh = Math.max(document.documentElement.clientHeight || 0, window.innerHeight || 0);
+
+ var rects = [...element.getClientRects()].filter(bb => {
+ var center_x = bb.left + bb.width / 2;
+ var center_y = bb.top + bb.height / 2;
+ var elAtCenter = document.elementFromPoint(center_x, center_y);
+
+ if (!elAtCenter) return false;
+ return elAtCenter === element || element.contains(elAtCenter)
+ }).map(bb => {
+ const rect = {
+ left: Math.max(0, bb.left),
+ top: Math.max(0, bb.top),
+ right: Math.min(vw, bb.right),
+ bottom: Math.min(vh, bb.bottom)
+ };
+ return {
+ ...rect,
+ width: rect.right - rect.left,
+ height: rect.bottom - rect.top
+ }
+ });
+ // var rects = [];
+ var area = rects.reduce((acc, rect) => acc + rect.width * rect.height, 0);
+
+ const tagName = element.tagName.toLowerCase?.() || "";
+ let isClickable = ((element.onclick != null) || window.getComputedStyle(element).cursor == "pointer");
+
+ // Insert area elements that provide click functionality to an img.
+ if (tagName === "img") {
+ let mapName = element.getAttribute("usemap");
+ if (mapName) {
+ const imgClientRects = element.getClientRects();
+ mapName = mapName.replace(/^#/, "").replace('"', '\\"');
+ const map = document.querySelector(`map[name=\"${mapName}\"]`);
+ if (map && (imgClientRects.length > 0)) isClickable = true;
+ }
+ }
+
+ if (!isClickable) {
+ const role = element.getAttribute("role");
+ const clickableRoles = [
+ "button",
+ "tab",
+ "link",
+ "checkbox",
+ "menuitem",
+ "menuitemcheckbox",
+ "menuitemradio",
+ "radio",
+ ];
+ if (role != null && clickableRoles.includes(role.toLowerCase())) {
+ isClickable = true;
+ } else {
+ const contentEditable = element.getAttribute("contentEditable");
+ if (
+ contentEditable != null &&
+ ["", "contenteditable", "true"].includes(contentEditable.toLowerCase())
+ ) {
+ isClickable = true;
+ }
+ }
+ }
+
+ // Check for jsaction event listeners on the element.
+ if (!isClickable && element.hasAttribute("jsaction")) {
+ const jsactionRules = element.getAttribute("jsaction").split(";");
+ for (let jsactionRule of jsactionRules) {
+ const ruleSplit = jsactionRule.trim().split(":");
+ if ((ruleSplit.length >= 1) && (ruleSplit.length <= 2)) {
+ const [eventType, namespace, actionName] = ruleSplit.length === 1
+ ? ["click", ...ruleSplit[0].trim().split("."), "_"]
+ : [ruleSplit[0], ...ruleSplit[1].trim().split("."), "_"];
+ if (!isClickable) {
+ isClickable = (eventType === "click") && (namespace !== "none") && (actionName !== "_");
+ }
+ }
+ }
+ }
+
+ if (!isClickable) {
+ const clickableTags = [
+ "input",
+ "textarea",
+ "select",
+ "button",
+ "a",
+ "iframe",
+ "video",
+ "object",
+ "embed",
+ "details"
+ ];
+ isClickable = clickableTags.includes(tagName);
+ }
+
+ if (!isClickable) {
+ if (tagName === "label")
+ isClickable = (element.control != null) && !element.control.disabled;
+ else if (tagName === "img")
+ isClickable = ["zoom-in", "zoom-out"].includes(element.style.cursor);
+ }
+
+ // An element with a class name containing the text "button" might be clickable. However, real
+ // clickables are often wrapped in elements with such class names. So, when we find clickables
+ // based only on their class name, we mark them as unreliable.
+ const className = element.getAttribute("class");
+ if (!isClickable && className && className.toLowerCase().includes("button")) {
+ isClickable = true;
+ }
+
+ // Elements with tabindex are sometimes useful, but usually not. We can treat them as second
+ // class citizens when it improves UX, so take special note of them.
+ const tabIndexValue = element.getAttribute("tabindex");
+ const tabIndex = tabIndexValue ? parseInt(tabIndexValue) : -1;
+ if (!isClickable && !(tabIndex < 0) && !isNaN(tabIndex)) {
+ isClickable = true;
+ }
+
+ const idValue = element.getAttribute("id");
+ const id = idValue ? idValue.toLowerCase() : "";
+ if (isClickable && area == 0) {
+ const textValue = element.textContent.trim().replace(/\s{2,}/g, ' ');
+ clickable_msg = `${tagName}[id=${id}] ${isClickable} (${area}) ${textValue}`
+ }
+
+ return {
+ element: element,
+ include: isClickable,
+ area,
+ rects,
+ text: element.textContent.trim().replace(/\s{2,}/g, ' ')
+ };
+ }).filter(item =>
+ item.include && (item.area >= 1)
+ );
+
+ items = items.filter(x => !items.some(y => x.element.contains(y.element) && !(x == y)))
+
+ items.forEach(item => {
+ item.element.classList.add('possible-clickable-element');
+ });
+}
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/element_info.js b/VAB-WebArena-Lite/new/html_tools/scripts/element_info.js
new file mode 100755
index 0000000..c396220
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/element_info.js
@@ -0,0 +1,34 @@
+() => {
+ function getElementInfo(element) {
+ return {
+ "bid": element.getAttribute("data-backend-node-id") || "",
+ "label": element.getAttribute("data-label-id") || "",
+ "tag": element.tagName.toLowerCase?.() || "",
+ "area": JSON.parse("[" + (element.getAttribute("data-bbox") || "") + "]"),
+ "text": element.innerText?.trim().replace(/\s{2,}/g, " ") || "",
+ "id": element.getAttribute("id") || "",
+ "role": element.getAttribute("role") || "",
+ "aria-label": element.getAttribute("aria-label") || "",
+ "href": element.getAttribute("href") || "",
+ };
+ }
+
+ var all_items = Array.prototype.slice.call(
+ document.querySelectorAll("*")
+ ).map((element) => {
+ return getElementInfo(element);
+ });
+
+ var clickable_items = Array.prototype.slice.call(
+ document.querySelectorAll("*")
+ ).filter(
+ element => element.getAttribute("data-label-id")
+ ).map((element) => {
+ return getElementInfo(element);
+ });
+
+ return {
+ all_elements: all_items,
+ clickable_elements: clickable_items
+ };
+}
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/label.js b/VAB-WebArena-Lite/new/html_tools/scripts/label.js
new file mode 100755
index 0000000..285ca89
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/label.js
@@ -0,0 +1,74 @@
+(packet) => {
+ function int2str(index) {
+ var str = "";
+ while (index >= 0) {
+ str = String.fromCharCode(65 + index % 26) + str;
+ index = Math.floor(index / 26) - 1;
+ }
+ return str;
+ };
+
+ selector = packet.selector
+ index = packet.startIndex
+ var items = Array.prototype.slice.call(
+ document.querySelectorAll(selector)
+ );
+
+ var vw = Math.max(document.documentElement.clientWidth || 0, window.innerWidth || 0);
+ var vh = Math.max(document.documentElement.clientHeight || 0, window.innerHeight || 0);
+
+ items = items.filter(
+ x => !items.some(y => x.contains(y) && !(x == y))
+ ).map(element => {
+ var bb = element.getClientRects();
+ var rect = {
+ left: 0,
+ top: 0,
+ right: 0,
+ bottom: 0,
+ width: 0,
+ height: 0
+ };
+ var keep = false;
+ var text = "", id = -1;
+ if (bb.length > 0) {
+ bb = bb[0];
+ rect = {
+ left: Math.max(0, bb.left),
+ top: Math.max(0, bb.top),
+ right: Math.min(vw, bb.right),
+ bottom: Math.min(vh, bb.bottom)
+ };
+ rect = {
+ ...rect,
+ width: rect.right - rect.left,
+ height: rect.bottom - rect.top
+ };
+ if (rect.width > 0 || rect.height > 0) {
+ keep = true;
+ if (index >= 0) {
+ // id = int2str(index++);
+ id = index++;
+ element.setAttribute("data-label-id", id);
+ }
+ var childNodes = element.childNodes;
+
+ for (var i = 0; i < childNodes.length; i++) {
+ if (childNodes[i].nodeType == Node.TEXT_NODE) {
+ text += childNodes[i].textContent;
+ }
+ }
+ }
+ }
+
+ return {
+ keep: true,
+ id,
+ rects: rect,
+ tag: element.tagName.toLowerCase?.() || "",
+ text,//: element.innerText?.trim().replace(/\s{2,}/g, " ") || ""
+ };
+ }).filter(x => x.keep);
+
+ return [items, index];
+}
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/label_marker.js b/VAB-WebArena-Lite/new/html_tools/scripts/label_marker.js
new file mode 100755
index 0000000..6c55af5
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/label_marker.js
@@ -0,0 +1,65 @@
+(items) => {
+ function getRandomColor() {
+ var letters = '0123456789ABCDEF';
+ var color = '#';
+ for (var i = 0; i < 6; i++) {
+ color += letters[Math.floor(Math.random() * 16)];
+ }
+ return color;
+ }
+
+ items.filter(
+ item => item.id != ""
+ ).forEach((item) => {
+ const bbox = item.rects;
+ const id_string = `dom-marker-id-${index}`;
+
+ index = item.id;
+
+ outerElement = document.createElement("div");
+ outerElement.classList.add("our-dom-marker");
+ // var borderColor = getRandomColor();
+ var borderColor = "#FFFF00";
+ outerElement.style.outline = `2px dashed ${borderColor}`;
+ outerElement.style.position = "fixed";
+ outerElement.style.left = bbox.left - 2 + "px";
+ outerElement.style.top = bbox.top - 2 + "px";
+ outerElement.style.width = bbox.width + 4 + "px";
+ outerElement.style.height = bbox.height + 4 + "px";
+ outerElement.style.pointerEvents = "none";
+ outerElement.style.boxSizing = "border-box";
+ outerElement.style.zIndex = 2147483647;
+
+ innerElement = document.createElement("div");
+ innerElement.classList.add("our-dom-marker");
+ innerElement.style.outline = `2px dashed #222288`;
+ innerElement.style.position = "fixed";
+ innerElement.style.left = bbox.left + "px";
+ innerElement.style.top = bbox.top + "px";
+ innerElement.style.width = bbox.width + "px";
+ innerElement.style.height = bbox.height + "px";
+ innerElement.style.pointerEvents = "none";
+ innerElement.style.boxSizing = "border-box";
+ innerElement.style.zIndex = 2147483647;
+
+ // Add floating label at the corner
+ var label = document.createElement("span");
+ var topPosition = 25;
+ if (bbox.top < 25) topPosition = bbox.top;
+ label.textContent = index;
+ label.style.position = "absolute";
+ label.style.top = `-${topPosition}px`;
+ label.style.left = "0px";
+ label.style.background = borderColor;
+ label.style.color = "black";
+ label.style.padding = "2px 4px";
+ label.style.fontSize = "16px";
+ label.style.borderRadius = "2px";
+ label.style.fontWeight = "bold";
+ outerElement.appendChild(label);
+
+ document.body.appendChild(outerElement);
+ document.body.appendChild(innerElement);
+ })
+ return items;
+}
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/scripts/prepare.js b/VAB-WebArena-Lite/new/html_tools/scripts/prepare.js
new file mode 100755
index 0000000..9e76a62
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/scripts/prepare.js
@@ -0,0 +1,83 @@
+() => {
+ // mark backend node id
+ var vw = Math.max(document.documentElement.clientWidth || 0, window.innerWidth || 0);
+ var vh = Math.max(document.documentElement.clientHeight || 0, window.innerHeight || 0);
+
+ var backendId = 0;
+ Array.prototype.slice.call(
+ document.querySelectorAll("*")
+ ).forEach((element) => {
+ element.setAttribute("data-backend-node-id", backendId);
+ backendId++;
+
+ var tag = element.tagName.toLowerCase?.() || "";
+ var bb = element.getClientRects();
+ var rect = {
+ left: 0,
+ top: 0,
+ right: 0,
+ bottom: 0,
+ width: 0,
+ height: 0
+ };
+
+ if (bb.length > 0) {
+ bb = bb[0];
+ // rect = {
+ // left: Math.round(Math.max(0, bb.left) * 100) / 100,
+ // top: Math.round(Math.max(0, bb.top) * 100) / 100,
+ // right: Math.round(Math.min(vw, bb.right) * 100) / 100,
+ // bottom: Math.round(Math.min(vh, bb.bottom) * 100) / 100
+ // };
+ rect = {
+ left: (Math.round(bb.left) * 100) / 100,
+ top: (Math.round(bb.top) * 100) / 100,
+ right: (Math.round(bb.right) * 100) / 100,
+ bottom: (Math.round(bb.bottom) * 100) / 100
+ };
+ rect = {
+ ...rect,
+ width: Math.round((rect.right - rect.left) * 100) / 100,
+ height: Math.round((rect.bottom - rect.top) * 100) / 100
+ };
+
+ element.setAttribute("data-bbox", `${rect.left},${rect.top},${rect.width},${rect.height}`);
+ }
+
+ if (element.hasChildNodes()) {
+ let children = Array.prototype.slice.call(element.childNodes);
+ var texts = children.filter(
+ (node) => node.nodeType == Node.TEXT_NODE
+ ).map(
+ (node) => node.textContent.trim().replace(/\s{2,}/g, " ") || ""
+ ).filter(
+ (text) => text.length > 0
+ )
+ element.setAttribute("data-text", texts.join(","));
+ }
+
+ // fix select issue
+ if (tag == "select") {
+ var value = element.value;
+ var text = element.options[element.selectedIndex]?.text || "";
+ element.setAttribute("data-value", value);
+ element.setAttribute("data-text", text);
+ element.options[element.selectedIndex]?.setAttribute("data-status", "selected");
+ }
+
+ if (tag == "input") {
+ var input_type = element.getAttribute("type") || "";
+ if (input_type == "checkbox") {
+ var status = element.checked? "checked" : "not-checked";
+ element.setAttribute("data-status", status);
+ }
+ }
+ });
+
+ // fix input and textarea issue
+ Array.prototype.slice.call(
+ document.querySelectorAll("input, textarea")
+ ).forEach(element => {
+ element.setAttribute("data-value", element.value);
+ });
+}
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/utils.py b/VAB-WebArena-Lite/new/html_tools/utils.py
new file mode 100755
index 0000000..9e0be65
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/utils.py
@@ -0,0 +1,101 @@
+from lxml import html
+def get_xpath_top_down(element: html.HtmlElement, id_column: str='temp_id', label_column: str='temp_clickable_label', path: str='', order: int=0,
+ in_svg: bool=False, temp_id: int=0) -> tuple[int, dict[str, str], dict[str]]:
+ used_labels, i2xpath = {}, {}
+ # path
+ tag = element.tag.lower()
+ in_svg = in_svg or (tag == 'svg')
+
+ if not in_svg and 'id' in element.attrib:
+ node_id = element.attrib['id']
+ path = f'//*[@id="{node_id}"]'
+ else:
+ suffix = f'[{order}]' if order > 0 else ''
+ prefix = f'*[name()="{tag}"]' if in_svg else tag
+ path = path + '/' + prefix + suffix
+
+ # add temp id
+ element.attrib[id_column] = str(temp_id)
+ ori_label = element.attrib.get(label_column, '')
+ if ori_label != '':
+ used_labels[ori_label] = True
+
+ bid = str(temp_id)
+ i2xpath[bid] = path
+ i2xpath[path] = bid
+ i2xpath[f'xpath/{path}'] = bid
+ i2xpath[f'xpath=/{path}'] = bid
+
+ temp_id += 1
+
+ # traverse node
+ children = element.getchildren()
+ tag_dict = {}
+ id_list = []
+ for child in children:
+ ctag = child.tag.lower()
+ if ctag not in tag_dict:
+ tag_dict[ctag] = 0
+ tag_dict[ctag] += 1
+ id_list.append(tag_dict[ctag])
+
+ for cid, child in zip(id_list, children):
+ ctag = child.tag.lower()
+ cod = cid if tag_dict[ctag] > 1 else 0
+ temp_id, i2x, ulabels = get_xpath_top_down(child, id_column, label_column, path, cod, in_svg, temp_id)
+ i2xpath.update(i2x)
+ used_labels.update(ulabels)
+
+ return temp_id, i2xpath, used_labels
+
+def print_html_object(obj: str='') -> str:
+ tab_cnt = 0
+ result, content, sep = '', '', ''
+ last_is_left, last_is_right = False, False
+ for ch in obj:
+ if ch == '<':
+ result += '\n'
+ if len(content.strip()) > 0:
+ result += sep + content.strip() + '\n'
+ result += sep + '<'
+
+ tab_cnt += 1
+ sep = ' ' * tab_cnt
+
+ content = ''
+ last_is_right = False
+ last_is_left = True
+ elif ch == '>':
+ if last_is_left:
+ result += content
+ else:
+ if last_is_right:
+ result += '\n'
+ if len(content.strip()) > 0:
+ result += sep + content.strip() + '\n'
+
+ tab_cnt -= 1
+ sep = ' ' * tab_cnt
+
+ if not last_is_left:
+ result += sep
+
+ result += '>'
+ content = ''
+
+ last_is_right = True
+ last_is_left = False
+ else:
+ content += ch
+
+ return result
+
+def rect2tuple(rect: str) -> tuple[int, int, int, int]:
+ if rect is None or type(rect) != type('str'):
+ return None
+ rect = rect.strip()
+ if rect.count(',') != 3:
+ return None
+ rect = rect.split(',')
+ rect = [float(r) for r in rect]
+ return tuple(rect)
diff --git a/VAB-WebArena-Lite/new/openai_utils.py b/VAB-WebArena-Lite/new/openai_utils.py
index 8374c7a..c9754f2 100644
--- a/VAB-WebArena-Lite/new/openai_utils.py
+++ b/VAB-WebArena-Lite/new/openai_utils.py
@@ -36,7 +36,6 @@ def retry_with_exponential_backoff( # type: ignore
# Initialize variables
num_retries = 0
delay = initial_delay
-
# Loop until a successful response or max_retries is hit or an exception is raised
while True:
try:
@@ -142,27 +141,32 @@ async def agenerate_from_openai_completion(
@retry_with_exponential_backoff
def generate_from_openai_completion(
prompt: str,
- engine: str,
+ model: str,
temperature: float,
max_tokens: int,
top_p: float,
- context_length: int,
stop_token: str | None = None,
+ api_key: str | None = None,
+ base_url: str | None = None
) -> str:
- if "OPENAI_API_KEY" not in os.environ:
+ if "OPENAI_API_KEY" not in os.environ and api_key is None:
raise ValueError(
"OPENAI_API_KEY environment variable must be set when using OpenAI API."
)
-
+ if api_key is not None:
+ client = OpenAI(api_key=api_key, base_url=base_url)
response = client.completions.create(
prompt=prompt,
- engine=engine,
+ model=model,
temperature=temperature,
max_tokens=max_tokens,
top_p=top_p,
stop=[stop_token],
)
- answer: str = response["choices"][0]["text"]
+ try:
+ answer: str = response["choices"][0]["text"]
+ except:
+ answer: str = response.choices[0].text
return answer
diff --git a/VAB-WebArena-Lite/new/p_webrl.json b/VAB-WebArena-Lite/new/p_webrl.json
new file mode 100644
index 0000000..c172455
--- /dev/null
+++ b/VAB-WebArena-Lite/new/p_webrl.json
@@ -0,0 +1,13 @@
+{
+ "intro": "",
+ "examples": [],
+ "template": "",
+ "meta_data": {
+ "observation": "webrl",
+ "action_type": "webrl_id",
+ "keywords": [],
+ "prompt_constructor": "WebRLPromptConstructor",
+ "answer_phrase": "",
+ "action_splitter": ""
+ }
+ }
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/processors.py b/VAB-WebArena-Lite/new/processors.py
new file mode 100644
index 0000000..d3f422d
--- /dev/null
+++ b/VAB-WebArena-Lite/new/processors.py
@@ -0,0 +1,1351 @@
+import json
+import pkgutil
+import re
+from collections import defaultdict
+from dataclasses import dataclass
+from io import BytesIO, StringIO
+from typing import Any, Optional, TypedDict, Union
+from urllib.parse import urljoin, urlparse
+
+import matplotlib.pyplot as plt
+import numpy as np
+import numpy.typing as npt
+import pandas as pd
+import playwright
+import requests
+from gymnasium import spaces
+from PIL import Image, ImageDraw, ImageFont
+from playwright.sync_api import CDPSession, Page, ViewportSize
+from .html_tools.fetch import get_parsed_html
+
+from browser_env.constants import (
+ ASCII_CHARSET,
+ FREQ_UNICODE_CHARSET,
+ IGNORED_ACTREE_PROPERTIES,
+ INJECTED_ATTR_NAME,
+ UTTERANCE_MAX_LENGTH,
+ BID_ATTR,
+ DATA_REGEXP,
+ IN_VIEWPORT_RATIO_THRESHOLD,
+)
+
+from .utils import (
+ AccessibilityTree,
+ AccessibilityTreeNode,
+ BrowserConfig,
+ BrowserInfo,
+ DOMNode,
+ DOMTree,
+ Observation,
+ png_bytes_to_numpy,
+)
+
+
+def remove_unicode(input_string):
+ # Define a regex pattern to match Unicode characters
+ unicode_pattern = re.compile(r"[^\x00-\x7F]+")
+
+ # Use the pattern to replace Unicode characters with an empty string
+ cleaned_string = unicode_pattern.sub("", input_string)
+
+ return cleaned_string
+
+
+class ObservationProcessor:
+ def process(self, page: Page) -> Observation:
+ raise NotImplementedError
+
+
+class ObservationMetadata(TypedDict):
+ obs_nodes_info: dict[str, Any]
+
+
+def create_empty_metadata() -> ObservationMetadata:
+ return {
+ "obs_nodes_info": {},
+ }
+
+
+def extract_data_items_from_aria(string: str) -> tuple[list[str], str]:
+ """
+ Utility function to extract temporary data stored in the "aria-roledescription" attribute of a node
+ """
+
+ match = DATA_REGEXP.fullmatch(string)
+ if not match:
+ return [], string
+
+ groups = match.groups()
+ data_items = groups[:-1]
+ original_aria = groups[-1]
+ return data_items, original_aria
+
+
+class TextObervationProcessor(ObservationProcessor):
+ def __init__(
+ self,
+ observation_type: str,
+ current_viewport_only: bool,
+ viewport_size: ViewportSize,
+ captioning_fn=None,
+ ):
+ self.observation_type = observation_type
+ self.current_viewport_only = current_viewport_only
+ self.viewport_size = viewport_size
+ self.observation_tag = "text"
+ self.meta_data = (
+ create_empty_metadata()
+ ) # use the store meta data of this observation type
+
+ if self.observation_type in [
+ "accessibility_tree_with_captioner",
+ "image_som",
+ ]:
+ self.captioning_fn = captioning_fn
+ # Cache captions.
+ self.url2caption = {}
+
+ def fetch_browser_info(
+ self,
+ page: Page,
+ ) -> BrowserInfo:
+ # extract domtree
+ client = page.context.new_cdp_session(page)
+ tree = client.send(
+ "DOMSnapshot.captureSnapshot",
+ {
+ "computedStyles": [],
+ "includeDOMRects": True,
+ "includePaintOrder": True,
+ },
+ )
+ client.detach()
+
+ # calibrate the bounds, in some cases, the bounds are scaled somehow
+ bounds = tree["documents"][0]["layout"]["bounds"]
+ b = bounds[0]
+ n = b[2] / self.viewport_size["width"]
+ bounds = [[x / n for x in bound] for bound in bounds]
+ tree["documents"][0]["layout"]["bounds"] = bounds
+ # add union bound placeholder
+ tree["documents"][0]["layout"]["unionBounds"] = [None for _ in bounds]
+
+ # extract browser info
+ win_upper_bound = page.evaluate("window.pageYOffset")
+ win_left_bound = page.evaluate("window.pageXOffset")
+ win_width = page.evaluate("window.screen.width")
+ win_height = page.evaluate("window.screen.height")
+ win_right_bound = win_left_bound + win_width
+ win_lower_bound = win_upper_bound + win_height
+ device_pixel_ratio = page.evaluate("window.devicePixelRatio")
+ assert device_pixel_ratio == 1.0, "devicePixelRatio is not 1.0"
+
+ config: BrowserConfig = {
+ "win_upper_bound": win_upper_bound,
+ "win_left_bound": win_left_bound,
+ "win_width": win_width,
+ "win_height": win_height,
+ "win_right_bound": win_right_bound,
+ "win_lower_bound": win_lower_bound,
+ "device_pixel_ratio": device_pixel_ratio,
+ }
+
+ # assert len(tree['documents']) == 1, "More than one document in the DOM tree"
+ info: BrowserInfo = {"DOMTree": tree, "config": config}
+
+ return info
+
+ @staticmethod
+ def get_bounding_client_rect(
+ client: CDPSession, backend_node_id: str
+ ) -> dict[str, Any]:
+ try:
+ remote_object = client.send(
+ "DOM.resolveNode", {"backendNodeId": int(backend_node_id)}
+ )
+ remote_object_id = remote_object["object"]["objectId"]
+ response = client.send(
+ "Runtime.callFunctionOn",
+ {
+ "objectId": remote_object_id,
+ "functionDeclaration": """
+ function() {
+ if (this.nodeType == 3) {
+ var range = document.createRange();
+ range.selectNode(this);
+ var rect = range.getBoundingClientRect().toJSON();
+ range.detach();
+ return rect;
+ } else {
+ return this.getBoundingClientRect().toJSON();
+ }
+ }
+ """,
+ "returnByValue": True,
+ },
+ )
+ return response
+ except Exception as e:
+ return {"result": {"subtype": "error"}}
+
+ @staticmethod
+ def get_element_in_viewport_ratio(
+ elem_left_bound: float,
+ elem_top_bound: float,
+ width: float,
+ height: float,
+ config: BrowserConfig,
+ ) -> float:
+ elem_right_bound = elem_left_bound + width
+ elem_lower_bound = elem_top_bound + height
+
+ win_left_bound = 0
+ win_right_bound = config["win_width"]
+ win_top_bound = 0
+ win_lower_bound = config["win_height"]
+
+ # Compute the overlap in x and y axes
+ overlap_width = max(
+ 0,
+ min(elem_right_bound, win_right_bound)
+ - max(elem_left_bound, win_left_bound),
+ )
+ overlap_height = max(
+ 0,
+ min(elem_lower_bound, win_lower_bound) - max(elem_top_bound, win_top_bound),
+ )
+
+ # Compute the overlap area
+ ratio = overlap_width * overlap_height / width * height
+ return ratio
+
+ def fetch_page_html(
+ self,
+ info: BrowserInfo,
+ page: Page,
+ current_viewport_only: bool,
+ ) -> DOMTree:
+ # adopted from [natbot](https://github.com/nat/natbot)
+ tree = info["DOMTree"]
+ strings = tree["strings"]
+ document = tree["documents"][0]
+ nodes = document["nodes"]
+
+ # make a dom tree that is easier to navigate
+ dom_tree: DOMTree = []
+ graph = defaultdict(list)
+ client = page.context.new_cdp_session(page)
+ for node_idx in range(len(nodes["nodeName"])):
+ cur_node: DOMNode = {
+ "nodeId": "",
+ "nodeType": "",
+ "nodeName": "",
+ "nodeValue": "",
+ "attributes": "",
+ "backendNodeId": "",
+ "parentId": "",
+ "childIds": [],
+ "cursor": 0,
+ "union_bound": None,
+ }
+
+ node_type_idx = nodes["nodeType"][node_idx]
+ node_type = "generic"
+ if node_type_idx >= 0 and node_type_idx < len(strings):
+ node_type = strings[node_type_idx]
+
+ node_name = strings[nodes["nodeName"][node_idx]]
+
+ node_value_idx = nodes["nodeValue"][node_idx]
+ node_value = ""
+ if node_value_idx >= 0 and node_value_idx < len(strings):
+ node_value = " ".join(strings[node_value_idx].split())
+
+ node_attributes = [strings[i] for i in nodes["attributes"][node_idx]]
+ node_attributes_str = ""
+ for i in range(0, len(node_attributes), 2):
+ a = node_attributes[i]
+ b = node_attributes[i + 1]
+ b = " ".join(b.split())
+ node_attributes_str += f'{a}="{b}" '
+ node_attributes_str = node_attributes_str.strip()
+
+ cur_node["nodeId"] = str(node_idx)
+ cur_node["nodeType"] = node_type
+ cur_node["nodeName"] = node_name
+ cur_node["nodeValue"] = node_value
+ cur_node["attributes"] = node_attributes_str
+ cur_node["backendNodeId"] = str(nodes["backendNodeId"][node_idx])
+ cur_node["parentId"] = str(nodes["parentIndex"][node_idx])
+
+ if cur_node["parentId"] != "-1":
+ graph[cur_node["parentId"]].append(str(cur_node["nodeId"]))
+
+ # get the bound
+ if cur_node["parentId"] == "-1":
+ cur_node["union_bound"] = [0.0, 0.0, 10.0, 10.0]
+ else:
+ response = self.get_bounding_client_rect(
+ client, cur_node["backendNodeId"]
+ )
+ if response.get("result", {}).get("subtype", "") == "error":
+ cur_node["union_bound"] = None
+ else:
+ x = response["result"]["value"]["x"]
+ y = response["result"]["value"]["y"]
+ width = response["result"]["value"]["width"]
+ height = response["result"]["value"]["height"]
+ cur_node["union_bound"] = [x, y, width, height]
+
+ dom_tree.append(cur_node)
+
+ client.detach()
+ # add parent children index to the node
+ for parent_id, child_ids in graph.items():
+ dom_tree[int(parent_id)]["childIds"] = child_ids
+
+ # remove the nodes that are not in the current viewport
+ if current_viewport_only:
+
+ def remove_node_in_graph(node: DOMNode) -> None:
+ # update the node information in the accessibility tree
+ node_id = node["nodeId"]
+ parent_id = node["parentId"]
+ child_ids = node["childIds"]
+
+ # update the children of the parent node
+ assert dom_tree[int(parent_id)]["parentId"] != "[REMOVED]"
+ # remove the nodeid from parent
+ index = dom_tree[int(parent_id)]["childIds"].index(node_id)
+ dom_tree[int(parent_id)]["childIds"].pop(index)
+
+ # Insert children_nodeids in the same location
+ for child_id in child_ids:
+ dom_tree[int(parent_id)]["childIds"].insert(index, child_id)
+ index += 1
+
+ # update children node's parent
+ for child_id in child_ids:
+ dom_tree[int(child_id)]["parentId"] = parent_id
+ # mark as removed
+ dom_tree[int(node_id)]["parentId"] = "[REMOVED]"
+
+ config = info["config"]
+ for cursor, node in enumerate(dom_tree):
+ if not node["union_bound"]:
+ remove_node_in_graph(node)
+ continue
+
+ [x, y, width, height] = node["union_bound"]
+
+ # invisible node
+ if width == 0.0 or height == 0.0:
+ remove_node_in_graph(node)
+ continue
+
+ in_viewport_ratio = self.get_element_in_viewport_ratio(
+ elem_left_bound=float(x),
+ elem_top_bound=float(y),
+ width=float(width),
+ height=float(height),
+ config=config,
+ )
+
+ if in_viewport_ratio < IN_VIEWPORT_RATIO_THRESHOLD:
+ remove_node_in_graph(node)
+
+ dom_tree = [
+ node for node in dom_tree if node.get("parentId", "-1") != "[REMOVED]"
+ ]
+
+ return dom_tree
+
+ @staticmethod
+ def parse_html(dom_tree: DOMTree) -> tuple[str, dict[str, Any]]:
+ """Parse the html tree into a string text"""
+
+ obs_nodes_info = {}
+ nodeid_to_cursor = {node["nodeId"]: idx for idx, node in enumerate(dom_tree)}
+
+ def dfs(node_cursor: int, depth: int) -> str:
+ tree_str = ""
+ node = dom_tree[node_cursor]
+ indent = "\t" * depth
+ valid_node = True
+ try:
+ node_str = f"[{node_cursor}] <{node['nodeName']}"
+ if node["attributes"]:
+ node_str += f" {node['attributes']}"
+ node_str += f"> {node['nodeValue']}"
+ valid_node = bool(node["attributes"] or node["nodeValue"])
+
+ if valid_node:
+ obs_nodes_info[str(node_cursor)] = {
+ "backend_id": node["backendNodeId"],
+ "union_bound": node["union_bound"],
+ "text": node_str,
+ }
+ tree_str += f"{indent}{node_str}\n"
+
+ except Exception as e:
+ valid_node = False
+
+ for child_ids in node["childIds"]:
+ child_cursor = nodeid_to_cursor[child_ids]
+ child_depth = depth + 1 if valid_node else depth
+ child_str = dfs(child_cursor, child_depth)
+ tree_str += child_str
+
+ return tree_str
+
+ html = dfs(0, 0)
+ return html, obs_nodes_info
+
+ def fetch_page_accessibility_tree(
+ self,
+ page: Page,
+ info: BrowserInfo,
+ current_viewport_only: bool,
+ ) -> AccessibilityTree:
+ client = page.context.new_cdp_session(page)
+ accessibility_tree: AccessibilityTree = client.send(
+ "Accessibility.getFullAXTree", {}
+ )["nodes"]
+
+ # a few nodes are repeated in the accessibility tree
+ seen_ids = set()
+ _accessibility_tree = []
+ for node in accessibility_tree:
+ if node["nodeId"] not in seen_ids:
+ _accessibility_tree.append(node)
+ seen_ids.add(node["nodeId"])
+ accessibility_tree = _accessibility_tree
+
+ nodeid_to_cursor = {}
+ for cursor, node in enumerate(accessibility_tree):
+ nodeid_to_cursor[node["nodeId"]] = cursor
+ # usually because the node is not visible etc
+ if "backendDOMNodeId" not in node:
+ node["union_bound"] = None
+ continue
+ backend_node_id = str(node["backendDOMNodeId"])
+ if node["role"]["value"] == "RootWebArea":
+ # always inside the viewport
+ node["union_bound"] = [0.0, 0.0, 10.0, 10.0]
+ else:
+ response = self.get_bounding_client_rect(
+ client,
+ backend_node_id
+ )
+ if response.get("result", {}).get("subtype", "") == "error":
+ node["union_bound"] = None
+ else:
+ x = response["result"]["value"]["x"]
+ y = response["result"]["value"]["y"]
+ width = response["result"]["value"]["width"]
+ height = response["result"]["value"]["height"]
+ node["union_bound"] = [x, y, width, height]
+
+ client.detach()
+ # filter nodes that are not in the current viewport
+ if current_viewport_only:
+
+ def remove_node_in_graph(node: AccessibilityTreeNode) -> None:
+ # update the node information in the accessibility tree
+ nodeid = node["nodeId"]
+ node_cursor = nodeid_to_cursor[nodeid]
+ parent_nodeid = node["parentId"]
+ children_nodeids = node["childIds"]
+ parent_cursor = nodeid_to_cursor[parent_nodeid]
+ # update the children of the parent node
+ assert (
+ accessibility_tree[parent_cursor].get("parentId", "Root")
+ is not None
+ )
+ # remove the nodeid from parent's childIds
+ index = accessibility_tree[parent_cursor]["childIds"].index(nodeid)
+ accessibility_tree[parent_cursor]["childIds"].pop(index)
+ # Insert children_nodeids in the same location
+ for child_nodeid in children_nodeids:
+ accessibility_tree[parent_cursor]["childIds"].insert(
+ index, child_nodeid
+ )
+ index += 1
+ # update children node's parent
+ for child_nodeid in children_nodeids:
+ child_cursor = nodeid_to_cursor[child_nodeid]
+ accessibility_tree[child_cursor]["parentId"] = parent_nodeid
+ # mark as removed
+ accessibility_tree[node_cursor]["parentId"] = "[REMOVED]"
+
+ config = info["config"]
+ for node in accessibility_tree:
+ if not node["union_bound"]:
+ remove_node_in_graph(node)
+ continue
+
+ [x, y, width, height] = node["union_bound"]
+
+ # invisible node
+ if width == 0 or height == 0:
+ remove_node_in_graph(node)
+ continue
+
+ in_viewport_ratio = self.get_element_in_viewport_ratio(
+ elem_left_bound=float(x),
+ elem_top_bound=float(y),
+ width=float(width),
+ height=float(height),
+ config=config,
+ )
+
+ if in_viewport_ratio < IN_VIEWPORT_RATIO_THRESHOLD:
+ remove_node_in_graph(node)
+
+ accessibility_tree = [
+ node
+ for node in accessibility_tree
+ if node.get("parentId", "Root") != "[REMOVED]"
+ ]
+
+ return accessibility_tree
+
+ @staticmethod
+ def parse_accessibility_tree(
+ accessibility_tree: AccessibilityTree,
+ ) -> tuple[str, dict[str, Any]]:
+ """Parse the accessibility tree into a string text"""
+ node_id_to_idx = {}
+ for idx, node in enumerate(accessibility_tree):
+ node_id_to_idx[node["nodeId"]] = idx
+
+ obs_nodes_info = {}
+
+ def dfs(idx: int, obs_node_id: str, depth: int) -> str:
+ tree_str = ""
+ node = accessibility_tree[idx]
+ indent = "\t" * depth
+ valid_node = True
+ try:
+ role = node["role"]["value"]
+ name = node["name"]["value"]
+ node_str = f"[{obs_node_id}] {role} {repr(name)}"
+ properties = []
+ for property in node.get("properties", []):
+ try:
+ if property["name"] in IGNORED_ACTREE_PROPERTIES:
+ continue
+ properties.append(
+ f'{property["name"]}: {property["value"]["value"]}'
+ )
+ except KeyError:
+ pass
+
+ if properties:
+ node_str += " " + " ".join(properties)
+
+ # check valid
+ if not node_str.strip():
+ valid_node = False
+
+ # empty generic node
+ if not name.strip():
+ if not properties:
+ if role in [
+ "generic",
+ "img",
+ "list",
+ "strong",
+ "paragraph",
+ "banner",
+ "navigation",
+ "Section",
+ "LabelText",
+ "Legend",
+ "listitem",
+ ]:
+ valid_node = False
+ elif role in ["listitem"]:
+ valid_node = False
+
+ if valid_node:
+ tree_str += f"{indent}{node_str}"
+ obs_nodes_info[obs_node_id] = {
+ "backend_id": node["backendDOMNodeId"],
+ "union_bound": node["union_bound"],
+ "text": node_str,
+ }
+
+ except Exception as e:
+ valid_node = False
+
+ for _, child_node_id in enumerate(node["childIds"]):
+ if child_node_id not in node_id_to_idx:
+ continue
+ # mark this to save some tokens
+ child_depth = depth + 1 if valid_node else depth
+ child_str = dfs(
+ node_id_to_idx[child_node_id], child_node_id, child_depth
+ )
+ if child_str.strip():
+ if tree_str.strip():
+ tree_str += "\n"
+ tree_str += child_str
+
+ return tree_str
+
+ tree_str = dfs(0, accessibility_tree[0]["nodeId"], 0)
+ return tree_str, obs_nodes_info
+
+ @staticmethod
+ def clean_accesibility_tree(tree_str: str) -> str:
+ """further clean accesibility tree"""
+ clean_lines: list[str] = []
+ for line in tree_str.split("\n"):
+ # remove statictext if the content already appears in the previous line
+ if "statictext" in line.lower():
+ prev_lines = clean_lines[-3:]
+ pattern = r"\[\d+\] StaticText (.+)"
+
+ match = re.search(pattern, line, re.DOTALL)
+ if match:
+ static_text = match.group(1)[1:-1] # remove the quotes
+ if static_text and all(
+ static_text not in prev_line for prev_line in prev_lines
+ ):
+ clean_lines.append(line)
+ else:
+ clean_lines.append(line)
+
+ return "\n".join(clean_lines)
+
+ def fetch_image_related(self, page: Page, browser_info: BrowserInfo) -> str:
+ # Check if the current page is an image url
+ if page.url.endswith((".jpg", ".jpeg", ".png")):
+ print("NOTE: We are on an image page!!!")
+ # Load image from current url and run captioning on it.
+ if page.url not in self.url2caption and self.captioning_fn is not None:
+ try:
+ image = Image.open(requests.get(page.url, stream=True).raw)
+ caption = self.captioning_fn([image])[0].strip()
+ self.url2caption[page.url] = remove_unicode(caption)
+ except Exception as e:
+ print("L579 WARNING: ", e)
+ content = self.url2caption.get(page.url, "Image")
+
+ else:
+ if self.captioning_fn is not None:
+ images = page.query_selector_all("img")
+ image_urls = []
+ for image in images:
+ try:
+ image_url = image.get_attribute("src")
+ if not image_url.startswith(("http://", "https://", "www.")):
+ image_url = urljoin(page.url, image_url)
+ if image_url not in self.url2caption:
+ image_urls.append(image_url)
+ except Exception as e:
+ print("L604 WARNING: ", e)
+
+ # Run image captioning on image_url pixels. This is for models which use captioning as a baseline.
+ if len(image_urls) > 0:
+ image_pixels = []
+ valid_urls = []
+ for url in image_urls:
+ if "data:image/svg" in url:
+ continue
+ else:
+ try:
+ image = Image.open(requests.get(url, stream=True).raw)
+ image_pixels.append(image)
+ valid_urls.append(url)
+ except Exception as e:
+ print("L616 WARNING: ", e)
+
+ # Caption images.
+ if image_pixels:
+ # Run in batches of 4.
+ bs = 4
+ captions = []
+ for i in range(0, len(image_pixels), bs):
+ try:
+ captions.extend(
+ self.captioning_fn(image_pixels[i : i + bs])
+ )
+ except Exception as e:
+ print("L628 WARNING: ", e)
+ captions.extend([""] * len(image_pixels[i : i + bs]))
+ assert len(valid_urls) == len(
+ captions
+ ), f"len(images)={len(valid_urls)}, len(captions)={len(captions)}"
+ for image_url, caption in zip(valid_urls, captions):
+ self.url2caption[image_url] = remove_unicode(
+ caption.strip()
+ )
+
+ image_idx = 0
+ for image in images:
+ try:
+ original_alt = image.get_attribute("alt") or ""
+ image_url = image.get_attribute("src")
+ if not image_url.startswith(("http://", "https://", "www.")):
+ image_url = urljoin(page.url, image_url)
+
+ updated_alt = original_alt
+
+ if image_url in self.url2caption:
+ if self.url2caption[image_url] not in updated_alt:
+ updated_alt = f"{updated_alt}, description: {self.url2caption[image_url]}"
+ elif "data:image/svg" not in image_url:
+ print(f"WARNING: {image_url} not in self.url2caption")
+
+ if "url:" not in updated_alt:
+ updated_alt = f"{updated_alt}, url: {image_url}"
+
+ safe_updated_alt = json.dumps(updated_alt)
+ image.evaluate(f"node => node.alt = {safe_updated_alt}")
+ except Exception as e:
+ print("L653 WARNING:", e)
+
+ if self.observation_type == "accessibility_tree_with_captioner":
+ frame_ax_trees = self.fetch_page_accessibility_tree(
+ page,
+ browser_info,
+ current_viewport_only=self.current_viewport_only
+ )
+ content, obs_nodes_info = self.parse_accessibility_tree(frame_ax_trees)
+ content = self.clean_accesibility_tree(content)
+ self.obs_nodes_info = obs_nodes_info
+ self.meta_data["obs_nodes_info"] = obs_nodes_info
+ else:
+ content = "" # Not used for SoM
+
+ return content
+
+ def process(self, page: Page) -> str:
+ # get the tab info
+ open_tabs = page.context.pages
+ try:
+ tab_titles = [tab.title() for tab in open_tabs]
+ current_tab_idx = open_tabs.index(page)
+ for idx in range(len(open_tabs)):
+ if idx == current_tab_idx:
+ tab_titles[idx] = f"Tab {idx} (current): {open_tabs[idx].title()}"
+ else:
+ tab_titles[idx] = f"Tab {idx}: {open_tabs[idx].title()}"
+ tab_title_str = " | ".join(tab_titles)
+ except Exception:
+ tab_title_str = " | ".join([f"Tab {idx}" for idx in range(len(open_tabs))])
+
+ try:
+ browser_info = self.fetch_browser_info(page)
+ except Exception:
+ page.wait_for_load_state("load", timeout=500)
+ browser_info = self.fetch_browser_info(page)
+
+ if self.observation_type == "html":
+ dom_tree = self.fetch_page_html(
+ browser_info,
+ page,
+ self.current_viewport_only,
+ )
+ content, obs_nodes_info = self.parse_html(dom_tree)
+ self.obs_nodes_info = obs_nodes_info
+ self.meta_data["obs_nodes_info"] = obs_nodes_info
+
+ elif self.observation_type == "accessibility_tree":
+ accessibility_tree = self.fetch_page_accessibility_tree(
+ page,
+ browser_info,
+ self.current_viewport_only,
+ )
+ content, obs_nodes_info = self.parse_accessibility_tree(accessibility_tree)
+ content = self.clean_accesibility_tree(content)
+ self.obs_nodes_info = obs_nodes_info
+ self.meta_data["obs_nodes_info"] = obs_nodes_info
+
+ elif self.observation_type in [
+ "accessibility_tree_with_captioner",
+ "image_som",
+ ]:
+ content = self.fetch_image_related(
+ page,
+ browser_info,
+ )
+
+ elif self.observation_type == "":
+ content = ""
+
+ else:
+ raise ValueError(f"Invalid observation type: {self.observation_type}")
+
+ self.browser_config = browser_info["config"]
+ content = f"{tab_title_str}\n\n{content}"
+
+ return content
+
+ def get_element_center(self, element_id: str) -> tuple[float, float]:
+ node_info = self.obs_nodes_info[element_id]
+ node_bound = node_info["union_bound"]
+ x, y, width, height = node_bound
+ center_x = x + width / 2
+ center_y = y + height / 2
+ return (
+ center_x / self.viewport_size["width"],
+ center_y / self.viewport_size["height"],
+ )
+
+
+class ImageObservationProcessor(ObservationProcessor):
+ def __init__(
+ self,
+ observation_type: str,
+ viewport_size: Optional[ViewportSize] = None,
+ ):
+ self.observation_type = observation_type
+ self.observation_tag = "image"
+ self.viewport_size = viewport_size
+ self.meta_data = create_empty_metadata()
+
+ def get_page_bboxes(self, page: Page) -> list[list[float]]:
+ """JavaScript code to return bounding boxes and other metadata from HTML elements."""
+ js_script = """
+ (() => {
+ const interactableSelectors = [
+ 'a[href]:not(:has(img))', 'a[href] img', 'button', 'input:not([type="hidden"])', 'textarea', 'select',
+ '[tabindex]:not([tabindex="-1"])', '[contenteditable="true"]', '[role="button"]', '[role="link"]',
+ '[role="checkbox"]', '[role="menuitem"]', '[role="tab"]', '[draggable="true"]',
+ '.btn', 'a[href="/notifications"]', 'a[href="/submit"]', '.fa.fa-star.is-rating-item', 'input[type="checkbox"]'
+
+ ];
+
+ const textSelectors = ['p', 'span', 'div:not(:has(*))', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'article'];
+ const modifiedTextSelectors = textSelectors.map(selector =>
+ `:not(${interactableSelectors.join(', ')}):not(style) > ${selector}`
+ );
+
+ const combinedSelectors = [...interactableSelectors, ...modifiedTextSelectors];
+ const elements = document.querySelectorAll(combinedSelectors.join(', '));
+
+ const pixelRatio = window.devicePixelRatio;
+ let csvContent = "ID,Element,Top,Right,Bottom,Left,Width,Height,Alt,Class,Id,TextContent,Interactable\\n";
+ let counter = 1;
+
+ elements.forEach(element => {
+ const rect = element.getBoundingClientRect();
+ if (rect.width === 0 || rect.height === 0) return;
+ let altText = element.getAttribute('alt') || '';
+ altText = altText.replace(/"/g, ''); // Escape double quotes in alt text
+ const classList = element.className || '';
+ const id = element.id || '';
+ let textContent = element.textContent || '';
+ textContent = textContent.replace(/"/g, ''); // Escape double quotes in textContent
+
+ // Determine if the element is interactable
+ const isInteractable = interactableSelectors.some(selector => element.matches(selector));
+
+ const dataString = [
+ counter, element.tagName, (rect.top + window.scrollY) * pixelRatio,
+ (rect.right + window.scrollX) * pixelRatio, (rect.bottom + window.scrollY) * pixelRatio,
+ (rect.left + window.scrollX) * pixelRatio, rect.width * pixelRatio, rect.height * pixelRatio,
+ altText, classList, id, textContent, isInteractable
+ ].map(value => `"${value}"`).join(",");
+
+ csvContent += dataString + "\\n";
+ counter++;
+ });
+
+ return csvContent;
+ })();
+ """
+ # Save the bbox as a CSV
+ csv_content = page.evaluate(js_script)
+ return csv_content
+
+ def draw_bounding_boxes(
+ self,
+ data_string,
+ screenshot_img,
+ viewport_size=None,
+ add_ids=True,
+ bbox_color=None,
+ min_width=8,
+ min_height=8,
+ bbox_padding=0,
+ bbox_border=2,
+ plot_ids=None,
+ ):
+ """
+ min_width and min_height: Minimum dimensions of the bounding box to be plotted.
+ """
+ # Read CSV data
+ df = pd.read_csv(StringIO(data_string), delimiter=",", quotechar='"')
+ df["Area"] = df["Width"] * df["Height"]
+ # Remove bounding boxes that are clipped.
+ b_x, b_y = (
+ self.browser_config["win_left_bound"],
+ self.browser_config["win_upper_bound"],
+ )
+ if viewport_size is not None:
+ df = df[
+ (df["Bottom"] - b_y >= 0)
+ & (df["Top"] - b_y <= viewport_size["height"])
+ & (df["Right"] - b_x >= 0)
+ & (df["Left"] - b_x <= viewport_size["width"])
+ ]
+ viewport_area = viewport_size["width"] * viewport_size["height"]
+ # Filter out bounding boxes that too large (more than 80% of the viewport)
+ df = df[df["Area"] <= 0.8 * viewport_area]
+
+ # Open the screenshot image
+ img = screenshot_img.copy()
+ draw = ImageDraw.Draw(img)
+
+ # Load a TTF font with a larger size
+ font_path = "media/SourceCodePro-SemiBold.ttf"
+ font_size, padding = 16, 2
+ font = ImageFont.truetype(font_path, font_size)
+
+ # Create a color cycle using one of the categorical color palettes in matplotlib
+ color_cycle = plt.rcParams["axes.prop_cycle"].by_key()["color"]
+ bbox_id2visid = {}
+ bbox_id2desc = {}
+ index = 0
+ id2center = {}
+ existing_text_rectangles = []
+ text_to_draw = []
+ # Provide [id] textContent inputs to the model as text.
+ text_content_elements = []
+ text_content_text = set() # Store text of interactable elements
+
+ # Iterate through each row in the CSV and draw bounding boxes
+ for _, row in df.iterrows():
+ if not row["Interactable"]:
+ content = ""
+ # Add image alt-text to the text representation.
+ if row["Element"] == "IMG" and pd.notna(row["Alt"]):
+ content += row["Alt"]
+ # Add HTML textContent (if any) to the text representation.
+ if pd.notna(row["TextContent"]):
+ content += (
+ row["TextContent"].strip().replace("\n", "").replace("\t", "")
+ )[
+ :200
+ ] # Limit to 200 characters to avoid having too much text
+
+ # Check if the text is a CSS selector
+ if content and not (content.startswith(".") and "{" in content):
+ # Add elements which are not interactable as StaticText
+ if content not in text_content_text:
+ text_content_elements.append(f"[] [StaticText] [{content}]")
+ text_content_text.add(content)
+ continue
+
+ if (plot_ids is not None) and (row["ID"] not in plot_ids):
+ continue
+
+ unique_id = str(index + 1)
+ bbox_id2visid[row["ID"]] = (
+ unique_id # map the bounding box ID to the unique character ID
+ )
+ top, right, bottom, left, width, height = (
+ row["Top"],
+ row["Right"],
+ row["Bottom"],
+ row["Left"],
+ row["Width"],
+ row["Height"],
+ )
+ left, right, top, bottom = left - b_x, right - b_x, top - b_y, bottom - b_y
+ id2center[unique_id] = (
+ (left + right) / 2,
+ (bottom + top) / 2,
+ width,
+ height,
+ )
+
+ if width >= min_width and height >= min_height:
+ # Get the next color in the cycle
+ color = bbox_color or color_cycle[index % len(color_cycle)]
+ draw.rectangle(
+ [
+ left - bbox_padding,
+ top - bbox_padding,
+ right + bbox_padding,
+ bottom + bbox_padding,
+ ],
+ outline=color,
+ width=bbox_border,
+ )
+ bbox_id2desc[row["ID"]] = color
+
+ # Draw the text on top of the rectangle
+ if add_ids:
+ # Calculate list of possible text positions
+ text_positions = [
+ (left - font_size, top - font_size), # Top-left corner
+ (
+ left,
+ top - font_size,
+ ), # A little to the right of the top-left corner
+ (right, top - font_size), # Top-right corner
+ (
+ right - font_size - 2 * padding,
+ top - font_size,
+ ), # A little to the left of the top-right corner
+ (left - font_size, bottom), # Bottom-left corner
+ (
+ left,
+ bottom,
+ ), # A little to the right of the bottom-left corner
+ (
+ right - font_size - 2 * padding,
+ bottom,
+ ), # A little to the left of the bottom-right corner
+ (
+ left,
+ bottom,
+ ), # A little to the right of the bottom-left corner
+ (
+ right - font_size - 2 * padding,
+ bottom,
+ ), # A little to the left of the bottom-right corner
+ ]
+ text_width = draw.textlength(unique_id, font=font)
+ text_height = font_size # Assume the text is one line
+
+ if viewport_size is not None:
+ for text_position in text_positions:
+ new_text_rectangle = [
+ text_position[0] - padding,
+ text_position[1] - padding,
+ text_position[0] + text_width + padding,
+ text_position[1] + text_height + padding,
+ ]
+
+ # Check if the new text rectangle is within the viewport
+ if (
+ new_text_rectangle[0] >= 0
+ and new_text_rectangle[1] >= 0
+ and new_text_rectangle[2] <= viewport_size["width"]
+ and new_text_rectangle[3] <= viewport_size["height"]
+ ):
+ # If the rectangle is within the viewport, check for overlaps
+ overlaps = False
+ for existing_rectangle in existing_text_rectangles:
+ if self.rectangles_overlap(
+ new_text_rectangle,
+ existing_rectangle,
+ padding * 2,
+ ):
+ overlaps = True
+ break
+
+ if not overlaps:
+ break
+ else:
+ # If the rectangle is outside the viewport, try the next position
+ continue
+ else:
+ # If none of the corners work, move the text rectangle by a fixed amount
+ text_position = (
+ text_positions[0][0] + padding,
+ text_positions[0][1],
+ )
+ new_text_rectangle = [
+ text_position[0] - padding,
+ text_position[1] - padding,
+ text_position[0] + text_width + padding,
+ text_position[1] + text_height + padding,
+ ]
+
+ existing_text_rectangles.append(new_text_rectangle)
+ text_to_draw.append(
+ (new_text_rectangle, text_position, unique_id, color)
+ )
+
+ content = ""
+ if row["Element"] == "IMG" and pd.notna(row["Alt"]):
+ content += row["Alt"]
+ if pd.notna(row["TextContent"]):
+ content += (
+ row["TextContent"]
+ .strip()
+ .replace("\n", "")
+ .replace("\t", "")
+ )[
+ :200
+ ] # Limit to 200 characters
+ text_content_elements.append(
+ f"[{unique_id}] [{row['Element']}] [{content}]"
+ )
+ if content in text_content_text:
+ # Remove text_content_elements with content
+ text_content_elements = [
+ element
+ for element in text_content_elements
+ if element.strip() != content
+ ]
+ text_content_text.add(content)
+
+ index += 1
+
+ for text_rectangle, text_position, unique_id, color in text_to_draw:
+ # Draw a background rectangle for the text
+ draw.rectangle(text_rectangle, fill=color)
+ draw.text(text_position, unique_id, font=font, fill="white")
+
+ content_str = "\n".join(text_content_elements)
+ return img, id2center, content_str
+
+ def rectangles_overlap(self, rect1, rect2, padding):
+ """
+ Check if two rectangles overlap.
+ Each rectangle is represented as a list [x1, y1, x2, y2].
+ """
+ return not (
+ rect1[2] < rect2[0] + padding
+ or rect1[0] > rect2[2] - padding
+ or rect1[1] > rect2[3] - padding
+ or rect1[3] < rect2[1] + padding
+ )
+
+ def process(self, page: Page) -> npt.NDArray[np.uint8]:
+ try:
+ browser_info = self.fetch_browser_info(page)
+ except Exception:
+ page.wait_for_load_state("load", timeout=500)
+ browser_info = self.fetch_browser_info(page)
+
+ self.browser_config = browser_info["config"]
+
+ if self.observation_type == "image_som":
+ # Produce the SoM image, with bounding boxes
+ try:
+ screenshot_bytes = page.screenshot()
+ som_bboxes = self.get_page_bboxes(page)
+ screenshot_img = Image.open(BytesIO(screenshot_bytes))
+ bbox_img, id2center, content_str = self.draw_bounding_boxes(
+ som_bboxes,
+ screenshot_img,
+ viewport_size=self.viewport_size,
+ )
+ self.som_id_info = id2center
+ self.meta_data["obs_nodes_info"] = id2center
+ screenshot_som = np.array(bbox_img)
+ return screenshot_som, content_str
+ except:
+ page.wait_for_event("load")
+ screenshot_bytes = page.screenshot()
+ som_bboxes = self.get_page_bboxes(page)
+ screenshot_img = Image.open(BytesIO(screenshot_bytes))
+ bbox_img, id2center, content_str = self.draw_bounding_boxes(
+ som_bboxes,
+ screenshot_img,
+ viewport_size=self.viewport_size,
+ )
+ self.som_id_info = id2center
+ self.meta_data["obs_nodes_info"] = id2center
+ screenshot_som = np.array(bbox_img)
+ return screenshot_som, content_str
+ else:
+ try:
+ screenshot = png_bytes_to_numpy(page.screenshot())
+ except:
+ page.wait_for_event("load")
+ screenshot = png_bytes_to_numpy(page.screenshot())
+ return screenshot, ""
+
+ def fetch_browser_info(self, page: Page) -> BrowserInfo:
+ client = page.context.new_cdp_session(page)
+ # extract domtree
+ tree = client.send(
+ "DOMSnapshot.captureSnapshot",
+ {
+ "computedStyles": [],
+ "includeDOMRects": True,
+ "includePaintOrder": True,
+ },
+ )
+ client.detach()
+ # calibrate the bounds, in some cases, the bounds are scaled somehow
+ bounds = tree["documents"][0]["layout"]["bounds"]
+ b = bounds[0]
+ n = b[2] / self.viewport_size["width"]
+ bounds = [[x / n for x in bound] for bound in bounds]
+ tree["documents"][0]["layout"]["bounds"] = bounds
+ # add union bound placeholder
+ tree["documents"][0]["layout"]["unionBounds"] = [None for _ in bounds]
+
+ # extract browser info
+ win_upper_bound = page.evaluate("window.pageYOffset")
+ win_left_bound = page.evaluate("window.pageXOffset")
+ win_width = page.evaluate("window.screen.width")
+ win_height = page.evaluate("window.screen.height")
+ win_right_bound = win_left_bound + win_width
+ win_lower_bound = win_upper_bound + win_height
+ device_pixel_ratio = page.evaluate("window.devicePixelRatio")
+ assert device_pixel_ratio == 1.0, "devicePixelRatio is not 1.0"
+
+ config: BrowserConfig = {
+ "win_upper_bound": win_upper_bound,
+ "win_left_bound": win_left_bound,
+ "win_width": win_width,
+ "win_height": win_height,
+ "win_right_bound": win_right_bound,
+ "win_lower_bound": win_lower_bound,
+ "device_pixel_ratio": device_pixel_ratio,
+ }
+
+ # assert len(tree['documents']) == 1, "More than one document in the DOM tree"
+ info: BrowserInfo = {"DOMTree": tree, "config": config}
+
+ return info
+
+ def get_element_center(self, element_id: str) -> tuple[float, float]:
+ if not self.observation_type == "image_som":
+ raise ValueError(
+ "get_element_center() is only supported for 'image_som' observation type."
+ )
+
+ browser_config = self.browser_config
+ center_x, center_y, width, height = self.som_id_info[element_id]
+ return (
+ center_x / self.viewport_size["width"],
+ center_y / self.viewport_size["height"],
+ )
+
+
+class TextObervationProcessorWebRL(TextObervationProcessor):
+ def __init__(
+ self,
+ observation_type: str,
+ current_viewport_only: bool,
+ viewport_size: ViewportSize,
+ captioning_fn=None,
+ ):
+ super().__init__(
+ observation_type,
+ current_viewport_only,
+ viewport_size,
+ captioning_fn,
+ )
+
+ def process(self, page: Page) -> str:
+ # get the tab info
+ page_info = get_parsed_html(page)
+ html = page_info["html"]
+ from bs4 import BeautifulSoup
+ soup = BeautifulSoup(html, 'html.parser')
+ obs_nodes_info = {}
+ for tag in soup.find_all(True):
+ if tag.has_attr('id') and tag.has_attr('data-bbox'):
+ backend_id = tag['id']
+ union_bound = tag['data-bbox']
+ union_bound = [float(num) for num in union_bound.split(',')]
+ obs_nodes_info[str(backend_id)] = {
+ "backend_id": backend_id,
+ "union_bound": union_bound,
+ "text": str(tag)
+ }
+ self.obs_nodes_info = obs_nodes_info
+ self.meta_data["obs_nodes_info"] = obs_nodes_info
+ return html
+
+ def get_element_center(self, element_id: str, page: Page=None) -> tuple[float, float]:
+
+ if page is not None:
+ element = page.query_selector(f"[data-label-id='{element_id}']")
+ bbox = element.bounding_box()
+ relative_bbox = (bbox['x'], bbox['y'], bbox['x'] + bbox['width'], bbox['y'] + bbox['height'])
+ center_x = (relative_bbox[0] + relative_bbox[2]) / 2
+ center_y = (relative_bbox[1] + relative_bbox[3]) / 2
+ else:
+ node_info = self.obs_nodes_info[element_id]
+ node_bound = node_info["union_bound"]
+ x, y, width, height = node_bound
+ center_x = x + width / 2
+ center_y = y + height / 2
+
+ return (
+ center_x / self.viewport_size["width"],
+ center_y / self.viewport_size["height"],
+ )
+
+class ObservationHandler:
+ """Main entry point to access all observation processor"""
+
+ def __init__(
+ self,
+ main_observation_type: str,
+ text_observation_type: str,
+ image_observation_type: str,
+ current_viewport_only: bool,
+ viewport_size: ViewportSize,
+ captioning_fn=None,
+ ) -> None:
+ self.main_observation_type = main_observation_type
+ if text_observation_type == "webrl":
+ self.text_processor = TextObervationProcessorWebRL(
+ text_observation_type,
+ current_viewport_only,
+ viewport_size,
+ captioning_fn,
+ )
+ else:
+ self.text_processor = TextObervationProcessor(
+ text_observation_type,
+ current_viewport_only,
+ viewport_size,
+ captioning_fn,
+ )
+ self.image_processor = ImageObservationProcessor(
+ image_observation_type, viewport_size
+ )
+ self.viewport_size = viewport_size
+
+ def get_observation_space(self) -> spaces.Dict:
+ text_space = spaces.Text(
+ min_length=0,
+ max_length=UTTERANCE_MAX_LENGTH,
+ charset=ASCII_CHARSET + FREQ_UNICODE_CHARSET,
+ )
+
+ image_space = spaces.Box(
+ # Each position stores the RGB values. Note the swapped axes (height first).
+ np.zeros(
+ (self.viewport_size["height"], self.viewport_size["width"], 3),
+ dtype=np.uint8,
+ ),
+ np.ones(
+ (self.viewport_size["height"], self.viewport_size["width"], 3),
+ dtype=np.uint8,
+ )
+ * 255.0,
+ dtype=np.uint8,
+ )
+
+ return spaces.Dict({"text": text_space, "image": image_space})
+
+ def get_observation(self, page: Page) -> dict[str, Observation]:
+ text_obs = self.text_processor.process(page)
+ image_obs, content_str = self.image_processor.process(page)
+ if content_str != "":
+ text_obs = content_str
+ return {"text": text_obs, "image": image_obs}
+
+ def get_observation_metadata(self) -> dict[str, ObservationMetadata]:
+ return {
+ "text": self.text_processor.meta_data,
+ "image": self.image_processor.meta_data,
+ }
+
+ @property
+ def action_processor(self) -> ObservationProcessor:
+ """Return the main processor that is associated with the action space"""
+ if self.main_observation_type == "text":
+ return self.text_processor
+ elif self.main_observation_type == "image":
+ return self.image_processor
+ else:
+ raise ValueError("Invalid main observation type")
diff --git a/VAB-WebArena-Lite/new/prompt_constructor.py b/VAB-WebArena-Lite/new/prompt_constructor.py
index 9cd5a2a..fc59b06 100644
--- a/VAB-WebArena-Lite/new/prompt_constructor.py
+++ b/VAB-WebArena-Lite/new/prompt_constructor.py
@@ -499,3 +499,59 @@ class MultimodalCoTPromptConstructor(CoTPromptConstructor):
raise NotImplementedError(
f"Provider {self.lm_config.provider} not implemented"
)
+
+class WebRLPromptConstructor(PromptConstructor):
+ """The agent will direct predict the action"""
+
+ def __init__(
+ self,
+ instruction_path: str | Path,
+ lm_config: lm_config.LMConfig,
+ tokenizer: Tokenizer,
+ ):
+ super().__init__(instruction_path, lm_config, tokenizer)
+
+ def construct(
+ self,
+ trajectory: Trajectory,
+ intent: str,
+ meta_data: dict[str, Any] = {},
+ ) -> APIInput:
+ """Construct prompt given the trajectory"""
+ state_info: StateInfo = trajectory[-1] # type: ignore[assignment]
+
+ obs = state_info["observation"][self.obs_modality]
+ max_obs_length = self.lm_config.gen_config["max_obs_length"]
+ if max_obs_length:
+ if self.lm_config.provider == "google":
+ print("NOTE: This is a Gemini model, so we use characters instead of tokens for max_obs_length.")
+ obs = obs[:max_obs_length]
+ else:
+ try:
+ obs = self.tokenizer.decode(self.tokenizer.encode(obs)[:max_obs_length]) # type: ignore[arg-type]
+ except:
+ print("NOTE: There is no available tokenizer, so we use characters instead of tokens for max_obs_length.")
+ obs = obs[:max_obs_length]
+
+ turn_num = len(meta_data["action_history"])
+ if turn_num == 1:
+ previous_action_str = []
+ else:
+ previous_action_str = meta_data["action_history"][1:]
+
+ index = turn_num - 1
+ history = ""
+ for i in range(index - 1, -1, -1):
+ if i == 0:
+ history = f"Round {i}\n\n<|eot_id|><|start_header_id|>user<|end_header_id|>\n{intent}\n\n<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n{previous_action_str[i]}\n\n" + history
+ else:
+ history = f"Round {i}\n\n<|eot_id|><|start_header_id|>user<|end_header_id|>\n** Simplified html **\n\n<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n{previous_action_str[i]}\n\n" + history
+ if len(history) + len(obs) > (16384 - 512):
+ obs = obs[:(16384 - 512)-len(history)]
+ current_turn = f"Round {index}\n\n<|eot_id|><|start_header_id|>user<|end_header_id|>\n{obs}\n\n<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n"
+ prompt = f"Task Instruction: {intent}\n\n{history}{current_turn}"
+
+ return prompt
+
+ def extract_action(self, response: str) -> str:
+ return response
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/run.py b/VAB-WebArena-Lite/new/run.py
index e406e2c..6f16161 100644
--- a/VAB-WebArena-Lite/new/run.py
+++ b/VAB-WebArena-Lite/new/run.py
@@ -13,11 +13,13 @@ import tempfile
import time
from pathlib import Path
from typing import List
+import cv2
+import shutil
import openai
import requests
import torch
-from PIL import Image
+from PIL import Image, ImageDraw, ImageFont
from agent import (
PromptAgent,
@@ -62,6 +64,34 @@ formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
+def text_wrap(text, font, max_width):
+ lines = []
+ paragraphs = text.split('\n') # 按照 \n 分割文本为段落
+ for paragraph in paragraphs:
+ words = paragraph.split(' ')
+ line = ''
+ for word in words:
+ # 临时行
+ test_line = f"{line} {word}".strip()
+ # 获取临时行的宽度
+ test_line_bbox = font.getbbox(test_line)
+ test_line_width = test_line_bbox[2] - test_line_bbox[0]
+ if test_line_width <= max_width:
+ # 如果临时行的宽度不超过图片宽度，继续添加单词
+ line = test_line
+ else:
+ # 如果超过了最大宽度，保存当前行，开始新的一行
+ lines.append(line)
+ line = word
+ # 添加每段的最后一行
+ if line:
+ lines.append(line)
+ # 每个段落后添加一个空行，以保留段落的换行
+ lines.append('')
+ # 移除最后一个空行（不需要额外的空行）
+ if lines[-1] == '':
+ lines.pop()
+ return lines
def config() -> argparse.Namespace:
parser = argparse.ArgumentParser(
@@ -88,6 +118,7 @@ def config() -> argparse.Namespace:
"html",
"image",
"image_som",
+ "webrl",
],
default="accessibility_tree",
help="Observation type",
@@ -176,6 +207,9 @@ def config() -> argparse.Namespace:
# logging related
parser.add_argument("--result_dir", type=str, default="")
+
+ # if use self-deployed model
+ parser.add_argument("--planner_ip", type=str, default=None)
args = parser.parse_args()
# check the whether the action space is compatible with the observation space
@@ -196,7 +230,7 @@ def config() -> argparse.Namespace:
def early_stop(
- trajectory: Trajectory, max_steps: int, thresholds: dict[str, int]
+ trajectory: Trajectory, max_steps: int, thresholds: dict[str, int], actions=None
) -> tuple[bool, str]:
"""Check whether need to stop early"""
@@ -228,28 +262,38 @@ def early_stop(
if len(action_seq) == 0:
return False, ""
- last_action: Action = action_seq[-1]
+ if actions is None:
+ last_action: Action = action_seq[-1]
+ if last_action["action_type"] != ActionTypes.TYPE:
+ if len(last_k_actions) >= k:
+ if all(
+ [
+ is_equivalent(action, last_action)
+ for action in last_k_actions
+ ]
+ ):
+ return True, f"Same action for {k} times"
+ else:
+ # check the action sequence
+ if (
+ sum([is_equivalent(action, last_action) for action in action_seq])
+ >= k
+ ):
+ return True, f"Same typing action for {k} times"
+ return False, ""
- if last_action["action_type"] != ActionTypes.TYPE:
+ else:
+ last_k_actions = actions[-k:]
+ last_action = actions[-1]
if len(last_k_actions) >= k:
if all(
[
- is_equivalent(action, last_action)
+ action == last_action
for action in last_k_actions
]
):
return True, f"Same action for {k} times"
-
- else:
- # check the action sequence
- if (
- sum([is_equivalent(action, last_action) for action in action_seq])
- >= k
- ):
- return True, f"Same typing action for {k} times"
-
- return False, ""
-
+ return False, ""
def update_action_history(path: str, task_id: int, actions: List[str], score: float=-0.1):
obj = {
@@ -387,18 +431,22 @@ def test(
obs, info = env.reset(options={"config_file": config_file})
state_info: StateInfo = {"observation": obs, "info": info}
trajectory.append(state_info)
-
meta_data = {"action_history": ["None"]}
out_path = os.path.join(args.result_dir, "actions", f"{task_id}.json")
actions = []
+ os.makedirs(os.path.join(args.result_dir, 'screehshots'), exist_ok=True)
+ if os.path.exists(os.path.join(args.result_dir, 'screehshots', f"{task_id}")):
+ shutil.rmtree(os.path.join(args.result_dir, 'screehshots', f"{task_id}"))
+ os.makedirs(os.path.join(args.result_dir, 'screehshots', f"{task_id}"))
+
while True:
update_action_history(out_path, task_id, actions=actions)
-
+ # If no actions variable is passed, the behavior of early_stop is the same as the original one.
early_stop_flag, stop_info = early_stop(
- trajectory, max_steps, early_stop_thresholds
+ trajectory, max_steps, early_stop_thresholds, actions
)
-
+
if early_stop_flag:
action = create_stop_action(f"Early stop: {stop_info}")
else:
@@ -407,14 +455,14 @@ def test(
trajectory,
intent,
images=images,
- meta_data=meta_data,
+ meta_data=meta_data
)
except ValueError as e:
# get the error message
action = create_stop_action(f"ERROR: {str(e)}")
-
+
trajectory.append(action)
-
+
action_str = get_action_description(
action,
state_info["info"]["observation_metadata"],
@@ -426,13 +474,50 @@ def test(
render_helper.render(
action, state_info, meta_data, args.render_screenshot
)
+
+ current_screenshot = os.path.join(args.result_dir, 'screehshots', f"{task_id}", f"{len(actions)}.png")
+ _ = env.page.viewport_size
+ env.page.screenshot(path="/dev/null")
+ env.page.screenshot(path=current_screenshot)
+ element_id = action["element_id"]
+ if element_id != "":
+ element = env.page.query_selector(f"[data-label-id='{element_id}']")
+ if element:
+ bbox = element.bounding_box()
+ bbox = [int(bbox['x']), int(bbox['y']), int(bbox['width']),int(bbox['height'])]
+ image = cv2.imread(current_screenshot)
+ cv2.rectangle(image, (bbox[0], bbox[1]), (bbox[0] + bbox[2], bbox[1] + bbox[3]), (0, 255, 0), 2)
+ cv2.circle(image, (int(bbox[0] + bbox[2] / 2), int(bbox[1] + bbox[3] / 2)), radius=0, color=(0, 255, 0), thickness=2)
+ cv2.imwrite(current_screenshot, image)
+ font_path = "/workspace/qzh/Arial.ttf" # 使用FreeSans字体
+ font_size = 30 # 你可以调整这个值来增大或缩小字体
+ image = Image.open(current_screenshot)
+ font = ImageFont.truetype(font_path, font_size)
+ draw = ImageDraw.Draw(image)
+ image_width = image.width
+ wrapped_text = text_wrap(action_str, font, image_width)
+ line_height = font.getbbox('hg')[3] - font.getbbox('hg')[1]
+ text_height = line_height * len(wrapped_text)
+ new_image_height = image.height + text_height + 20 # 20 is extra white space
+ new_image = Image.new('RGB', (image.width, new_image_height), color=(255, 255, 255)) # white background
+ draw_new = ImageDraw.Draw(new_image)
+ y_text = 10 # Initial position of text from top
+ for line in wrapped_text:
+ text_bbox = draw_new.textbbox((0, 0), line, font=font)
+ text_width = text_bbox[2] - text_bbox[0]
+ text_position = ((image_width - text_width) // 2, y_text) # Center text horizontally
+ draw_new.text(text_position, line, font=font, fill=(0, 0, 0)) # black text
+ y_text += line_height # move to next line
+ new_image.paste(image, (0, text_height + 20))
+ new_image.save(current_screenshot)
+
meta_data["action_history"].append(action_str)
actions.append(action_str)
- print(action_str)
+ print('Action String: ', action_str)
if action["action_type"] == ActionTypes.STOP:
break
-
+
obs, _, terminated, _, info = env.step(action)
state_info = {"observation": obs, "info": info}
trajectory.append(state_info)
@@ -540,7 +625,7 @@ if __name__ == "__main__":
os.environ["TOKENIZERS_PARALLELISM"] = "false"
args = config()
- args.sleep_after_execution = 2.5
+ args.sleep_after_execution = 3.0
prepare(args)
test_config_base_dir = args.test_config_base_dir
diff --git a/VAB-WebArena-Lite/new/score.py b/VAB-WebArena-Lite/new/score.py
new file mode 100644
index 0000000..95235e5
--- /dev/null
+++ b/VAB-WebArena-Lite/new/score.py
@@ -0,0 +1,86 @@
+import os, json, sys, copy
+
+USE_TASKS = [i for i in range(165)]
+
+def get_result(res_dict, src="all"):
+ if len(res_dict) == 0:
+ return ''
+
+ success_id = [k for k, v in res_dict.items() if v >= 1.0]
+ score = len(success_id)
+ finish_count = len(res_dict)
+ pacc, acc = score / finish_count * 100, score / TASKS * 100
+
+ print(sorted(success_id))
+
+ meta = """
+--------
+src file: {}
+successed: {:3} / {:4} (812)
+partial accuracy: {:7}
+overall accuracy: {:7}
+--------
+""".format(src, int(score), finish_count, round(pacc, 2), round(acc, 2))
+
+ print(meta)
+
+def export_result(res_dict, src=".", note=["1.0", "0.0"], show_all=False):
+ out_string = ""
+ for id in USE_TASKS:
+ # with open(f"Pipeline/config_files/{id}.json", "r") as f:
+ # jd = json.load(f)
+
+ # if "map" in jd["sites"]:
+ # continue
+ if id in res_dict:
+ if res_dict[id] >= 1.0:
+ out_string += note[0]
+ else:
+ out_string += note[1]
+ elif show_all:
+ out_string += note[1]
+ out_string += "\n"
+
+ with open(os.path.join(src, 'export.txt'), 'w') as f:
+ f.write(out_string)
+
+TASKS = 165
+
+files = sys.argv[1]
+file_list = files.split(',')
+
+all_result = {}
+
+for src in file_list:
+ path = os.path.join(src, 'actions')
+
+ result = {}
+ finished = os.listdir(path)
+
+ for file in finished:
+ if not file.endswith('.json'):
+ continue
+ with open(os.path.join(path, file), 'r') as f:
+ data = json.load(f)
+
+ if not isinstance(data, dict):
+ continue
+
+ task_id = data.get('task_id', 1000)
+ # if task_id >= TASKS:
+ # continue
+
+ task_score = data.get('score', 0)
+ if task_score < 0:
+ continue
+
+ result[task_id] = task_score
+ if task_id not in all_result or task_score > all_result[task_id]:
+ all_result[task_id] = task_score
+
+ get_result(result, src)
+ export_result(result, src=src)
+
+if len(file_list) > 1:
+ get_result(all_result)
+export_result(all_result, show_all=True)
diff --git a/VAB-WebArena-Lite/new/test_webarena_lite.raw.json b/VAB-WebArena-Lite/new/test_webarena_lite.raw.json
index c77225a..adb1822 100644
--- a/VAB-WebArena-Lite/new/test_webarena_lite.raw.json
+++ b/VAB-WebArena-Lite/new/test_webarena_lite.raw.json
@@ -187,7 +187,7 @@
"string_match"
],
"reference_answers": {
- "must_include": [
+ "fuzzy_match": [
"0"
]
},
@@ -774,8 +774,8 @@
"string_match"
],
"reference_answers": {
- "fuzzy_match": [
- "914km"
+ "must_include": [
+ "914"
]
},
"reference_url": "",
@@ -1139,7 +1139,7 @@
"string_match"
],
"reference_answers": {
- "must_include": [
+ "fuzzy_match": [
"0"
]
},
@@ -1679,7 +1679,7 @@
"string_match"
],
"reference_answers": {
- "exact_match": "N/A"
+ "fuzzy_match": ["N/A"]
},
"reference_url": "",
"program_html": [],
@@ -2605,7 +2605,7 @@
"string_match"
],
"reference_answers": {
- "exact_match": "yjlou"
+ "must_include": "yjlou"
},
"reference_url": "",
"program_html": [],
@@ -3712,7 +3712,7 @@
"string_match"
],
"reference_answers": {
- "exact_match": "N/A"
+ "fuzzy_match": "N/A"
},
"reference_url": "",
"program_html": [],
diff --git a/VAB-WebArena-Lite/new/tokenizers.py b/VAB-WebArena-Lite/new/tokenizers.py
index 2234040..b6e8e74 100644
--- a/VAB-WebArena-Lite/new/tokenizers.py
+++ b/VAB-WebArena-Lite/new/tokenizers.py
@@ -1,15 +1,18 @@
from typing import Any
import tiktoken
-from transformers import LlamaTokenizer # type: ignore
+from transformers import LlamaTokenizer, AutoTokenizer # type: ignore
class Tokenizer(object):
def __init__(self, provider: str, model_name: str) -> None:
if provider == "openai":
- self.tokenizer = tiktoken.encoding_for_model(model_name)
+ try:
+ self.tokenizer = tiktoken.encoding_for_model(model_name)
+ except: # The provider is in openai format but the model is a finetuned model
+ self.tokenizer = None
elif provider == "huggingface":
- self.tokenizer = LlamaTokenizer.from_pretrained(model_name)
+ self.tokenizer = LlamaTokenizer.from_pretrained(model_name, trust_remote_code=True)
# turn off adding special tokens automatically
self.tokenizer.add_special_tokens = False # type: ignore[attr-defined]
self.tokenizer.add_bos_token = False # type: ignore[attr-defined]
diff --git a/VAB-WebArena-Lite/new/utils.py b/VAB-WebArena-Lite/new/utils.py
index 4558585..2a8b87c 100644
--- a/VAB-WebArena-Lite/new/utils.py
+++ b/VAB-WebArena-Lite/new/utils.py
@@ -21,6 +21,8 @@ APIInput = str | list[Any] | dict[str, Any]
def call_llm(
lm_config: lm_config.LMConfig,
prompt: APIInput,
+ api_key = None,
+ base_url = None
) -> str:
response: str
if lm_config.provider == "openai":
@@ -39,11 +41,13 @@ def call_llm(
assert isinstance(prompt, str)
response = generate_from_openai_completion(
prompt=prompt,
- engine=lm_config.model,
+ model=lm_config.model,
temperature=lm_config.gen_config["temperature"],
max_tokens=lm_config.gen_config["max_tokens"],
top_p=lm_config.gen_config["top_p"],
stop_token=lm_config.gen_config["stop_token"],
+ api_key=api_key,
+ base_url=base_url
)
else:
raise ValueError(
diff --git a/VAB-WebArena-Lite/new/wa_parallel_run_webrl.sh b/VAB-WebArena-Lite/new/wa_parallel_run_webrl.sh
new file mode 100644
index 0000000..f8c4784
--- /dev/null
+++ b/VAB-WebArena-Lite/new/wa_parallel_run_webrl.sh
@@ -0,0 +1,97 @@
+#!/bin/bash
+DATASET='webarena' # TODO: select from ['webarena', 'visualwebarena']
+result_dir='' # TODO: set your result_dir
+provider='openai' # TODO: select from ['openai', 'finetune', ...]
+model='' # TODO: assign model name, which is used for action generation
+planner_ip='' # TODO: ip address of the model you are deploying (only if you are deploying your own model using e.g. vllm)
+instruction_path='agent/prompts/jsons/p_webrl.json' # e.g., agent/prompts/jsons/p_cot_id_actree_2s.json
+test_config_base_dir='config_files/wa/test_webarena_lite' # e.g., config_files/wa/test_webarena_lite
+temperature=0.0
+
+SERVER='' # TODO: your server address
+MAP_SERVER='' # TODO: the server address for MAP tasks
+OPENAI_API_KEY='' # TODO: if you test OpenAI APIs
+OPENAI_ORGANIZATION=''
+CONDA_ENV_NAME='' # TODO: the name of your conda environment for testing WebArena
+
+ENV_VARIABLES="export DATASET=${DATASET}; export SHOPPING='http://${SERVER}:7770';export SHOPPING_ADMIN='http://${SERVER}:7780/admin';export REDDIT='http://${SERVER}:9999';export GITLAB='http://${SERVER}:8023';export MAP='http://${MAP_SERVER}:3000';export WIKIPEDIA='http://${SERVER}:8888/wikipedia_en_all_maxi_2022-05/A/User:The_other_Kiwix_guy/Landing';export HOMEPAGE='http://${SERVER}:4399';export OPENAI_API_KEY=${OPENAI_API_KEY};export OPENAI_ORGANIZATION=${OPENAI_ORGANIZATION}"
+
+# get the number of tmux panes
+num_panes=$(tmux list-panes | wc -l)
+
+# calculate how many panes need to be created
+let "panes_to_create = 7 - num_panes"
+
+# array of tmux commands to create each pane
+tmux_commands=(
+ 'tmux split-window -h'
+ 'tmux split-window -v'
+ 'tmux select-pane -t 0; tmux split-window -v'
+ 'tmux split-window -v'
+ 'tmux select-pane -t 3; tmux split-window -v'
+ 'tmux select-pane -t 5; tmux split-window -v'
+)
+
+# create panes up to 7
+for ((i=0; i<$panes_to_create; i++)); do
+ eval ${tmux_commands[$i]}
+done
+
+#!/bin/bash
+
+# Function to run a job
+run_job() {
+ tmux select-pane -t $1
+ COMMAND="python run.py \
+ --instruction_path ${instruction_path} \
+ --test_start_idx $2 \
+ --test_end_idx $3 \
+ --result_dir ${result_dir} \
+ --test_config_base_dir ${test_config_base_dir} \
+ --provider ${provider} \
+ --model ${model} \
+ --mode completion \
+ --planner_ip ${planner_ip} \
+ --stop_token \"<|eot_id|>\" \
+ --temperature ${temperature} \
+ --max_obs_length 0 \
+ --max_tokens 2048 \
+ --viewport_width 1280 \
+ --viewport_height 720 \
+ --parsing_failure_th 5 \
+ --repeating_action_failure_th 5 \
+ --action_set_tag webrl_id --observation_type webrl"
+ tmux send-keys "tmux set mouse on; conda activate ${CONDA_ENV_NAME}; ${ENV_VARIABLES}; until ${COMMAND}; do echo 'crashed' >&2; sleep 1; done" C-m
+ sleep 3
+}
+
+TOLERANCE=2
+run_batch() {
+ args=("$@") # save all arguments in an array
+ num_jobs=${#args[@]} # get number of arguments
+
+ for ((i=1; i<$num_jobs; i++)); do
+ run_job $i ${args[i-1]} ${args[i]}
+ done
+
+ # Wait for all jobs to finish
+ while tmux list-panes -F "#{pane_pid} #{pane_current_command}" | grep -q python; do
+ sleep 100 # wait for 10 seconds before checking again
+ done
+
+ # Run checker
+ while ! python scripts/check_error_runs.py ${result_dir} --delete_errors --tolerance ${TOLERANCE}; do
+ echo "Check failed, rerunning jobs..."
+ for ((i=1; i<$num_jobs; i++)); do
+ run_job $i ${args[i-1]} ${args[i]}
+ done
+
+ # Wait for all jobs to finish
+ while tmux list-panes -F "#{pane_pid} #{pane_current_command}" | grep -q python; do
+ sleep 100 # wait for 10 seconds before checking again
+ done
+ done
+
+}
+run_batch 0 28 56 84 112 140 165
+
diff --git a/VAB-WebArena-Lite/replace.sh b/VAB-WebArena-Lite/replace.sh
index add244c..7a674f0 100644
--- a/VAB-WebArena-Lite/replace.sh
+++ b/VAB-WebArena-Lite/replace.sh
@@ -14,6 +14,14 @@ cp -f new/generate_test_data.py visualwebarena/scripts/generate_test_data.py
cp -f new/run.py visualwebarena/run.py
cp -f new/agent.py visualwebarena/agent/agent.py
cp -f new/prompt_constructor.py visualwebarena/agent/prompts/prompt_constructor.py
+cp -f new/p_webrl.json visualwebarena/agent/prompts/jsons/p_webrl.json
+
+# browser_env
+cp -f new/actions.py visualwebarena/browser_env/actions.py
+cp -f new/envs.py visualwebarena/browser_env/envs.py
+cp -f new/helper_functions_browser.py visualwebarena/browser_env/helper_functions.py
+cp -f new/processors.py visualwebarena/browser_env/processors.py
+cp -rf new/html_tools visualwebarena/browser_env/
# llms
cp -f new/utils.py visualwebarena/llms/utils.py
@@ -22,15 +30,19 @@ cp -f new/lm_config.py visualwebarena/llms/lm_config.py
cp -f new/tokenizers.py visualwebarena/llms/tokenizers.py
cp -f new/api_utils.py visualwebarena/llms/providers/api_utils.py
cp -f new/openai_utils.py visualwebarena/llms/providers/openai_utils.py
+cp -f new/utils.py visualwebarena/llms/utils.py
# eval
cp -f new/evaluators.py visualwebarena/evaluation_harness/evaluators.py
-cp -f new/helper_functions.py visualwebarena/evaluation_harness/helper_functions.py
+cp -f new/helper_functions_eval.py visualwebarena/evaluation_harness/helper_functions.py
# misc
cp -f README.md visualwebarena/README.md
cp -f new/wa_parallel_run.sh visualwebarena/wa_parallel_run.sh
+cp -f new/score.py visualwebarena/score.py
+cp -f new/wa_parallel_run_webrl.sh visualwebarena/wa_parallel_run_webrl.sh
+
# 3. remove temporary files
mv visualwebarena/* .
rm -rf new
\n"
+
+ # meta data
+ new_content += f"{meta_data['action_history'][-1]}
\n"
+
+ # action
+ action_str = get_render_action(
+ action,
+ info["observation_metadata"],
+ action_set_tag=self.action_set_tag,
+ )
+ # with yellow background
+ action_str = f"{action_str}
"
+ new_content += f"{action_str}\n"
+
+ # add new content
+ self.render_file.seek(0)
+ html = self.render_file.read()
+ html_body = re.findall(r"(.*?)", html, re.DOTALL)[0]
+ html_body += new_content
+
+ html = HTML_TEMPLATE.format(body=html_body)
+ self.render_file.seek(0)
+ self.render_file.truncate()
+ self.render_file.write(html)
+ self.render_file.flush()
+
+ def close(self) -> None:
+ self.render_file.close()
diff --git a/VAB-WebArena-Lite/new/helper_functions.py b/VAB-WebArena-Lite/new/helper_functions_eval.py
similarity index 100%
rename from VAB-WebArena-Lite/new/helper_functions.py
rename to VAB-WebArena-Lite/new/helper_functions_eval.py
diff --git a/VAB-WebArena-Lite/new/html_tools/__init__.py b/VAB-WebArena-Lite/new/html_tools/__init__.py
new file mode 100755
index 0000000..6aac581
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/__init__.py
@@ -0,0 +1,7 @@
+from .identifier import IdentifierTool
+from .prompt import HtmlPrompt
+from .html_parser import HtmlParser
+
+from .utils import print_html_object
+from .configs import basic_attrs, mind2web_keep_attrs
+from .fetch import get_parsed_html
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/configs/__init__.py b/VAB-WebArena-Lite/new/html_tools/configs/__init__.py
new file mode 100755
index 0000000..b900a67
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/configs/__init__.py
@@ -0,0 +1,3 @@
+from .html_prompt import prompts
+from .config import basic_attrs, mind2web_keep_attrs, miniwob_attrs
+from .config import config_meta
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/configs/config.py b/VAB-WebArena-Lite/new/html_tools/configs/config.py
new file mode 100755
index 0000000..d7da96c
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/configs/config.py
@@ -0,0 +1,56 @@
+basic_attrs = [
+ 'title',
+ 'value',
+ 'type',
+ 'placeholder',
+ 'selected',
+ 'data-value',
+ 'data-text',
+ 'data-testid',
+ 'data-label',
+ 'data-bbox',
+ 'data-status'
+]
+
+mind2web_keep_attrs = [
+ 'alt',
+ 'aria_description',
+ 'aria_label',
+ 'aria_role',
+ 'input_checked',
+ 'input_value',
+ 'label',
+ 'name',
+ 'option_selected',
+ 'placeholder',
+ 'role',
+ 'text_value',
+ 'title',
+ 'type',
+ 'value',
+]
+
+miniwob_attrs = [
+ 'id',
+ 'type',
+ 'value',
+]
+
+config_meta = """
+======= Configs =======
+Columns:
+ - id: {id_attr}
+ - label: {label_attr}
+Position: {use_position}
+ - window: {window_size}
+ - rect_dict: {rect}
+Keep:
+ - parents: {parent_chain}
+ - attrs: {keep_attrs}
+ - elems: {keep_elem}
+ - obs_elem: {obs_elem}
+Generator:
+ - prompt: {prompt_name}
+ - label: {identifier_name}
+========================
+"""
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/configs/html_prompt.py b/VAB-WebArena-Lite/new/html_tools/configs/html_prompt.py
new file mode 100755
index 0000000..904d021
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/configs/html_prompt.py
@@ -0,0 +1,22 @@
+refine_prompt = {
+ 'dom': '<{tag}{label}|{attr}{content}{subtree} >',
+ 'label': '[{label}]',
+ 'attr': '{attr}',
+ 'attr_splitter': '; ',
+ 'subtree_splitter': ' ',
+}
+
+xml_prompt = {
+ 'dom': '<{tag}{label}{attr}>{content}{subtree} ',
+ 'label': ' id="{label}"',
+ 'attr': '{key}="{attr}"',
+ 'attr_splitter': ' ',
+ 'subtree_splitter': ' ',
+}
+
+prompts = {
+ 'refine': refine_prompt,
+ 'xml': xml_prompt,
+ 'new_data': refine_prompt,
+}
+
\ No newline at end of file
diff --git a/VAB-WebArena-Lite/new/html_tools/fetch.py b/VAB-WebArena-Lite/new/html_tools/fetch.py
new file mode 100755
index 0000000..25d5637
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/fetch.py
@@ -0,0 +1,108 @@
+import os
+import json
+import base64
+from .html_parser import HtmlParser
+from .configs import basic_attrs
+from .scripts import *
+
+def get_window(page):
+ x = page.evaluate("window.scrollX")
+ y = page.evaluate("window.scrollY")
+ w = page.evaluate("window.innerWidth")
+ h = page.evaluate("window.innerHeight")
+ return (x, y, w, h)
+
+def modify_page(page):
+ page.wait_for_timeout(500)
+
+ try:
+ page.evaluate(remove_id_script)
+ except:
+ pass
+
+ packet = {
+ "raw_html": page.evaluate("document.documentElement.outerHTML"),
+ "window": get_window(page)
+ }
+
+ page.evaluate(prepare_script)
+ page.wait_for_timeout(100)
+
+ img_bytes = page.screenshot(path="debug_info/screenshot_raw.png")
+ raw_image = base64.b64encode(img_bytes).decode()
+
+ page.evaluate(clickable_checker_script)
+ page.wait_for_timeout(50)
+
+ # get all clickable elements
+ start_id = 0
+ items, start_id = page.evaluate(label_script, {
+ "selector": ".possible-clickable-element",
+ "startIndex": start_id
+ })
+ page.wait_for_timeout(50)
+
+ # mark our own labels and get the images
+ items = page.evaluate(label_marker_script, items)
+ page.wait_for_timeout(100)
+ img_bytes = page.screenshot(path="debug_info/marked.png")
+ marked_image = base64.b64encode(img_bytes).decode()
+
+ # remove markers on the page
+ page.evaluate(remove_label_mark_script)
+
+ packet.update({
+ "raw_image": raw_image,
+ "marked_image": marked_image,
+ "modified_html": page.evaluate("document.documentElement.outerHTML")
+ })
+
+ # element_info, include "all_elements" and "clickable_elements"
+ element_info = page.evaluate(element_info_script)
+ page.wait_for_timeout(100)
+ packet.update(element_info)
+ return packet
+
+def save_debug_info(packet):
+ with open("debug_info/raw.html", "w") as f:
+ f.write(packet["modified_html"])
+ with open("debug_info/parsed.html", "w") as f:
+ f.write(packet["html"])
+ with open("debug_info/all_element.json", "w") as f:
+ f.write(json.dumps(packet["all_elements"]))
+
+def get_parsed_html(page):
+ if not os.path.exists("debug_info"):
+ os.makedirs("debug_info")
+
+ print("parsing html...")
+
+ packet = modify_page(page)
+ raw_html = packet["modified_html"]
+
+ args = {
+ "use_position": True,
+ "rect_dict": {},
+ "window_size": packet["window"],
+ "id-attr": "data-backend-node-id",
+ "label_attr": "data-label-id",
+ "label_generator": "order",
+ "regenerate_label": False,
+ "attr_list": basic_attrs,
+ "prompt": "xml",
+ "dataset": "pipeline"
+ }
+
+ hp = HtmlParser(raw_html, args)
+ res = hp.parse_tree()
+ page_html = res.get("html", "")
+
+ packet["html"] = page_html
+
+ # for debug
+ save_debug_info(packet)
+
+ print("parsing finished.")
+
+ return packet
+
diff --git a/VAB-WebArena-Lite/new/html_tools/html_parser.py b/VAB-WebArena-Lite/new/html_tools/html_parser.py
new file mode 100755
index 0000000..080342f
--- /dev/null
+++ b/VAB-WebArena-Lite/new/html_tools/html_parser.py
@@ -0,0 +1,447 @@
+from lxml import html
+import time, copy, random
+import json, re, os
+
+from .identifier import IdentifierTool
+from .prompt import HtmlPrompt
+from .configs import config_meta
+from .utils import get_xpath_top_down, rect2tuple
+
+class HtmlParser():
+ def __init__(self, ctx: str, args: dict[str]={}) -> None:
+ stt = time.time()
+ self.dom_tree = self.ctx2tree(ctx)
+ # tool related
+ self.bids2label = {}
+ self.bids2xpath = {}
+ self.used_labels = {}
+
+ # parse args
+ self.parse_args(args)
+ self.init_time = time.time() - stt
+
+ def parse_args(self, args: dict[str]={}) -> None:
+ def attr_check(attr, type_model='str'):
+ if attr is None:
+ return False
+ attr_type = type(attr)
+ if attr_type != type(type_model):
+ return False
+ if attr_type == type('str') and len(attr) == 0:
+ return False
+ return True
+
+ args = {} if args is None else args
+
+ # [Position] use_pos: False -> use full page, otherwise use window_size
+ dataset = args.get('dataset', '')
+ use_position = args.get('use_position', False)
+ window_size = args.get('window_size', None)
+ rect = args.get('rect_dict', None)
+ if use_position:
+ if not attr_check(window_size, ()):
+ raise ValueError('window_size must be set when use_position is True')
+ if not attr_check(rect, {}):
+ raise ValueError('rect_dict must be set when use_position is True')
+
+ if not attr_check(rect, {}):
+ rect = {}
+
+ # [Label] for vimium is temp_clickable_label, otherwise keep all of it
+ label_attr = args.get('label_attr', '')
+ get_new_label = args.get('regenerate_label', False)
+ label_method = args.get('label_generator', None)
+ regen_label = not attr_check(label_method)
+
+ # [id] for mind2web is backend_node_id, for normal website use our method
+ id_attr = args.get('id_attr', '')
+ regen_id = not attr_check(id_attr)
+
+ if regen_id:
+ id_attr = 'temp_id'
+
+ # [attributes]
+ keep_attrs = args.get('attr_list', [])
+ if not attr_check(keep_attrs, []):
+ keep_attrs = []
+
+ # [Tags] for clickable elem, keep: must keep, obs: keep if follow specific rule
+ parent_chain = args.get('parent_chain', False)
+ keep_elem = args.get('keep_elem', [])
+ obs_elem = args.get('obs_elem', [])
+
+ # sanity check
+ self.set_args(use_position, window_size, rect, label_attr, id_attr, keep_attrs, keep_elem, obs_elem, parent_chain, get_new_label, dataset)
+
+ # [Prompt]
+ prompt = args.get('prompt', None)
+ self.prompt = HtmlPrompt(prompt)
+
+ # traverse and get special data
+ if regen_id or regen_label:
+ self.mark_id()
+
+ if get_new_label:
+ self.used_labels = {}
+
+ self.identifier = IdentifierTool(label_method, self.used_labels)
+
+ def set_args(self, use_position: bool=False, window_size: tuple=(), rect_dict: dict[str]={}, label_attr: str='',
+ id_attr: str='', keep_attrs: list[str]=[], keep_elem: list[str]=[], obs_elem: list[str]=[],
+ parent_chain: bool=False, get_new_label: bool=False, dataset: str='') -> None:
+
+ self.use_position = use_position
+ self.window_size = window_size
+ self.rect = rect_dict
+ self.label_attr = label_attr
+ self.id_attr = id_attr
+ self.keep_attrs = keep_attrs
+ self.keep = keep_elem
+ self.obs = obs_elem
+ self.parent_chain = parent_chain
+ self.get_new_label = get_new_label
+ self.dataset = dataset
+
+ def get_config(self):
+ config = {
+ 'id_attr': self.id_attr,
+ 'keep_attrs': self.keep_attrs[:5],
+ 'label_attr': self.label_attr,
+ 'use_position': self.use_position,
+ 'window_size': self.window_size,
+ 'rect': dict(list(self.rect.items())[:3]),
+ 'keep_elem': self.keep[:5],
+ 'obs_elem': self.obs[:5],
+ 'parent_chain': self.parent_chain,
+ 'prompt_name': self.prompt.name,
+ 'identifier_name': self.identifier.name
+ }
+
+ return config, config_meta.format(**config)
+
+ def update_rect_dict(self, rect_dict: dict[str]={}) -> None:
+ self.rect = rect_dict
+
+ @staticmethod
+ def ctx2tree(ctx: str) -> html.HtmlElement:
+ # remove useless tags, eg. style and script
+ ctx = re.sub('', '', ctx)
+ ctx = re.sub('