pulse/scripts/demo/capture.py

"""Regenerates docs/img/* from a LOCAL demo instance (fictional data). See scripts/demo/README.md."""

# flake8: noqa: E501  (demo data and canned output contain long literal lines)

import asyncio
import io
from PIL import Image
from playwright.async_api import async_playwright
import pathlib

OUT = str(pathlib.Path(__file__).resolve().parents[2] / "docs" / "img")
B = "http://127.0.0.1:8400"
H = {
    "Remote-User": "alex.rivera",
    "Remote-Name": "Alex Rivera",
    "Remote-Email": "alex@example.com",
    "Remote-Groups": "admin,employee",
}
TOP = (
    "document.activeElement&&document.activeElement.blur();"
    "document.documentElement.style.scrollBehavior='auto';window.scrollTo(0,0)"
)


async def mk(p, theme="dark", w=1440, h=900):
    b = await p.chromium.launch()
    c = await b.new_context(viewport={"width": w, "height": h}, extra_http_headers=H)
    await c.add_init_script(
        f"try{{localStorage.setItem('lt_theme','{theme}');sessionStorage.setItem('lt_boot','1')}}catch(e){{}}"
    )
    return b, c


async def go(pg, path, wait=3500):
    await pg.goto(B + path)
    await pg.wait_for_timeout(wait)
    await pg.evaluate(TOP)
    await pg.wait_for_timeout(300)


async def shots():
    async with async_playwright() as p:
        b, c = await mk(p)
        pg = await c.new_page()
        errs = []
        pg.on("pageerror", lambda e: errs.append(str(e)[:100]))
        for path, name in (
            ("/", "dashboard"),
            ("/workers", "workers"),
            ("/workflows", "workflows"),
            ("/executions", "executions"),
            ("/quick", "quick-command"),
            ("/scheduler", "scheduler"),
        ):
            await go(pg, path, 5000 if path == "/" else 3500)
            await pg.screenshot(path=f"{OUT}/{name}.png")
            print(name, "ok")
        print("errors", errs[:3])
        await b.close()


async def open_exec(pg, text, idx=0, wait=1500):
    await go(pg, "/executions", 3000)
    await pg.locator("tr", has_text=text).nth(idx).click()
    await pg.wait_for_timeout(wait)


async def scroll_modal(pg, y):
    await pg.evaluate(
        "(y)=>{const m=[...document.querySelectorAll('*')].filter(e=>e.scrollHeight>e.clientHeight+40&&['auto','scroll'].includes(getComputedStyle(e).overflowY)&&e.clientHeight>200); m.forEach(e=>e.scrollTop=y)}",
        y,
    )
    await pg.wait_for_timeout(400)


async def shots2():
    async with async_playwright() as p:
        b, c = await mk(p)
        pg = await c.new_page()
        await open_exec(pg, "Rolling restart", 1)
        await pg.screenshot(path=f"{OUT}/execution-detail.png")
        await scroll_modal(pg, 700)
        await pg.screenshot(path=f"{OUT}/execution-detail-steps.png")
        await open_exec(pg, "Check backup mount", 0)
        await pg.screenshot(path=f"{OUT}/execution-failed.png")
        await open_exec(pg, "Rolling restart", 0, 2500)
        await pg.screenshot(path=f"{OUT}/execution-awaiting-approval.png")
        # workflow editor
        await go(pg, "/workflows", 3000)
        await pg.locator("tr", has_text="Rolling restart").get_by_text("EDIT", exact=False).first.click()
        await pg.wait_for_timeout(1200)
        await pg.screenshot(path=f"{OUT}/workflow-editor.png")
        # quick command with output
        await go(pg, "/quick", 3000)
        try:
            await pg.get_by_text("Multiple Workers").click()
            await pg.wait_for_timeout(300)
            await pg.get_by_text("SELECT ALL", exact=False).first.click()
            await pg.wait_for_timeout(300)
        except Exception as e:
            print("multi fail", str(e)[:60])
        await pg.locator("textarea").first.fill('echo "== $(hostname) =="; uptime; df -h / | tail -1')
        await pg.wait_for_timeout(300)
        await pg.screenshot(path=f"{OUT}/quick-command.png")
        await pg.keyboard.press("Control+Enter")
        await pg.wait_for_timeout(3500)
        await pg.screenshot(path=f"{OUT}/quick-command-result.png")
        # palette
        await go(pg, "/", 3500)
        await pg.keyboard.press("Control+k")
        await pg.wait_for_timeout(500)
        await pg.keyboard.type("exec", delay=70)
        await pg.wait_for_timeout(700)
        await pg.screenshot(path=f"{OUT}/command-palette.png", clip={"x": 320, "y": 60, "width": 800, "height": 460})
        await b.close()
        b, c = await mk(p, "light")
        pg = await c.new_page()
        await go(pg, "/", 4500)
        await pg.screenshot(path=f"{OUT}/dashboard-light.png")
        await b.close()


async def gif():
    frames = []
    async with async_playwright() as p:
        b, c = await mk(p, "dark", 1280, 720)
        pg = await c.new_page()

        async def snap(n=1, ms=0):
            if ms:
                await pg.wait_for_timeout(ms)
            im = Image.open(io.BytesIO(await pg.screenshot())).convert("RGB").resize((960, 540), Image.LANCZOS)
            for _ in range(n):
                frames.append(im)

        await go(pg, "/", 4500)
        await snap(3)
        await go(pg, "/workers", 2500)
        await snap(2)
        await go(pg, "/workflows", 2500)
        await snap(2)
        await go(pg, "/executions", 3500)
        await pg.locator("tr", has_text="Rolling restart").nth(1).wait_for()
        await snap(2)
        await pg.locator("tr", has_text="Rolling restart").nth(1).click()
        await snap(3, 1200)
        await pg.evaluate(
            "(y)=>{[...document.querySelectorAll('*')].filter(e=>e.scrollHeight>e.clientHeight+40&&['auto','scroll'].includes(getComputedStyle(e).overflowY)&&e.clientHeight>200).forEach(e=>e.scrollTop=y)}",
            700,
        )
        await snap(3, 500)
        await go(pg, "/executions", 3500)
        await pg.locator("tr", has_text="Rolling restart").nth(1).wait_for()
        await pg.locator("tr", has_text="Rolling restart").nth(0).click()
        await snap(3, 1800)
        await go(pg, "/quick", 2500)
        await pg.get_by_text("Multiple Workers").click()
        await pg.get_by_text("SELECT ALL", exact=False).first.click()
        await pg.locator("textarea").first.fill("uptime; df -h / | tail -1")
        await snap(2, 300)
        await pg.keyboard.press("Control+Enter")
        await snap(3, 2500)
        await go(pg, "/", 2500)
        await pg.keyboard.press("Control+k")
        await snap(2, 500)
        for ch in "exec":
            await pg.keyboard.type(ch)
            await snap(1, 200)
        await snap(3, 400)
        await b.close()
    pal = [f.quantize(colors=96, method=Image.MEDIANCUT, dither=Image.NONE) for f in frames]
    pal[0].save(OUT + "/demo.gif", save_all=True, append_images=pal[1:], duration=1100, loop=0, optimize=True)
    print("gif", len(frames))


async def main():
    await shots()
    await shots2()
    await gif()


asyncio.run(main())