#-----------------------------------------------------------------------------
# Copyright (c) 2012 - 2024, 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 with
Bokeh Documents and Sessions.
'''
#-----------------------------------------------------------------------------
# Boilerplate
#-----------------------------------------------------------------------------
from __future__ import annotations
import logging # isort:skip
log = logging.getLogger(__name__)
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Standard library imports
from typing import TYPE_CHECKING, Callable, Sequence
# Bokeh imports
from ..core.types import ID
from ..util.tornado import _CallbackGroup
if TYPE_CHECKING:
from tornado.ioloop import IOLoop
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
__all__ = (
'NextTickCallback',
'PeriodicCallback',
'SessionCallback',
'TimeoutCallback',
)
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
Callback = Callable[[], None]
[docs]
class SessionCallback:
''' 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/10548
self._callback: Callback = callback
@property
def id(self) -> ID:
''' A unique ID for this callback
'''
return self._id
@property
def callback(self) -> Callback:
''' The callable that this callback wraps.
'''
return self._callback
#-----------------------------------------------------------------------------
# General API
#-----------------------------------------------------------------------------
[docs]
class NextTickCallback(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]
class PeriodicCallback(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
@property
def period(self) -> int:
''' The period time (in milliseconds) that this callback should
repeat execution at.
'''
return self._period
[docs]
class TimeoutCallback(SessionCallback):
''' Represent a callback to execute once on the ``IOLoop`` after a specified
time interval passes.
'''
_timeout: int
[docs]
def __init__(self, callback: Callback, timeout: int, *, callback_id: ID) -> None:
'''
Args:
callback (callable) :
timeout (int) :
id (ID) :
'''
super().__init__(callback=callback, callback_id=callback_id)
self._timeout = timeout
@property
def timeout(self) -> int:
''' The timeout (in milliseconds) that the callback should run after.
'''
return self._timeout
#-----------------------------------------------------------------------------
# Dev API
#-----------------------------------------------------------------------------
class DocumentCallbackGroup:
'''
'''
def __init__(self, io_loop: IOLoop) -> None:
'''
'''
self._group = _CallbackGroup(io_loop)
def remove_all_callbacks(self) -> None:
'''
'''
self._group.remove_all_callbacks()
def add_session_callbacks(self, callbacks: Sequence[SessionCallback]) -> None:
'''
'''
for cb in callbacks:
self.add_session_callback(cb)
def add_session_callback(self, callback_obj: SessionCallback) -> None:
'''
'''
if isinstance(callback_obj, PeriodicCallback):
self._group.add_periodic_callback(callback_obj.callback, callback_obj.period, callback_obj.id)
elif isinstance(callback_obj, TimeoutCallback):
self._group.add_timeout_callback(callback_obj.callback, callback_obj.timeout, callback_obj.id)
elif isinstance(callback_obj, NextTickCallback):
self._group.add_next_tick_callback(callback_obj.callback, callback_obj.id)
else:
raise ValueError(f"Expected callback of type PeriodicCallback, TimeoutCallback, NextTickCallback, got: {callback_obj.callback}")
def remove_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:
if isinstance(callback_obj, PeriodicCallback):
self._group.remove_periodic_callback(callback_obj.id)
elif isinstance(callback_obj, TimeoutCallback):
self._group.remove_timeout_callback(callback_obj.id)
elif isinstance(callback_obj, NextTickCallback):
self._group.remove_next_tick_callback(callback_obj.id)
except ValueError:
pass
#-----------------------------------------------------------------------------
# Private API
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------
