Suppress window warnings in node

altair_saver/_utils.py

@@ -6,7 +6,7 @@
 import subprocess
 import sys
 import tempfile
-from typing import IO, Iterator, List, Optional, Union
+from typing import Callable, IO, Iterator, List, Optional, Union
 
 import altair as alt
 
@@ -168,31 +168,53 @@ def extract_format(fp: Union[IO, str]) -> str:
 
 
 def check_output_with_stderr(
-    cmd: Union[str, List[str]], shell: bool = False, input: Optional[bytes] = None
+    cmd: Union[str, List[str]],
+    shell: bool = False,
+    input: Optional[bytes] = None,
+    stderr_filter: Callable[[str], bool] = None,
 ) -> bytes:
     """Run a command in a subprocess, printing stderr to sys.stderr.
 
-    Arguments are passed directly to subprocess.run().
+    This function exists because normally, stderr from subprocess in the notebook
+    is printed to the terminal rather than to the notebook itself.
 
-    This is important because subprocess stderr in notebooks is printed to the
-    terminal rather than the notebook.
+    Parameters
+    ----------
+    cmd, shell, input :
+        Arguments are passed directly to `subprocess.run()`.
+    stderr_filter : function(str)->bool (optional)
+        If provided, this function is used to filter stderr lines from display.
+
+    Returns
+    -------
+    result : bytes
+        The stdout from the command
+
+    Raises
+    ------
+    subprocess.CalledProcessError : if the called process returns a non-zero exit code.
     """
     try:
         ps = subprocess.run(
             cmd,
             shell=shell,
+            input=input,
             check=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            input=input,
         )
     except subprocess.CalledProcessError as err:
-        if err.stderr:
-            sys.stderr.write(err.stderr.decode())
-            sys.stderr.flush()
+        stderr = err.stderr
         raise
     else:
-        if ps.stderr:
-            sys.stderr.write(ps.stderr.decode())
-            sys.stderr.flush()
+        stderr = ps.stderr
         return ps.stdout
+    finally:
+        s = stderr.decode()
+        if stderr_filter:
+            s = "\n".join(filter(stderr_filter, s.splitlines()))
+        if s:
+            if not s.endswith("\n"):
+                s += "\n"
+            sys.stderr.write(s)
+            sys.stderr.flush()

altair_saver/savers/_node.py

@@ -1,7 +1,7 @@
 import functools
 import json
 import shutil
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from altair_saver.types import JSONDict, MimebundleContent
 from altair_saver._utils import check_output_with_stderr
@@ -33,34 +33,54 @@ def exec_path(name: str) -> str:
     raise ExecutableNotFound(name)
 
 
-def vl2vg(spec: JSONDict) -> JSONDict:
+def vl2vg(
+    spec: JSONDict, stderr_filter: Optional[Callable[[str], bool]] = None
+) -> JSONDict:
     """Compile a Vega-Lite spec into a Vega spec."""
     vl2vg = exec_path("vl2vg")
     vl_json = json.dumps(spec).encode()
-    vg_json = check_output_with_stderr([vl2vg], input=vl_json)
+    vg_json = check_output_with_stderr(
+        [vl2vg], input=vl_json, stderr_filter=stderr_filter
+    )
     return json.loads(vg_json)
 
 
-def vg2png(spec: JSONDict, vega_cli_options: Optional[List[str]] = None) -> bytes:
+def vg2png(
+    spec: JSONDict,
+    vega_cli_options: Optional[List[str]] = None,
+    stderr_filter: Optional[Callable[[str], bool]] = None,
+) -> bytes:
     """Generate a PNG image from a Vega spec."""
     vg2png = exec_path("vg2png")
     vg_json = json.dumps(spec).encode()
-    return check_output_with_stderr([vg2png, *(vega_cli_options or [])], input=vg_json)
+    return check_output_with_stderr(
+        [vg2png, *(vega_cli_options or [])], input=vg_json, stderr_filter=stderr_filter
+    )
 
 
-def vg2pdf(spec: JSONDict, vega_cli_options: Optional[List[str]] = None) -> bytes:
+def vg2pdf(
+    spec: JSONDict,
+    vega_cli_options: Optional[List[str]] = None,
+    stderr_filter: Optional[Callable[[str], bool]] = None,
+) -> bytes:
     """Generate a PDF image from a Vega spec."""
     vg2pdf = exec_path("vg2pdf")
     vg_json = json.dumps(spec).encode()
-    return check_output_with_stderr([vg2pdf, *(vega_cli_options or [])], input=vg_json)
+    return check_output_with_stderr(
+        [vg2pdf, *(vega_cli_options or [])], input=vg_json, stderr_filter=stderr_filter
+    )
 
 
-def vg2svg(spec: JSONDict, vega_cli_options: Optional[List[str]] = None) -> str:
+def vg2svg(
+    spec: JSONDict,
+    vega_cli_options: Optional[List[str]] = None,
+    stderr_filter: Optional[Callable[[str], bool]] = None,
+) -> str:
     """Generate an SVG image from a Vega spec."""
     vg2svg = exec_path("vg2svg")
     vg_json = json.dumps(spec).encode()
     return check_output_with_stderr(
-        [vg2svg, *(vega_cli_options or [])], input=vg_json
+        [vg2svg, *(vega_cli_options or [])], input=vg_json, stderr_filter=stderr_filter
     ).decode()
 
 
@@ -82,6 +102,12 @@ def __init__(
         self._vega_cli_options = vega_cli_options or []
         super().__init__(spec=spec, mode=mode, **kwargs)
 
+    _stderr_ignore = ["WARN Can not resolve event source: window"]
+
+    @classmethod
+    def _stderr_filter(cls, line: str) -> bool:
+        return line not in cls._stderr_ignore
+
     @classmethod
     def enabled(cls) -> bool:
         try:
@@ -96,15 +122,27 @@ def _serialize(self, fmt: str, content_type: str) -> MimebundleContent:
         spec = self._spec
 
         if self._mode == "vega-lite":
-            spec = vl2vg(spec)
+            spec = vl2vg(spec, stderr_filter=self._stderr_filter)
 
         if fmt == "vega":
             return spec
         elif fmt == "png":
-            return vg2png(spec, vega_cli_options=self._vega_cli_options)
+            return vg2png(
+                spec,
+                vega_cli_options=self._vega_cli_options,
+                stderr_filter=self._stderr_filter,
+            )
         elif fmt == "svg":
-            return vg2svg(spec, vega_cli_options=self._vega_cli_options)
+            return vg2svg(
+                spec,
+                vega_cli_options=self._vega_cli_options,
+                stderr_filter=self._stderr_filter,
+            )
         elif fmt == "pdf":
-            return vg2pdf(spec, vega_cli_options=self._vega_cli_options)
+            return vg2pdf(
+                spec,
+                vega_cli_options=self._vega_cli_options,
+                stderr_filter=self._stderr_filter,
+            )
         else:
             raise ValueError(f"Unrecognized format: {fmt!r}")

altair_saver/savers/tests/test_node.py

@@ -6,11 +6,13 @@
 from PIL import Image
 from PyPDF2 import PdfFileReader
 import pytest
+from _pytest.capture import SysCapture
 from _pytest.monkeypatch import MonkeyPatch
 
 from altair_saver import NodeSaver
 from altair_saver._utils import fmt_to_mimetype
 from altair_saver.savers import _node
+from altair_saver.types import JSONDict
 
 
 def get_testcases() -> Iterator[Tuple[str, Dict[str, Any]]]:
@@ -31,6 +33,19 @@ def get_testcases() -> Iterator[Tuple[str, Dict[str, Any]]]:
         yield case, {"vega-lite": vl, "vega": vg, "svg": svg, "png": png, "pdf": pdf}
 
 
+@pytest.fixture
+def interactive_spec() -> JSONDict:
+    return {
+        "data": {"values": [{"x": 1, "y": 1}]},
+        "mark": "point",
+        "encoding": {
+            "x": {"field": "x", "type": "quantitative"},
+            "y": {"field": "y", "type": "quantitative"},
+        },
+        "selection": {"zoon": {"type": "interval", "bind": "scales"}},
+    }
+
+
 def get_modes_and_formats() -> Iterator[Tuple[str, str]]:
     for mode in ["vega", "vega-lite"]:
         for fmt in NodeSaver.valid_formats[mode]:
@@ -89,3 +104,25 @@ def exec_path(name: str) -> str:
 
     monkeypatch.setattr(_node, "exec_path", exec_path)
     assert NodeSaver.enabled() is enabled
+
+
+@pytest.mark.parametrize("suppress_warnings", [True, False])
+def test_stderr_suppression(
+    interactive_spec: JSONDict,
+    suppress_warnings: bool,
+    monkeypatch: MonkeyPatch,
+    capsys: SysCapture,
+) -> None:
+    message = NodeSaver._stderr_ignore[0]
+
+    # Window resolve warnings are emitted by the vega CLI when an interactive chart
+    # is saved, and are suppressed by default.
+    if not suppress_warnings:
+        monkeypatch.setattr(NodeSaver, "_stderr_ignore", [])
+
+    NodeSaver(interactive_spec).save(fmt="png")
+    captured = capsys.readouterr()
+    if suppress_warnings:
+        assert message not in captured.err
+    else:
+        assert message in captured.err

altair_saver/tests/test_utils.py

@@ -6,7 +6,7 @@
 from typing import Any
 
 import pytest
-from _pytest.capture import SysCaptureBinary
+from _pytest.capture import SysCapture
 
 from altair_saver.types import JSONDict
 from altair_saver._utils import (
@@ -133,20 +133,27 @@ def test_infer_mode_from_spec(mode: str, spec: JSONDict) -> None:
     assert infer_mode_from_spec(spec) == mode
 
 
-def test_check_output_with_stderr(capsysbinary: SysCaptureBinary) -> None:
-    output = check_output_with_stderr(
-        r'>&2 echo "the error" && echo "the output"', shell=True
-    )
-    assert output == b"the output\n"
-    captured = capsysbinary.readouterr()
-    assert captured.out == b""
-    assert captured.err == b"the error\n"
+@pytest.mark.parametrize("cmd_error", [True, False])
+@pytest.mark.parametrize("use_filter", [True, False])
+def test_check_output_with_stderr(
+    capsys: SysCapture, use_filter: bool, cmd_error: bool
+) -> None:
+    cmd = r'>&2 echo "first error\nsecond error" && echo "the output"'
+    stderr_filter = None if not use_filter else lambda line: line.startswith("second")
 
+    if cmd_error:
+        cmd += r" && exit 1"
+        with pytest.raises(subprocess.CalledProcessError) as err:
+            check_output_with_stderr(cmd, shell=True, stderr_filter=stderr_filter)
+        assert err.value.stderr == b"first error\nsecond error\n"
+    else:
+        output = check_output_with_stderr(cmd, shell=True, stderr_filter=stderr_filter)
+        assert output == b"the output\n"
+
+    captured = capsys.readouterr()
+    assert captured.out == ""
 
-def test_check_output_with_stderr_exit_1(capsysbinary: SysCaptureBinary) -> None:
-    with pytest.raises(subprocess.CalledProcessError) as err:
-        check_output_with_stderr(r'>&2 echo "the error" && exit 1', shell=True)
-    assert err.value.stderr == b"the error\n"
-    captured = capsysbinary.readouterr()
-    assert captured.out == b""
-    assert captured.err == b"the error\n"
+    if use_filter:
+        assert captured.err == "second error\n"
+    else:
+        assert captured.err == "first error\nsecond error\n"