"""Handles javascript remote function calls"""
import logging
log = logging.Logger(__name__)
_socketio = None
_socketio_ready = False
[docs]def set_socketio(socket):
global _socketio
_socketio = socket
[docs]def socketio():
return _socketio
[docs]def register_event(event, handler, namespace='/'):
"""Register a socketio event to a python handler
Parameters
----------
event: str
Name of the event
handler: call
Function to call when the event is fired
"""
assert callable(handler), f'{handler} is not callable'
return socketio().on(event, namespace)(handler)
_pending_bind = []
_pending_attr = []
_pending_props = []
_pending_append_child = []
_pending_display = []
[docs]def display_vega(id, spec):
if not _socketio_ready:
_pending_display.append((id, spec))
log.debug(f'display_vega {id}')
socketio().emit('display_vega', dict(
id=id,
spec=spec
))
[docs]def send_new_data_vega(id, name, new_values):
socketio().emit('stream_data_' + id, dict(
name=name,
new_values=new_values
))
[docs]def set_attribute(id, attribute, value):
"""Set the attribute of an element on the webpage
Parameters
----------
id: str
id of the DOM element
attribute: str
name of the attribute to set
value: json
new value of the attribute
"""
if not _socketio_ready:
_pending_attr.append((id, attribute, value))
log.debug(f'set_attribute {attribute} of {id}')
socketio().emit('set_attribute', dict(
id=id,
attribute=attribute,
value=value
))
[docs]def append_child(id, node):
"""Set the attribute of an element on the webpage
Parameters
----------
id: str
id of the DOM element
node: str
child node
"""
if not _socketio_ready:
_pending_append_child.append((id, node))
log.debug(f'append_child to {id}')
socketio().emit('append_child', dict(
id=id,
node=node
))
[docs]def set_property(id, property, value):
"""Set the attribute of an element on the webpage
Parameters
----------
id: str
id of the DOM element
property: str
name of the property to set
value: json
new value of the property
"""
if not _socketio_ready:
_pending_props.append((id, property, value))
log.debug(f'set_property {property} of {id}')
socketio().emit('set_property', dict(
id=id,
property=property,
value=value
))
[docs]def redirect(url):
"""Set the attribute of an element on the webpage
Parameters
----------
url: str
url to redirect the client to
"""
log.debug(f'redirect ot {url}')
socketio().emit('redirect', dict(url=url))
[docs]def get_element_size(id, callback):
"""Get the size of an element inside the webpage
Parameters
----------
id: str
id of the DOM element
callback: Call
Function to call with the size information `{width: w, height: h}`
"""
log.debug(f'get_element_size of {id}')
socketio().emit('get_size', dict(
id=id
))
register_event(f'get_size_{id}', callback)
[docs]def bind(id, event, handler, attribute=None, property=None):
"""Bind an element event to a handler and return a property of an attribute of the element
Parameters
----------
id: str
id of the DOM element
event: str
Name of the event we are listening too.
The full list of supported events can be found `here <https://www.w3schools.com/jsref/dom_obj_event.asp>`_.
handler: call
function to callback when the event is fired
attribute: str
Attribute of the element to return
property: str
Property of the element to return
"""
assert callable(handler), f'{handler} is not callable'
if not _socketio_ready:
_pending_bind.append((id, event, handler, attribute, property))
log.debug(f'binding `{id}` with `{event}` to `{handler}`')
# ask javascript to listen to events for a particular kind of event on our element
socketio().emit('bind', {'id': id, 'event': event, 'attribute': attribute, 'property': property})
# when the event happen js will send us back the innerHTML of that element
register_event(f'bind_{event}_{id}', handler)
return
[docs]def handshake_event():
"""Called when socketIO connects to the server"""
global _socketio_ready, _pending_attr, _pending_bind, _pending_display, _pending_props, _pending_append_child
_socketio_ready = True
log.info('SocketIO connected')
if _pending_attr:
for arg in _pending_attr:
set_attribute(*arg)
_pending_attr = []
if _pending_bind:
for arg in _pending_bind:
bind(*arg)
_pending_bind = []
if _pending_display:
for arg in _pending_display:
display_vega(*arg)
_pending_display = []
if _pending_props:
for arg in _pending_props:
set_property(*arg)
_pending_props = []
if _pending_append_child:
for arg in _pending_append_child:
append_child(*arg)
_pending_append_child = []
[docs]def disconnect_event():
"""Called when socketIO disconnects from the server"""
global _socketio_ready
_socketio_ready = False
log.info('SocketIO disconnected')