#-----------------------------------------------------------------------------# Copyright (c) Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------'''Functions for helping with serialization and deserialization ofBokeh objects.Certain NumPy array dtypes can be serialized to a binary format forperformance and efficiency. The list of supported dtypes is:{binary_array_types}'''#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsimportdatetimeasdtimportuuidfromfunctoolsimportlru_cachefromthreadingimportLockfromtypingimportTYPE_CHECKING,Any,TypeGuard# External importsimportnumpyasnp# Bokeh importsfrom..core.typesimportIDfrom..settingsimportsettingsfrom.stringsimportformat_docstringifTYPE_CHECKING:importnumpy.typingasnptimportpandasaspd#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------@lru_cache(None)def_compute_datetime_types()->set[type]:importpandasaspdresult={dt.time,dt.datetime,np.datetime64}result.add(pd.Timestamp)result.add(pd.Timedelta)result.add(pd.Period)result.add(type(pd.NaT))returnresultdef__getattr__(name:str)->Any:ifname=="DATETIME_TYPES":return_compute_datetime_types()raiseAttributeErrorBINARY_ARRAY_TYPES={np.dtype(np.bool_),np.dtype(np.uint8),np.dtype(np.int8),np.dtype(np.uint16),np.dtype(np.int16),np.dtype(np.uint32),np.dtype(np.int32),#np.dtype(np.uint64),#np.dtype(np.int64),np.dtype(np.float32),np.dtype(np.float64),}NP_EPOCH=np.datetime64(0,'ms')NP_MS_DELTA=np.timedelta64(1,'ms')DT_EPOCH=dt.datetime.fromtimestamp(0,tz=dt.timezone.utc)__doc__=format_docstring(__doc__,binary_array_types="\n".join(f"* ``np.{x}``"forxinBINARY_ARRAY_TYPES))__all__=('array_encoding_disabled','convert_date_to_datetime','convert_datetime_array','convert_datetime_type','convert_timedelta_type','is_datetime_type','is_timedelta_type','make_globally_unique_css_safe_id','make_globally_unique_id','make_id','transform_array','transform_series',)#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------
[docs]defis_datetime_type(obj:Any)->TypeGuard[dt.time|dt.datetime|np.datetime64]:''' Whether an object is any date, time, or datetime type recognized by Bokeh. Args: obj (object) : the object to test Returns: bool : True if ``obj`` is a datetime type '''_dt_tuple=tuple(_compute_datetime_types())returnisinstance(obj,_dt_tuple)
[docs]defis_timedelta_type(obj:Any)->TypeGuard[dt.timedelta|np.timedelta64]:''' Whether an object is any timedelta type recognized by Bokeh. Args: obj (object) : the object to test Returns: bool : True if ``obj`` is a timedelta type '''returnisinstance(obj,dt.timedelta|np.timedelta64)
[docs]defconvert_date_to_datetime(obj:dt.date)->float:''' Convert a date object to a datetime Args: obj (date) : the object to convert Returns: datetime '''return(dt.datetime(*obj.timetuple()[:6],tzinfo=dt.timezone.utc)-DT_EPOCH).total_seconds()*1000
[docs]defconvert_timedelta_type(obj:dt.timedelta|np.timedelta64)->float:''' Convert any recognized timedelta value to floating point absolute milliseconds. Args: obj (object) : the object to convert Returns: float : milliseconds '''ifisinstance(obj,dt.timedelta):returnobj.total_seconds()*1000.elifisinstance(obj,np.timedelta64):returnfloat(obj/NP_MS_DELTA)raiseValueError(f"Unknown timedelta object: {obj!r}")
# The Any here should be pd.NaT | pd.Period but mypy chokes on that for some reason
[docs]defconvert_datetime_type(obj:Any|pd.Timestamp|pd.Timedelta|dt.datetime|dt.date|dt.time|np.datetime64)->float:''' Convert any recognized date, time, or datetime value to floating point milliseconds since epoch. Args: obj (object) : the object to convert Returns: float : milliseconds '''importpandasaspd# Pandas NaTifobjispd.NaT:returnnp.nan# Pandas Periodifisinstance(obj,pd.Period):returnobj.to_timestamp().value/10**6.0# Pandas Timestampifisinstance(obj,pd.Timestamp):returnobj.value/10**6.0# Pandas Timedeltaelifisinstance(obj,pd.Timedelta):returnobj.value/10**6.0# Datetime (datetime is a subclass of date)elifisinstance(obj,dt.datetime):diff=obj.replace(tzinfo=dt.timezone.utc)-DT_EPOCHreturndiff.total_seconds()*1000# XXX (bev) ideally this would not be here "dates are not datetimes"# Dateelifisinstance(obj,dt.date):returnconvert_date_to_datetime(obj)# NumPy datetime64elifisinstance(obj,np.datetime64):epoch_delta=obj-NP_EPOCHreturnfloat(epoch_delta/NP_MS_DELTA)# Timeelifisinstance(obj,dt.time):return(obj.hour*3600+obj.minute*60+obj.second)*1000+obj.microsecond/1000.0raiseValueError(f"unknown datetime object: {obj!r}")
[docs]defconvert_datetime_array(array:npt.NDArray[Any])->npt.NDArray[np.floating[Any]]:''' Convert NumPy datetime arrays to arrays to milliseconds since epoch. Args: array : (obj) A NumPy array of datetime to convert If the value passed in is not a NumPy array, it will be returned as-is. Returns: array '''defconvert(array:npt.NDArray[Any])->npt.NDArray[Any]:returnnp.where(np.isnat(array),np.nan,array.astype("int64")/1000.0)# not quite correct, truncates to ms..ifarray.dtype.kind=="M":returnconvert(array.astype("datetime64[us]"))elifarray.dtype.kind=="m":returnconvert(array.astype("timedelta64[us]"))# XXX (bev) special case dates, not greatelifarray.dtype.kind=="O"andlen(array)>0andisinstance(array[0],dt.date):try:returnconvert(array.astype("datetime64[us]"))exceptException:passreturnarray
[docs]defmake_id()->ID:''' Return a new unique ID for a Bokeh object. Normally this function will return simple monotonically increasing integer IDs (as strings) for identifying Bokeh objects within a Document. However, if it is desirable to have globally unique for every object, this behavior can be overridden by setting the environment variable ``BOKEH_SIMPLE_IDS=no``. Returns: str '''global_simple_idifsettings.simple_ids():with_simple_id_lock:_simple_id+=1returnID(f"p{_simple_id}")else:returnmake_globally_unique_id()
[docs]defmake_globally_unique_id()->ID:''' Return a globally unique UUID. Some situations, e.g. id'ing dynamically created Divs in HTML documents, always require globally unique IDs. Returns: str '''returnID(str(uuid.uuid4()))
[docs]defmake_globally_unique_css_safe_id()->ID:''' Return a globally unique CSS-safe UUID. Some situations, e.g. id'ing dynamically created Divs in HTML documents, always require globally unique IDs. ID generated with this function can be used in APIs like ``document.querySelector("#id")``. Returns: str '''max_iter=100for_iinrange(0,max_iter):id=make_globally_unique_id()ifid[0].isalpha():returnidreturnID(f"bk-{make_globally_unique_id()}")
[docs]defarray_encoding_disabled(array:npt.NDArray[Any])->bool:''' Determine whether an array may be binary encoded. The NumPy array dtypes that can be encoded are: {binary_array_types} Args: array (np.ndarray) : the array to check Returns: bool '''# disable binary encoding for non-supported dtypesreturnarray.dtypenotinBINARY_ARRAY_TYPES
[docs]deftransform_array(array:npt.NDArray[Any])->npt.NDArray[Any]:''' Transform a ndarray into a serializable ndarray. Converts un-serializable dtypes and returns JSON serializable format Args: array (np.ndarray) : a NumPy array to be transformed Returns: ndarray '''array=convert_datetime_array(array)# XXX: as long as we can't support 64-bit integers, try to convert# to 32-bits. If not possible, let the serializer convert to a less# efficient representation and/or deal with any error messaging.def_cast_if_can(array:npt.NDArray[Any],dtype:type[Any])->npt.NDArray[Any]:info=np.iinfo(dtype)ifnp.any((array<info.min)|(info.max<array)):returnarrayelse:returnarray.astype(dtype,casting="unsafe")ifarray.dtype==np.dtype(np.int64):array=_cast_if_can(array,np.int32)elifarray.dtype==np.dtype(np.uint64):array=_cast_if_can(array,np.uint32)ifisinstance(array,np.ma.MaskedArray):array=array.filled(np.nan)# type: ignore # filled is untypedifnotarray.flags["C_CONTIGUOUS"]:array=np.ascontiguousarray(array)returnarray
[docs]deftransform_series(series:pd.Series[Any]|pd.Index[Any]|pd.api.extensions.ExtensionArray)->npt.NDArray[Any]:''' Transforms a Pandas series into serialized form Args: series (pd.Series) : the Pandas series to transform Returns: ndarray '''importpandasaspd# not checking for pd here, this function should only be called if it# is already known that series is a Pandas Series typeifisinstance(series,pd.PeriodIndex):vals=series.to_timestamp().valueselse:vals=series.to_numpy()returnvals
#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Private API#-----------------------------------------------------------------------------_simple_id=999_simple_id_lock=Lock()#-----------------------------------------------------------------------------# Code#-----------------------------------------------------------------------------