| """ |
| Lightpanda Headless CDP Engine for Miniswardbower. |
| |
| Integrates lightpanda-io/browser (AI-native headless browser in Zig) |
| via Chrome DevTools Protocol (CDP) WebSocket endpoint. |
| Bypasses GPU rasterization and layout overhead for 9x-60x faster |
| DOM interactions, ultra-low memory (<30MB RAM), and instant startup (<15ms). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| import os |
| import socket |
| import subprocess |
| import tempfile |
| import time |
| from pathlib import Path |
| from typing import Optional, Tuple |
| from playwright.async_api import Browser, BrowserContext, Page, Playwright, async_playwright |
|
|
|
|
| class LightpandaServerManager: |
| """Manages the background lifecycle of the Lightpanda CDP server process.""" |
|
|
| def __init__( |
| self, |
| bin_path: Optional[str] = None, |
| host: str = "127.0.0.1", |
| port: Optional[int] = None, |
| log_level: str = "warn", |
| ): |
| self.bin_path = bin_path or "/home/psikosen/canopy/bin/lightpanda" |
| self.host = host |
| self.port = port |
| self.log_level = log_level |
| self._process: Optional[subprocess.Popen] = None |
| self._cdp_ws_url: Optional[str] = None |
| self._log = None |
|
|
| def command(self) -> list[str]: |
| return [str(self.bin_path), "serve", "--host", self.host, |
| "--port", str(self.port), "--log-level", self.log_level] |
|
|
| @staticmethod |
| def find_free_port() -> int: |
| """Finds an available TCP port assigned by the OS kernel.""" |
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: |
| s.bind(("127.0.0.1", 0)) |
| return s.getsockname()[1] |
|
|
| def is_available(self) -> bool: |
| """Checks if the lightpanda binary exists and is executable.""" |
| path = Path(self.bin_path) |
| return path.exists() and os.access(str(path), os.X_OK) |
|
|
| async def start(self) -> str: |
| """ |
| Starts the lightpanda serve process and waits until the CDP endpoint is responsive. |
| Returns the WebSocket CDP endpoint URL (e.g. ws://127.0.0.1:9222). |
| """ |
| if not self.is_available(): |
| raise FileNotFoundError(f"CDP browser binary not found or not executable at: {self.bin_path}") |
|
|
| if self.port is None: |
| self.port = self.find_free_port() |
|
|
| |
| self._log = tempfile.TemporaryFile(mode="w+") |
| self._process = subprocess.Popen( |
| self.command(), |
| stdout=self._log, |
| stderr=self._log, |
| text=True, |
| start_new_session=True, |
| ) |
|
|
| |
| self._cdp_ws_url = f"ws://{self.host}:{self.port}" |
| connected = False |
| deadline = time.monotonic() + 5.0 |
| while time.monotonic() < deadline: |
| if self._process.poll() is not None: |
| self._log.seek(0) |
| err = self._log.read()[-4000:] |
| code = self._process.returncode |
| self.stop() |
| raise RuntimeError(f"CDP browser exited with code {code}: {err}") |
| try: |
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: |
| s.settimeout(0.2) |
| if s.connect_ex((self.host, self.port)) == 0: |
| connected = True |
| break |
| except Exception: |
| pass |
| await asyncio.sleep(0.02) |
|
|
| if not connected: |
| self.stop() |
| raise TimeoutError(f"Timed out waiting for CDP browser at {self._cdp_ws_url}") |
|
|
| return self._cdp_ws_url |
|
|
| def stop(self) -> None: |
| """Stops the lightpanda server process gracefully.""" |
| if self._process is not None: |
| try: |
| self._process.terminate() |
| self._process.wait(timeout=2.0) |
| except Exception: |
| try: |
| self._process.kill() |
| self._process.wait(timeout=2.0) |
| except Exception: |
| pass |
| finally: |
| self._process = None |
| if self._log is not None: |
| self._log.close() |
| self._log = None |
|
|
| def get_memory_rss_mb(self) -> float: |
| """Returns the resident set size (RSS) memory of the Lightpanda daemon in MB.""" |
| if self._process is None or self._process.poll() is not None: |
| return 0.0 |
| try: |
| with open(f"/proc/{self._process.pid}/statm", "r") as f: |
| pages = int(f.read().split()[1]) |
| page_size = os.sysconf("SC_PAGE_SIZE") |
| return (pages * page_size) / (1024 * 1024) |
| except Exception: |
| return 0.0 |
|
|
|
|
| class LightpandaDriver: |
| """ |
| Playwright CDP Driver connected to Lightpanda. |
| Provides standard Playwright browser and page interfaces. |
| """ |
|
|
| def __init__(self, bin_path: Optional[str] = None): |
| self.server_mgr = LightpandaServerManager(bin_path=bin_path) |
| self._playwright: Optional[Playwright] = None |
| self._browser: Optional[Browser] = None |
| self._context: Optional[BrowserContext] = None |
| self._page: Optional[Page] = None |
|
|
| async def start(self) -> Tuple[Browser, BrowserContext, Page]: |
| """Launches Lightpanda daemon and connects Playwright via connect_over_cdp.""" |
| try: |
| ws_url = await self.server_mgr.start() |
| self._playwright = await async_playwright().start() |
| self._browser = await self._playwright.chromium.connect_over_cdp(ws_url, timeout=10000) |
| self._page = await self._browser.new_page() |
| self._context = self._page.context |
| return self._browser, self._context, self._page |
| except BaseException: |
| await self.stop() |
| raise |
|
|
| async def stop(self) -> None: |
| """Closes Playwright connection and terminates Lightpanda server.""" |
| try: |
| if self._page is not None: |
| await self._page.close() |
| except Exception: |
| pass |
|
|
| try: |
| if self._browser is not None: |
| await self._browser.close() |
| except Exception: |
| pass |
|
|
| try: |
| if self._playwright is not None: |
| await self._playwright.stop() |
| except Exception: |
| pass |
|
|
| self.server_mgr.stop() |
| self._page = None |
| self._context = None |
| self._browser = None |
| self._playwright = None |
|
|
