"""Document routes: create, read, edit, import, report, render, export, save.
Every write names the revision it read: ``If-Match`` on ``PUT`` of a whole
document, a ``revision`` field on the smaller edits. A stale write answers
409 with the current revision, and the client re-reads (roadmap, section
3.2). Each accepted edit is recorded in the document's provenance; a run of
edits of the same kind from the same client is folded into one entry, so a
drawing session does not leave hundreds.
"""
from __future__ import annotations
import errno
from typing import Annotated, Any, Literal
from fastapi import APIRouter, Header, HTTPException, Query, Response
from pydantic import BaseModel, ConfigDict, Field
from vibesketch.chem.relax import DEFAULT_FMAX, DEFAULT_MAX_STEPS
from vibesketch.document.io import DocumentError, from_data, new_document, with_edit
from vibesketch.document.model import (
Amount,
Conformer,
Constraint,
Document,
JobSpec,
Name,
Recipe,
RecipeAnalysis,
Species,
Surface,
)
from vibesketch.document.spectra import SpectrumExport
from vibesketch.inputs.neb import NebSpec
from vibesketch.server.store import Conflict, Entry, NotFound, Store
__all__ = ["make_router"]
Actor = Literal["human", "agent", "system"]
class DocumentResponse(BaseModel):
id: str
revision: int
document: dict[str, Any]
warnings: list[str]
#: The file the document saves to, by name only, or ``None``.
file: str | None
#: Whether the file is behind the current revision.
dirty: bool
#: What the QVF job container the document was opened from says about its
#: calculation (``vibesketch.document.qvf``), or ``None``.
result: dict[str, Any] | None = None
class DocumentSummary(BaseModel):
id: str
revision: int
file: str | None
dirty: bool
species: int
reactions: int
class Editor(BaseModel):
"""Who is editing; recorded in the provenance."""
model_config = ConfigDict(extra="forbid")
actor: Actor = "agent"
client: Annotated[str, Field(pattern=r"^[a-z0-9][a-z0-9_.:+-]{0,63}$")] = "rest"
class ResultViewRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
view: Literal["orbitals", "density", "mode", "spectrum"]
artifact: str | None = None
mode: int | None = Field(default=None, ge=1)
orbital: str | None = None
class ResultFileRequest(Editor):
revision: int
name: str
data: str
class GeometryReturnRequest(Editor):
data: str
class CreateRequest(Editor):
document: dict[str, Any] | None = None
format: str | None = None
data: str | None = None
#: ``base64`` for binary data, such as a QVF archive.
encoding: Literal["text", "base64"] = "text"
class CanvasRequest(Editor):
revision: int
ket: dict[str, Any]
class QvfOverwriteRequest(BaseModel):
"""The existing destination's bytes, never a filesystem path."""
model_config = ConfigDict(extra="forbid")
data: str
class QvfOverwriteResponse(BaseModel):
allowed: bool
reason: str | None
class ImportRequest(Editor):
revision: int
format: str
data: str
mode: Literal["append", "replace"] = "append"
#: ``base64`` for binary data, such as a QVF archive.
encoding: Literal["text", "base64"] = "text"
def _payload(data: str | None, encoding: str) -> str | bytes:
"""The request's data, decoded from base64 when it says so."""
if encoding != "base64":
return data or ""
import base64
import binascii
try:
return base64.b64decode(data or "", validate=True)
except binascii.Error as error:
raise HTTPException(422, {"errors": [f"the data is not valid base64: {error}"]}) from None
class RevisionRequest(Editor):
"""An edit that needs nothing but the revision it is based on."""
revision: int
class NanostructureRequest(RevisionRequest):
"""A nanostructure to generate and add as a species (#137)."""
parameters: dict[str, Any]
picture_style: Literal["lattice", "schematic", "end-on", "unrolled"] | None = None
picture_label: bool = False
species_id: str | None = None
label: str | None = None
charge: int = 0
multiplicity: int | None = Field(default=None, ge=1)
class NanostructureResponse(DocumentResponse):
#: The new species' key, description, formula and size.
nanostructure: dict[str, Any]
class CapRequest(RevisionRequest):
"""Cap a labelled structure's attachment atoms to make a species (#139)."""
cap: Literal["H", "CH3"]
class SurfaceRequest(RevisionRequest):
parameters: Surface
at: tuple[float, float] | None = None
width: float = Field(default=8, ge=1, le=100, allow_inf_nan=False)
height: float = Field(default=3, ge=1, le=100, allow_inf_nan=False)
class PictureRequest(RevisionRequest):
"""A nanostructure picture to put on the canvas (#138)."""
style: Literal["lattice", "schematic", "end-on", "unrolled"] = "schematic"
label: bool = False
rear: Literal["light", "omit"] = "light"
brk: Literal["auto", "on", "off"] = Field(default="auto", alias="break")
width: float = Field(default=8.0, ge=1.0, le=100.0)
class AttachRequest(RevisionRequest):
"""Move a labelled structure onto a nanostructure picture's anchor (#138)."""
species: str
anchor: str = "upper-end"
node: int | None = Field(default=None, ge=0)
class HalfReactionRequest(RevisionRequest):
"""An electrode half-reaction to draw beside the drawing (#141)."""
reactants: list[Any] = Field(min_length=1)
products: list[Any] = Field(min_length=1)
electrons: dict[str, Any] | None = None
conditions: list[str] = Field(default_factory=list)
arrow: str = "forward"
over: list[Any] = Field(default_factory=list)
class AmountsRequest(RevisionRequest):
"""The amounts entered for a reaction's participants; an empty list clears them."""
amounts: list[Amount]
class HighlightRequest(RevisionRequest):
"""Mark atoms of a species, numbered from 1 as in the species report."""
species: str
atoms: list[int] = Field(min_length=1)
colour: str = "orange"
class SelectionHighlightRequest(RevisionRequest):
"""Mark the atoms the canvas has selected, at their points in its own KET."""
ket: dict[str, Any]
points: list[tuple[float, float]] = Field(min_length=1)
colour: str = "orange"
class ClearHighlightsRequest(RevisionRequest):
species: str | None = None
class EnumerationRequest(RevisionRequest):
core: str | None = None
class AbbreviationsRequest(RevisionRequest):
"""Contract or expand abbreviations: of one species or all, of some labels or all."""
species: str | None = None
labels: list[str] = Field(default_factory=list)
class HighlightPointsRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
ket: dict[str, Any]
class TlcRequest(RevisionRequest):
"""The lanes of a TLC plate to draw beside the drawing."""
lanes: list[dict[str, Any]]
class TemplateLibraryRequest(RevisionRequest):
entry: str
scaffold: bool = False
class ReactionLibraryRequest(RevisionRequest):
entry: str
kind: Literal["example", "scaffold"] = "example"
class ApparatusRequest(RevisionRequest):
"""A laboratory apparatus pictogram to draw, and where."""
kind: str
at: tuple[float, float] | None = None
scale: float = 1.0
label: str | None = None
class NmrLabelsRequest(RevisionRequest):
"""Place a run's NMR shifts as labels beside its species' structure."""
run: str
reference: str
nuclei: list[str] = Field(default_factory=lambda: ["H", "C"])
class StereoLabelsRequest(RevisionRequest):
"""Write the drawn structures' CIP descriptors as labels, or remove them."""
clear: bool = False
class PropertyTextRequest(RevisionRequest):
"""Which lines the property text under a structure shows."""
fields: list[Literal["formula", "mass", "analysis", "logp", "tpsa"]] | None = None
class ResultImportRequest(RevisionRequest):
run: str
format: Literal["orca", "qvf", "orca-hess"]
source_name: str
data: str = Field(max_length=48 * 1024 * 1024)
replace: bool = False
class ResultReportRequest(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
group: str | None = None
runs: list[str] | None = None
reactions: list[str] | None = None
coefficients: dict[str, dict[str, dict[str, float]]] = Field(default_factory=dict)
format: Literal["json", "svg", "qvf", "csv", "markdown", "latex"] = "json"
class ExternalChainRequest(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
data: str = Field(max_length=90 * 1024 * 1024)
sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
reference_contract: Literal["real_restricted_cc"]
format: Literal["json", "csv", "markdown", "latex"] = "json"
class CampaignEnergyRequest(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
stage: str
reactions: list[str] | None = None
coefficients: dict[str, dict[str, dict[str, float]]] = Field(default_factory=dict)
format: Literal["json", "svg", "qvf", "csv", "markdown", "latex"] = "json"
class TidyRequest(Editor):
"""Lay the scheme out; rows longer than ``width`` bond lengths turn down."""
revision: int
width: float = Field(default=24.0, ge=4.0)
class EmbedRequest(Editor):
"""Which species to give 3D structures, and how many conformers each."""
revision: int
#: Species ids; all of them when left out.
species: list[str] | None = None
count: int = Field(default=1, ge=1, le=20)
#: Re-embed species that have conformers already.
replace: bool = False
class Embed3dRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
smiles: str
count: int = Field(default=1, ge=1, le=20)
class RelaxRequest(Editor):
"""Which species to relax, on which force field, and what to hold."""
revision: int
#: Species ids; all of them when left out. With constraints, exactly one.
species: list[str] | None = None
constraints: list[Constraint] | None = None
force_field: Literal["mmff94", "uff"] = "mmff94"
#: Convergence: the largest force component, in kcal/mol/angstrom.
fmax: float = Field(default=DEFAULT_FMAX, gt=0)
max_steps: int = Field(default=DEFAULT_MAX_STEPS, ge=0, le=100_000)
class RelaxResponse(DocumentResponse):
#: Per relaxed species id: energies, steps, convergence and the constraint checks.
relaxed: dict[str, dict[str, Any]]
class Relax3dRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
smiles: str
conformer: Conformer
constraints: list[Constraint] | None = None
force_field: Literal["mmff94", "uff"] = "mmff94"
fmax: float = Field(default=DEFAULT_FMAX, gt=0)
max_steps: int = Field(default=DEFAULT_MAX_STEPS, ge=0, le=100_000)
class InputsRequest(BaseModel):
"""One writer and one specification for several species of the document."""
model_config = ConfigDict(extra="forbid")
program: str
spec: JobSpec
#: Species ids; all of them when left out.
species: list[str] | None = None
#: ``json`` for the files in the answer, ``zip`` for one archive to download.
format: Literal["json", "zip"] = "json"
class MappingRequest(BaseModel):
"""Pinned atom pairs for a reaction's mapping, reactant:atom=product:atom."""
model_config = ConfigDict(extra="forbid")
pins: list[str] = Field(default_factory=list)
class NebRequest(BaseModel):
"""A reaction-path input: the program, the band and the level of theory, and any pins."""
model_config = ConfigDict(extra="forbid")
program: Literal["orca", "vibeqc"] = "orca"
spec: NebSpec
pins: list[str] = Field(default_factory=list)
#: ``json`` for the files in the answer, ``zip`` for one archive to download.
format: Literal["json", "zip"] = "json"
class SubmitRequest(Editor):
"""One program and specification for species of the document, sent to vibe-queue."""
revision: int
program: str
spec: JobSpec
#: Species ids; all of them when neither this nor ``reactions`` is given.
species: list[str] | None = None
#: Reaction ids whose species (reactants, reagents, products, each once)
#: are submitted too; ``[]`` means every reaction, the whole scheme.
reactions: list[str] | None = None
#: vibe-qc jobs as QVF job containers (the default), or as run scripts.
container: bool = True
class SubmitResponse(DocumentResponse):
#: ``{"job", "species", "run", "input"}`` per run started; ``input`` is
#: ``container``, ``script`` or ``directory``.
submitted: list[dict[str, str]]
#: Per species id, why it was not submitted.
refused: dict[str, list[str]]
#: Per species id, why a run script went instead of a QVF job container.
fallbacks: dict[str, str]
#: The group the started runs form; ``None`` when nothing was submitted.
group: str | None
class EnergyRecipeRequest(Editor):
revision: int
id: str
analysis: RecipeAnalysis
spec: JobSpec
program: Literal["orca", "vibeqc"]
class EnergyResultRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
runs: list[str]
figure: bool = False
class RecipeRequest(Editor):
"""A recipe to save on the document, replacing one with the same id."""
revision: int
recipe: Recipe
class BaselineRecipeRequest(Editor):
"""The publication baseline recipe for species, reactions or the whole scheme."""
revision: int
id: str = "recipe1"
name: str | None = None
reference_contract: Literal["all_tested_channels", "real_restricted_cc"] = "all_tested_channels"
profile: Literal["baseline", "dd"] = "baseline"
#: Species ids the recipe applies to.
species: list[str] | None = None
#: Reaction ids whose species it applies to; ``[]`` means every reaction.
reactions: list[str] | None = None
class CompileRequest(Editor):
"""Where and how to compile a recipe into a vibe-queue campaign."""
revision: int
#: A new, empty directory, as an absolute path on the server's machine.
directory: str
id: str | None = None
host: str | None = None
program: str | None = "vibeqc"
python: str = "python3"
#: The container runner command; ``{python}`` stands for the interpreter.
runner: list[str] | None = None
cpus: int = Field(default=1, ge=1, le=4096)
mem_mb: int | None = Field(default=None, ge=1, le=16 * 2**20)
wall_time_seconds: int | None = Field(default=None, strict=True, ge=1, le=31536000)
check_wall_time_seconds: int | None = Field(default=None, strict=True, ge=1, le=31536000)
stage_runtimes: dict[str, dict[str, Any]] | None = None
#: The operator's acknowledgement of the recipe's MACE model licence.
accept_model_license: bool = False
class CompileResponse(DocumentResponse):
campaign: dict[str, Any]
run_command: str
class RevalidateRequest(Editor):
revision: int
directory: str
node: str
reviewer: str
reason: str
class ReviewRequest(RevalidateRequest):
criteria: dict[str, Any]
warnings: list[str]
evidence: list[str]
class RenderStagesRequest(BaseModel):
"""Which species to render, where, and how big."""
model_config = ConfigDict(extra="forbid")
species: str
#: A new, empty directory, as an absolute path on the server's machine.
directory: str
width: int = Field(default=900, ge=16, le=8000)
height: int = Field(default=600, ge=16, le=8000)
publication: bool = False
images: bool = True
class RetryRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
node: str
class RecipeResponse(DocumentResponse):
#: The saved recipe with every stage resolved per species, as ``GET .../recipes`` gives it.
report: list[dict[str, Any]]
class ViewRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
species: str
class SpeciesRequest(Editor):
"""Change what the drawing cannot say about a species."""
revision: int
label: str | None = None
multiplicity: int | None = Field(default=None, ge=1)
#: A name the user gives; ``""`` removes it.
name: str | None = None
def _response(entry: Entry) -> DocumentResponse:
return DocumentResponse(
id=entry.id,
revision=entry.document.revision,
document=entry.document.to_json_data(),
warnings=entry.warnings,
file=entry.path.name if entry.path else None,
dirty=entry.dirty,
result=entry.result,
)
def _summary(entry: Entry) -> DocumentSummary:
return DocumentSummary(
id=entry.id,
revision=entry.document.revision,
file=entry.path.name if entry.path else None,
dirty=entry.dirty,
species=len(entry.document.species),
reactions=len(entry.document.reactions),
)
def _with_edit(document: Document, editor: Editor, summary: str) -> Document:
"""``document`` with this edit recorded, folded into the last one if alike."""
return with_edit(document, actor=editor.actor, client=editor.client, summary=summary)
[docs]
def make_router(store: Store) -> APIRouter:
router = APIRouter(prefix="/api", tags=["documents"])
def entry_or_404(identifier: str) -> Entry:
try:
return store.get(identifier)
except NotFound:
raise HTTPException(404, f"no document {identifier!r}") from None
def commit(
entry: Entry, document: Document, *, revision: int, warnings: list[str] | None = None
) -> DocumentResponse:
try:
return _response(
store.replace(entry.id, document, expected=revision, warnings=warnings)
)
except Conflict as error:
raise HTTPException(409, {"message": str(error), "revision": error.current}) from None
@router.get("/session")
def session() -> dict[str, Any]:
"""What the browser needs at start-up: the documents, the one to open, capabilities."""
from vibesketch import viewer
from vibesketch.document.qvf import qvf_available
from vibesketch.inputs import writers
from vibesketch.naming import naming_available
from vibesketch.naming.opsin import opsin_available
from vibesketch.queue import queue_status
return {
"documents": [_summary(e).model_dump() for e in store.entries()],
"default": store.default_id,
"naming": naming_available(),
"opsin": opsin_available(),
"qvf": qvf_available(),
"viewer": viewer.mode(),
"queue": queue_status(),
"writers": [w.name for w in writers()],
}
@router.get("/documents")
def list_documents() -> list[DocumentSummary]:
return [_summary(e) for e in store.entries()]
@router.post("/qvf/overwrite-check")
def qvf_overwrite_check(request: QvfOverwriteRequest) -> QvfOverwriteResponse:
"""Check base64 QVF bytes before replacement; callers must check for races.
This inspects content only and neither grants filesystem access nor
writes a file. Malformed archives are refused; malformed base64 is 422.
"""
from vibesketch.document.qvf import overwrite_refusal
data = _payload(request.data, "base64")
assert isinstance(data, bytes)
reason = overwrite_refusal(data)
return QvfOverwriteResponse(allowed=reason is None, reason=reason)
@router.post("/documents", status_code=201)
def create_document(request: CreateRequest) -> DocumentResponse:
"""A new document: empty, from a document's JSON, or imported from ``format``/``data``."""
from vibesketch.document.formats import ImportFailed, import_into
warnings: list[str] = []
result: dict[str, Any] | None = None
if request.document is not None:
try:
document = from_data(request.document)
except DocumentError as error:
raise HTTPException(422, {"errors": error.errors}) from None
elif request.format is not None:
try:
data = _payload(request.data, request.encoding)
derived = import_into(None, data, request.format, mode="replace")
except ImportFailed as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document, warnings, result = derived.document, derived.warnings, derived.result
else:
document = new_document()
entry = store.add(document)
entry.warnings = warnings
entry.result = result
return _response(entry)
@router.get("/documents/{identifier}")
def get_document(identifier: str, response: Response) -> DocumentResponse:
entry = entry_or_404(identifier)
response.headers["ETag"] = f'"{entry.document.revision}"'
return _response(entry)
@router.put("/documents/{identifier}")
def put_document(
identifier: str,
document: Document,
if_match: Annotated[str | None, Header()] = None,
) -> DocumentResponse:
"""Replace the whole document; ``If-Match`` names the revision it is based on."""
entry_or_404(identifier)
if if_match is None:
raise HTTPException(428, "send If-Match with the revision this document is based on")
try:
expected = int(if_match.strip('"W/ '))
except ValueError:
raise HTTPException(
400, f"If-Match must be a revision number, not {if_match!r}"
) from None
try:
stored = store.replace(identifier, document, expected=expected)
except Conflict as error:
raise HTTPException(412, {"message": str(error), "revision": error.current}) from None
return _response(stored)
@router.put("/documents/{identifier}/canvas")
def put_canvas(identifier: str, request: CanvasRequest) -> DocumentResponse:
"""Replace the drawing; species and reactions are derived from it."""
from vibesketch.document.derive import derive
from vibesketch.document.formats import _atoms_in
from vibesketch.document.ket import read_scene
from vibesketch.document.model import Canvas
from vibesketch.limits import check_atoms, check_depth
entry = entry_or_404(identifier)
try:
# The same bounds as an import; a drawing is no more trusted than a file (#69).
check_depth(request.ket, "the drawing")
check_atoms(_atoms_in(request.ket), "the drawing")
canvas = Canvas(format="ket", data=request.ket)
read_scene(canvas.data)
except (ValueError, TypeError, AttributeError, KeyError, IndexError) as error:
reason = f"not a drawing vibe-sketcher can use: {error}"
raise HTTPException(422, {"errors": [reason]}) from None
base = entry.document.model_copy(update={"canvas": canvas})
derived = derive(base)
document = _with_edit(derived.document, request, "Edited the drawing")
return commit(entry, document, revision=request.revision, warnings=derived.warnings)
@router.post("/documents/{identifier}/import")
def import_document(identifier: str, request: ImportRequest) -> DocumentResponse:
"""Draw structures read from ``format`` onto the canvas."""
from vibesketch.document.formats import ImportFailed, import_into
entry = entry_or_404(identifier)
try:
data = _payload(request.data, request.encoding)
derived = import_into(entry.document, data, request.format, mode=request.mode)
except ImportFailed as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document = _with_edit(derived.document, request, f"Imported {request.format}")
if derived.result is not None:
entry.result = derived.result
return commit(entry, document, revision=request.revision, warnings=derived.warnings)
@router.post("/documents/{identifier}/clean")
def clean_document(identifier: str, request: RevisionRequest) -> DocumentResponse:
"""Standard bond lengths and angles for every molecule; the drawing is kept.
A molecule that cleaning cannot handle without changing its
stereochemistry, or that is not a plain structure, is left as drawn
and named in ``warnings``.
"""
from vibesketch.document.clean import clean
entry = entry_or_404(identifier)
derived, report = clean(entry.document)
document = _with_edit(derived.document, request, "Cleaned up the drawing")
warnings = [*derived.warnings, *report.left]
return commit(entry, document, revision=request.revision, warnings=warnings)
@router.post("/documents/{identifier}/tidy")
def tidy_document(identifier: str, request: TidyRequest) -> DocumentResponse:
"""The reaction scheme laid out as papers draw it; its reactions are kept.
Molecules are cleaned and placed on their arrows' line, reagents above
each arrow, conditions below, compound numbers under their structures,
steps in their order, branches off the row. A scheme that would read as
different reactions, or that branches in a way the layout cannot
express, is refused with 422 and the reason; nothing changes.
"""
from vibesketch.document.scheme import tidy_scheme
entry = entry_or_404(identifier)
derived, report = tidy_scheme(entry.document, width=request.width)
if not report.tidied:
raise HTTPException(422, {"errors": [f"left the scheme as drawn: {report.reason}"]})
document = _with_edit(derived.document, request, "Tidied the scheme")
warnings = [*derived.warnings, *report.left]
return commit(entry, document, revision=request.revision, warnings=warnings)
@router.post("/documents/{identifier}/embed")
def embed_document(identifier: str, request: EmbedRequest) -> DocumentResponse:
"""3D structures (ETKDG, then MMFF94) for the species, kept in the document.
The drawn stereochemistry is checked on every conformer. A species
that cannot be embedded keeps what it had and is named in
``warnings`` with the reason; so is a stereocentre the drawing leaves
open. When nothing changes, the revision stays.
"""
from vibesketch.document.embed import UnknownSpecies, embed
entry = entry_or_404(identifier)
try:
document, report = embed(
entry.document, request.species, count=request.count, replace=request.replace
)
except UnknownSpecies as error:
raise HTTPException(404, str(error)) from None
warnings = [*report.warnings, *(f"{i}: {why}" for i, why in report.failed.items())]
if not report.embedded:
if request.revision != entry.document.revision:
raise HTTPException(
409,
{"message": "the document has changed", "revision": entry.document.revision},
)
entry.warnings = warnings
return _response(entry)
summary = f"Generated 3D structures: {', '.join(report.embedded)}"
document = _with_edit(document, request, summary)
return commit(entry, document, revision=request.revision, warnings=warnings)
@router.post("/documents/{identifier}/relax")
def relax_document(identifier: str, request: RelaxRequest) -> RelaxResponse:
"""Relax 3D structures with FIRE on a force field, constraints held exactly.
Each relaxed species' first conformer is replaced, as ``fire+mmff94``
or ``fire+uff``; ``relaxed`` reports energies, steps, convergence and
the constraint checks per species. A species that cannot be relaxed
keeps what it had and is named in ``warnings`` with the reason. When
nothing changes, the revision stays.
"""
from vibesketch.document.embed import UnknownSpecies
from vibesketch.document.relax import relax
entry = entry_or_404(identifier)
try:
document, report = relax(
entry.document,
request.species,
constraints=request.constraints,
force_field=request.force_field,
fmax=request.fmax,
max_steps=request.max_steps,
)
except UnknownSpecies as error:
raise HTTPException(404, str(error)) from None
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
warnings = [*report.warnings, *(f"{i}: {why}" for i, why in report.failed.items())]
relaxed = {
identifier: {k: v for k, v in result.to_dict().items() if k != "conformer"}
for identifier, result in report.results.items()
}
if not report.relaxed:
if request.revision != entry.document.revision:
raise HTTPException(
409,
{"message": "the document has changed", "revision": entry.document.revision},
)
entry.warnings = warnings
return RelaxResponse(**_response(entry).model_dump(), relaxed={})
summary = f"Relaxed 3D structures with FIRE: {', '.join(report.relaxed)}"
document = _with_edit(document, request, summary)
state = commit(entry, document, revision=request.revision, warnings=warnings)
return RelaxResponse(**state.model_dump(), relaxed=relaxed)
@router.post("/documents/{identifier}/inputs", tags=["inputs"])
def document_inputs(identifier: str, request: InputsRequest) -> Response:
"""Inputs for several species with one writer and one specification.
Species the writer refuses are listed under ``refused`` with the
reasons (in ``REFUSED.txt`` inside a zip) and the rest are written;
when every species is refused, the answer is 422. Species need their
3D structures first (``POST /api/documents/{id}/embed``).
"""
import io
import json
import zipfile
from vibesketch.inputs import UnknownReference, UnknownWriter, write_inputs
entry = entry_or_404(identifier)
try:
batch = write_inputs(
entry.document, program=request.program, spec=request.spec, species=request.species
)
except UnknownWriter as error:
raise HTTPException(422, {"program": error.name, "reasons": [str(error)]}) from None
except UnknownReference as error:
raise HTTPException(404, str(error)) from None
reasons = [f"{sid}: {why}" for sid, whys in batch.refused.items() for why in whys]
if not batch.bundles:
raise HTTPException(422, {"program": request.program, "reasons": reasons})
files = [
{
"species": bundle.species,
"name": f.name,
"media_type": f.media_type,
"content": f.content,
}
for bundle in batch.bundles
for f in bundle.files
]
if request.format == "json":
body = json.dumps(
{"program": request.program, "files": files, "refused": batch.refused}
)
return Response(body, media_type="application/json")
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
for file in files:
archive.writestr(file["name"], file["content"])
if reasons:
archive.writestr("REFUSED.txt", "\n".join(reasons) + "\n")
name = f"{identifier}-{request.program}-inputs.zip"
return Response(
buffer.getvalue(),
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{name}"'},
)
@router.put("/documents/{identifier}/reactions/{reaction}/amounts")
def set_amounts(identifier: str, reaction: str, request: AmountsRequest) -> DocumentResponse:
"""Replace the amounts entered for a reaction's stoichiometry table."""
from vibesketch.document.stoichiometry import StoichiometryRefused, with_amounts
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
updated = with_amounts(entry.document, reaction, request.amounts)
except StoichiometryRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(updated, request, f"Entered the amounts of {reaction}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/reactions/{reaction}/stoichiometry")
def reaction_stoichiometry(
identifier: str,
reaction: str,
format: Literal["json", "csv", "markdown", "latex"] = "json",
) -> Response:
"""The worked-out stoichiometry table, as JSON or as a CSV, Markdown or LaTeX table."""
import json
from vibesketch.document.stoichiometry import (
MEDIA_TYPES,
StoichiometryRefused,
stoichiometry,
stoichiometry_table,
)
document = entry_or_404(identifier).document
try:
if format == "json":
body = json.dumps(stoichiometry(document, reaction), allow_nan=False)
return Response(body, media_type="application/json")
return Response(
stoichiometry_table(document, reaction, format), media_type=MEDIA_TYPES[format]
)
except StoichiometryRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/reactions/{reaction}/mapping", tags=["inputs"])
def reaction_mapping_of(
identifier: str, reaction: str, request: MappingRequest
) -> dict[str, Any]:
"""Which reactant atom becomes which product atom, for review before a path input.
The pairing breaks and forms the fewest bonds; ``ambiguous`` with
``alternatives`` when equally cheap pairings describe different
reactions, and ``refusals`` saying what stops an input. Pins
(``m1:3=m3:5``) correct it.
"""
from vibesketch.inputs import UnknownReference
from vibesketch.inputs.neb import reaction_mapping
entry = entry_or_404(identifier)
try:
return reaction_mapping(entry.document, reaction, request.pins)
except UnknownReference as error:
raise HTTPException(404, str(error)) from None
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/reactions/{reaction}/neb", tags=["inputs"])
def reaction_neb(identifier: str, reaction: str, request: NebRequest) -> Response:
"""Nudged-elastic-band inputs for a reaction, with the mapping they rest on.
ORCA gets NEB-TS, NEB-CI or NEB with the product end as an XYZ file;
vibe-qc a run_neb script (no NEB-TS). A refusal is 422 with the
``reasons`` and the ``mapping``.
"""
import io
import json
import zipfile
from vibesketch.inputs import UnknownReference, WriterRefusal
from vibesketch.inputs.neb import neb_inputs, reaction_mapping
entry = entry_or_404(identifier)
try:
bundle, mapping = neb_inputs(
entry.document, reaction, request.program, request.spec, request.pins
)
except UnknownReference as error:
raise HTTPException(404, str(error)) from None
except WriterRefusal as error:
detail = {
"program": error.program,
"reasons": list(error.reasons),
"mapping": reaction_mapping(entry.document, reaction, request.pins),
}
raise HTTPException(422, detail) from None
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
files = [
{"name": f.name, "media_type": f.media_type, "content": f.content} for f in bundle.files
]
if request.format == "json":
body = json.dumps({"program": request.program, "files": files, "mapping": mapping})
return Response(body, media_type="application/json")
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
for file in files:
archive.writestr(file["name"], file["content"])
name = f"{identifier}-{reaction}-{request.program}-neb.zip"
return Response(
buffer.getvalue(),
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{name}"'},
)
@router.get("/queue", tags=["queue"])
def queue() -> dict[str, Any]:
"""Whether vibe-queue is installed and serving, and which programs it can run."""
from vibesketch.queue import queue_status
return queue_status()
@router.post("/documents/{identifier}/submit", tags=["queue"])
def submit_jobs(identifier: str, request: SubmitRequest) -> SubmitResponse:
"""Write inputs for the species and submit them to vibe-queue.
A vibe-qc job goes as a QVF job container that keeps the sketch with
its result (``container`` false sends run scripts); when none can be
built or pinned, the script goes and ``fallbacks`` says why. Species
without a 3D structure are embedded first. The jobs and their runs
are recorded on the document; species the writer refuses are listed
under ``refused``. 503 when vibe-queue is missing or cannot run the
program.
"""
from vibesketch.document.embed import UnknownSpecies, embed
from vibesketch.inputs import UnknownReference, UnknownWriter
from vibesketch.queue import QueueUnavailable, reaction_species, submit
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(
409, {"message": "the document has changed", "revision": entry.document.revision}
)
document = entry.document
reactions = None if request.reactions is None else list(request.reactions)
if reactions == []:
reactions = [r.id for r in document.reactions]
try:
wanted = list(request.species or [])
if reactions is not None:
wanted += reaction_species(document, reactions)
elif not wanted:
wanted = [s.id for s in document.species]
missing = [s.id for s in document.species if s.id in wanted and not s.conformers]
if missing:
document, _ = embed(document, missing)
submitted = submit(
document,
program=request.program,
spec=request.spec,
species=request.species,
reactions=reactions,
container=request.container,
)
except (UnknownSpecies, UnknownReference) as error:
raise HTTPException(404, str(error)) from None
except UnknownWriter as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
except QueueUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
runs = ", ".join(row["run"] for row in submitted.submitted) or "none"
summary = (
f"Submitted to vibe-scheduler as group {submitted.group}: {runs}"
if submitted.group
else f"Submitted to vibe-scheduler: {runs}"
)
edited = _with_edit(submitted.document, request, summary)
state = commit(entry, edited, revision=request.revision, warnings=entry.warnings)
return SubmitResponse(
**state.model_dump(),
submitted=submitted.submitted,
refused=submitted.refused,
fallbacks=submitted.fallbacks,
group=submitted.group,
)
@router.get("/energy-recipes/constants")
def energy_constants() -> dict[str, Any]:
from vibesketch.document.electrochemical_recipes import constants
return constants()
@router.post("/documents/{identifier}/energy-recipes")
def energy_recipe_add(identifier: str, request: EnergyRecipeRequest) -> DocumentResponse:
from vibesketch.document.electrochemical_recipes import with_energy_recipe
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
document = with_energy_recipe(
entry.document, request.id, request.analysis, request.spec, request.program
)
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(document, request, f"Saved energy recipe {request.id}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.post("/documents/{identifier}/energy-recipes/{recipe}/jobs")
def energy_recipe_jobs(
identifier: str, recipe: str, request: RevisionRequest
) -> dict[str, Any]:
from vibesketch.document.electrochemical_recipes import compile_energy_recipe
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
document, bundle = compile_energy_recipe(entry.document, recipe)
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
state = commit(
entry,
_with_edit(document, request, f"Compiled energy recipe {recipe}"),
revision=request.revision,
warnings=entry.warnings,
)
return {**state.model_dump(), "bundle": bundle}
@router.post("/documents/{identifier}/energy-recipes/{recipe}/results")
def energy_recipe_results(identifier: str, recipe: str, request: EnergyResultRequest) -> Any:
from vibesketch.document.electrochemical_recipes import (
energy_recipe_figure,
energy_recipe_result,
)
document = entry_or_404(identifier).document
try:
if request.figure:
return Response(
energy_recipe_figure(document, recipe, request.runs), media_type="image/svg+xml"
)
return energy_recipe_result(document, recipe, request.runs)
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
@router.get("/documents/{identifier}/recipes", tags=["recipes"])
def document_recipes(identifier: str, recipe: str | None = None) -> list[dict[str, Any]]:
"""The document's staged recipes, each stage resolved per species with its refusals."""
from vibesketch.document.recipe import UnknownRecipe, recipe_report
try:
return recipe_report(entry_or_404(identifier).document, recipe)
except UnknownRecipe as error:
raise HTTPException(404, str(error)) from None
@router.put("/documents/{identifier}/recipes/{recipe}", tags=["recipes"])
def put_recipe(identifier: str, recipe: str, request: RecipeRequest) -> RecipeResponse:
"""Save a recipe on the document, replacing the one with its id."""
from vibesketch.document.recipe import recipe_report, set_recipe
entry = entry_or_404(identifier)
if request.recipe.id != recipe:
raise HTTPException(
422, {"errors": [f"the recipe's id {request.recipe.id!r} is not {recipe!r}"]}
)
try:
document = set_recipe(entry.document, request.recipe)
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document = _with_edit(document, request, f"Saved recipe {recipe}")
state = commit(entry, document, revision=request.revision, warnings=entry.warnings)
return RecipeResponse(**state.model_dump(), report=recipe_report(document, recipe))
@router.post("/documents/{identifier}/recipes/baseline", tags=["recipes"])
def baseline_recipe_route(identifier: str, request: BaselineRecipeRequest) -> RecipeResponse:
"""Save the publication baseline (MACE, PBE, Hessian, DLPNO) for the chosen species."""
from vibesketch.document.recipe import baseline_recipe, recipe_report, set_recipe
from vibesketch.inputs import UnknownReference
from vibesketch.queue import reaction_species
entry = entry_or_404(identifier)
document = entry.document
try:
targets = list(request.species or [])
if request.reactions is not None:
targets += [
s for s in reaction_species(document, request.reactions or None)
if s not in targets
] # fmt: skip
if not targets:
raise HTTPException(422, {"errors": ["name species or reactions for the recipe"]})
document = set_recipe(
document,
baseline_recipe(
targets,
identifier=request.id,
name=request.name,
reference_contract=request.reference_contract,
profile=request.profile,
),
)
except UnknownReference as error:
raise HTTPException(404, str(error)) from None
except ValueError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document = _with_edit(document, request, f"Saved recipe {request.id}")
state = commit(entry, document, revision=request.revision, warnings=entry.warnings)
return RecipeResponse(**state.model_dump(), report=recipe_report(document, request.id))
@router.post("/documents/{identifier}/recipes/{recipe}/remove", tags=["recipes"])
def remove_recipe_route(
identifier: str, recipe: str, request: RevisionRequest
) -> DocumentResponse:
"""Remove a recipe from the document."""
from vibesketch.document.recipe import UnknownRecipe, remove_recipe
entry = entry_or_404(identifier)
try:
document = remove_recipe(entry.document, recipe)
except UnknownRecipe as error:
raise HTTPException(404, str(error)) from None
document = _with_edit(document, request, f"Removed recipe {recipe}")
return commit(entry, document, revision=request.revision, warnings=entry.warnings)
@router.post("/documents/{identifier}/recipes/{recipe}/campaigns", tags=["recipes"])
def compile_recipe(identifier: str, recipe: str, request: CompileRequest) -> CompileResponse:
"""Compile a recipe into a vibe-queue campaign in a new directory, recorded on it.
422 with every reason when the recipe cannot run as asked (a refused
stage, a missing conformer, an unacknowledged MACE licence, a
directory in use); the answer carries the ``vq campaign run`` command.
"""
from pathlib import Path
from vibesketch.document.campaign import CampaignRefused, compile_campaign
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(
409, {"message": "the document has changed", "revision": entry.document.revision}
)
if not Path(request.directory).is_absolute():
raise HTTPException(422, {"errors": ["directory must be an absolute path"]})
try:
compiled = compile_campaign(
entry.document,
recipe,
request.directory,
campaign_id=request.id,
host=request.host,
program=request.program,
python=request.python,
runner=request.runner,
cpus=request.cpus,
mem_mb=request.mem_mb,
accept_model_license=request.accept_model_license,
stage_runtimes=request.stage_runtimes,
wall_time_seconds=request.wall_time_seconds,
check_wall_time_seconds=request.check_wall_time_seconds,
)
except CampaignRefused as error:
raise HTTPException(422, {"errors": error.reasons}) from None
summary = f"Compiled campaign {compiled.record.id} from recipe {recipe}"
document = _with_edit(compiled.document, request, summary)
state = commit(
entry,
document,
revision=request.revision,
warnings=[*entry.warnings, *compiled.warnings],
)
return CompileResponse(
**state.model_dump(),
campaign=compiled.record.model_dump(mode="json", exclude_none=True),
run_command=compiled.run_command,
)
@router.post(
"/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/revalidate", tags=["recipes"]
)
def campaign_revalidate(
identifier: str,
recipe: str,
campaign: str,
request: RevalidateRequest,
) -> CompileResponse:
from pathlib import Path
from vibesketch.document.review import revalidate_campaign
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"message": "the document has changed"})
if not Path(request.directory).is_absolute():
raise HTTPException(422, {"errors": ["directory must be an absolute path"]})
try:
result = revalidate_campaign(
entry.document,
recipe,
campaign,
request.directory,
node=request.node,
reviewer=request.reviewer,
reason=request.reason,
)
except (ValueError, OSError, KeyError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document = _with_edit(
result.document, request, f"Revalidated campaign {campaign} as {result.record.id}"
)
state = commit(entry, document, revision=request.revision, warnings=result.warnings)
return CompileResponse(
**state.model_dump(),
campaign=result.record.model_dump(mode="json", exclude_none=True),
run_command=result.run_command,
)
@router.post(
"/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/review", tags=["recipes"]
)
def campaign_review(
identifier: str,
recipe: str,
campaign: str,
request: ReviewRequest,
) -> CompileResponse:
from pathlib import Path
from vibesketch.document.review import review_campaign
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"message": "the document has changed"})
if not Path(request.directory).is_absolute():
raise HTTPException(422, {"errors": ["directory must be an absolute path"]})
try:
result = review_campaign(
entry.document,
recipe,
campaign,
request.directory,
node=request.node,
reviewer=request.reviewer,
reason=request.reason,
criteria=request.criteria,
warnings=request.warnings,
evidence=request.evidence,
)
except (ValueError, OSError, KeyError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
document = _with_edit(
result.document, request, f"Reviewed campaign {campaign} as {result.record.id}"
)
state = commit(entry, document, revision=request.revision, warnings=result.warnings)
return CompileResponse(
**state.model_dump(),
campaign=result.record.model_dump(mode="json", exclude_none=True),
run_command=result.run_command,
)
@router.get("/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}", tags=["recipes"])
def campaign_state(identifier: str, recipe: str, campaign: str) -> dict[str, Any]:
"""Each stage's process state in vibe-queue and its scientific acceptance, apart."""
from vibesketch.document.campaign import (
CampaignUnavailable,
UnknownCampaign,
campaign_report,
)
from vibesketch.document.recipe import UnknownRecipe
try:
return campaign_report(entry_or_404(identifier).document, recipe, campaign)
except (UnknownRecipe, UnknownCampaign) as error:
raise HTTPException(404, str(error)) from None
except CampaignUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/external-chain-report", tags=["recipes"])
def external_chain_report(identifier: str, request: ExternalChainRequest) -> Response:
import base64
import binascii
from vibesketch.document.external_results import external_chain_export
from vibesketch.document.results import ResultRefused
try:
data = base64.b64decode(request.data, validate=True)
result = external_chain_export(
entry_or_404(identifier).document,
data,
request.sha256,
request.reference_contract,
request.format,
)
except (ResultRefused, ValueError, binascii.Error) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(
result,
media_type={
"json": "application/json",
"csv": "text/csv",
"markdown": "text/markdown",
"latex": "application/x-latex",
}[request.format],
)
@router.post(
"/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/energies", tags=["recipes"]
)
def campaign_energy_report(
identifier: str, recipe: str, campaign: str, request: CampaignEnergyRequest
) -> Response:
from vibesketch.document.campaign import CampaignUnavailable
from vibesketch.document.campaign_results import CampaignSelection, campaign_energy_export
from vibesketch.document.qvf import QvfUnavailable
from vibesketch.document.results import ResultRefused
from vibesketch.render import RenderRefused, RenderUnavailable
selection = CampaignSelection(
recipe=recipe, campaign=campaign, **request.model_dump(exclude={"format"})
)
try:
data = campaign_energy_export(
entry_or_404(identifier).document, selection, request.format
)
except (ResultRefused, RenderRefused, KeyError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
except (CampaignUnavailable, QvfUnavailable, RenderUnavailable) as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
media_types = {
"json": "application/json",
"csv": "text/csv",
"markdown": "text/markdown",
"latex": "application/x-latex",
"svg": "image/svg+xml",
"qvf": "application/octet-stream",
}
return Response(data, media_type=media_types[request.format])
@router.post(
"/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/renders", tags=["recipes"]
)
def campaign_renders(
identifier: str, recipe: str, campaign: str, request: RenderStagesRequest
) -> dict[str, Any]:
"""Same-orientation renders of a species' start and accepted stages.
From the campaign's verified artifacts only; the images are vibe-view's
when it is installed. Answers with the record written as
``renders.json``, the files and the warnings; 422 when an artifact is
not the one vibe-queue verified, or the directory is in use.
"""
from pathlib import Path
from vibesketch.document.renders import RendersRefused, render_stages
entry = entry_or_404(identifier)
if not Path(request.directory).is_absolute():
raise HTTPException(422, {"errors": ["directory must be an absolute path"]})
try:
renders = render_stages(
entry.document, recipe, campaign, request.species, request.directory,
size=(request.width, request.height), publication=request.publication,
images=request.images,
) # fmt: skip
except RendersRefused as error:
raise HTTPException(422, {"errors": error.reasons}) from None
return {
"directory": str(renders.directory),
"files": renders.files,
"record": renders.record,
"warnings": renders.warnings,
}
@router.post(
"/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/retry", tags=["recipes"]
)
def campaign_retry(
identifier: str, recipe: str, campaign: str, request: RetryRequest
) -> dict[str, Any]:
"""Authorise the next attempt of a failed node; the next controller pass sends it."""
from vibesketch.document.campaign import (
CampaignRefused,
CampaignUnavailable,
UnknownCampaign,
retry_node,
)
from vibesketch.document.recipe import UnknownRecipe
try:
attempt = retry_node(entry_or_404(identifier).document, recipe, campaign, request.node)
except (UnknownRecipe, UnknownCampaign) as error:
raise HTTPException(404, str(error)) from None
except CampaignRefused as error:
raise HTTPException(409, {"errors": error.reasons}) from None
except CampaignUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
return {"node": request.node, "attempt": attempt, "retry_of": attempt - 1}
@router.post("/documents/{identifier}/results", tags=["queue"])
def import_run_result(identifier: str, request: ResultImportRequest) -> DocumentResponse:
from vibesketch.document.qvf import QvfUnavailable
from vibesketch.document.results import ResultRefused, import_result
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
payload = _payload(request.data, "base64")
assert isinstance(payload, bytes)
try:
updated = import_result(
entry.document,
request.run,
payload,
format=request.format,
source_name=request.source_name,
replace=request.replace,
)
except ResultRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
except QvfUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(updated, request, f"Imported result for {request.run}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/runs/{run_id}/views")
def result_views(identifier: str, run_id: str) -> dict[str, Any]:
from vibesketch.result_views import capabilities
try:
return capabilities(entry_or_404(identifier).document, run_id)
except (ValueError, OSError) as error:
raise HTTPException(422, str(error)) from None
@router.post("/documents/{identifier}/runs/{run_id}/view")
def show_result_view(
identifier: str, run_id: str, request: ResultViewRequest
) -> dict[str, str]:
from vibesketch.viewer import ViewerUnavailable, show_result
try:
return show_result(
entry_or_404(identifier).document,
run_id,
document_id=identifier,
**request.model_dump(),
)
except ViewerUnavailable as error:
raise HTTPException(503, str(error)) from None
except (ValueError, OSError) as error:
raise HTTPException(422, str(error)) from None
@router.post("/documents/{identifier}/runs/{run_id}/view-files")
def attach_result_file(
identifier: str, run_id: str, request: ResultFileRequest
) -> DocumentResponse:
from vibesketch.result_views import attach_file
from vibesketch.viewer import ViewerUnavailable
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
data = _payload(request.data, "base64")
assert isinstance(data, bytes)
try:
updated = attach_file(entry.document, run_id, data, request.name)
except ViewerUnavailable as error:
raise HTTPException(503, str(error)) from None
except (ValueError, OSError) as error:
raise HTTPException(422, str(error)) from None
return commit(
entry,
_with_edit(updated, request, f"Attached viewer data for {run_id}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.post("/documents/{identifier}/results/report", tags=["queue"])
def reaction_results(identifier: str, request: ResultReportRequest) -> Response:
import json
from vibesketch.document.qvf import QvfUnavailable, to_qvf
from vibesketch.document.results import (
ResultRefused,
Selection,
reaction_energies,
result_figure,
)
from vibesketch.render import RenderRefused, RenderUnavailable
document = entry_or_404(identifier).document
selection = Selection.model_validate(request.model_dump(exclude={"format"}))
try:
if request.format in {"csv", "markdown", "latex"}:
from vibesketch.document.tables import MEDIA_TYPES, reaction_table
return Response(
reaction_table(document, selection, request.format),
media_type=MEDIA_TYPES[request.format],
)
if request.format == "svg":
return Response(result_figure(document, selection), media_type="image/svg+xml")
if request.format == "qvf":
return Response(
to_qvf(document, result_selection=selection),
media_type="application/octet-stream",
)
return Response(
json.dumps(reaction_energies(document, selection), allow_nan=False),
media_type="application/json",
)
except (ResultRefused, RenderRefused) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
except (QvfUnavailable, RenderUnavailable) as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/results/spectrum", tags=["queue"])
def run_spectrum(identifier: str, request: SpectrumExport) -> Response:
"""A run's IR, Raman or UV-Vis line list, or its broadened curve, as text."""
from vibesketch.document.spectra import MEDIA_TYPES, SpectrumRefused, export_spectrum
try:
text = export_spectrum(entry_or_404(identifier).document, request)
except SpectrumRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(text, media_type=MEDIA_TYPES[request.format])
@router.get("/documents/{identifier}/results/nmr", tags=["queue"])
def run_nmr(
identifier: str,
run: str,
reference: str | None = None,
nucleus: Annotated[list[str] | None, Query()] = None,
format: Literal["json", "csv", "markdown", "latex"] = "json",
) -> Response:
"""A run's NMR shieldings and, against a reference run, shifts; JSON or a table."""
import json
from vibesketch.document.nmr import MEDIA_TYPES, NmrRefused, nmr_shifts, nmr_table
document = entry_or_404(identifier).document
try:
if format == "json":
found = nmr_shifts(document, run, reference=reference, nuclei=nucleus)
return Response(json.dumps(found, allow_nan=False), media_type="application/json")
data = nmr_table(document, run, reference=reference, nuclei=nucleus, format=format)
except NmrRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(data, media_type=MEDIA_TYPES[format])
@router.post("/documents/{identifier}/results/nmr-labels", tags=["queue"])
def place_nmr_labels(identifier: str, request: NmrLabelsRequest) -> DocumentResponse:
"""Place a run's NMR shifts as labels beside its species' drawn structure."""
from vibesketch.document.nmr import NmrRefused, with_nmr_labels
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
labelled = with_nmr_labels(
entry.document, request.run, reference=request.reference, nuclei=request.nuclei
)
except NmrRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(labelled, request, f"Placed the NMR shifts of {request.run}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/runs", tags=["queue"])
def document_runs(identifier: str) -> list[dict[str, Any]]:
"""The document's runs with their state in vibe-queue, newest first."""
from vibesketch.queue import QueueUnavailable, run_states
try:
return run_states(entry_or_404(identifier).document)
except QueueUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.get("/documents/{identifier}/groups", tags=["queue"])
def document_groups(identifier: str) -> list[dict[str, Any]]:
"""The document's run groups with their progress, newest first."""
from vibesketch.queue import QueueUnavailable, group_states
try:
return group_states(entry_or_404(identifier).document)
except QueueUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/groups/{group}/cancel", tags=["queue"])
def cancel_group_route(identifier: str, group: str) -> dict[str, Any]:
"""Cancel the group's runs that have not finished; answers with those cancelled."""
from vibesketch.queue import QueueUnavailable, cancel_group
document = entry_or_404(identifier).document
try:
cancelled = cancel_group(document, group)
except KeyError:
raise HTTPException(404, f"no run of the document belongs to group {group!r}") from None
except QueueUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
return {"group": group, "cancelled": cancelled}
@router.post("/queue/runs/{run_id}/cancel", tags=["queue"])
def cancel_run(run_id: str) -> dict[str, str]:
"""Cancel a run that has not finished."""
from vibesketch.queue import QueueUnavailable, cancel
try:
return {"run": run_id, "message": cancel(run_id)}
except KeyError:
raise HTTPException(404, f"vibe-scheduler has no run {run_id!r}") from None
except ValueError as error:
raise HTTPException(409, str(error)) from None
except QueueUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/view")
def view_species(identifier: str, request: ViewRequest) -> dict[str, str]:
"""Show a species in vibe-view: sent through the hand-off when vibe-view
started this server, else opened with ``vibe-view open``.
The species needs its 3D structure (``POST .../embed``); the answer
names the route taken and the QVF written.
"""
from vibesketch import viewer
from vibesketch.document.qvf import QvfError
entry = entry_or_404(identifier)
try:
return viewer.show(entry.document, request.species, document_id=identifier)
except KeyError:
raise HTTPException(404, f"the document has no species {request.species!r}") from None
except QvfError as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
except viewer.ViewerUnavailable as error:
raise HTTPException(503, {"errors": [str(error)]}) from None
@router.post("/documents/{identifier}/geometry-return")
def geometry_return(identifier: str, request: GeometryReturnRequest) -> DocumentResponse:
"""Accept a base64 QVF return only for this document's unchanged species graph."""
import base64
import binascii
from vibesketch.document.geometry_return import return_geometry
entry = entry_or_404(identifier)
revision = entry.document.revision
try:
data = base64.b64decode(request.data, validate=True)
document = return_geometry(entry.document, data, document_id=identifier)
except (ValueError, KeyError, binascii.Error) as error:
raise HTTPException(409, str(error)) from None
document = _with_edit(document, request, "Returned geometry from viewer")
return commit(entry, document, revision=revision, warnings=entry.warnings)
@router.patch("/documents/{identifier}/species/{species}")
def patch_species(identifier: str, species: str, request: SpeciesRequest) -> DocumentResponse:
"""Set a species' label, multiplicity or name."""
entry = entry_or_404(identifier)
data = entry.document.to_json_data()
target = next((s for s in data["species"] if s["id"] == species), None)
if target is None:
raise HTTPException(404, f"the document has no species {species!r}")
changed = []
if request.label is not None:
if request.label.strip():
target["label"] = request.label.strip()
else:
target.pop("label", None)
changed.append("label")
if request.multiplicity is not None:
if target.get("smiles"):
from vibesketch.chem.species import identity, mol_from_smiles
electrons = identity(mol_from_smiles(target["smiles"])).electrons
unpaired = request.multiplicity - 1
if unpaired > electrons or (electrons - unpaired) % 2:
raise HTTPException(
422,
{
"errors": [
f"multiplicity {request.multiplicity} is impossible with "
f"{electrons} electrons"
]
},
)
target["multiplicity"] = request.multiplicity
changed.append("multiplicity")
if request.name is not None:
if request.name.strip():
target["name"] = Name(
value=request.name.strip(), source="user", confidence="high"
).model_dump()
else:
target.pop("name", None)
changed.append("name")
try:
document = from_data(data)
except DocumentError as error:
raise HTTPException(422, {"errors": error.errors}) from None
document = _with_edit(document, request, f"Changed {species}: {', '.join(changed)}")
return commit(entry, document, revision=request.revision, warnings=entry.warnings)
@router.get("/documents/{identifier}/report")
def document_report(identifier: str, names: bool = True) -> list[dict[str, Any]]:
"""Formula, masses, identifiers and the checked name of each species."""
from vibesketch.chem.report import report
return [row.to_dict() for row in report(entry_or_404(identifier).document, names=names)]
def _highlighted(
entry: Entry, request: RevisionRequest, change: Any, summary: str
) -> DocumentResponse:
from vibesketch.document.highlights import HighlightRefused
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
updated = change(entry.document)
except HighlightRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(updated, request, summary),
revision=request.revision,
warnings=entry.warnings,
)
@router.post("/documents/{identifier}/highlights")
def highlight_atoms(identifier: str, request: HighlightRequest) -> DocumentResponse:
"""Mark atoms of a species in a colour of the palette."""
from vibesketch.document.highlights import with_highlight
def change(document: Document) -> Document:
atoms = [a - 1 for a in request.atoms]
return with_highlight(document, request.species, atoms, request.colour)
return _highlighted(
entry_or_404(identifier), request, change, f"Marked atoms of {request.species}"
)
@router.post("/documents/{identifier}/highlights/selection")
def highlight_selection(
identifier: str, request: SelectionHighlightRequest
) -> DocumentResponse:
"""Mark the atoms selected on the canvas, found at their points in the canvas's KET."""
from vibesketch.document.highlights import with_selection_highlighted
def change(document: Document) -> Document:
return with_selection_highlighted(document, request.ket, request.points, request.colour)
return _highlighted(entry_or_404(identifier), request, change, "Marked selected atoms")
@router.post("/documents/{identifier}/highlights/clear")
def clear_highlights(identifier: str, request: ClearHighlightsRequest) -> DocumentResponse:
"""Remove every highlight, or those of one species."""
from vibesketch.document.highlights import without_highlights
def change(document: Document) -> Document:
return without_highlights(document, request.species)
return _highlighted(entry_or_404(identifier), request, change, "Cleared highlights")
@router.post("/documents/{identifier}/highlights/points")
def highlight_points(identifier: str, request: HighlightPointsRequest) -> list[dict[str, Any]]:
"""Where the marked atoms are drawn in the canvas's KET, by colour, to show them."""
from vibesketch.document.highlights import highlight_points as points
return points(entry_or_404(identifier).document, request.ket)
@router.post("/documents/{identifier}/enumerate-rgroups")
def enumerate_rgroups_request(identifier: str, request: EnumerationRequest) -> DocumentResponse:
from vibesketch.document.enumeration import EnumerationRefused, enumerate_rgroups
from vibesketch.limits import LimitExceeded
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
updated = enumerate_rgroups(entry.document, request.core)
except (EnumerationRefused, LimitExceeded) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(updated, request, "Enumerated all R-group combinations"),
revision=request.revision,
warnings=entry.warnings,
)
def _abbreviated(
entry: Entry, request: AbbreviationsRequest, change: Any, summary: str
) -> DocumentResponse:
from vibesketch.document.abbreviations import AbbreviationRefused
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
updated = change(entry.document, request.species, request.labels or None)
except AbbreviationRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
if request.species:
summary += f" in {request.species}"
return commit(
entry,
_with_edit(updated, request, summary),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/abbreviations")
def list_abbreviations(identifier: str) -> dict[str, Any]:
"""The labels that can be contracted, and those contracted in each species."""
from vibesketch.document.abbreviations import abbreviation_labels, abbreviations_of
return {
"available": abbreviation_labels(),
"species": abbreviations_of(entry_or_404(identifier).document),
}
@router.post("/documents/{identifier}/abbreviations")
def contract_abbreviations(identifier: str, request: AbbreviationsRequest) -> DocumentResponse:
"""Draw groups such as Ph, Boc or OTf as their label; the structure is unchanged."""
from vibesketch.document.abbreviations import with_abbreviations
return _abbreviated(
entry_or_404(identifier), request, with_abbreviations, "Contracted abbreviations"
)
@router.post("/documents/{identifier}/abbreviations/expand")
def expand_abbreviations(identifier: str, request: AbbreviationsRequest) -> DocumentResponse:
"""Draw contracted abbreviations as their atoms again."""
from vibesketch.document.abbreviations import without_abbreviations
return _abbreviated(
entry_or_404(identifier), request, without_abbreviations, "Expanded abbreviations"
)
@router.get("/template-library")
def template_library(q: str = "", category: str = "", kind: str = "") -> list[dict[str, Any]]:
from vibesketch.document.template_library import list_templates
return [e.model_dump(mode="json") for e in list_templates(q, category=category, kind=kind)]
@router.get("/template-library/{template_id}/preview")
def template_preview(template_id: str, scaffold: bool = False) -> Response:
from vibesketch.document import new_document
from vibesketch.document.formats import ImportFailed
from vibesketch.document.reaction_library import ReactionRefused
from vibesketch.document.template_library import insert_template
from vibesketch.render import render_svg
try:
document = insert_template(new_document(), template_id, scaffold=scaffold).document
except (ReactionRefused, ImportFailed) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(render_svg(document), media_type="image/svg+xml")
@router.post("/documents/{identifier}/template-library")
def add_library_template(identifier: str, request: TemplateLibraryRequest) -> DocumentResponse:
from vibesketch.document.formats import ImportFailed
from vibesketch.document.reaction_library import ReactionRefused
from vibesketch.document.template_library import insert_template
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
result = insert_template(entry.document, request.entry, scaffold=request.scaffold)
except (ReactionRefused, ImportFailed) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(result.document, request, f"Inserted template {request.entry}"),
revision=request.revision,
warnings=[*entry.warnings, *result.warnings],
)
@router.get("/reaction-library")
def reaction_library(q: str = "") -> list[dict[str, Any]]:
"""Search the bundled offline catalogue, including aliases and source links."""
from vibesketch.document.reaction_library import list_reactions
return [entry.model_dump(mode="json") for entry in list_reactions(q)]
@router.get("/reaction-library/{reaction_id}/preview")
def reaction_preview(
reaction_id: str, kind: Literal["example", "scaffold"] = "example"
) -> Response:
import json
from vibesketch.document.formats import import_into
from vibesketch.document.reaction_library import ReactionRefused, reaction_fragment
from vibesketch.render import render_svg
try:
fragment = reaction_fragment(reaction_id, kind=kind)
document = import_into(None, json.dumps(fragment), "ket").document
except ReactionRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(render_svg(document), media_type="image/svg+xml")
@router.post("/documents/{identifier}/reaction-library")
def add_library_reaction(identifier: str, request: ReactionLibraryRequest) -> DocumentResponse:
from vibesketch.document.formats import ImportFailed
from vibesketch.document.reaction_library import ReactionRefused, insert_reaction
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
result = insert_reaction(entry.document, request.entry, kind=request.kind)
except (ReactionRefused, ImportFailed) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(
result.document, request, f"Inserted reaction {request.entry} ({request.kind})"
),
revision=request.revision,
warnings=[*entry.warnings, *result.warnings],
)
@router.post("/documents/{identifier}/surfaces")
def add_surface(identifier: str, request: SurfaceRequest) -> DocumentResponse:
from vibesketch.document.surfaces import SurfaceRefused, with_surface
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
drawn = with_surface(
entry.document,
request.parameters,
at=request.at,
width=request.width,
height=request.height,
)
except (SurfaceRefused, ValueError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(drawn, request, f"Drew surface {request.parameters.id}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/surfaces")
def document_surfaces(identifier: str) -> list[dict[str, Any]]:
from vibesketch.document.surfaces import records
return [
s.model_dump(mode="json", exclude_none=True)
for s in records(entry_or_404(identifier).document)
]
@router.post("/documents/{identifier}/nanostructures")
def add_nanostructure(identifier: str, request: NanostructureRequest) -> NanostructureResponse:
"""Generate a nanotube, graphene flake, nanoribbon or fullerene and add it as a species."""
from vibesketch.document.nanostructures import (
NanostructureRefused,
with_nanostructure,
)
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
made, summary = with_nanostructure(
entry.document,
request.parameters,
species_id=request.species_id,
label=request.label,
charge=request.charge,
multiplicity=request.multiplicity,
picture_style=request.picture_style,
picture_label=request.picture_label,
)
except NanostructureRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
response = commit(
entry,
_with_edit(made, request, f"Added {summary['description']}"),
revision=request.revision,
warnings=entry.warnings,
)
return NanostructureResponse(**response.model_dump(), nanostructure=summary)
@router.post("/documents/{identifier}/species/{species}/picture")
def add_picture(identifier: str, species: str, request: PictureRequest) -> DocumentResponse:
"""Put a picture of a nanostructure species on the canvas, as one object."""
from vibesketch.document.nanostructures import NanostructureRefused, with_picture
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
drawn = with_picture(
entry.document,
species,
request.style,
label=request.label,
rear=request.rear,
brk=request.brk,
width=request.width,
)
except NanostructureRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(drawn, request, f"Drew {species} ({request.style})"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/pictures")
def document_pictures(identifier: str) -> list[dict[str, Any]]:
"""The nanostructure pictures on the canvas with their boxes and anchors."""
from vibesketch.document.nanostructures import pictures
return pictures(entry_or_404(identifier).document)
@router.post("/documents/{identifier}/generics/{generic}/attach")
def attach_generic(identifier: str, generic: str, request: AttachRequest) -> DocumentResponse:
"""Move a labelled structure onto an anchor of a nanostructure picture."""
from vibesketch.document.nanostructures import NanostructureRefused, attach_to_picture
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
moved = attach_to_picture(
entry.document, generic, request.species, request.anchor, node=request.node
)
except NanostructureRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(moved, request, f"Attached {generic} to {request.species}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.post("/documents/{identifier}/half-reactions")
def add_half_reaction(identifier: str, request: HalfReactionRequest) -> DocumentResponse:
"""Draw an electrode half-reaction: states, coefficients, electrons and conditions."""
from vibesketch.document.electrochem import HalfReactionRefused, with_half_reaction
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
drawn, warnings = with_half_reaction(
entry.document,
request.reactants,
request.products,
electrons=request.electrons,
conditions=request.conditions,
arrow=request.arrow,
over=request.over,
)
except HalfReactionRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(drawn, request, "Drew a half-reaction"),
revision=request.revision,
warnings=warnings,
)
@router.get("/documents/{identifier}/balance")
def document_balance(identifier: str) -> list[dict[str, Any]]:
"""Each reaction's atoms and charge on both sides, electrons included, and the imbalance."""
from vibesketch.document.electrochem import balance
document = entry_or_404(identifier).document
return [dict(balance(document, r.id)) for r in document.reactions]
@router.post("/documents/{identifier}/tlc")
def add_tlc_plate(identifier: str, request: TlcRequest) -> DocumentResponse:
"""Draw a TLC plate, lanes with their spots' Rf values, beside the drawing."""
from vibesketch.document.tlc import TlcRefused, with_tlc_plate
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
drawn = with_tlc_plate(entry.document, request.lanes)
except TlcRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(drawn, request, "Drew a TLC plate"),
revision=request.revision,
warnings=entry.warnings,
)
@router.post("/documents/{identifier}/apparatus")
def add_apparatus(identifier: str, request: ApparatusRequest) -> DocumentResponse:
"""Draw a laboratory apparatus pictogram, beside the drawing or at ``at``."""
from vibesketch.document.apparatus import ApparatusRefused, with_apparatus
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
drawn = with_apparatus(
entry.document, request.kind, request.at, scale=request.scale, label=request.label
)
except ApparatusRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(drawn, request, f"Drew a {request.kind}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/generics")
def document_generics(identifier: str) -> dict[str, Any]:
"""The R-group cores, bracketed groups and R-group definitions the drawing shows."""
from vibesketch.document.generics import generics_report
return generics_report(entry_or_404(identifier).document)
@router.post("/documents/{identifier}/generics/{generic}/cap")
def cap_generic(identifier: str, generic: str, request: CapRequest) -> DocumentResponse:
"""Make a species of a labelled structure by capping its labels with H or CH3."""
from vibesketch.document.generics import CapRefused, cap_labelled
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
capped, _ = cap_labelled(entry.document, generic, request.cap)
except CapRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(capped, request, f"Capped {generic} with {request.cap}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/stereo")
def document_stereo(identifier: str) -> dict[str, Any]:
"""Each species' CIP descriptors, undefined stereo and enhanced-stereo groups."""
from vibesketch.document.stereo import stereo_report
return stereo_report(entry_or_404(identifier).document)
@router.post("/documents/{identifier}/stereo-labels")
def label_stereo(identifier: str, request: StereoLabelsRequest) -> DocumentResponse:
"""Write the CIP descriptors into the drawing as labels, or clear them."""
from vibesketch.document.stereo import with_stereo_labels
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
labelled = with_stereo_labels(entry.document, clear=request.clear)
summary = "Cleared the stereo labels" if request.clear else "Labelled the stereochemistry"
return commit(
entry,
_with_edit(labelled, request, summary),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/properties")
def species_properties(
identifier: str, format: Literal["csv", "markdown", "latex"] = "csv"
) -> Response:
"""Formula, masses, elemental analysis and calculated descriptors per species."""
from vibesketch.document.properties import (
MEDIA_TYPES,
PropertiesRefused,
property_table,
)
try:
data = property_table(entry_or_404(identifier).document, format)
except PropertiesRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return Response(data, media_type=MEDIA_TYPES[format])
@router.post("/documents/{identifier}/species/{species}/property-text")
def place_property_text(
identifier: str, species: str, request: PropertyTextRequest
) -> DocumentResponse:
"""Place a species' properties as text under its structure."""
from vibesketch.document.properties import PropertiesRefused, with_property_text
entry = entry_or_404(identifier)
if request.revision != entry.document.revision:
raise HTTPException(409, {"revision": entry.document.revision})
try:
placed = with_property_text(entry.document, species, request.fields)
except PropertiesRefused as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return commit(
entry,
_with_edit(placed, request, f"Placed the properties of {species}"),
revision=request.revision,
warnings=entry.warnings,
)
@router.get("/documents/{identifier}/export/{fmt}")
def export_document(
identifier: str,
fmt: str,
download: Annotated[bool, Query()] = False,
style: str | None = None,
dpi: Annotated[int, Query(ge=36, le=2400)] = 300,
stereo_labels: bool = False,
sequences: bool = False,
) -> Response:
"""The document in ``fmt``: qvf, vsk, ket, mol, sdf, rxn, smiles, inchi, helm,
fasta, svg, png, pdf or chemfig; ``stereo_labels`` prints CIP descriptors and
``sequences`` species built from a sequence as it, in figures."""
return _export(
entry_or_404(identifier).document,
fmt,
download,
stem=identifier,
style=style,
dpi=dpi,
stereo_labels=stereo_labels,
sequences=sequences,
)
@router.post("/documents/{identifier}/save")
def save_document(identifier: str) -> DocumentResponse:
"""Write the document back to the file the server opened it from."""
entry_or_404(identifier)
try:
return _response(store.save(identifier))
except ValueError as error:
raise HTTPException(409, str(error)) from None
except OSError as error:
reason = error.strerror or "the operating system refused the write"
status = 507 if error.errno in (errno.ENOSPC, errno.EDQUOT) else 500
raise HTTPException(
status,
f"Could not complete saving: {reason}. Your edits remain open. "
"Retry after checking disk space and file permissions, or download a copy.",
) from None
@router.delete("/documents/{identifier}", status_code=204)
def delete_document(identifier: str) -> Response:
try:
store.remove(identifier)
except NotFound:
raise HTTPException(404, f"no document {identifier!r}") from None
return Response(status_code=204)
# ── Stateless: the same chemistry without storing anything ───────────
@router.post("/convert", tags=["chemistry"])
def convert(request: CreateRequest) -> dict[str, Any]:
"""Read ``format``/``data`` into a new document, without storing it."""
from vibesketch.document.formats import ImportFailed, import_into
if request.format is None:
raise HTTPException(422, {"errors": ["give a format and data"]})
try:
data = _payload(request.data, request.encoding)
derived = import_into(None, data, request.format, mode="replace")
except ImportFailed as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return {"document": derived.document.to_json_data(), "warnings": derived.warnings}
@router.post("/clean", tags=["chemistry"])
def clean_for(document: Document) -> dict[str, Any]:
"""The document with its drawing cleaned, without storing it."""
from vibesketch.document.clean import clean
derived, report = clean(document)
return {
"document": derived.document.to_json_data(),
"cleaned": report.cleaned,
"warnings": [*derived.warnings, *report.left],
}
@router.post("/tidy", tags=["chemistry"])
def tidy_for(
document: Document, width: Annotated[float, Query(ge=4.0)] = 24.0
) -> dict[str, Any]:
"""The document with its scheme laid out, without storing it; 422 when refused."""
from vibesketch.document.scheme import tidy_scheme
derived, report = tidy_scheme(document, width=width)
if not report.tidied:
raise HTTPException(422, {"errors": [f"left the scheme as drawn: {report.reason}"]})
return {
"document": derived.document.to_json_data(),
"rows": report.rows,
"cleaned": report.cleaned,
"warnings": [*derived.warnings, *report.left],
}
@router.post("/chem/embed3d", tags=["chemistry"])
def embed_smiles(request: Embed3dRequest) -> dict[str, Any]:
"""3D conformers for one structure given as SMILES, best first."""
from vibesketch.chem.embed import EmbedFailed, embed_3d
try:
result = embed_3d(
request.smiles, count=request.count, attempts=max(10, 3 * request.count)
)
except (EmbedFailed, ValueError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return {
"conformers": [c.model_dump(mode="json") for c in result.conformers],
"energies_kcal_mol": result.energies,
"open_stereo": result.open_stereo,
"warnings": result.warnings,
}
@router.post("/chem/relax", tags=["chemistry"])
def relax_conformer(request: Relax3dRequest) -> dict[str, Any]:
"""One conformer of a structure relaxed with FIRE on a force field."""
from vibesketch.chem.relax import RelaxFailed, relax_3d
try:
species = Species(id="species", smiles=request.smiles, conformers=[request.conformer])
result = relax_3d(
species,
constraints=request.constraints,
force_field=request.force_field,
fmax=request.fmax,
max_steps=request.max_steps,
)
except (RelaxFailed, ValueError) as error:
raise HTTPException(422, {"errors": [str(error)]}) from None
return result.to_dict()
@router.post("/report", tags=["chemistry"])
def report_for(document: Document, names: bool = True) -> list[dict[str, Any]]:
from vibesketch.chem.report import report
return [row.to_dict() for row in report(document, names=names)]
@router.post("/export/{fmt}", tags=["chemistry"])
def export_for(
document: Document,
fmt: str,
download: bool = False,
style: str | None = None,
dpi: Annotated[int, Query(ge=36, le=2400)] = 300,
stereo_labels: bool = False,
sequences: bool = False,
) -> Response:
return _export(
document,
fmt,
download,
stem="document",
style=style,
dpi=dpi,
stereo_labels=stereo_labels,
sequences=sequences,
)
return router
def _export(
document: Document,
fmt: str,
download: bool,
*,
stem: str,
style: str | None = None,
dpi: int = 300,
stereo_labels: bool = False,
sequences: bool = False,
) -> Response:
from vibesketch.document.formats import ExportRefused, export
from vibesketch.render import RenderRefused, RenderUnavailable
try:
payload = export(
document, fmt, style=style, dpi=dpi, stereo_labels=stereo_labels, sequences=sequences
)
except (ExportRefused, RenderRefused) as error:
raise HTTPException(422, {"format": fmt, "reasons": [str(error)]}) from None
except RenderUnavailable as error:
raise HTTPException(503, {"format": fmt, "reasons": [str(error)]}) from None
headers = {}
if download:
headers["Content-Disposition"] = f'attachment; filename="{stem}.{payload.extension}"'
return Response(payload.content, media_type=payload.media_type, headers=headers)