"""Showing a species in vibe-view, the family's 3D viewer.
Two routes, both carrying the structure as a QVF of its own (one structure,
its charge and multiplicity; :func:`vibesketch.document.qvf.species_qvf`):
- **The hand-off.** When vibe-view's Sketch button started this server, it
named a private directory in ``VIBEVIEW_SKETCH_HANDOFF`` (vibe-view #77,
vibe-sketch #1). Sending writes the QVF, then a record
``{"schema": "vibe-view-sketch-handoff/1", "path": "<absolute path>"}``
under a name that does not end in ``.json``, renamed into place so the
viewer never reads half a record. Names sort in the order of sending. A
directory that has gone means the viewer has closed.
- **vibe-view's own command.** Otherwise, when vibe-view is installed (the
``vibe-view`` command, or the ``vibeview`` package beside vibe-sketch),
``vibe-view open FILE`` opens the QVF in its viewer.
Neither imports vibe-view; the coupling is the command line and the record.
"""
from __future__ import annotations
import contextlib
import importlib.util
import json
import os
import shutil
import socket
import subprocess
import sys
import time
from pathlib import Path
from typing import Literal
from vibesketch.document.model import Document
__all__ = ["HANDOFF_ENV", "HANDOFF_SCHEMA", "ViewerUnavailable", "mode", "show"]
HANDOFF_ENV = "VIBEVIEW_SKETCH_HANDOFF"
HANDOFF_SCHEMA = "vibe-view-sketch-handoff/1"
_START_TIMEOUT = 45.0
Mode = Literal["handoff", "launch"]
[docs]
class ViewerUnavailable(RuntimeError):
"""No vibe-view to show the structure in."""
def _free_port() -> int:
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
return int(sock.getsockname()[1])
def _log_tail(path: Path) -> str:
with path.open("rb") as stream:
stream.seek(max(0, path.stat().st_size - 16384))
return stream.read(16384).decode("utf-8", errors="replace")
def _launch(
command: list[str],
path: Path,
return_dir: Path | None = None,
*,
view: str | None = None,
mode: int | None = None,
orbital: str | None = None,
) -> None:
"""Wait for this child's ready announcement, not an unrelated listener.
vibe-view's CLI announces ``vibe-view ready at URL`` after building its
application and binding. Keep its diagnostics in a file: a pipe would
block the long-lived child once the sketcher stops reading it.
"""
log = path.with_suffix(".launch.log")
process: subprocess.Popen[bytes] | None = None
started = False
try:
port = _free_port()
ready = f"vibe-view ready at http://127.0.0.1:{port}"
with log.open("ab") as output:
process = subprocess.Popen(
[
*command,
"open",
str(path),
"--host",
"127.0.0.1",
"--port",
str(port),
*(["--return-dir", str(return_dir)] if return_dir is not None else []),
*(["--view", view] if view is not None else []),
*(["--mode", str(mode)] if mode is not None else []),
*(["--orbital", orbital] if orbital is not None else []),
],
stdin=subprocess.DEVNULL,
stdout=output,
stderr=subprocess.STDOUT,
env={**os.environ, "PYTHONUNBUFFERED": "1"},
start_new_session=True,
)
deadline = time.monotonic() + _START_TIMEOUT
while True:
status = process.poll()
diagnostics = _log_tail(log)
if status is not None:
detail = diagnostics[-4000:].strip()
raise ViewerUnavailable(
f"vibe-view exited during startup (exit {status}). "
f"See {log}." + (f"\n{detail}" if detail else "")
)
if ready in diagnostics.splitlines():
started = True
return
if time.monotonic() >= deadline:
raise ViewerUnavailable(
f"vibe-view did not report readiness within {_START_TIMEOUT:g} seconds. "
f"See {log}. Check the viewer installation, or start vibe-sketch "
"from vibe-view's Sketch button."
)
time.sleep(0.1)
except OSError as error:
raise ViewerUnavailable(f"Could not start vibe-view: {error}. See {log}.") from error
finally:
# A successful viewer is independent of the sketcher. Only clean up
# our failed/unfinished child; never stop an existing viewer.
if process is not None and not started:
with contextlib.suppress(ProcessLookupError):
process.terminate()
try:
process.wait(timeout=3)
except subprocess.TimeoutExpired:
with contextlib.suppress(ProcessLookupError):
process.kill()
process.wait()
def _command() -> list[str] | None:
"""How to run vibe-view's CLI here, or ``None``."""
executable = shutil.which("vibe-view")
if executable is not None:
return [executable]
try:
if importlib.util.find_spec("vibeview") is not None:
return [sys.executable, "-m", "vibeview.cli"]
except (ImportError, ValueError):
pass
return None
[docs]
def mode() -> Mode | None:
"""How a species would reach vibe-view now: ``handoff``, ``launch`` or ``None``."""
directory = os.environ.get(HANDOFF_ENV)
if directory:
return "handoff" if Path(directory).is_dir() else None
return "launch" if _command() is not None else None
def _cache() -> Path:
base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache")
folder = Path(base) / "vibe-sketch" / "views"
folder.mkdir(parents=True, exist_ok=True)
return folder
[docs]
def show(
document: Document,
species_id: str,
*,
document_id: str | None = None,
return_dir: Path | None = None,
) -> dict[str, str]:
"""Show one species in vibe-view; returns how and the QVF written.
Raises ``KeyError`` for an unknown species,
:class:`~vibesketch.document.qvf.QvfError` for one without a 3D structure,
and :class:`ViewerUnavailable` with what to do when there is no viewer.
"""
from vibesketch.document.qvf import species_qvf
if return_dir is not None:
return_dir = return_dir.resolve()
if not return_dir.is_dir():
raise ViewerUnavailable("the requested return directory does not exist")
data = species_qvf(document, species_id, document_id=document_id)
stamp = f"{time.time_ns():020d}"
directory = os.environ.get(HANDOFF_ENV)
if directory:
if return_dir is not None and os.environ.get("VIBEVIEW_SKETCH_RETURN") != str(return_dir):
raise ViewerUnavailable(
"the existing viewer must name this return directory with VIBEVIEW_SKETCH_RETURN"
)
handoff = Path(directory)
if not handoff.is_dir():
raise ViewerUnavailable("vibe-view, which started this sketcher, has closed")
folder = handoff / "structures"
folder.mkdir(exist_ok=True)
path = (folder / f"{stamp}-{species_id}.qvf").resolve()
path.write_bytes(data)
record = handoff / f"{stamp}-{species_id}.part"
record.write_text(json.dumps({"schema": HANDOFF_SCHEMA, "path": str(path)}))
record.replace(handoff / f"{stamp}-{species_id}.json")
return {"mode": "handoff", "path": str(path)}
command = _command()
if command is None:
raise ViewerUnavailable(
"vibe-view is not installed; install it from its checkout, or start vibe-sketch "
"from vibe-view's Sketch button"
)
path = (_cache() / f"{stamp}-{species_id}.qvf").resolve()
path.write_bytes(data)
if return_dir is None:
_launch(command, path)
else:
_launch(command, path, return_dir)
return {"mode": "launch", "path": str(path)}
def show_result(
document: Document,
run_id: str,
view: str,
*,
document_id: str | None = None,
artifact: str | None = None,
mode: int | None = None,
orbital: str | None = None,
) -> dict[str, str]:
"""Open the selected run's retained data using the shared result-view contract."""
from typing import cast
from vibesketch.result_views import KINDS, View, prepare
if view not in KINDS:
raise ValueError("view must be orbitals, density, mode or spectrum")
directory = os.environ.get(HANDOFF_ENV)
command = _command() if not directory else None
if not directory and command is None:
raise ViewerUnavailable("vibe-view is not installed; install vibe-view to inspect results")
data = prepare(
document,
run_id,
cast(View, view),
document_id=document_id,
artifact=artifact,
mode=mode,
orbital=orbital,
)
stamp = f"{time.time_ns():020d}-result"
if directory:
handoff = Path(directory)
if not handoff.is_dir():
raise ViewerUnavailable("the viewer hand-off directory is gone")
folder = handoff / "structures"
folder.mkdir(exist_ok=True)
path = (folder / f"{stamp}.qvf").resolve()
path.write_bytes(data)
record: dict[str, object] = {"schema": HANDOFF_SCHEMA, "path": str(path), "view": view}
if mode is not None:
record["mode"] = mode
if orbital is not None:
record["orbital"] = orbital
temporary = handoff / f"{stamp}.part"
temporary.write_text(json.dumps(record))
temporary.replace(handoff / f"{stamp}.json")
return {"mode": "handoff", "path": str(path), "run": run_id, "view": view}
assert command is not None
path = (_cache() / f"{stamp}.qvf").resolve()
path.write_bytes(data)
_launch(command, path, view=view, mode=mode, orbital=orbital)
return {"mode": "launch", "path": str(path), "run": run_id, "view": view}