Source code for matplotlib.backends.backend_nbagg

"""Interactive figures in the IPython notebook"""
# Note: There is a notebook in
# lib/matplotlib/backends/web_backend/nbagg_uat.ipynb to help verify
# that changes made maintain expected behaviour.

from base64 import b64encode
import io
import json
import pathlib
import uuid

from IPython.display import display, Javascript, HTML
try:
    # Jupyter/IPython 4.x or later
    from ipykernel.comm import Comm
except ImportError:
    # Jupyter/IPython 3.x or earlier
    from IPython.kernel.comm import Comm

from matplotlib import cbook, is_interactive
from matplotlib._pylab_helpers import Gcf
from matplotlib.backend_bases import _Backend, NavigationToolbar2
from matplotlib.backends.backend_webagg_core import (
    FigureCanvasWebAggCore, FigureManagerWebAgg, NavigationToolbar2WebAgg,
    TimerTornado)


[docs]def connection_info(): """ Return a string showing the figure and connection status for the backend. This is intended as a diagnostic tool, and not for general use. """ result = [ '{fig} - {socket}'.format( fig=(manager.canvas.figure.get_label() or "Figure {}".format(manager.num)), socket=manager.web_sockets) for manager in Gcf.get_all_fig_managers() ] if not is_interactive(): result.append(f'Figures pending show: {len(Gcf.figs)}') return '\n'.join(result)
# Note: Version 3.2 and 4.x icons # http://fontawesome.io/3.2.1/icons/ # http://fontawesome.io/ # the `fa fa-xxx` part targets font-awesome 4, (IPython 3.x) # the icon-xxx targets font awesome 3.21 (IPython 2.x) _FONT_AWESOME_CLASSES = { 'home': 'fa fa-home icon-home', 'back': 'fa fa-arrow-left icon-arrow-left', 'forward': 'fa fa-arrow-right icon-arrow-right', 'zoom_to_rect': 'fa fa-square-o icon-check-empty', 'move': 'fa fa-arrows icon-move', 'download': 'fa fa-floppy-o icon-save', None: None }
[docs]class FigureManagerNbAgg(FigureManagerWebAgg): ToolbarCls = NavigationIPy def __init__(self, canvas, num): self._shown = False FigureManagerWebAgg.__init__(self, canvas, num)
[docs] def display_js(self): # XXX How to do this just once? It has to deal with multiple # browser instances using the same kernel (require.js - but the # file isn't static?). display(Javascript(FigureManagerNbAgg.get_javascript()))
[docs] def show(self): if not self._shown: self.display_js() self._create_comm() else: self.canvas.draw_idle() self._shown = True
[docs] def reshow(self): """ A special method to re-show the figure in the notebook. """ self._shown = False self.show()
@property def connected(self): return bool(self.web_sockets)
[docs] @classmethod def get_javascript(cls, stream=None): if stream is None: output = io.StringIO() else: output = stream super().get_javascript(stream=output) output.write((pathlib.Path(__file__).parent / "web_backend/js/nbagg_mpl.js") .read_text(encoding="utf-8")) if stream is None: return output.getvalue()
def _create_comm(self): comm = CommSocket(self) self.add_web_socket(comm) return comm
[docs] def destroy(self): self._send_event('close') # need to copy comms as callbacks will modify this list for comm in list(self.web_sockets): comm.on_close() self.clearup_closed()
[docs] def clearup_closed(self): """Clear up any closed Comms.""" self.web_sockets = {socket for socket in self.web_sockets if socket.is_open()} if len(self.web_sockets) == 0: self.canvas.close_event()
[docs] def remove_comm(self, comm_id): self.web_sockets = {socket for socket in self.web_sockets if socket.comm.comm_id != comm_id}
[docs]class FigureCanvasNbAgg(FigureCanvasWebAggCore): _timer_cls = TimerTornado
[docs]class CommSocket: """ Manages the Comm connection between IPython and the browser (client). Comms are 2 way, with the CommSocket being able to publish a message via the send_json method, and handle a message with on_message. On the JS side figure.send_message and figure.ws.onmessage do the sending and receiving respectively. """ def __init__(self, manager): self.supports_binary = None self.manager = manager self.uuid = str(uuid.uuid4()) # Publish an output area with a unique ID. The javascript can then # hook into this area. display(HTML("<div id=%r></div>" % self.uuid)) try: self.comm = Comm('matplotlib', data={'id': self.uuid}) except AttributeError as err: raise RuntimeError('Unable to create an IPython notebook Comm ' 'instance. Are you in the IPython ' 'notebook?') from err self.comm.on_msg(self.on_message) manager = self.manager self._ext_close = False def _on_close(close_message): self._ext_close = True manager.remove_comm(close_message['content']['comm_id']) manager.clearup_closed() self.comm.on_close(_on_close)
[docs] def is_open(self): return not (self._ext_close or self.comm._closed)
[docs] def on_close(self): # When the socket is closed, deregister the websocket with # the FigureManager. if self.is_open(): try: self.comm.close() except KeyError: # apparently already cleaned it up? pass
[docs] def send_json(self, content): self.comm.send({'data': json.dumps(content)})
[docs] def send_binary(self, blob): # The comm is ascii, so we always send the image in base64 # encoded data URL form. data = b64encode(blob).decode('ascii') data_uri = "data:image/png;base64,{0}".format(data) self.comm.send({'data': data_uri})
[docs] def on_message(self, message): # The 'supports_binary' message is relevant to the # websocket itself. The other messages get passed along # to matplotlib as-is. # Every message has a "type" and a "figure_id". message = json.loads(message['content']['data']) if message['type'] == 'closing': self.on_close() self.manager.clearup_closed() elif message['type'] == 'supports_binary': self.supports_binary = message['value'] else: self.manager.handle_json(message)
@_Backend.export class _BackendNbAgg(_Backend): FigureCanvas = FigureCanvasNbAgg FigureManager = FigureManagerNbAgg @staticmethod def new_figure_manager_given_figure(num, figure): canvas = FigureCanvasNbAgg(figure) manager = FigureManagerNbAgg(canvas, num) if is_interactive(): manager.show() figure.canvas.draw_idle() canvas.mpl_connect('close_event', lambda event: Gcf.destroy(manager)) return manager @staticmethod def trigger_manager_draw(manager): manager.show() @staticmethod def show(block=None): ## TODO: something to do when keyword block==False ? from matplotlib._pylab_helpers import Gcf managers = Gcf.get_all_fig_managers() if not managers: return interactive = is_interactive() for manager in managers: manager.show() # plt.figure adds an event which makes the figure in focus the # active one. Disable this behaviour, as it results in # figures being put as the active figure after they have been # shown, even in non-interactive mode. if hasattr(manager, '_cidgcf'): manager.canvas.mpl_disconnect(manager._cidgcf) if not interactive: Gcf.figs.pop(manager.num, None)