Skip to content

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Proof of concept code for serving interactive matplotlib figures to the webbrowser
import serve_figure
import numpy as np
from numpy import ma
from matplotlib import pyplot as plt
n = 12
x = np.linspace(-1.5,1.5,n)
y = np.linspace(-1.5,1.5,n*2)
X,Y = np.meshgrid(x,y);
Qx = np.cos(Y) - np.cos(X)
Qz = np.sin(Y) + np.sin(X)
Qx = (Qx + 1.1)
Z = np.sqrt(X**2 + Y**2)/5;
Z = (Z - Z.min()) / (Z.max() - Z.min())
# The color array can include masked values:
Zm = ma.masked_where(np.fabs(Qz) < 0.5*np.amax(Qz), Z)
fig = plt.figure()
ax = fig.add_subplot(121)
ax.set_axis_bgcolor("#bdb76b")
ax.pcolormesh(Qx,Qz,Z, shading='gouraud')
ax.set_title('Without masked values')
ax = fig.add_subplot(122)
ax.set_axis_bgcolor("#bdb76b")
col = ax.pcolormesh(Qx,Qz,Zm,shading='gouraud')
ax.set_title('With masked values')
serve_figure.serve_figure(fig, port=8888)
import serve_figure
import numpy as np
from numpy import ma
from matplotlib import pyplot as plt
n = 30
x = np.linspace(-1.5,1.5,n)
fig = plt.figure()
ax = fig.add_subplot(111)
ax.set_axis_bgcolor("#bdb76b")
ax.plot(x, np.sin(x))
ax.set_title('Without masked values')
serve_figure.serve_figure(fig, port=8888)
function GUID () {
var S4 = function ()
{
return Math.floor(
Math.random() * 0x10000 /* 65536 */
).toString(16);
};
return (
S4() + S4() + "-" +
S4() + "-" +
S4() + "-" +
S4() + "-" +
S4() + S4() + S4()
);
};
window.onload = function() {
if (!("WebSocket" in window)) {
alert("WebSocket not supported");
return;
}
var message = document.getElementById("message");
control_ws = new WebSocket("ws://localhost:8888/event");
control_ws.onmessage = function (evt) {
var msg = JSON.parse(evt.data);
message.textContent = msg[0];
};
var canvas = document.getElementById("myCanvas");
var context = canvas.getContext("2d");
imageObj = new Image();
imageObj.onload = function() {
context.drawImage(imageObj, 0, 0);
};
var image_ws = new WebSocket("ws://localhost:8888/image");
image_ws.onopen = function() {image_ws.send(1)};
image_ws.onmessage = function (evt) {
imageObj.src = evt.data;
}
};
function mouse_event(event, name) {
control_ws.send(JSON.stringify({type: name, x: event.clientX, y: event.clientY, button: event.button}));
}
import json
import time
import datetime
import tornado.web
import tornado.ioloop
import tornado.websocket
import numpy as np
import matplotlib
matplotlib.use('Agg')
from matplotlib import _png
from matplotlib import backend_bases
import cStringIO
png_buffer = cStringIO.StringIO()
html = """
<html>
<head>
<script src="static/mpl.js"></script>
<body>
<canvas id="myCanvas" width="800" height="600"
onmousedown="mouse_event(event, 'button_press')"
onmouseup="mouse_event(event, 'button_release')"
onmousemove="mouse_event(event, 'motion_notify')">
</canvas>
<div id="message">MESSAGE</div>
</body>
</html>
"""
class IndexPage(tornado.web.RequestHandler):
def get(self):
self.write(html)
image_socket = None
def serve_figure(fig, port=8888):
# The panning and zooming is handled by the toolbar, (strange enough),
# so we need to create a dummy one.
class Toolbar(backend_bases.NavigationToolbar2):
def _init_toolbar(self):
self.message = ''
self.needs_draw = True
def set_message(self, message):
self.message = message
def dynamic_update(self):
if self.needs_draw is False:
Image.image_number += 1
self.needs_draw = True
toolbar = Toolbar(fig.canvas)
# Set pan mode -- it's the most interesting one
toolbar.pan()
def RateLimited(maxPerSecond):
"Based on http://stackoverflow.com/a/667706/1200039"
min_time = 1.0 / float(maxPerSecond)
def decorate(func):
# these are lists so we can modify them below
# sort of a poor-man's nonlocal keyword
timeout = [0.0]
pending = [False]
def rateLimitedFunction(*args,**kwargs):
# called with no pending calls: run function
# called with with pending call: do nothing
# called with no pending calls, but within the window of the last call: set timeout for pending call
curr_time = time.time()
if pending[0]:
return
else:
def ff():
timeout[0] = time.time() + min_time
ret = func(*args, **kwargs)
pending[0] = False
return ret
ioloop = tornado.ioloop.IOLoop.instance()
pending[0] = ioloop.add_timeout(datetime.timedelta(seconds=max(0, timeout[0] - curr_time)), ff)
return rateLimitedFunction
return decorate
class Image(tornado.websocket.WebSocketHandler):
last_buffer = None
image_number = 0
def open(self):
global image_socket
image_socket = self
self.init=True
def on_message(self, message):
self.refresh()
def close(self):
global image_socket
self.init = False
image_socket = None
@RateLimited(5)
def refresh(self):
if not self.init: return
if fig.canvas.toolbar.needs_draw:
fig.canvas.draw()
fig.canvas.toolbar.needs_draw = False
renderer = fig.canvas.get_renderer()
buffer = np.array(
np.frombuffer(renderer.buffer_rgba(0,0), dtype=np.uint32),
copy=True)
buffer = buffer.reshape((renderer.height, renderer.width))
last_buffer = self.last_buffer
if last_buffer is not None:
diff = buffer != last_buffer
if not np.any(diff):
output = np.zeros((1, 1))
else:
output = np.where(diff, buffer, 0)
else:
output = buffer
png_buffer.reset()
png_buffer.truncate()
#global_timer()
_png.write_png(output.tostring(),
output.shape[1], output.shape[0],
png_buffer)
#print global_timer
datauri = "data:image/png;base64,{0}".format(png_buffer.getvalue().encode("base64").replace("\n", ""))
self.write_message(datauri)
self.last_buffer = buffer
class Event(tornado.websocket.WebSocketHandler):
def open(self):
print "Opened Event connection"
def on_message(self, message):
global image_socket
message = json.loads(message)
type = message['type']
if type != 'poll':
x = int(message['x'])
y = int(message['y'])
y = fig.canvas.get_renderer().height - y
# Javascript button numbers and matplotlib button numbers are
# off by 1
button = int(message['button']) + 1
# The right mouse button pops up a context menu, which doesn't
# work very well, so use the middle mouse button instead
if button == 2:
button = 3
if type == 'button_press':
fig.canvas.button_press_event(x, y, button)
elif type == 'button_release':
fig.canvas.button_release_event(x, y, button)
elif type == 'motion_notify':
fig.canvas.motion_notify_event(x, y)
# The response is:
# [message (str), needs_draw (bool) ]
self.write_message(
json.dumps(
[fig.canvas.toolbar.message,
fig.canvas.toolbar.needs_draw]))
if fig.canvas.toolbar.needs_draw:
image_socket.refresh()
def on_close(self):
print "Event websocket closed"
application = tornado.web.Application([
(r"/", IndexPage),
(r"/image", Image),
(r"/event", Event),
(r'/static/(.*)', tornado.web.StaticFileHandler, {'path': '.'}),
])
application.listen(port)
tornado.ioloop.IOLoop.instance().start()
class Timer(object):
def __init__(self, name="", reset=False):
self.start = time.time()
self.name = name
self.reset = reset
def __call__(self, reset=None):
if reset is None:
reset = self.reset
old_time = self.start
new_time = time.time()
if reset:
self.start = new_time
return new_time - old_time
def __repr__(self):
return str(self.name)+" %s ms"%(int(self()*1000))
global_timer=Timer("Global timer", reset=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.