#-----------------------------------------------------------------------------
# Copyright (c) 2012 - 2024, Anaconda, Inc., and Bokeh Contributors.
# All rights reserved.
#
# The full license is in the file LICENSE.txt, distributed with this software.
#-----------------------------------------------------------------------------
''' Provide functions and classes to help with various JS and CSS compilation.
'''
#-----------------------------------------------------------------------------
# Boilerplate
#-----------------------------------------------------------------------------
from __future__ import annotations
import logging # isort:skip
log = logging.getLogger(__name__)
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Standard library imports
import hashlib
import json
import os
import re
import sys
from os.path import (
abspath,
dirname,
exists,
isabs,
join,
)
from pathlib import Path
from subprocess import PIPE, Popen
from typing import Any, Callable, Sequence
# Bokeh imports
from ..core.has_props import HasProps
from ..settings import settings
from .strings import snakify
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
__all__ = (
'AttrDict',
'bundle_all_models',
'bundle_models',
'calc_cache_key',
'CompilationError',
'CustomModel',
'FromFile',
'get_cache_hook',
'Implementation',
'Inline',
'JavaScript',
'Less',
'nodejs_compile',
'nodejs_version',
'npmjs_version',
'set_cache_hook',
'TypeScript',
)
#-----------------------------------------------------------------------------
# General API
#-----------------------------------------------------------------------------
[docs]
class AttrDict(dict[str, Any]):
''' Provide a dict subclass that supports access by named attributes.
'''
def __getattr__(self, key: str) -> Any:
return self[key]
[docs]
class CompilationError(RuntimeError):
''' A ``RuntimeError`` subclass for reporting JS compilation errors.
'''
def __init__(self, error: dict[str, str] | str) -> None:
super().__init__()
if isinstance(error, dict):
self.line = error.get("line")
self.column = error.get("column")
self.message = error.get("message")
self.text = error.get("text", "")
self.annotated = error.get("annotated")
else:
self.text = error
def __str__(self) -> str:
return "\n" + self.text.strip()
bokehjs_dir = settings.bokehjsdir()
nodejs_min_version = (18, 0, 0)
def nodejs_version() -> str | None:
return _version(_run_nodejs)
def npmjs_version() -> str | None:
return _version(_run_npmjs)
def nodejs_compile(code: str, lang: str = "javascript", file: str | None = None) -> AttrDict:
compilejs_script = join(bokehjs_dir, "js", "compiler.js")
output = _run_nodejs([compilejs_script], dict(code=code, lang=lang, file=file, bokehjs_dir=bokehjs_dir))
lines = output.split("\n")
for i, line in enumerate(lines):
if not line.startswith("LOG"):
break
else:
print(line)
obj = json.loads("\n".join(lines[i:]))
if isinstance(obj, dict):
return AttrDict(obj)
raise CompilationError(obj)
[docs]
class Implementation:
''' Base class for representing Bokeh custom model implementations.
'''
file: str | None = None
code: str
@property
def lang(self) -> str:
raise NotImplementedError()
[docs]
class Inline(Implementation):
''' Base class for representing Bokeh custom model implementations that may
be given as inline code in some language.
Args:
code (str) :
The source code for the implementation
file (str, optional)
A file path to a file containing the source text (default: None)
'''
def __init__(self, code: str, file: str|None = None) -> None:
self.code = code
self.file = file
[docs]
class TypeScript(Inline):
''' An implementation for a Bokeh custom model in TypeScript
Example:
.. code-block:: python
class MyExt(Model):
__implementation__ = TypeScript(""" <TypeScript code> """)
'''
@property
def lang(self) -> str:
return "typescript"
[docs]
class JavaScript(Inline):
''' An implementation for a Bokeh custom model in JavaScript
Example:
.. code-block:: python
class MyExt(Model):
__implementation__ = JavaScript(""" <JavaScript code> """)
'''
@property
def lang(self) -> str:
return "javascript"
[docs]
class Less(Inline):
''' An implementation of a Less CSS style sheet.
'''
@property
def lang(self) -> str:
return "less"
[docs]
class FromFile(Implementation):
''' A custom model implementation read from a separate source file.
Args:
path (str) :
The path to the file containing the extension source code
'''
def __init__(self, path: str) -> None:
with open(path, encoding="utf-8") as f:
self.code = f.read()
self.file = path
@property
def lang(self) -> str:
if self.file is not None:
if self.file.endswith(".ts"):
return "typescript"
if self.file.endswith(".js"):
return "javascript"
if self.file.endswith((".css", ".less")):
return "less"
raise ValueError(f"unknown file type {self.file}")
#: recognized extensions that can be compiled
exts = (".ts", ".js", ".css", ".less")
[docs]
class CustomModel:
''' Represent a custom (user-defined) Bokeh model.
'''
def __init__(self, cls: type[HasProps]) -> None:
self.cls = cls
@property
def name(self) -> str:
return self.cls.__name__
@property
def full_name(self) -> str:
name = self.cls.__module__ + "." + self.name
return name.replace("__main__.", "")
@property
def file(self) -> str | None:
module = sys.modules[self.cls.__module__]
if hasattr(module, "__file__") and (file := module.__file__) is not None:
return abspath(file)
else:
return None
@property
def path(self) -> str:
path = getattr(self.cls, "__base_path__", None)
if path is not None:
return path
elif self.file is not None:
return dirname(self.file)
else:
return os.getcwd()
@property
def implementation(self) -> Implementation:
impl = getattr(self.cls, "__implementation__")
if isinstance(impl, str):
if "\n" not in impl and impl.endswith(exts):
impl = FromFile(impl if isabs(impl) else join(self.path, impl))
else:
impl = TypeScript(impl)
if isinstance(impl, Inline) and impl.file is None:
file = f"{self.file + ':' if self.file else ''}{self.name}.ts"
impl = impl.__class__(impl.code, file)
return impl
@property
def dependencies(self) -> dict[str, str]:
return getattr(self.cls, "__dependencies__", {})
@property
def module(self) -> str:
return f"custom/{snakify(self.full_name)}"
[docs]
def get_cache_hook() -> Callable[[CustomModel, Implementation], AttrDict | None]:
'''Returns the current cache hook used to look up the compiled
code given the CustomModel and Implementation'''
return _CACHING_IMPLEMENTATION
[docs]
def set_cache_hook(hook: Callable[[CustomModel, Implementation], AttrDict | None]) -> None:
'''Sets a compiled model cache hook used to look up the compiled
code given the CustomModel and Implementation'''
global _CACHING_IMPLEMENTATION
_CACHING_IMPLEMENTATION = hook
[docs]
def calc_cache_key(custom_models: dict[str, CustomModel]) -> str:
''' Generate a key to cache a custom extension implementation with.
There is no metadata other than the Model classes, so this is the only
base to generate a cache key.
We build the model keys from the list of ``model.full_name``. This is
not ideal but possibly a better solution can be found found later.
'''
model_names = {model.full_name for model in custom_models.values()}
encoded_names = ",".join(sorted(model_names)).encode('utf-8')
return hashlib.sha256(encoded_names).hexdigest()
_bundle_cache: dict[str, str] = {}
[docs]
def bundle_models(models: Sequence[type[HasProps]] | None) -> str | None:
"""Create a bundle of selected `models`. """
custom_models = _get_custom_models(models)
if custom_models is None:
return None
key = calc_cache_key(custom_models)
bundle = _bundle_cache.get(key, None)
if bundle is None:
try:
_bundle_cache[key] = bundle = _bundle_models(custom_models)
except CompilationError as error:
print("Compilation failed:", file=sys.stderr)
print(str(error), file=sys.stderr)
sys.exit(1)
return bundle
[docs]
def bundle_all_models() -> str | None:
"""Create a bundle of all models. """
return bundle_models(None)
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Private API
#-----------------------------------------------------------------------------
_plugin_umd = \
"""\
(function(root, factory) {
factory(root["Bokeh"]);
})(this, function(Bokeh) {
let define;
return %(content)s;
});
"""
# XXX: this is (almost) the same as bokehjs/src/js/plugin-prelude.js
_plugin_prelude = \
"""\
(function outer(modules, entry) {
if (Bokeh != null) {
return Bokeh.register_plugin(modules, entry);
} else {
throw new Error("Cannot find Bokeh. You have to load it prior to loading plugins.");
}
})
"""
_plugin_template = \
"""\
%(prelude)s\
({
"custom/main": function(require, module, exports) {
const models = {
%(exports)s
};
require("base").register_models(models);
module.exports = models;
},
%(modules)s
}, "custom/main");
"""
_style_template = \
"""\
(function() {
const head = document.getElementsByTagName('head')[0];
const style = document.createElement('style');
style.type = 'text/css';
const css = %(css)s;
if (style.styleSheet) {
style.styleSheet.cssText = css;
} else {
style.appendChild(document.createTextNode(css));
}
head.appendChild(style);
}());
"""
_export_template = \
""""%(name)s": require("%(module)s").%(name)s"""
_module_template = \
""""%(module)s": function(require, module, exports) {\n%(source)s\n}"""
def _detect_nodejs() -> str:
nodejs_path = settings.nodejs_path()
nodejs_paths = [nodejs_path] if nodejs_path is not None else ["nodejs", "node"]
for nodejs_path in nodejs_paths:
try:
proc = Popen([nodejs_path, "--version"], stdout=PIPE, stderr=PIPE)
(stdout, _) = proc.communicate()
except OSError:
continue
if proc.returncode != 0:
continue
match = re.match(r"^v(\d+)\.(\d+)\.(\d+).*$", stdout.decode("utf-8"))
if match is not None:
version = tuple(int(v) for v in match.groups())
if version >= nodejs_min_version:
return nodejs_path
# if we've reached here, no valid version was found
version_repr = ".".join(str(x) for x in nodejs_min_version)
raise RuntimeError(f'node.js v{version_repr} or higher is needed to allow compilation of custom models ' +
'("conda install nodejs" or follow https://nodejs.org/en/download/)')
_nodejs = None
_npmjs = None
def _nodejs_path() -> str:
global _nodejs
if _nodejs is None:
_nodejs = _detect_nodejs()
return _nodejs
def _npmjs_path() -> str:
global _npmjs
if _npmjs is None:
_npmjs = join(dirname(_nodejs_path()), "npm")
if sys.platform == "win32":
_npmjs += '.cmd'
return _npmjs
def _crlf_cr_2_lf(s: str) -> str:
return re.sub(r"\\r\\n|\\r|\\n", r"\\n", s)
def _run(app: str, argv: list[str], input: dict[str, Any] | None = None) -> str:
proc = Popen([app, *argv], stdout=PIPE, stderr=PIPE, stdin=PIPE)
(stdout, errout) = proc.communicate(input=None if input is None else json.dumps(input).encode())
if proc.returncode != 0:
raise RuntimeError(errout.decode('utf-8'))
else:
return _crlf_cr_2_lf(stdout.decode('utf-8'))
def _run_nodejs(argv: list[str], input: dict[str, Any] | None = None) -> str:
return _run(_nodejs_path(), argv, input)
def _run_npmjs(argv: list[str], input: dict[str, Any] | None = None) -> str:
return _run(_npmjs_path(), argv, input)
def _version(run_app: Callable[[list[str], dict[str, Any] | None], str]) -> str | None:
try:
version = run_app(["--version"], None) # explicit None to make mypy happy
except RuntimeError:
return None
else:
return version.strip()
def _model_cache_no_op(model: CustomModel, implementation: Implementation) -> AttrDict | None:
"""Return cached compiled implementation"""
return None
_CACHING_IMPLEMENTATION = _model_cache_no_op
def _get_custom_models(models: Sequence[type[HasProps]] | None) -> dict[str, CustomModel] | None:
"""Returns CustomModels for models with a custom `__implementation__`"""
custom_models: dict[str, CustomModel] = dict()
for cls in models or HasProps.model_class_reverse_map.values():
impl = getattr(cls, "__implementation__", None)
if impl is not None:
model = CustomModel(cls)
custom_models[model.full_name] = model
return custom_models if custom_models else None
def _compile_models(custom_models: dict[str, CustomModel]) -> dict[str, AttrDict]:
"""Returns the compiled implementation of supplied `models`. """
ordered_models = sorted(custom_models.values(), key=lambda model: model.full_name)
custom_impls = {}
dependencies: list[tuple[str, str]] = []
for model in ordered_models:
dependencies.extend(list(model.dependencies.items()))
if dependencies:
dependencies = sorted(dependencies, key=lambda name_version: name_version[0])
_run_npmjs(["install", "--no-progress"] + [ name + "@" + version for (name, version) in dependencies ])
for model in ordered_models:
impl = model.implementation
compiled = _CACHING_IMPLEMENTATION(model, impl)
if compiled is None:
compiled = nodejs_compile(impl.code, lang=impl.lang, file=impl.file)
if "error" in compiled:
raise CompilationError(compiled.error)
custom_impls[model.full_name] = compiled
return custom_impls
def _bundle_models(custom_models: dict[str, CustomModel]) -> str:
""" Create a JavaScript bundle with selected `models`. """
exports = []
modules = []
lib_dir = Path(bokehjs_dir) / "js" / "lib"
known_modules: set[str] = set()
for path in lib_dir.rglob("*.d.ts"):
s = str(path.relative_to(lib_dir))
if s.endswith(".d.ts"):
s = s[:-5] # TODO: removesuffix() (Py 3.9+)
s = s.replace(os.path.sep, "/")
known_modules.add(s)
custom_impls = _compile_models(custom_models)
extra_modules = {}
def resolve_modules(to_resolve: set[str], root: str) -> dict[str, str]:
resolved = {}
for module in to_resolve:
if module.startswith(("./", "../")):
def mkpath(module: str, ext: str = "") -> str:
return abspath(join(root, *module.split("/")) + ext)
if module.endswith(exts):
path = mkpath(module)
if not exists(path):
raise RuntimeError("no such module: %s" % module)
else:
for ext in exts:
path = mkpath(module, ext)
if exists(path):
break
else:
raise RuntimeError("no such module: %s" % module)
impl = FromFile(path)
compiled = nodejs_compile(impl.code, lang=impl.lang, file=impl.file)
if impl.lang == "less":
code = _style_template % dict(css=json.dumps(compiled.code))
deps = []
else:
code = compiled.code
deps = compiled.deps
sig = hashlib.sha256(code.encode('utf-8')).hexdigest()
resolved[module] = sig
deps_map = resolve_deps(deps, dirname(path))
if sig not in extra_modules:
extra_modules[sig] = True
modules.append((sig, code, deps_map))
else:
index = module + ("" if module.endswith("/") else "/") + "index"
if index not in known_modules:
raise RuntimeError("no such module: %s" % module)
return resolved
def resolve_deps(deps : list[str], root: str) -> dict[str, str]:
custom_modules = {model.module for model in custom_models.values()}
missing = set(deps) - known_modules - custom_modules
return resolve_modules(missing, root)
for model in custom_models.values():
compiled = custom_impls[model.full_name]
deps_map = resolve_deps(compiled.deps, model.path)
exports.append((model.name, model.module))
modules.append((model.module, compiled.code, deps_map))
# sort everything by module name
exports = sorted(exports, key=lambda spec: spec[1])
modules = sorted(modules, key=lambda spec: spec[0])
bare_modules = []
for i, (module, code, deps) in enumerate(modules):
for name, ref in deps.items():
code = code.replace("""require("%s")""" % name, """require("%s")""" % ref)
code = code.replace("""require('%s')""" % name, """require('%s')""" % ref)
bare_modules.append((module, code))
sep = ",\n"
rendered_exports = sep.join(_export_template % dict(name=name, module=module) for (name, module) in exports)
rendered_modules = sep.join(_module_template % dict(module=module, source=code) for (module, code) in bare_modules)
content = _plugin_template % dict(prelude=_plugin_prelude, exports=rendered_exports, modules=rendered_modules)
return _plugin_umd % dict(content=content)
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------
