#-----------------------------------------------------------------------------# Copyright (c) Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------''' Provide classes to represent callback code that can be associate withBokeh Documents and Sessions.'''#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsfromtypingimportTYPE_CHECKING,Callable,Sequence# Bokeh importsfrom..core.typesimportIDfrom..util.tornadoimport_CallbackGroupifTYPE_CHECKING:fromtornado.ioloopimportIOLoop#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=('NextTickCallback','PeriodicCallback','SessionCallback','TimeoutCallback',)#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------Callback=Callable[[],None]
[docs]classSessionCallback:''' A base class for callback objects associated with Bokeh Documents and Sessions. '''_id:ID
[docs]def__init__(self,callback:Callback,*,callback_id:ID)->None:''' Args: callback (callable) : id (ID) : '''self._id=callback_id# specify type here until this is released: https://github.com/python/mypy/pull/10548self._callback:Callback=callback
@propertydefid(self)->ID:''' A unique ID for this callback '''returnself._id@propertydefcallback(self)->Callback:''' The callable that this callback wraps. '''returnself._callback
#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------
[docs]classNextTickCallback(SessionCallback):''' Represent a callback to execute on the next ``IOLoop`` "tick". '''
[docs]def__init__(self,callback:Callback,*,callback_id:ID)->None:''' Args: callback (callable) : id (ID) : '''super().__init__(callback=callback,callback_id=callback_id)
[docs]classPeriodicCallback(SessionCallback):''' Represent a callback to execute periodically on the ``IOLoop`` at a specified periodic time interval. '''_period:int
[docs]def__init__(self,callback:Callback,period:int,*,callback_id:ID)->None:''' Args: callback (callable) : period (int) : id (ID) : '''super().__init__(callback=callback,callback_id=callback_id)self._period=period
@propertydefperiod(self)->int:''' The period time (in milliseconds) that this callback should repeat execution at. '''returnself._period
[docs]classTimeoutCallback(SessionCallback):''' Represent a callback to execute once on the ``IOLoop`` after a specified time interval passes. '''_timeout:int
@propertydeftimeout(self)->int:''' The timeout (in milliseconds) that the callback should run after. '''returnself._timeout
#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------classDocumentCallbackGroup:''' '''def__init__(self,io_loop:IOLoop)->None:''' '''self._group=_CallbackGroup(io_loop)defremove_all_callbacks(self)->None:''' '''self._group.remove_all_callbacks()defadd_session_callbacks(self,callbacks:Sequence[SessionCallback])->None:''' '''forcbincallbacks:self.add_session_callback(cb)defadd_session_callback(self,callback_obj:SessionCallback)->None:''' '''ifisinstance(callback_obj,PeriodicCallback):self._group.add_periodic_callback(callback_obj.callback,callback_obj.period,callback_obj.id)elifisinstance(callback_obj,TimeoutCallback):self._group.add_timeout_callback(callback_obj.callback,callback_obj.timeout,callback_obj.id)elifisinstance(callback_obj,NextTickCallback):self._group.add_next_tick_callback(callback_obj.callback,callback_obj.id)else:raiseValueError(f"Expected callback of type PeriodicCallback, TimeoutCallback, NextTickCallback, got: {callback_obj.callback}")defremove_session_callback(self,callback_obj:SessionCallback)->None:''' '''# we may be called multiple times because of multiple# views on a document - the document has to notify that# the callback was removed even if only one view invoked# it. So we need to silently no-op if we're already# removed.try:ifisinstance(callback_obj,PeriodicCallback):self._group.remove_periodic_callback(callback_obj.id)elifisinstance(callback_obj,TimeoutCallback):self._group.remove_timeout_callback(callback_obj.id)elifisinstance(callback_obj,NextTickCallback):self._group.remove_next_tick_callback(callback_obj.id)exceptValueError:pass#-----------------------------------------------------------------------------# Private API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Code#-----------------------------------------------------------------------------