#!/usr/bin/env python
# Copyright (c) 2018, DIANA-HEP
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# * Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# * Neither the name of the copyright holder nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import errno
import getpass
import json
import os
import re
import socket
import sys
import threading
import time
import webbrowser
if sys.version_info[0] <= 2:
import SimpleHTTPServer
import SocketServer
from urllib2 import urlopen
from urlparse import urlparse
class BrokenPipeError(Exception): pass
else:
import http.server as SimpleHTTPServer
import socketserver as SocketServer
from urllib.request import urlopen
from urllib.parse import urlparse
unicode = str
__version__ = "1.0.14"
version = __version__
version_info = tuple(re.split(r"[-\.]", __version__))
class Canvas(object):
"""A general Vega canvas that can be viewed anywhere on the internet.
Args:
title (string or None): title for this canvas, can be changed, used as a file name for saved graphics.
initial (string, dict, or None; URL or JSON): first Vega graphic to display, defaults to "plot goes here".
host (string): host name to bind to, default is "0.0.0.0" for any address.
port (integer): port to bind to, default is 0 for any open port.
verbose (boolean): if True (default), print messages when web browsers connect or disconnect.
vega (string or None): Vega version to request from cdn.jsdelivr.net or None to use an standalone copy.
vegalite (string or None): Vega-Lite version to request from cdn.jsdelivr.net or None to use an standalone copy.
vegaembed (string or None): Vega-Embed version to request from cdn.jsdelivr.net or None to use an standalone copy.
Attributes:
connection (dict): web browser URL and possibly terminal command (for TunnelCanvas).
title (string): current title.
spec (string or dict; URL or JSON): current Vega graphic.
httpd (socketserver.ThreadingTCPServer): web server object.
ip (string): IP address of server as seen from outside ("localhost" for LocalCanvas and TunnelCanvas).
host (string): actual host used by web server.
port (string): actual port used by web server.
thread (threading.Thread): thread in which the web server is running.
connected (list of strings): currently connected web browser clients.
closed (boolean): True if the server has shut down, False otherwise.
"""
def __init__(self, title=None, initial=None, host="0.0.0.0", port=0, verbose=True, vega="5.4.0", vegalite="3.3.0", vegaembed="4.2.0"):
self._lock = threading.Lock()
if title is None:
self._title = "VegaScope"
else:
self._title = title
if initial is None:
self.spec = Canvas._default
else:
self.spec = initial
self._action = None
self._actionevent = threading.Event()
canvas = self
if vega == "":
vega = None
if vegalite == "":
vegalite = None
if vegaembed == "":
vegaembed = None
class FakeFile(object):
@property
def closed(self):
return True
def close(self):
pass
def flush(self):
pass
class HTTPHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
def do_HEAD(self):
pass
def do_GET(self):
if self.path == "/":
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
vegalibs = """
""".format("vega.min.js" if vega is None else "https://cdn.jsdelivr.net/npm/vega@" + vega,
"vega-lite.min.js" if vegalite is None else "https://cdn.jsdelivr.net/npm/vega-lite@" + vegalite,
"vega-embed.min.js" if vegaembed is None else "https://cdn.jsdelivr.net/npm/vega-embed@" + vegaembed)
page = canvas._template.replace("VEGAVIEW", "#vegaview").replace("VEGALIBS", vegalibs).replace("TITLE", canvas._title).replace("SPEC", canvas._spec).encode("utf-8")
self.wfile.write(page)
elif self.path == "/update":
self.send_response(200)
self.send_header("Content-type", "text/event-stream")
self.end_headers()
client = self.client_address[0]
canvas._connected.add(client)
if canvas.verbose:
sys.stdout.write("{0} connected\n".format(client))
sys.stdout.flush()
title = canvas._title
spec = canvas._spec
try:
while not self.wfile.closed:
time.sleep(0.1)
with canvas._lock:
if canvas._spec is None:
break
elif spec != canvas._spec or title != canvas._title or canvas._action is not None:
spec = canvas._spec
title = canvas._title
self.wfile.write(b"data: " + json.dumps({"title": title, "spec": json.loads(spec), "action": canvas._action}).encode("utf-8") + b"\n\n")
canvas._action = None
canvas._actionevent.set()
else:
self.wfile.write(":\n\n".encode("utf-8"))
self.wfile.flush()
except socket.error as err:
if isinstance(err, BrokenPipeError) or err[0] == errno.EPIPE:
self.wfile = FakeFile()
else:
raise
finally:
canvas._connected.discard(client)
if canvas.verbose:
sys.stdout.write("{0} disconnected\n".format(client))
sys.stdout.flush()
elif self.path == "/vega.min.js":
self.send_response(200)
self.send_header("Content-type", "application/javascript")
self.end_headers()
self.wfile.write(canvas._vega)
elif self.path == "/vega-lite.min.js":
self.send_response(200)
self.send_header("Content-type", "application/javascript")
self.end_headers()
self.wfile.write(canvas._vegalite)
elif self.path == "/vega-embed.min.js":
self.send_response(200)
self.send_header("Content-type", "application/javascript")
self.end_headers()
self.wfile.write(canvas._vegaembed)
def log_request(self, code="-", size="-"):
pass
def log_error(self, format, *args):
pass
def log_message(self, format, *args):
pass
self._httpd = SocketServer.ThreadingTCPServer((host, port), HTTPHandler)
self._httpd.daemon_threads = True
self._host, self._port = self._httpd.server_address
self._thread = threading.Thread(name=self._title, target=self._httpd.serve_forever)
self._thread.daemon = True
self._thread.start()
self.verbose = verbose
self._connected = set()
self._launch()
def _launch(self):
if self.verbose:
self.how()
@property
def title(self):
return self._title
@title.setter
def title(self, value):
self._specify(value, None, None)
@property
def spec(self):
return self._spec
@spec.setter
def spec(self, value):
self._specify(None, value, None)
def __call__(self, spec):
"""Update the Vega graphic to spec (string or dict; URL or JSON).
"""
self._specify(None, spec, None)
def png(self, spec, title=None):
"""Update the Vega graphic to spec, optionally set a title, and make connected web clients save the image as PNG.
Note that pop-up blockers won't let remote servers save files on your computer without your permission. You will probably have to respond to the pop-up blocker's notice to enable pop-ups and try again (once per host/port combination).
This method blocks until the PNG request has been sent, so it can be called in a loop to make lots of files.
Args:
spec (string or dict; URL or JSON): new Vega graphic.
title (string or None): new title.
"""
self._specify(title, spec, "png")
self._actionevent.wait()
def svg(self, spec, title=None):
"""Update the Vega graphic to spec, optionally set a title, and make connected web clients save the image as SVG.
Note that pop-up blockers won't let remote servers save files on your computer without your permission. You will probably have to respond to the pop-up blocker's notice to enable pop-ups and try again (once per host/port combination).
This method blocks until the SVG request has been sent, so it can be called in a loop to make lots of files.
SVG files can be converted into clean, rescalable PDFs.
Args:
spec (string or dict; URL or JSON): new Vega graphic.
title (string or None): new title.
"""
self._specify(title, spec, "svg")
self._actionevent.wait()
def _specify(self, title, spec, action):
with self._lock:
if title is not None:
if isinstance(title, (unicode, str)):
self._title = title
else:
raise TypeError("title must be a string")
if spec is not None:
if isinstance(spec, bytes):
spec = spec.decode("utf-8")
if isinstance(spec, (unicode, str)):
p = urlparse(spec)
if p.scheme != "":
self._spec = json.dumps(spec) # spec is a URL; wrap it with quotes for JSON
else:
self._spec = json.dumps(json.loads(spec), allow_nan=False) # not a URL; ensure that it's JSON and a one-liner
else:
# PdVega
if spec.__class__.__module__.startswith("pdvega") and hasattr(spec, "spec"):
spec = spec.spec
# Altair
elif hasattr(spec, "to_json") and callable(spec.to_json):
spec = json.loads(spec.to_json())
self._spec = json.dumps(spec, allow_nan=False) # spec is an object; encode it as JSON
if action is not None:
self._action = action
@property
def httpd(self):
return self._httpd
@property
def host(self):
return self._host
@property
def port(self):
return self._port
@property
def thread(self):
return self._thread
@property
def connected(self):
return sorted(self._connected)
def close(self):
"""Shut down the web server, disconnecting all client browsers.
"""
self._spec = None
if hasattr(self, "_httpd"):
self._httpd.shutdown()
self._httpd.server_close()
@property
def closed(self):
return self._spec is None
def __del__(self):
if not self.closed:
self.close()
def __enter__(self, *args, **kwds):
return self
def __exit__(self, *args, **kwds):
if not self.closed:
self.close()
@property
def connection(self):
return {"browser": "http://{ip}:{port}".format(ip=self.ip, port=self._port)}
def how(self):
"""Print instructions for connecting to the web server.
Called automatically by the constructor unless verbose=False.
"""
connection = self.connection
if "terminal" in connection:
sys.stdout.write("Type into terminal: " + connection["terminal"] + "\n")
sys.stdout.write("Point web browser at: " + connection["browser"] + "\n")
sys.stdout.flush()
@property
def ip(self):
return urlopen("https://v4.ident.me").read().decode("ascii").strip()
class LocalCanvas(Canvas):
"""A Vega canvas that can only be viewed by the machine on which it is running.
"""
__doc__ += "\n".join(Canvas.__doc__.split("\n")[1:])
def __init__(self, title=None, initial=None, port=0, verbose=True, newtab=True, vega="5.4.0", vegalite="3.3.0", vegaembed="4.2.0"):
self._newtab = newtab
super(LocalCanvas, self).__init__(title=title, initial=initial, host="localhost", port=port, verbose=verbose, vega=vega, vegalite=vegalite, vegaembed=vegaembed)
def _launch(self):
if self.verbose and not self._newtab:
self.how()
if self._newtab:
webbrowser.open_new_tab(self.connection["browser"])
@property
def ip(self):
return "localhost"
class TunnelCanvas(Canvas):
"""A Vega canvas that can only be viewed by the machine on which it is running or through an ssh tunnel.
"""
__doc__ += "\n".join(Canvas.__doc__.split("\n")[1:])
def __init__(self, title=None, initial=None, port=0, verbose=True, vega="5.4.0", vegalite="3.3.0", vegaembed="4.2.0"):
super(TunnelCanvas, self).__init__(title=title, initial=initial, host="localhost", port=port, verbose=verbose, vega=vega, vegalite=vegalite, vegaembed=vegaembed)
@property
def connection(self):
return {"terminal": "ssh -L {port}:localhost:{port} {user}@{ip}".format(port=self._port, user=getpass.getuser(), ip=self.ip),
"browser": "http://localhost:{port}".format(port=self._port)}
# This is the global canvas instance used by entrypoint-based renderers
_entrypoint_renderer_canvas = None
def _vegalite_renderer_entry_point(spec, embed_options=None):
import altair
global _entrypoint_renderer_canvas
if embed_options is not None:
import warnings
warnings.warn("embed_options is not yet supported & will be ignored")
if _entrypoint_renderer_canvas is None:
_entrypoint_renderer_canvas = LocalCanvas(vega=altair.v3.VEGA_VERSION, vegalite=altair.v3.VEGALITE_VERSION, vegaembed=altair.v3.VEGAEMBED_VERSION)
_entrypoint_renderer_canvas(spec)
browser = _entrypoint_renderer_canvas.connection['browser']
return {'text/plain': 'Rendered at {0}'.format(browser)}
Canvas._default = {
"$schema": "https://vega.github.io/schema/vega-lite/v2.json",
"data": {"values": [{"x": "plot goes here"}]},
"mark": "text",
"encoding": {"text": {"field": "x"}},
"config": {"text": {"baseline": "middle", "align": "center"}}
}
Canvas._template = u"""
VEGALIBS
TITLE