justinandClaude Opus 5
2a35397e99
The demo tooling only existed in a session scratch directory, which is
temporary. This puts it in the repo so the video can be rebuilt.
Terminal only: tmux drives a two pane session, asciinema records it, agg
renders it, ffmpeg encodes it. The left pane runs the loop against a live
ZVM, the right pane polls the VPG journal so the tagged checkpoint appears
on camera as it lands.
Narration is synthesised per beat and aligned to marks the driver writes
while it runs, rather than to predicted timings, so a slow API call or an
FLR mount that takes longer than usual does not drift the audio. Two
things that has to respect are written down in the README: agg's
idle-time-limit must exceed the longest pause or it compresses idle time
and breaks the wall-clock mapping, and each beat holds for its narration
length so no line is cut off.
demo_win.json carries live guest credentials, so only an example with
placeholders is committed and the real file is gitignored, along with the
generated wav, cast, gif and mp4.
The xAI key path and voice id come from the environment now instead of
being hardcoded to one machine.
Also records the narration gotchas that cost time: mapping an acronym to
run-together phonetics ({"VM": "vee em"}) is spoken as one word, "vem";
prose written for the page sounds robotic read aloud; and volumedetect
reports no samples when aimed at a file whose first stream is video,
which makes a working audio track look silent.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_016yVfC5nvZowoLFnEGWhLGn
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Right pane: live view of a VPG journal, tagged checkpoints highlighted."""
|
|
from __future__ import annotations
|
|
import asyncio, os, sys, time
|
|
os.environ.setdefault("ZERTO_REWIND_CONFIG", "/home/justin/github/zerto-ai-rewind/config.json")
|
|
from zerto_rewind_mcp.config import load_config
|
|
from zerto_rewind_mcp.client import ZertoClient
|
|
from zerto_rewind_mcp.checkpoints import checkpoint_tag, checkpoint_id
|
|
from zerto_rewind_mcp.util import pick
|
|
|
|
VPG_ID = sys.argv[1]
|
|
VPG_NAME = sys.argv[2] if len(sys.argv) > 2 else VPG_ID
|
|
ROWS = int(os.environ.get("JOURNAL_ROWS", "10"))
|
|
|
|
DIM, RST, BOLD = "\033[2m", "\033[0m", "\033[1m"
|
|
CYAN, GREEN, YEL = "\033[36m", "\033[32m", "\033[33m"
|
|
|
|
|
|
def _wrap_tag(tag, width=52, maxlines=3):
|
|
tag = tag.replace("; Used for File Level Restore", " [FLR]")
|
|
out = []
|
|
while tag and len(out) < maxlines:
|
|
out.append(tag[:width]); tag = tag[width:]
|
|
return out
|
|
|
|
|
|
def ts_of(row):
|
|
return str(pick(row, "TimeStamp", "Timestamp", "timestamp") or "")
|
|
|
|
|
|
def render(rows, spin, tagged):
|
|
out = ["\033[H\033[2J"]
|
|
out.append(f"{BOLD}{CYAN} ZERTO JOURNAL {RST}{BOLD}{VPG_NAME}{RST}\n")
|
|
out.append(f"{DIM} vpg {VPG_ID}{RST}\n")
|
|
out.append(f"{DIM} {len(rows)} checkpoints{RST} {YEL}{len(tagged)} ai-tagged{RST} {DIM}{spin}{RST}\n\n")
|
|
|
|
out.append(f"{BOLD} AI TAGS{RST}\n")
|
|
if not tagged:
|
|
out.append(f"{DIM} (none yet){RST}\n")
|
|
for row in tagged[-4:]:
|
|
t = ts_of(row).replace("T", " ").replace(".000Z", "")[11:]
|
|
out.append(f"{GREEN}{BOLD} * {t} cp {checkpoint_id(row)}{RST}\n")
|
|
for seg in _wrap_tag(checkpoint_tag(row)):
|
|
out.append(f"{GREEN} {seg}{RST}\n")
|
|
|
|
out.append(f"\n{BOLD} NEWEST{RST}\n")
|
|
for row in rows[-ROWS:]:
|
|
tag = checkpoint_tag(row)
|
|
t = ts_of(row).replace("T", " ").replace(".000Z", "")
|
|
cid = checkpoint_id(row)
|
|
t = t[11:]
|
|
if tag.startswith("ai:"):
|
|
out.append(f"{GREEN}{BOLD} * {t} cp {cid} <-- new{RST}\n")
|
|
elif tag:
|
|
out.append(f"{DIM} {t} cp {cid} {tag[:20]}{RST}\n")
|
|
else:
|
|
out.append(f"{DIM} {t} cp {cid}{RST}\n")
|
|
sys.stdout.write("".join(out))
|
|
sys.stdout.flush()
|
|
|
|
|
|
async def main():
|
|
s = load_config()
|
|
c = ZertoClient(base_url=s["zerto_url"], username=s["username"], password=s["password"],
|
|
client_id=s.get("client_id", "zerto-client"), verify_tls=bool(s.get("verify_tls")))
|
|
spinner = "|/-\\"
|
|
i = 0
|
|
while True:
|
|
try:
|
|
rows = await c.list_checkpoints(VPG_ID)
|
|
tagged = [r for r in rows if checkpoint_tag(r).startswith("ai:")]
|
|
render(rows, spinner[i % 4], tagged)
|
|
except Exception as exc:
|
|
sys.stdout.write(f"\n journal poll error: {exc}\n")
|
|
sys.stdout.flush()
|
|
i += 1
|
|
await asyncio.sleep(3)
|
|
|
|
asyncio.run(main())
|