diff --git a/docs/_exts/plr_devices/__init__.py b/docs/_exts/plr_devices/__init__.py index 7ace642a06d..7653dc31691 100644 --- a/docs/_exts/plr_devices/__init__.py +++ b/docs/_exts/plr_devices/__init__.py @@ -1 +1,5 @@ -from .directive import setup # re-export for Sphinx +"""Sphinx directives that render the device registry (``docs/_static/devices.json``). + +``conf.py`` loads ``plr_devices.directive``, the submodule holding ``setup``, so that +``plr_devices.data`` and ``plr_devices.html`` stay importable without Sphinx. +""" diff --git a/docs/_exts/plr_devices/data.py b/docs/_exts/plr_devices/data.py new file mode 100644 index 00000000000..0618aa0668b --- /dev/null +++ b/docs/_exts/plr_devices/data.py @@ -0,0 +1,247 @@ +"""Loading and validation of the device registry (``devices.json``).""" + +import json +import zlib +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence + + +class DeviceRegistryError(ValueError): + """A registry entry is malformed.""" + + +# Controlled vocabularies. They keep near-synonyms from drifting apart ("sealer" vs "heat +# sealer"), which would split the filter chips in two. Adding a genuinely new kind or capability +# means adding it here as well, deliberately. + +KINDS = ( + "arm", + "barcode scanner", + "bulk dispenser", + "centrifuge", + "centrifuge loader", + "decapper", + "delidder", + "fan", + "flow cytometer", + "heater shaker", + "liquid handler", + "microscope", + "peeler", + "plate reader", + "plate washer", + "pump", + "qPCR machine", + "scale", + "sealer", + "shaker", + "storage", + "temperature controller", + "thermocycler", + "tilter", +) + +CAPABILITIES = ( + "absorbance", + "active cooling", + "air filtration", + "arm", + "barcode reading", + "centrifuging", + "decapping", + "delidding", + "dispensing", + "flow cytometry", + "fluorescence", + "fluorescence polarization", + "grinding", + "heating", + "liquid handling", + "luminescence", + "microscopy", + "peeling", + "plate washing", + "pumping", + "qPCR", + "sealing", + "shaking", + "storage", + "thermocycling", + "tilting", + "time-resolved fluorescence", + "weighing", +) + +# Ordered from least to most complete; the table renders statuses in this order. +STATUSES = ("wip", "basic", "mostly", "full") + +STATUS_LABELS = { + "wip": "WIP", + "basic": "Basic", + "mostly": "Mostly", + "full": "Full", +} + +STATUS_DESCRIPTIONS = { + "wip": "Work in progress.", + "basic": "Core functionality is available.", + "mostly": "Most capabilities are available, but some known commands are still missing.", + "full": "Comprehensive support (at least 90% of capabilities), with documentation.", +} + +API_VERSIONS = ("v0", "v1") + +API_VERSION_DESCRIPTIONS = { + "v0": "Driver still lives under pylabrobot.legacy and is being migrated.", + "v1": "Driver uses the current API.", +} + +REQUIRED_FIELDS = ("id", "vendor", "name", "kind", "status") + +OPTIONAL_FIELDS = ( + "capabilities", + "api", + "api_version", + "code_slug", + "doc_slug", + "manager", + "oem", + "notes", +) + +_ALL_FIELDS = set(REQUIRED_FIELDS) | set(OPTIONAL_FIELDS) + + +class Device(Dict[str, Any]): + """A single registry entry: a plain dict with a few derived conveniences.""" + + @property + def id(self) -> str: + return str(self["id"]) + + @property + def title(self) -> str: + return f"{self['vendor']} {self['name']}" + + @property + def status_label(self) -> str: + return STATUS_LABELS.get(str(self["status"]), str(self["status"])) + + +def capability_hue(capability: str) -> int: + """A stable hue (0-359) for a capability, so badge colors survive new capabilities.""" + return zlib.crc32(capability.encode("utf-8")) % 360 + + +def _validate(device: Any, index: int, seen_ids: Dict[str, int], path: Path) -> Device: + where = f"{path}[{index}]" + + if not isinstance(device, dict): + raise DeviceRegistryError( + f"{where}: device entries must be JSON objects, got {type(device).__name__}" + ) + + missing = [f for f in REQUIRED_FIELDS if not device.get(f)] + if missing: + raise DeviceRegistryError(f"{where}: missing required field(s): {', '.join(missing)}") + + unknown = sorted(set(device) - _ALL_FIELDS) + if unknown: + raise DeviceRegistryError(f"{where} ({device['id']}): unknown field(s): {', '.join(unknown)}") + + device_id = device["id"] + if device_id in seen_ids: + raise DeviceRegistryError( + f"{where}: duplicate device id {device_id!r} (first seen at index {seen_ids[device_id]})" + ) + seen_ids[device_id] = index + + if device["kind"] not in KINDS: + raise DeviceRegistryError( + f"{where} ({device_id}): kind {device['kind']!r} is not a known kind. Add it to KINDS in " + f"{Path(__file__).name} if it is genuinely new." + ) + + if device["status"] not in STATUSES: + raise DeviceRegistryError( + f"{where} ({device_id}): status {device['status']!r} is not one of {', '.join(STATUSES)}" + ) + + api_version = device.get("api_version") + if api_version is not None and api_version not in API_VERSIONS: + raise DeviceRegistryError( + f"{where} ({device_id}): api_version {api_version!r} is not one of {', '.join(API_VERSIONS)}" + ) + + for field in ("manager", "oem"): + value = device.get(field) + if value is not None and not str(value).startswith(("http://", "https://")): + raise DeviceRegistryError(f"{where} ({device_id}): {field} must be a URL, got {value!r}") + + capabilities = device.get("capabilities", []) + if not isinstance(capabilities, list) or not all(isinstance(c, str) for c in capabilities): + raise DeviceRegistryError(f"{where} ({device_id}): capabilities must be a list of strings") + + unknown_capabilities = [c for c in capabilities if c not in CAPABILITIES] + if unknown_capabilities: + raise DeviceRegistryError( + f"{where} ({device_id}): unknown capability/capabilities: {', '.join(unknown_capabilities)}. " + f"Add to CAPABILITIES in {Path(__file__).name} if genuinely new." + ) + + return Device(device) + + +def load_devices(path: Path) -> List[Device]: + """Read and validate the registry at ``path``, sorted by vendor and then name.""" + + if not path.is_file(): + raise DeviceRegistryError(f"device registry not found: {path}") + + with open(path, encoding="utf-8") as f: + try: + raw = json.load(f) + except json.JSONDecodeError as e: + raise DeviceRegistryError(f"{path}: invalid JSON: {e}") from e + + if not isinstance(raw, list): + raise DeviceRegistryError(f"{path}: expected a JSON array of devices, got {type(raw).__name__}") + + seen_ids: Dict[str, int] = {} + devices = [_validate(d, i, seen_ids, path) for i, d in enumerate(raw)] + devices.sort(key=lambda d: (str(d["vendor"]).lower(), str(d["name"]).lower())) + return devices + + +def registry_path(app_or_env) -> Path: + """Absolute path of the registry. Accepts either the app or the build environment.""" + base = getattr(app_or_env, "confdir", None) or app_or_env.srcdir + return Path(base) / app_or_env.config.plr_devices_json + + +def get_devices(app) -> List[Device]: + """Return the registry for this build, loading it on first use.""" + + devices = getattr(app, "plr_devices", None) + if devices is None: + devices = load_devices(registry_path(app)) + app.plr_devices = devices + return devices + + +def get_device(app, device_id: str) -> Optional[Device]: + for device in get_devices(app): + if device["id"] == device_id: + return device + return None + + +def filter_devices(devices: Sequence[Device], filters: Dict[str, str]) -> List[Device]: + """Keep devices matching every filter.""" + + def matches(device: Device, field: str, wanted: str) -> bool: + if field == "capabilities": + return wanted.lower() in {c.lower() for c in device.get("capabilities", [])} + return str(device.get(field, "")).lower() == wanted.lower() + + return [d for d in devices if all(matches(d, f, w) for f, w in filters.items() if w)] diff --git a/docs/_exts/plr_devices/directive.py b/docs/_exts/plr_devices/directive.py index d5977537150..a948f7102e1 100644 --- a/docs/_exts/plr_devices/directive.py +++ b/docs/_exts/plr_devices/directive.py @@ -1,124 +1,218 @@ -"""Sphinx directive that renders a 'Supported hardware' table from devices.json. +"""Sphinx directives that render the device registry (``devices.json``). -Usage in MyST markdown:: +``device-table`` renders a searchable table of every device in the registry, or of +the subset selected by its options:: - ```{supported-devices} shaking + ```{device-table} ``` -Or with multiple capabilities:: + ```{device-table} + :kind: plate reader + :capabilities: absorbance, luminescence + ``` + +``device-card`` renders one device as a card, by registry id:: - ```{supported-devices} heating, shaking + ```{device-card} curiox-ht2000 ``` -The directive filters devices.json to rows where the device's ``capabilities`` -list intersects with the requested set, then renders a native docutils table -styled by the active Sphinx theme. +Both are usable anywhere MyST is parsed, including markdown cells of the notebooks +under ``docs/user_guide``. """ -import json from pathlib import Path from docutils import nodes -from docutils.parsers.rst import Directive - - -_DEVICES = None - - -def _load_devices(): - global _DEVICES - if _DEVICES is None: - json_path = Path(__file__).resolve().parents[2] / "_static" / "devices.json" - with open(json_path, encoding="utf-8") as f: - _DEVICES = json.load(f) - return _DEVICES - - -class SupportedDevices(Directive): - """Render a table of devices that have the requested capabilities.""" - - has_content = False - required_arguments = 1 - optional_arguments = 0 - final_argument_whitespace = True - - def run(self): - requested = {c.strip() for c in self.arguments[0].split(",")} - devices = _load_devices() - matches = [ - d for d in devices if requested & set(d.get("capabilities", [])) - ] - - if not matches: - para = nodes.paragraph( - text=f"No supported devices found for: {', '.join(requested)}" - ) - return [para] - - matches.sort(key=lambda d: (d["vendor"], d["name"])) - - # Build a native docutils table - table = nodes.table() - table["classes"].append("table") - - tgroup = nodes.tgroup(cols=4) - table += tgroup - for _ in range(4): - tgroup += nodes.colspec() - - # Header - thead = nodes.thead() - tgroup += thead - header_row = nodes.row() - thead += header_row - for title in ("Device", "Vendor", "Status", "Links"): - entry = nodes.entry() - entry += nodes.paragraph(text=title) - header_row += entry - - # Body - tbody = nodes.tbody() - tgroup += tbody - for d in matches: - row = nodes.row() - tbody += row - - # Device name (bold) - name_entry = nodes.entry() - name_entry += nodes.strong(text=d["name"]) - row += name_entry - - # Vendor - vendor_entry = nodes.entry() - vendor_entry += nodes.paragraph(text=d["vendor"]) - row += vendor_entry - - # Status - status_entry = nodes.entry() - status_entry += nodes.paragraph(text=d.get("status", "")) - row += status_entry - - # Links - links_entry = nodes.entry() - link_nodes = [] - if d.get("docs"): - ref = nodes.reference("", "docs", refuri=d["docs"]) - link_nodes.append(ref) - if d.get("oem"): - if link_nodes: - link_nodes.append(nodes.Text(" · ")) - ref = nodes.reference("", "oem", refuri=d["oem"]) - link_nodes.append(ref) - if link_nodes: - para = nodes.paragraph() - for n in link_nodes: - para += n - links_entry += para - row += links_entry - - return [table] +from docutils.parsers.rst import Directive, directives +from sphinx.errors import ExtensionError +from sphinx.util import logging + +from .data import ( + DeviceRegistryError, + filter_devices, + get_device, + get_devices, + registry_path, +) +from .html import render_card, render_table + +logger = logging.getLogger(__name__) + + +class device_placeholder(nodes.General, nodes.Element): + """Replaced with rendered HTML once docnames can be resolved to URIs.""" + + +def _note_registry_dependency(directive): + """Rebuild pages that use a directive whenever devices.json changes.""" + env = directive.state.document.settings.env + env.note_dependency(str(registry_path(env))) + + +def _flag(raw): + if raw is None or raw.strip() == "": + return True + return raw.strip().lower() not in ("false", "no", "off", "0") + + +class DeviceTable(Directive): + """Render a searchable table of registry devices.""" + + has_content = False + required_arguments = 0 + optional_arguments = 0 + option_spec = { + "capabilities": directives.unchanged, + "vendor": directives.unchanged, + "kind": directives.unchanged, + "status": directives.unchanged, + "search": directives.unchanged, + "filters": directives.unchanged, + } + + def run(self): + _note_registry_dependency(self) + node = device_placeholder("") + node["kind"] = "table" + node["filters"] = { + field: self.options.get(field, "").strip() + for field in ("capabilities", "vendor", "kind", "status") + } + node["search"] = _flag(self.options.get("search")) + node["filters_ui"] = _flag(self.options.get("filters")) + return [node] + + +class DeviceCard(Directive): + """Render one device from the registry as a card.""" + + has_content = False + required_arguments = 1 + optional_arguments = 0 + final_argument_whitespace = True + option_spec: dict = {} + + def run(self): + _note_registry_dependency(self) + node = device_placeholder("") + node["kind"] = "card" + node["device_id"] = self.arguments[0].strip() + node.line = self.lineno + return [node] + + +def _doc_uri_factory(app, fromdocname): + """Resolve a device's ``doc_slug`` to a URI relative to the page being rendered.""" + + env = app.builder.env + prefix = app.config.plr_devices_doc_prefix + + def doc_uri(doc_slug): + docname = prefix + doc_slug + if docname not in env.found_docs: + logger.warning( + "device registry: doc_slug %r does not name a page (looked for %r)", + doc_slug, + docname, + location=fromdocname, + ) + return None + try: + return app.builder.get_relative_uri(fromdocname, docname) + except Exception: # builders without relative URIs, e.g. `dummy` + return None + + return doc_uri + + +def _code_base_from_context(app): + """Point code links at the same repository and branch as the theme's "edit this page" links.""" + context = app.config.html_context + return "https://github.com/{github_user}/{github_repo}/blob/{github_version}".format(**context) + + +def _code_uri_factory(app, fromdocname): + """Resolve a device's ``code_slug`` to a URL for its driver's source.""" + + base = (app.config.plr_devices_code_base or _code_base_from_context(app)).rstrip("/") + source_root = Path(app.confdir).parent / app.config.plr_devices_code_root + + def code_uri(code_slug): + target = source_root / code_slug + if not (target.is_dir() or target.with_suffix(".py").is_file()): + logger.warning( + "device registry: code_slug %r is not a module or package under %s", + code_slug, + source_root, + location=fromdocname, + ) + return None + return f"{base}/{app.config.plr_devices_code_root}/{code_slug}" + + return code_uri + + +def _resolve_card_device(app, node, fromdocname): + device = get_device(app, node["device_id"]) + if device is None: + logger.warning( + "device-card: no device with id %r in the device registry", + node["device_id"], + location=(fromdocname, node.line), + ) + return device + + +def _render(app, doctree, fromdocname): + doc_uri = _doc_uri_factory(app, fromdocname) + code_uri = _code_uri_factory(app, fromdocname) + + for index, node in enumerate(list(doctree.findall(device_placeholder))): + if node["kind"] == "table": + devices = filter_devices(get_devices(app), node["filters"]) + html = render_table( + devices, + doc_uri, + code_uri, + table_id=f"plr-devices-{index}", + search=node["search"], + filters=node["filters_ui"], + ) + else: + device = _resolve_card_device(app, node, fromdocname) + if device is None: + node.parent.remove(node) + continue + html = render_card(device, doc_uri, code_uri) + + node.replace_self(nodes.raw("", html, format="html")) + + +def _load_registry(app): + """Load and validate the registry up front, so errors surface before any page is read.""" + app.plr_devices = None + try: + get_devices(app) + except DeviceRegistryError as e: + raise ExtensionError(str(e)) from e def setup(app): - app.add_directive("supported-devices", SupportedDevices) - return {"version": "0.2", "parallel_read_safe": True} + app.add_config_value("plr_devices_json", "_static/devices.json", "env") + # doc_slug is relative to this docname prefix; code_slug to this directory of the repository. + app.add_config_value("plr_devices_doc_prefix", "user_guide/", "env") + app.add_config_value("plr_devices_code_root", "pylabrobot", "env") + # Empty means: the repository and branch in html_context. + app.add_config_value("plr_devices_code_base", "", "env") + + app.add_node(device_placeholder) + app.add_directive("device-table", DeviceTable) + app.add_directive("device-card", DeviceCard) + + app.add_css_file("plr_devices.css") + app.add_js_file("plr_devices.js") + + app.connect("builder-inited", _load_registry) + app.connect("doctree-resolved", _render) + + return {"version": "1.0", "parallel_read_safe": True, "parallel_write_safe": True} diff --git a/docs/_exts/plr_devices/html.py b/docs/_exts/plr_devices/html.py new file mode 100644 index 00000000000..dca457471fb --- /dev/null +++ b/docs/_exts/plr_devices/html.py @@ -0,0 +1,389 @@ +"""HTML fragments for device tables and device cards.""" + +from html import escape +from typing import Callable, Dict, List, Optional, Sequence, Tuple + +from .data import ( + API_VERSION_DESCRIPTIONS, + STATUS_DESCRIPTIONS, + STATUS_LABELS, + Device, + capability_hue, +) + +# Turns a device's doc_slug into a URI relative to the page being rendered, and its code_slug +# into a source URL. Either may return None, in which case that link is left out. +DocURI = Callable[[str], Optional[str]] +CodeURI = Callable[[str], Optional[str]] + + +# Inline styles for a card that has to stand on its own, outside a page that loads +# plr_devices.css: a notebook opened in VS Code, JupyterLab, nbviewer or GitHub. Those renderers +# strip