Andrew/api factor

src/py/CHANGELOG.txt

@@ -1,3 +1,4 @@
+- Add new api functions start/stop_sync_server
 v1.0.0
 - Add warning if using incompatible Plotly version
 

src/py/kaleido/__init__.py

@@ -4,22 +4,42 @@
 Please see the README.md for more information and a quickstart.
 """
 
-import asyncio
-import queue
-from threading import Thread
+from __future__ import annotations
 
 from choreographer.cli import get_chrome, get_chrome_sync
 
+from . import _sync_server
 from ._page_generator import PageGenerator
 from .kaleido import Kaleido
 
+_global_server = _sync_server.GlobalKaleidoServer()
+
+
+def start_sync_server(*args, **kwargs):
+    """
+    Start a kaleido server which will process all sync generation requests.
+
+    Only one server can be started at a time.
+
+    This wrapper function takes the exact same arguments as kaleido.Kaleido().
+    """
+    _global_server.open(*args, **kwargs)
+
+
+def stop_sync_server():
+    """Stop the kaleido server. It can be restarted."""
+    _global_server.close()
+
+
 __all__ = [
     "Kaleido",
     "PageGenerator",
     "calc_fig",
     "calc_fig_sync",
     "get_chrome",
     "get_chrome_sync",
+    "start_sync_server",
+    "stop_sync_server",
     "write_fig",
     "write_fig_from_object",
     "write_fig_from_object_sync",
@@ -120,36 +140,25 @@ async def write_fig_from_object(
         )
 
 
-def _async_thread_run(func, args, kwargs):
-    q = queue.Queue(maxsize=1)
-
-    def run(*args, **kwargs):
-        # func is a closure
-        try:
-            q.put(asyncio.run(func(*args, **kwargs)))
-        except BaseException as e:  # noqa: BLE001
-            q.put(e)
-
-    t = Thread(target=run, args=args, kwargs=kwargs)
-    t.start()
-    t.join()
-    res = q.get()
-    if isinstance(res, BaseException):
-        raise res
-    else:
-        return res
-
-
 def calc_fig_sync(*args, **kwargs):
     """Call `calc_fig` but blocking."""
-    return _async_thread_run(calc_fig, args=args, kwargs=kwargs)
+    if _global_server.is_running():
+        return _global_server.call_function("calc_fig", *args, **kwargs)
+    else:
+        return _sync_server.oneshot_async_run(calc_fig, args=args, kwargs=kwargs)
 
 
 def write_fig_sync(*args, **kwargs):
     """Call `write_fig` but blocking."""
-    _async_thread_run(write_fig, args=args, kwargs=kwargs)
+    if _global_server.is_running():
+        _global_server.call_function("write_fig", *args, **kwargs)
+    else:
+        _sync_server.oneshot_async_run(write_fig, args=args, kwargs=kwargs)
 
 
 def write_fig_from_object_sync(*args, **kwargs):
     """Call `write_fig_from_object` but blocking."""
-    _async_thread_run(write_fig_from_object, args=args, kwargs=kwargs)
+    if _global_server.is_running():
+        _global_server.call_function("write_fig_from_object", *args, **kwargs)
+    else:
+        _sync_server.oneshot_async_run(write_fig_from_object, args=args, kwargs=kwargs)

src/py/kaleido/_fig_tools.py

@@ -1,20 +1,36 @@
+from __future__ import annotations
+
 import glob
 import re
 from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, TypedDict
 
 import logistro
 
+if TYPE_CHECKING:
+    from typing_extensions import TypeGuard
+
 _logger = logistro.getLogger(__name__)
 
 # constants
 DEFAULT_EXT = "png"
 DEFAULT_SCALE = 1
 DEFAULT_WIDTH = 700
 DEFAULT_HEIGHT = 500
-SUPPORTED_FORMATS = ("png", "jpg", "jpeg", "webp", "svg", "json", "pdf")  # pdf and eps
+SUPPORTED_FORMATS = ("png", "jpg", "jpeg", "webp", "svg", "json", "pdf")
+FormatString = Literal["png", "jpg", "jpeg", "webp", "svg", "json", "pdf"]
+
+
+def _assert_format(ext: str) -> TypeGuard[FormatString]:
+    if ext not in SUPPORTED_FORMATS:
+        raise ValueError(f"File format {ext} is not supported.")
+    return True
+
+
+Figurish = Any  # Be nice to make it more specific, dictionary or something
 
 
-def _is_figurish(o):
+def _is_figurish(o) -> TypeGuard[Figurish]:
     valid = hasattr(o, "to_dict") or (isinstance(o, dict) and "data" in o)
     if not valid:
         _logger.debug(
@@ -43,32 +59,49 @@ def _get_figure_dimensions(layout, width, height):
 
 
 def _get_format(extension):
-    # Normalize format
     original_format = extension
     extension = extension.lower()
     if extension == "jpg":
-        extension = "jpeg"
+        return "jpeg"
 
     if extension not in SUPPORTED_FORMATS:
-        supported_formats_str = repr(list(SUPPORTED_FORMATS))
         raise ValueError(
             f"Invalid format '{original_format}'.\n"
-            f"    Supported formats: {supported_formats_str}",
+            f"    Supported formats: {SUPPORTED_FORMATS!s}",
         )
     return extension
 
 
-def to_spec(figure, layout_opts):
+# Input of to_spec
+class LayoutOpts(TypedDict, total=False):
+    format: FormatString | None
+    scale: int | float
+    height: int | float
+    width: int | float
+
+
+# Output of to_spec
+class Spec(TypedDict):
+    format: FormatString
+    width: int | float
+    height: int | float
+    scale: int | float
+    data: Figurish
+
+
+def to_spec(figure, layout_opts: LayoutOpts) -> Spec:
     # Get figure layout
     layout = figure.get("layout", {})
 
     for k, v in layout_opts.items():
         if k == "format":
             if v is not None and not isinstance(v, (str)):
-                raise TypeError(f"{v} must be string or None")
+                raise TypeError(
+                    f"{k} must be one of {SUPPORTED_FORMATS!s} or None, not {v}.",
+                )
         elif k in ("scale", "height", "width"):
             if v is not None and not isinstance(v, (float, int)):
-                raise TypeError(f"{v} must be numeric or None")
+                raise TypeError(f"{k} must be numeric or None, not {v}.")
         else:
             raise AttributeError(f"Unknown key in layout options, {k}")
 
@@ -90,7 +123,7 @@ def to_spec(figure, layout_opts):
     }
 
 
-def _next_filename(path, prefix, ext):
+def _next_filename(path, prefix, ext) -> str:
     default = 1 if (path / f"{prefix}.{ext}").exists() else 0
     re_number = re.compile(
         r"^" + re.escape(prefix) + r"\-(\d+)\." + re.escape(ext) + r"$",
@@ -106,7 +139,11 @@ def _next_filename(path, prefix, ext):
     return f"{prefix}.{ext}" if n == 1 else f"{prefix}-{n}.{ext}"
 
 
-def build_fig_spec(fig, path, opts):  #  noqa: C901
+def build_fig_spec(  #  noqa: C901, PLR0912
+    fig: Figurish,
+    path: Path | str | None,
+    opts: LayoutOpts | None,
+) -> tuple[Spec, Path]:
     if not opts:
         opts = {}
 
@@ -122,23 +159,27 @@ def build_fig_spec(fig, path, opts):  #  noqa: C901
         raise TypeError("Path should be a string or `pathlib.Path` object (or None)")
 
     if path and path.suffix and not opts.get("format"):
-        opts["format"] = path.suffix.lstrip(".")
+        ext = path.suffix.lstrip(".")
+        if _assert_format(ext):  # not strict necessary if but helps typeguard
+            opts["format"] = ext
 
     spec = to_spec(fig, opts)
 
     ext = spec["format"]
-    full_path = None
+
+    full_path: Path | None = None
+    directory: Path
     if not path:
-        directory = Path()
+        directory = Path()  # use current Path
     elif path and (not path.suffix or path.is_dir()):
         if not path.is_dir():
-            raise ValueError(f"Directories will not be created for you: {path}")
+            raise ValueError(f"Directory {path} not found. Please create it.")
         directory = path
     else:
         full_path = path
         if not full_path.parent.is_dir():
             raise RuntimeError(
-                f"Cannot reach path {path}. Are all directories created?",
+                f"Cannot reach path {path.parent}. Are all directories created?",
             )
     if not full_path:
         _logger.debug("Looking for title")

src/py/kaleido/_mocker.py

@@ -9,6 +9,7 @@
 from pathlib import Path
 from pprint import pp
 from random import sample
+from typing import TypedDict
 
 import logistro
 import orjson
@@ -39,8 +40,14 @@ def _get_jsons_in_paths(path: str | Path) -> list[Path]:
         raise TypeError("--input must be file or directory")
 
 
+class Param(TypedDict):
+    name: str
+    opts: dict[str, int | float]
+
+
 def _load_figures_from_paths(paths: list[Path]):
     # Set json
+    params: list[Param]
     for path in paths:
         if path.is_file():
             with path.open(encoding="utf-8") as file:
@@ -80,7 +87,10 @@ def _load_figures_from_paths(paths: list[Path]):
                                 for f in formats:
                                     params.append(
                                         {
-                                            "name": f"{path.stem}-{w}x{h}X{s}.{f}",
+                                            "name": (
+                                                f"{path.stem!s}-{w!s}"
+                                                f"x{h!s}X{s!s}.{f!s}"
+                                            ),
                                             "opts": {
                                                 "scale": s,
                                                 "width": w,

src/py/kaleido/_page_generator.py

@@ -60,7 +60,8 @@ def __init__(self, *, plotly=None, mathjax=None, others=None, force_cdn=False):
         elif not plotly:
             try:
                 # ruff: noqa: PLC0415
-                import plotly as pltly  # type: ignore [import-not-found]
+                # is this the best way to do this? can't we use importlib?
+                import plotly as pltly  # type: ignore[import-untyped]
 
                 plotly_path = (
                     Path(pltly.__file__).parent / "package_data" / "plotly.min.js"