The goal of this environment is to produce a prototype of a browser environment. In the end, we want to support a fully configurable browser environment with wide range of action spaces and observation spaces, both structured and unstructured. But in this prototype, we just support
| 14 | |
| 15 | |
| 16 | class AsyncScriptBrowserEnv(Env[npt.NDArray[np.uint8], Action]): |
| 17 | """ |
| 18 | The goal of this environment is to produce a prototype of a browser environment. |
| 19 | In the end, we want to support a fully configurable browser environment with wide |
| 20 | range of action spaces and observation spaces, both structured and unstructured. |
| 21 | But in this prototype, we just support action space specified by Playwright script, |
| 22 | and observation space is the html content of the page. |
| 23 | """ |
| 24 | |
| 25 | def __init__( |
| 26 | self, |
| 27 | max_page_length: int = 2048, |
| 28 | headless: bool = True, |
| 29 | slow_mo: int = 0, |
| 30 | timeout: int = 30000, |
| 31 | viewport_size: ViewportSize = {"width": 1280, "height": 720}, |
| 32 | ): |
| 33 | self.observation_space = Box( |
| 34 | 0, |
| 35 | 255, |
| 36 | (viewport_size["height"], viewport_size["width"], 4), |
| 37 | np.uint8, |
| 38 | ) |
| 39 | # TODO: make Space[Action] = ActionSpace |
| 40 | self.action_space = get_action_space() # type: ignore[assignment] |
| 41 | self.headless = headless |
| 42 | self.slow_mo = slow_mo |
| 43 | self.reset_finished = False |
| 44 | self.timeout = timeout |
| 45 | self.viewport_size = viewport_size |
| 46 | |
| 47 | async def setup(self, config_file: Path | None = None) -> None: |
| 48 | self.context_manager = async_playwright() |
| 49 | self.playwright = await self.context_manager.__aenter__() |
| 50 | self.browser = await self.playwright.chromium.launch( |
| 51 | headless=self.headless, slow_mo=self.slow_mo |
| 52 | ) |
| 53 | if config_file: |
| 54 | with open(config_file, "r") as f: |
| 55 | instance_config = json.load(f) |
| 56 | else: |
| 57 | instance_config = {} |
| 58 | |
| 59 | storage_state = instance_config.get("storage_state", None) |
| 60 | start_url = instance_config.get("start_url", None) |
| 61 | geolocation = instance_config.get("geolocation", None) |
| 62 | |
| 63 | self.context = await self.browser.new_context( |
| 64 | viewport=self.viewport_size, |
| 65 | storage_state=storage_state, |
| 66 | geolocation=geolocation, |
| 67 | device_scale_factor=1, |
| 68 | ) |
| 69 | self.page = await self.context.new_page() |
| 70 | if start_url: |
| 71 | await self.page.goto(start_url) |
| 72 | |
| 73 | async def areset( |
no outgoing calls