File size: 6,548 Bytes
ed79a7f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """
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()
# A file avoids blocking the child on a full, unread stderr pipe.
self._log = tempfile.TemporaryFile(mode="w+")
self._process = subprocess.Popen(
self.command(),
stdout=self._log,
stderr=self._log,
text=True,
start_new_session=True,
)
# Wait for socket to become connectable
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
|