chore: import upstream snapshot with attribution

browser_use/browser/python_highlights.py

@@ -0,0 +1,548 @@
+"""Python-based highlighting system for drawing bounding boxes on screenshots.
+
+This module replaces JavaScript-based highlighting with fast Python image processing
+to draw bounding boxes around interactive elements directly on screenshots.
+"""
+
+import asyncio
+import base64
+import io
+import logging
+import os
+
+from PIL import Image, ImageDraw, ImageFont
+
+from browser_use.dom.views import DOMSelectorMap, EnhancedDOMTreeNode
+from browser_use.observability import observe_debug
+from browser_use.utils import time_execution_async
+
+logger = logging.getLogger(__name__)
+
+# Font cache to prevent repeated font loading and reduce memory usage
+_FONT_CACHE: dict[tuple[str, int], ImageFont.FreeTypeFont | None] = {}
+
+# Cross-platform font paths
+_FONT_PATHS = [
+	'/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf',  # Linux (Debian/Ubuntu)
+	'/usr/share/fonts/TTF/DejaVuSans-Bold.ttf',  # Linux (Arch/Fedora)
+	'/System/Library/Fonts/Arial.ttf',  # macOS
+	'C:\\Windows\\Fonts\\arial.ttf',  # Windows
+	'arial.ttf',  # Windows (system path)
+	'Arial Bold.ttf',  # macOS alternative
+	'/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf',  # Linux alternative
+]
+
+
+def get_cross_platform_font(font_size: int) -> ImageFont.FreeTypeFont | None:
+	"""Get a cross-platform compatible font with caching to prevent memory leaks.
+
+	Args:
+	    font_size: Size of the font to load
+
+	Returns:
+	    ImageFont object or None if no system fonts are available
+	"""
+	# Use cache key based on font size
+	cache_key = ('system_font', font_size)
+
+	# Return cached font if available
+	if cache_key in _FONT_CACHE:
+		return _FONT_CACHE[cache_key]
+
+	# Try to load a system font
+	font = None
+	for font_path in _FONT_PATHS:
+		try:
+			font = ImageFont.truetype(font_path, font_size)
+			break
+		except OSError:
+			continue
+
+	# Cache the result (even if None) to avoid repeated attempts
+	_FONT_CACHE[cache_key] = font
+	return font
+
+
+def cleanup_font_cache() -> None:
+	"""Clean up the font cache to prevent memory leaks in long-running applications."""
+	global _FONT_CACHE
+	_FONT_CACHE.clear()
+
+
+# Color scheme for different element types
+ELEMENT_COLORS = {
+	'button': '#FF6B6B',  # Red for buttons
+	'input': '#4ECDC4',  # Teal for inputs
+	'select': '#45B7D1',  # Blue for dropdowns
+	'a': '#96CEB4',  # Green for links
+	'textarea': '#FF8C42',  # Orange for text areas (was yellow, now more visible)
+	'default': '#DDA0DD',  # Light purple for other interactive elements
+}
+
+# Element type mappings
+ELEMENT_TYPE_MAP = {
+	'button': 'button',
+	'input': 'input',
+	'select': 'select',
+	'a': 'a',
+	'textarea': 'textarea',
+}
+
+
+def get_element_color(tag_name: str, element_type: str | None = None) -> str:
+	"""Get color for element based on tag name and type."""
+	# Check input type first
+	if tag_name == 'input' and element_type:
+		if element_type in ['button', 'submit']:
+			return ELEMENT_COLORS['button']
+
+	# Use tag-based color
+	return ELEMENT_COLORS.get(tag_name.lower(), ELEMENT_COLORS['default'])
+
+
+def should_show_index_overlay(backend_node_id: int | None) -> bool:
+	"""Determine if index overlay should be shown."""
+	return backend_node_id is not None
+
+
+def draw_enhanced_bounding_box_with_text(
+	draw,  # ImageDraw.Draw - avoiding type annotation due to PIL typing issues
+	bbox: tuple[int, int, int, int],
+	color: str,
+	text: str | None = None,
+	font: ImageFont.FreeTypeFont | None = None,
+	element_type: str = 'div',
+	image_size: tuple[int, int] = (2000, 1500),
+	device_pixel_ratio: float = 1.0,
+) -> None:
+	"""Draw an enhanced bounding box with much bigger index containers and dashed borders."""
+	x1, y1, x2, y2 = bbox
+
+	# Draw dashed bounding box with pattern: 1 line, 2 spaces, 1 line, 2 spaces...
+	dash_length = 4
+	gap_length = 8
+	line_width = 2
+
+	# Helper function to draw dashed line
+	def draw_dashed_line(start_x, start_y, end_x, end_y):
+		if start_x == end_x:  # Vertical line
+			y = start_y
+			while y < end_y:
+				dash_end = min(y + dash_length, end_y)
+				draw.line([(start_x, y), (start_x, dash_end)], fill=color, width=line_width)
+				y += dash_length + gap_length
+		else:  # Horizontal line
+			x = start_x
+			while x < end_x:
+				dash_end = min(x + dash_length, end_x)
+				draw.line([(x, start_y), (dash_end, start_y)], fill=color, width=line_width)
+				x += dash_length + gap_length
+
+	# Draw dashed rectangle
+	draw_dashed_line(x1, y1, x2, y1)  # Top
+	draw_dashed_line(x2, y1, x2, y2)  # Right
+	draw_dashed_line(x2, y2, x1, y2)  # Bottom
+	draw_dashed_line(x1, y2, x1, y1)  # Left
+
+	# Draw much bigger index overlay if we have index text
+	if text:
+		try:
+			# Scale font size for appropriate sizing across different resolutions
+			img_width, img_height = image_size
+
+			css_width = img_width  # / device_pixel_ratio
+			# Much smaller scaling - 1% of CSS viewport width, max 16px to prevent huge highlights
+			base_font_size = max(10, min(20, int(css_width * 0.01)))
+			# Use shared font loading function with caching
+			big_font = get_cross_platform_font(base_font_size)
+			if big_font is None:
+				big_font = font  # Fallback to original font if no system fonts found
+
+			# Get text size with bigger font
+			if big_font:
+				bbox_text = draw.textbbox((0, 0), text, font=big_font)
+				text_width = bbox_text[2] - bbox_text[0]
+				text_height = bbox_text[3] - bbox_text[1]
+			else:
+				# Fallback for default font
+				bbox_text = draw.textbbox((0, 0), text)
+				text_width = bbox_text[2] - bbox_text[0]
+				text_height = bbox_text[3] - bbox_text[1]
+
+			# Scale padding appropriately for different resolutions
+			padding = max(4, min(10, int(css_width * 0.005)))  # 0.3% of CSS width, max 4px
+			element_width = x2 - x1
+			element_height = y2 - y1
+
+			# Container dimensions
+			container_width = text_width + padding * 2
+			container_height = text_height + padding * 2
+
+			# Position in top center - for small elements, place further up to avoid blocking content
+			# Center horizontally within the element
+			bg_x1 = x1 + (element_width - container_width) // 2
+
+			# Simple rule: if element is small, place index further up to avoid blocking icons
+			if element_width < 60 or element_height < 30:
+				# Small element: place well above to avoid blocking content
+				bg_y1 = max(0, y1 - container_height - 5)
+			else:
+				# Regular element: place inside with small offset
+				bg_y1 = y1 + 2
+
+			bg_x2 = bg_x1 + container_width
+			bg_y2 = bg_y1 + container_height
+
+			# Center the number within the index box with proper baseline handling
+			text_x = bg_x1 + (container_width - text_width) // 2
+			# Add extra vertical space to prevent clipping
+			text_y = bg_y1 + (container_height - text_height) // 2 - bbox_text[1]  # Subtract top offset
+
+			# Ensure container stays within image bounds
+			img_width, img_height = image_size
+			if bg_x1 < 0:
+				offset = -bg_x1
+				bg_x1 += offset
+				bg_x2 += offset
+				text_x += offset
+			if bg_y1 < 0:
+				offset = -bg_y1
+				bg_y1 += offset
+				bg_y2 += offset
+				text_y += offset
+			if bg_x2 > img_width:
+				offset = bg_x2 - img_width
+				bg_x1 -= offset
+				bg_x2 -= offset
+				text_x -= offset
+			if bg_y2 > img_height:
+				offset = bg_y2 - img_height
+				bg_y1 -= offset
+				bg_y2 -= offset
+				text_y -= offset
+
+			# Draw bigger background rectangle with thicker border
+			draw.rectangle([bg_x1, bg_y1, bg_x2, bg_y2], fill=color, outline='white', width=2)
+
+			# Draw white text centered in the index box
+			draw.text((text_x, text_y), text, fill='white', font=big_font or font)
+
+		except Exception as e:
+			logger.debug(f'Failed to draw enhanced text overlay: {e}')
+
+
+def draw_bounding_box_with_text(
+	draw,  # ImageDraw.Draw - avoiding type annotation due to PIL typing issues
+	bbox: tuple[int, int, int, int],
+	color: str,
+	text: str | None = None,
+	font: ImageFont.FreeTypeFont | None = None,
+) -> None:
+	"""Draw a bounding box with optional text overlay."""
+	x1, y1, x2, y2 = bbox
+
+	# Draw dashed bounding box
+	dash_length = 2
+	gap_length = 6
+
+	# Top edge
+	x = x1
+	while x < x2:
+		end_x = min(x + dash_length, x2)
+		draw.line([(x, y1), (end_x, y1)], fill=color, width=2)
+		draw.line([(x, y1 + 1), (end_x, y1 + 1)], fill=color, width=2)
+		x += dash_length + gap_length
+
+	# Bottom edge
+	x = x1
+	while x < x2:
+		end_x = min(x + dash_length, x2)
+		draw.line([(x, y2), (end_x, y2)], fill=color, width=2)
+		draw.line([(x, y2 - 1), (end_x, y2 - 1)], fill=color, width=2)
+		x += dash_length + gap_length
+
+	# Left edge
+	y = y1
+	while y < y2:
+		end_y = min(y + dash_length, y2)
+		draw.line([(x1, y), (x1, end_y)], fill=color, width=2)
+		draw.line([(x1 + 1, y), (x1 + 1, end_y)], fill=color, width=2)
+		y += dash_length + gap_length
+
+	# Right edge
+	y = y1
+	while y < y2:
+		end_y = min(y + dash_length, y2)
+		draw.line([(x2, y), (x2, end_y)], fill=color, width=2)
+		draw.line([(x2 - 1, y), (x2 - 1, end_y)], fill=color, width=2)
+		y += dash_length + gap_length
+
+	# Draw index overlay if we have index text
+	if text:
+		try:
+			# Get text size
+			if font:
+				bbox_text = draw.textbbox((0, 0), text, font=font)
+				text_width = bbox_text[2] - bbox_text[0]
+				text_height = bbox_text[3] - bbox_text[1]
+			else:
+				# Fallback for default font
+				bbox_text = draw.textbbox((0, 0), text)
+				text_width = bbox_text[2] - bbox_text[0]
+				text_height = bbox_text[3] - bbox_text[1]
+
+			# Smart positioning based on element size
+			padding = 5
+			element_width = x2 - x1
+			element_height = y2 - y1
+			element_area = element_width * element_height
+			index_box_area = (text_width + padding * 2) * (text_height + padding * 2)
+
+			# Calculate size ratio to determine positioning strategy
+			size_ratio = element_area / max(index_box_area, 1)
+
+			if size_ratio < 4:
+				# Very small elements: place outside in bottom-right corner
+				text_x = x2 + padding
+				text_y = y2 - text_height
+				# Ensure it doesn't go off screen
+				text_x = min(text_x, 1200 - text_width - padding)
+				text_y = max(text_y, 0)
+			elif size_ratio < 16:
+				# Medium elements: place in bottom-right corner inside
+				text_x = x2 - text_width - padding
+				text_y = y2 - text_height - padding
+			else:
+				# Large elements: place in center
+				text_x = x1 + (element_width - text_width) // 2
+				text_y = y1 + (element_height - text_height) // 2
+
+			# Ensure text stays within bounds
+			text_x = max(0, min(text_x, 1200 - text_width))
+			text_y = max(0, min(text_y, 800 - text_height))
+
+			# Draw background rectangle for maximum contrast
+			bg_x1 = text_x - padding
+			bg_y1 = text_y - padding
+			bg_x2 = text_x + text_width + padding
+			bg_y2 = text_y + text_height + padding
+
+			# Use white background with thick black border for maximum visibility
+			draw.rectangle([bg_x1, bg_y1, bg_x2, bg_y2], fill='white', outline='black', width=2)
+
+			# Draw bold dark text on light background for best contrast
+			draw.text((text_x, text_y), text, fill='black', font=font)
+
+		except Exception as e:
+			logger.debug(f'Failed to draw text overlay: {e}')
+
+
+def process_element_highlight(
+	element_id: int,
+	element: EnhancedDOMTreeNode,
+	draw,
+	device_pixel_ratio: float,
+	font,
+	filter_highlight_ids: bool,
+	image_size: tuple[int, int],
+) -> None:
+	"""Process a single element for highlighting."""
+	try:
+		# Use absolute_position coordinates directly
+		if not element.absolute_position:
+			return
+
+		bounds = element.absolute_position
+
+		# Scale coordinates from CSS pixels to device pixels for screenshot
+		# The screenshot is captured at device pixel resolution, but coordinates are in CSS pixels
+		x1 = int(bounds.x * device_pixel_ratio)
+		y1 = int(bounds.y * device_pixel_ratio)
+		x2 = int((bounds.x + bounds.width) * device_pixel_ratio)
+		y2 = int((bounds.y + bounds.height) * device_pixel_ratio)
+
+		# Ensure coordinates are within image bounds
+		img_width, img_height = image_size
+		x1 = max(0, min(x1, img_width))
+		y1 = max(0, min(y1, img_height))
+		x2 = max(x1, min(x2, img_width))
+		y2 = max(y1, min(y2, img_height))
+
+		# Skip if bounding box is too small or invalid
+		if x2 - x1 < 2 or y2 - y1 < 2:
+			return
+
+		# Get element color based on type
+		tag_name = element.tag_name if hasattr(element, 'tag_name') else 'div'
+		element_type = None
+		if hasattr(element, 'attributes') and element.attributes:
+			element_type = element.attributes.get('type')
+
+		color = get_element_color(tag_name, element_type)
+
+		# Get element index for overlay and apply filtering
+		backend_node_id = getattr(element, 'backend_node_id', None)
+		index_text = None
+
+		if backend_node_id is not None:
+			if filter_highlight_ids:
+				# Use the meaningful text that matches what the LLM sees
+				meaningful_text = element.get_meaningful_text_for_llm()
+				# Show ID only if meaningful text is less than 5 characters
+				if len(meaningful_text) < 3:
+					index_text = str(backend_node_id)
+			else:
+				# Always show ID when filter is disabled
+				index_text = str(backend_node_id)
+
+		# Draw enhanced bounding box with bigger index
+		draw_enhanced_bounding_box_with_text(
+			draw, (x1, y1, x2, y2), color, index_text, font, tag_name, image_size, device_pixel_ratio
+		)
+
+	except Exception as e:
+		logger.debug(f'Failed to draw highlight for element {element_id}: {e}')
+
+
+@observe_debug(ignore_input=True, ignore_output=True, name='create_highlighted_screenshot')
+@time_execution_async('create_highlighted_screenshot')
+async def create_highlighted_screenshot(
+	screenshot_b64: str,
+	selector_map: DOMSelectorMap,
+	device_pixel_ratio: float = 1.0,
+	viewport_offset_x: int = 0,
+	viewport_offset_y: int = 0,
+	filter_highlight_ids: bool = True,
+) -> str:
+	"""Create a highlighted screenshot with bounding boxes around interactive elements.
+
+	Args:
+	    screenshot_b64: Base64 encoded screenshot
+	    selector_map: Map of interactive elements with their positions
+	    device_pixel_ratio: Device pixel ratio for scaling coordinates
+	    viewport_offset_x: X offset for viewport positioning
+	    viewport_offset_y: Y offset for viewport positioning
+
+	Returns:
+	    Base64 encoded highlighted screenshot
+	"""
+	try:
+		# Decode screenshot
+		screenshot_data = base64.b64decode(screenshot_b64)
+		image = Image.open(io.BytesIO(screenshot_data)).convert('RGBA')
+
+		# Create drawing context
+		draw = ImageDraw.Draw(image)
+
+		# Load font using shared function with caching
+		font = get_cross_platform_font(12)
+		# If no system fonts found, font remains None and will use default font
+
+		# Process elements sequentially to avoid ImageDraw thread safety issues
+		# PIL ImageDraw is not thread-safe, so we process elements one by one
+		for element_id, element in selector_map.items():
+			process_element_highlight(element_id, element, draw, device_pixel_ratio, font, filter_highlight_ids, image.size)
+
+		# Convert back to base64
+		output_buffer = io.BytesIO()
+		try:
+			image.save(output_buffer, format='PNG')
+			output_buffer.seek(0)
+			highlighted_b64 = base64.b64encode(output_buffer.getvalue()).decode('utf-8')
+
+			logger.debug(f'Successfully created highlighted screenshot with {len(selector_map)} elements')
+			return highlighted_b64
+		finally:
+			# Explicit cleanup to prevent memory leaks
+			output_buffer.close()
+			if 'image' in locals():
+				image.close()
+
+	except Exception as e:
+		logger.error(f'Failed to create highlighted screenshot: {e}')
+		# Clean up on error as well
+		if 'image' in locals():
+			image.close()
+		# Return original screenshot on error
+		return screenshot_b64
+
+
+async def get_viewport_info_from_cdp(cdp_session) -> tuple[float, int, int]:
+	"""Get viewport information from CDP session.
+
+	Returns:
+	    Tuple of (device_pixel_ratio, scroll_x, scroll_y)
+	"""
+	try:
+		# Get layout metrics which includes viewport info and device pixel ratio
+		metrics = await cdp_session.cdp_client.send.Page.getLayoutMetrics(session_id=cdp_session.session_id)
+
+		# Extract viewport information
+		visual_viewport = metrics.get('visualViewport', {})
+		css_visual_viewport = metrics.get('cssVisualViewport', {})
+		css_layout_viewport = metrics.get('cssLayoutViewport', {})
+
+		# Calculate device pixel ratio
+		css_width = css_visual_viewport.get('clientWidth', css_layout_viewport.get('clientWidth', 1280.0))
+		device_width = visual_viewport.get('clientWidth', css_width)
+		device_pixel_ratio = device_width / css_width if css_width > 0 else 1.0
+
+		# Get scroll position in CSS pixels
+		scroll_x = int(css_visual_viewport.get('pageX', 0))
+		scroll_y = int(css_visual_viewport.get('pageY', 0))
+
+		return float(device_pixel_ratio), scroll_x, scroll_y
+
+	except Exception as e:
+		logger.debug(f'Failed to get viewport info from CDP: {e}')
+		return 1.0, 0, 0
+
+
+@time_execution_async('create_highlighted_screenshot_async')
+async def create_highlighted_screenshot_async(
+	screenshot_b64: str, selector_map: DOMSelectorMap, cdp_session=None, filter_highlight_ids: bool = True
+) -> str:
+	"""Async wrapper for creating highlighted screenshots.
+
+	Args:
+	    screenshot_b64: Base64 encoded screenshot
+	    selector_map: Map of interactive elements
+	    cdp_session: CDP session for getting viewport info
+	    filter_highlight_ids: Whether to filter element IDs based on meaningful text
+
+	Returns:
+	    Base64 encoded highlighted screenshot
+	"""
+	# Get viewport information if CDP session is available
+	device_pixel_ratio = 1.0
+	viewport_offset_x = 0
+	viewport_offset_y = 0
+
+	if cdp_session:
+		try:
+			device_pixel_ratio, viewport_offset_x, viewport_offset_y = await get_viewport_info_from_cdp(cdp_session)
+		except Exception as e:
+			logger.debug(f'Failed to get viewport info from CDP: {e}')
+
+	# Create highlighted screenshot with async processing
+	final_screenshot = await create_highlighted_screenshot(
+		screenshot_b64, selector_map, device_pixel_ratio, viewport_offset_x, viewport_offset_y, filter_highlight_ids
+	)
+
+	filename = os.getenv('BROWSER_USE_SCREENSHOT_FILE')
+	if filename:
+
+		def _write_screenshot():
+			try:
+				with open(filename, 'wb') as f:
+					f.write(base64.b64decode(final_screenshot))
+				logger.debug('Saved screenshot to ' + str(filename))
+			except Exception as e:
+				logger.warning(f'Failed to save screenshot to {filename}: {e}')
+
+		await asyncio.to_thread(_write_screenshot)
+	return final_screenshot
+
+
+# Export the cleanup function for external use in long-running applications
+__all__ = ['create_highlighted_screenshot', 'create_highlighted_screenshot_async', 'cleanup_font_cache']