gandalf/scripts/demo/capture.py

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

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

import asyncio
import io
import pathlib
import sys

from PIL import Image
from playwright.async_api import async_playwright

OUT = str(pathlib.Path(__file__).resolve().parents[2] / "docs" / "img")
B = "http://127.0.0.1:8500"
H = {
    "Remote-User": "alex.rivera",
    "Remote-Name": "Alex Rivera",
    "Remote-Email": "alex@example.com",
    "Remote-Groups": "admin",
}
TOP = (
    "document.activeElement&&document.activeElement.blur();"
    "document.documentElement.style.scrollBehavior='auto';window.scrollTo(0,0)"
)
AV = (
    '<svg xmlns="http://www.w3.org/2000/svg" width="64" height="64">'
    '<rect width="64" height="64" fill="#050a10"/>'
    '<text x="32" y="40" font-family="monospace" font-size="24" font-weight="700" fill="#00d4ff" text-anchor="middle">AR</text></svg>'
)


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}')}}catch(e){{}}")

    async def av(route):
        await route.fulfill(status=200, content_type="image/svg+xml", body=AV)

    await c.route("**/api/avatar*", av)
    return b, c


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


async def el(pg, sel, name, idx=0):
    loc = pg.locator(sel).nth(idx)
    await loc.scroll_into_view_if_needed()
    await pg.wait_for_timeout(300)
    await loc.screenshot(path=f"{OUT}/{name}.png")
    print(name, "ok")


async def select_port(pg, sw, idx):
    await pg.locator(f'[data-action="select-port"][data-switch="{sw}"][data-port-idx="{idx}"]').first.click()
    await pg.wait_for_timeout(1200)


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]))
        await go(pg, "/")
        await pg.screenshot(path=f"{OUT}/dashboard.png")
        print("dashboard ok")
        await el(pg, "#topology-diagram", "topology")
        await el(pg, ".lt-section-header >> text=ALERT QUEUE", "alerts-tmp") if False else None
        # host cards + unifi table via element crops
        hb = (
            await pg.locator("#hosts-grid, .host-grid, .hosts-grid").first.bounding_box()
            if await pg.locator("#hosts-grid, .host-grid, .hosts-grid").count()
            else None
        )
        if hb:
            await pg.locator("#hosts-grid, .host-grid, .hosts-grid").first.screenshot(path=f"{OUT}/host-cards.png")
            print("host-cards ok")
        await go(pg, "/links")
        await pg.screenshot(path=f"{OUT}/link-debug.png")
        names = await pg.evaluate("[...document.querySelectorAll('.link-host-name')].map(e=>e.textContent.trim())")
        print("hosts", names)
        for nm, fn in (
            ("large1", "link-debug-sfp-warning"),
            ("compute-storage-gpu-01", "link-debug-sfp"),
            ("Pro 24 PoE", "link-debug-switch"),
        ):
            i = next((k for k, n in enumerate(names) if nm in n), None)
            if i is not None:
                await el(pg, ".link-host-panel", fn, i)
        await go(pg, "/inspector")
        await pg.screenshot(path=f"{OUT}/inspector.png")
        await select_port(pg, "USW-Aggregation", 4)
        await pg.screenshot(path=f"{OUT}/inspector-port.png", full_page=False)
        await pg.locator('[data-action="run-diagnostic"]').first.click()
        await pg.wait_for_timeout(9000)
        await pg.evaluate("document.getElementById('diag-results')?.scrollIntoView({block:'start'})")
        await pg.wait_for_timeout(600)
        await pg.screenshot(path=f"{OUT}/inspector-diagnostics.png")
        print("diagnostics ok")
        await go(pg, "/suppressions")
        await pg.screenshot(path=f"{OUT}/suppressions.png", full_page=True)
        print("errors", errs[:3])
        await b.close()
        b, c = await mk(p, "light")
        pg = await c.new_page()
        await go(pg, "/")
        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, "/", 5000)
        await snap(3)
        await pg.mouse.wheel(0, 620)
        await snap(3, 600)
        await pg.mouse.wheel(0, 700)
        await snap(2, 600)
        await go(pg, "/links", 4500)
        await snap(3)
        await pg.mouse.wheel(0, 500)
        await snap(2, 500)
        await go(pg, "/inspector", 4000)
        await snap(2)
        await select_port(pg, "USW-Aggregation", 4)
        await snap(3)
        await pg.locator('[data-action="run-diagnostic"]').first.click()
        await snap(2, 1500)
        await snap(2, 7500)
        await pg.evaluate("document.getElementById('diag-results')?.scrollIntoView({block:'start'})")
        await snap(4, 600)
        await go(pg, "/suppressions", 4000)
        await snap(3)
        await b.close()
    pal = [f.quantize(colors=96, method=Image.MEDIANCUT, dither=Image.NONE) for f in frames]
    pal[0].save(f"{OUT}/demo.gif", save_all=True, append_images=pal[1:], duration=1100, loop=0, optimize=True)
    print("gif", len(frames))


async def main():
    if "gif" not in sys.argv:
        await shots()
    if "shots" not in sys.argv:
        await gif()


asyncio.run(main())