livekit--agents/livekit-plugins/livekit-plugins-browser/livekit/plugins/browser/page_actions.py

"""PageActions — typed input API for a CEF BrowserPage."""

from __future__ import annotations

import asyncio
from typing import Any

from livekit import rtc
from livekit.browser import BrowserPage  # type: ignore[import-untyped]

from ._keys import (
    CHAR,
    KEY_NAME_TO_VK,
    KEYUP,
    MOD_SHIFT,
    MODIFIER_MAP,
    NATIVE_KEY_CODES,
    NON_CHAR_KEYS,
    RAWKEYDOWN,
    SHIFTED_CHAR_TO_VK,
)

VK_SHIFT = 16


class PageActions:
    """Typed input API for a CEF BrowserPage with frame capture.

    Usage::

        actions = PageActions(page=page)
        await actions.left_click(100, 200)
        frame = actions.last_frame
    """

    def __init__(self, *, page: BrowserPage) -> None:
        self._page = page
        self._lock = asyncio.Lock()
        self._last_frame: rtc.VideoFrame | None = None
        self._page.on("paint", self._on_paint)

    def _on_paint(self, data: Any) -> None:
        self._last_frame = data.frame

    @property
    def last_frame(self) -> rtc.VideoFrame | None:
        return self._last_frame

    async def left_click(self, x: int, y: int, *, modifiers: str | None = None) -> None:
        async with self._lock:
            mod_keys = _parse_modifier_keys(modifiers)
            await self._page.send_mouse_move(x, y)
            await _press_modifiers(self._page, mod_keys)
            await self._page.send_mouse_click(x, y, 0, False, 1)
            await self._page.send_mouse_click(x, y, 0, True, 1)
            await _release_modifiers(self._page, mod_keys)

    async def right_click(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 2, False, 1)
            await self._page.send_mouse_click(x, y, 2, True, 1)

    async def double_click(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 0, False, 1)
            await self._page.send_mouse_click(x, y, 0, True, 1)
            await self._page.send_mouse_click(x, y, 0, False, 2)
            await self._page.send_mouse_click(x, y, 0, True, 2)

    async def triple_click(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 0, False, 1)
            await self._page.send_mouse_click(x, y, 0, True, 1)
            await self._page.send_mouse_click(x, y, 0, False, 2)
            await self._page.send_mouse_click(x, y, 0, True, 2)
            await self._page.send_mouse_click(x, y, 0, False, 3)
            await self._page.send_mouse_click(x, y, 0, True, 3)

    async def middle_click(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 1, False, 1)
            await self._page.send_mouse_click(x, y, 1, True, 1)

    async def mouse_move(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)

    async def left_click_drag(self, *, start_x: int, start_y: int, end_x: int, end_y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(start_x, start_y)
            await self._page.send_mouse_click(start_x, start_y, 0, False, 1)
            await asyncio.sleep(0.05)
            await self._page.send_mouse_move(end_x, end_y)
            await asyncio.sleep(0.05)
            await self._page.send_mouse_click(end_x, end_y, 0, True, 1)

    async def left_mouse_down(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 0, False, 1)

    async def left_mouse_up(self, x: int, y: int) -> None:
        async with self._lock:
            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_click(x, y, 0, True, 1)

    async def scroll(
        self,
        x: int,
        y: int,
        *,
        direction: str = "down",
        amount: int = 3,
    ) -> None:
        async with self._lock:
            pixels = amount * 120

            delta_x, delta_y = 0, 0
            if direction == "down":
                delta_y = -pixels
            elif direction == "up":
                delta_y = pixels
            elif direction == "left":
                delta_x = pixels
            elif direction == "right":
                delta_x = -pixels

            await self._page.send_mouse_move(x, y)
            await self._page.send_mouse_wheel(x, y, delta_x, delta_y)

    async def type_text(self, text: str) -> None:
        async with self._lock:
            for ch in text:
                char_code = ord(ch)
                shifted = False

                if ch.isalpha():
                    vk = ord(ch.upper())
                    shifted = ch.isupper()
                elif ch in SHIFTED_CHAR_TO_VK:
                    vk = SHIFTED_CHAR_TO_VK[ch]
                    shifted = True
                else:
                    vk = KEY_NAME_TO_VK.get(ch, 0)

                if not vk:
                    # Unknown character (e.g. unicode) — CHAR-only
                    await self._page.send_key_event(CHAR, 0, 0, 0, char_code)
                    await asyncio.sleep(0.01)
                    continue

                modifiers = MOD_SHIFT if shifted else 0
                nkc = NATIVE_KEY_CODES.get(vk, 0)

                if shifted:
                    shift_nkc = NATIVE_KEY_CODES.get(VK_SHIFT, 0)
                    await self._page.send_key_event(RAWKEYDOWN, MOD_SHIFT, VK_SHIFT, shift_nkc, 0)

                await self._page.send_key_event(RAWKEYDOWN, modifiers, vk, nkc, 0)
                await self._page.send_key_event(CHAR, modifiers, char_code, nkc, char_code)
                await self._page.send_key_event(KEYUP, modifiers, vk, 0, 0)

                if shifted:
                    await self._page.send_key_event(KEYUP, 0, VK_SHIFT, 0, 0)

                await asyncio.sleep(0.01)

    async def key(self, text: str) -> None:
        async with self._lock:
            await _send_key_combo(self._page, text)

    async def hold_key(self, text: str, *, duration: float = 0.5) -> None:
        async with self._lock:
            keys = [k.strip().lower() for k in text.split("+")]

            modifiers = 0
            for k in keys:
                if k in MODIFIER_MAP:
                    modifiers |= MODIFIER_MAP[k]
                vk = KEY_NAME_TO_VK.get(k, 0)
                nkc = NATIVE_KEY_CODES.get(vk, 0)
                await self._page.send_key_event(RAWKEYDOWN, modifiers, vk, nkc, 0)

            await asyncio.sleep(duration)

            for k in reversed(keys):
                vk = KEY_NAME_TO_VK.get(k, 0)
                await self._page.send_key_event(KEYUP, 0, vk, 0, 0)

    async def wait(self) -> None:
        async with self._lock:
            await asyncio.sleep(1)

    async def navigate(self, url: str) -> None:
        async with self._lock:
            await self._page.navigate(url)

    async def go_back(self) -> None:
        async with self._lock:
            await self._page.go_back()

    async def go_forward(self) -> None:
        async with self._lock:
            await self._page.go_forward()

    def aclose(self) -> None:
        self._page.off("paint", self._on_paint)


def _parse_modifier_keys(text: str | None) -> list[str]:
    if not text:
        return []
    return [part.strip().lower() for part in text.split("+") if part.strip()]


async def _press_modifiers(page: BrowserPage, keys: list[str]) -> None:
    modifiers = 0
    for k in keys:
        modifiers |= MODIFIER_MAP.get(k, 0)
        vk = KEY_NAME_TO_VK.get(k, 0)
        nkc = NATIVE_KEY_CODES.get(vk, 0)
        await page.send_key_event(RAWKEYDOWN, modifiers, vk, nkc, 0)


async def _release_modifiers(page: BrowserPage, keys: list[str]) -> None:
    for k in reversed(keys):
        vk = KEY_NAME_TO_VK.get(k, 0)
        await page.send_key_event(KEYUP, 0, vk, 0, 0)


async def _send_key_combo(page: BrowserPage, text: str) -> None:
    keys = [k.strip().lower() for k in text.split("+")]

    modifiers = 0
    main_keys: list[str] = []
    for k in keys:
        if k in MODIFIER_MAP:
            modifiers |= MODIFIER_MAP[k]
        else:
            main_keys.append(k)

    # press modifiers down
    for k in keys:
        if k in MODIFIER_MAP:
            vk = KEY_NAME_TO_VK.get(k, 0)
            nkc = NATIVE_KEY_CODES.get(vk, 0)
            await page.send_key_event(RAWKEYDOWN, modifiers, vk, nkc, 0)

    # press and release main keys
    for k in main_keys:
        vk = KEY_NAME_TO_VK.get(k, 0)
        if vk == 0 and len(k) == 1:
            vk = ord(k.upper())
        nkc = NATIVE_KEY_CODES.get(vk, 0)
        await page.send_key_event(RAWKEYDOWN, modifiers, vk, nkc, 0)
        if vk not in NON_CHAR_KEYS and len(k) == 1:
            char_code = ord(k)
            await page.send_key_event(CHAR, modifiers, char_code, nkc, char_code)
        await page.send_key_event(KEYUP, modifiers, vk, 0, 0)

    # release modifiers
    for k in reversed(keys):
        if k in MODIFIER_MAP:
            vk = KEY_NAME_TO_VK.get(k, 0)
            await page.send_key_event(KEYUP, 0, vk, 0, 0)