#-----------------------------------------------------------------------------# Copyright (c) Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------""" Serialization and deserialization utilities. """#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsimportbase64importdatetimeasdtimportsysfromarrayimportarrayasTypedArrayfrommathimportisinf,isnanfromtypesimportSimpleNamespacefromtypingimport(TYPE_CHECKING,Any,Callable,ClassVar,Generic,Literal,NoReturn,Sequence,TypeAlias,TypedDict,TypeVar,cast,)# External importsimportnumpyasnp# Bokeh importsfrom..util.dataclassesimport(Unspecified,dataclass,entries,is_dataclass,)from..util.dependenciesimportuses_pandasfrom..util.serializationimport(array_encoding_disabled,convert_datetime_type,convert_timedelta_type,is_datetime_type,is_timedelta_type,make_id,transform_array,transform_series,)from..util.warningsimportBokehUserWarning,warnfrom.typesimportIDifTYPE_CHECKING:importnumpy.typingasnptfromtyping_extensionsimportNotRequiredfrom..core.has_propsimportSetterfrom..modelimportModel#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=("Buffer","DeserializationError","Deserializer","Serializable","SerializationError","Serializer",)_MAX_SAFE_INT=2**53-1#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------AnyRep:TypeAlias=AnyclassRef(TypedDict):id:IDclassRefRep(TypedDict):type:Literal["ref"]id:IDclassSymbolRep(TypedDict):type:Literal["symbol"]name:strclassNumberRep(TypedDict):type:Literal["number"]value:Literal["nan","-inf","+inf"]|floatclassArrayRep(TypedDict):type:Literal["array"]entries:NotRequired[list[AnyRep]]ArrayRepLike:TypeAlias=ArrayRep|list[AnyRep]classSetRep(TypedDict):type:Literal["set"]entries:NotRequired[list[AnyRep]]classMapRep(TypedDict):type:Literal["map"]entries:NotRequired[list[tuple[AnyRep,AnyRep]]]classBytesRep(TypedDict):type:Literal["bytes"]data:Buffer|Ref|strclassSliceRep(TypedDict):type:Literal["slice"]start:int|Nonestop:int|Nonestep:int|NoneclassObjectRep(TypedDict):type:Literal["object"]name:strattributes:NotRequired[dict[str,AnyRep]]classObjectRefRep(TypedDict):type:Literal["object"]name:strid:IDattributes:NotRequired[dict[str,AnyRep]]ModelRep=ObjectRefRepByteOrder:TypeAlias=Literal["little","big"]DataType:TypeAlias=Literal["uint8","int8","uint16","int16","uint32","int32","float32","float64"]# "uint64", "int64"NDDataType:TypeAlias=Literal["bool"]|DataType|Literal["object"]classTypedArrayRep(TypedDict):type:Literal["typed_array"]array:BytesReporder:ByteOrderdtype:DataTypeclassNDArrayRep(TypedDict):type:Literal["ndarray"]array:BytesRep|ArrayRepLikeorder:ByteOrderdtype:NDDataTypeshape:list[int]
[docs]classSerializable:""" A mixin for making a type serializable. """defto_serializable(self,serializer:Serializer)->AnyRep:""" Converts this object to a serializable representation. """raiseNotImplementedError()
ObjID=int
[docs]classSerializer:""" Convert built-in and custom types into serializable representations. Not all built-in types are supported (e.g., decimal.Decimal due to lacking support for fixed point arithmetic in JavaScript). """_encoders:ClassVar[dict[type[Any],Encoder]]={}@classmethoddefregister(cls,type:type[Any],encoder:Encoder)->None:asserttypenotincls._encoders,f"'{type} is already registered"cls._encoders[type]=encoder_references:dict[ObjID,Ref]_deferred:bool_circular:dict[ObjID,Any]_buffers:list[Buffer]def__init__(self,*,references:set[Model]=set(),deferred:bool=True)->None:self._references={id(obj):obj.refforobjinreferences}self._deferred=deferredself._circular={}self._buffers=[]defhas_ref(self,obj:Any)->bool:returnid(obj)inself._referencesdefadd_ref(self,obj:Any,ref:Ref)->None:assertid(obj)notinself._referencesself._references[id(obj)]=refdefget_ref(self,obj:Any)->Ref|None:returnself._references.get(id(obj))@propertydefbuffers(self)->list[Buffer]:returnlist(self._buffers)defserialize(self,obj:Any)->Serialized[Any]:returnSerialized(self.encode(obj),self.buffers)defencode(self,obj:Any)->AnyRep:ref=self.get_ref(obj)ifrefisnotNone:returnrefident=id(obj)ifidentinself._circular:self.error("circular reference")self._circular[ident]=objtry:returnself._encode(obj)finally:delself._circular[ident]defencode_struct(self,**fields:Any)->dict[str,AnyRep]:return{key:self.encode(val)forkey,valinfields.items()ifvalisnotUnspecified}def_encode(self,obj:Any)->AnyRep:ifisinstance(obj,Serializable):returnobj.to_serializable(self)elif(encoder:=self._encoders.get(type(obj)))isnotNone:returnencoder(obj,self)elifobjisNone:returnNoneelifisinstance(obj,bool):returnself._encode_bool(obj)elifisinstance(obj,str):returnself._encode_str(obj)elifisinstance(obj,int):returnself._encode_int(obj)elifisinstance(obj,float):returnself._encode_float(obj)elifisinstance(obj,tuple):returnself._encode_tuple(obj)elifisinstance(obj,list):returnself._encode_list(obj)elifisinstance(obj,set):returnself._encode_set(obj)elifisinstance(obj,dict):returnself._encode_dict(obj)elifisinstance(obj,SimpleNamespace):returnself._encode_struct(obj)elifisinstance(obj,bytes):returnself._encode_bytes(obj)elifisinstance(obj,slice):returnself._encode_slice(obj)elifisinstance(obj,TypedArray):returnself._encode_typed_array(obj)elifisinstance(obj,np.ndarray):ifobj.shape!=():returnself._encode_ndarray(obj)else:returnself._encode(obj.item())elifis_dataclass(obj):returnself._encode_dataclass(obj)else:returnself._encode_other(obj)def_encode_bool(self,obj:bool)->AnyRep:returnobjdef_encode_str(self,obj:str)->AnyRep:returnobjdef_encode_int(self,obj:int)->AnyRep:if-_MAX_SAFE_INT<obj<=_MAX_SAFE_INT:returnobjelse:warn("out of range integer may result in loss of precision",BokehUserWarning)returnself._encode_float(float(obj))def_encode_float(self,obj:float)->NumberRep|float:ifisnan(obj):returnNumberRep(type="number",value="nan")elifisinf(obj):returnNumberRep(type="number",value="-inf"ifobj<0else"+inf")else:returnobjdef_encode_tuple(self,obj:tuple[Any,...])->ArrayRepLike:returnself._encode_list(list(obj))def_encode_list(self,obj:list[Any])->ArrayRepLike:return[self.encode(item)foriteminobj]def_encode_set(self,obj:set[Any])->SetRep:iflen(obj)==0:returnSetRep(type="set")else:returnSetRep(type="set",entries=[self.encode(entry)forentryinobj],)def_encode_dict(self,obj:dict[Any,Any])->MapRep:iflen(obj)==0:result=MapRep(type="map")else:result=MapRep(type="map",entries=[(self.encode(key),self.encode(val))forkey,valinobj.items()],)returnresultdef_encode_struct(self,obj:SimpleNamespace)->MapRep:returnself._encode_dict(obj.__dict__)def_encode_dataclass(self,obj:Any)->ObjectRep:cls=type(obj)module=cls.__module__name=cls.__qualname__.replace("<locals>.","")rep=ObjectRep(type="object",name=f"{module}.{name}",)attributes=list(entries(obj))ifattributes:rep["attributes"]={key:self.encode(val)forkey,valinattributes}returnrepdef_encode_bytes(self,obj:bytes|memoryview)->BytesRep:buffer=Buffer(make_id(),obj)data:Buffer|strifself._deferred:self._buffers.append(buffer)data=bufferelse:data=buffer.to_base64()returnBytesRep(type="bytes",data=data)def_encode_slice(self,obj:slice)->SliceRep:returnSliceRep(type="slice",start=self.encode(obj.start),stop=self.encode(obj.stop),step=self.encode(obj.step),)def_encode_typed_array(self,obj:TypedArray[Any])->TypedArrayRep:array=self._encode_bytes(memoryview(obj))typecode=obj.typecodeitemsize=obj.itemsizedefdtype()->DataType:matchtypecode:case"f":return"float32"case"d":return"float64"case"B"|"H"|"I"|"L"|"Q":matchobj.itemsize:case1:return"uint8"case2:return"uint16"case4:return"uint32"#case 8: return "uint64"case"b"|"h"|"i"|"l"|"q":matchobj.itemsize:case1:return"int8"case2:return"int16"case4:return"int32"#case 8: return "int64"self.error(f"can't serialize array with items of type '{typecode}@{itemsize}'")returnTypedArrayRep(type="typed_array",array=array,order=sys.byteorder,dtype=dtype(),)def_encode_ndarray(self,obj:npt.NDArray[Any])->NDArrayRep:array=transform_array(obj)data:ArrayRepLike|BytesRepdtype:NDDataTypeifarray_encoding_disabled(array):data=self._encode_list(array.flatten().tolist())dtype="object"else:data=self._encode_bytes(array.data)dtype=cast(NDDataType,array.dtype.name)returnNDArrayRep(type="ndarray",array=data,shape=list(array.shape),dtype=dtype,order=sys.byteorder,)def_encode_other(self,obj:Any)->AnyRep:# date/time values that get serialized as millisecondsifis_datetime_type(obj):returnconvert_datetime_type(obj)ifis_timedelta_type(obj):returnconvert_timedelta_type(obj)ifisinstance(obj,dt.date):returnobj.isoformat()# NumPy scalarsifnp.issubdtype(type(obj),np.floating):returnself._encode_float(float(obj))ifnp.issubdtype(type(obj),np.integer):returnself._encode_int(int(obj))ifnp.issubdtype(type(obj),np.bool_):returnself._encode_bool(bool(obj))# avoid importing pandas here unless it is actually in useifuses_pandas(obj):importpandasaspdifisinstance(obj,pd.Series|pd.Index|pd.api.extensions.ExtensionArray):returnself._encode_ndarray(transform_series(obj))elifobjispd.NA:returnNone# handle array libraries that support conversion to a numpy array (e.g. polars, PyTorch)ifhasattr(obj,"__array__")andisinstance(arr:=obj.__array__(),np.ndarray):returnself._encode_ndarray(arr)self.error(f"can't serialize {type(obj)}")deferror(self,message:str)->NoReturn:raiseSerializationError(message)
[docs]classDeserializer:""" Convert from serializable representations to built-in and custom types. """_decoders:ClassVar[dict[str,Decoder]]={}@classmethoddefregister(cls,type:str,decoder:Decoder)->None:asserttypenotincls._decoders,f"'{type} is already registered"cls._decoders[type]=decoder_references:dict[ID,Model]_setter:Setter|None_decoding:bool_buffers:dict[ID,Buffer]def__init__(self,references:Sequence[Model]|None=None,*,setter:Setter|None=None):self._references={obj.id:objforobjinreferencesor[]}self._setter=setterself._decoding=Falseself._buffers={}defhas_ref(self,obj:Model)->bool:returnobj.idinself._referencesdefdeserialize(self,obj:Any|Serialized[Any])->Any:ifisinstance(obj,Serialized):returnself.decode(obj.content,obj.buffers)else:returnself.decode(obj)defdecode(self,obj:AnyRep,buffers:list[Buffer]|None=None)->Any:ifbuffersisnotNone:forbufferinbuffers:self._buffers[buffer.id]=bufferifself._decoding:returnself._decode(obj)self._decoding=Truetry:returnself._decode(obj)finally:self._buffers.clear()self._decoding=Falsedef_decode(self,obj:AnyRep)->Any:ifisinstance(obj,dict):if"type"inobj:matchobj["type"]:casetypeiftypeinself._decoders:returnself._decoders[type](obj,self)case"ref":returnself._decode_ref(cast(Ref,obj))case"symbol":returnself._decode_symbol(cast(SymbolRep,obj))case"number":returnself._decode_number(cast(NumberRep,obj))case"array":returnself._decode_array(cast(ArrayRep,obj))case"set":returnself._decode_set(cast(SetRep,obj))case"map":returnself._decode_map(cast(MapRep,obj))case"bytes":returnself._decode_bytes(cast(BytesRep,obj))case"slice":returnself._decode_slice(cast(SliceRep,obj))case"typed_array":returnself._decode_typed_array(cast(TypedArrayRep,obj))case"ndarray":returnself._decode_ndarray(cast(NDArrayRep,obj))case"object":if"id"inobj:returnself._decode_object_ref(cast(ObjectRefRep,obj))else:returnself._decode_object(cast(ObjectRep,obj))casetype:self.error(f"unable to decode an object of type '{type}'")elif"id"inobj:returnself._decode_ref(cast(Ref,obj))else:return{key:self._decode(val)forkey,valinobj.items()}elifisinstance(obj,list):return[self._decode(entry)forentryinobj]else:returnobjdef_decode_ref(self,obj:Ref)->Model:id=obj["id"]instance=self._references.get(id)ifinstanceisnotNone:returninstanceelse:self.error(UnknownReferenceError(id))def_decode_symbol(self,obj:SymbolRep)->float:name=obj["name"]self.error(f"can't resolve named symbol '{name}'")# TODO: implement symbol resolutiondef_decode_number(self,obj:NumberRep)->float:value=obj["value"]returnfloat(value)ifisinstance(value,str)elsevaluedef_decode_array(self,obj:ArrayRep)->list[Any]:entries=obj.get("entries",[])return[self._decode(entry)forentryinentries]def_decode_set(self,obj:SetRep)->set[Any]:entries=obj.get("entries",[])return{self._decode(entry)forentryinentries}def_decode_map(self,obj:MapRep)->dict[Any,Any]:entries=obj.get("entries",[])return{self._decode(key):self._decode(val)forkey,valinentries}def_decode_bytes(self,obj:BytesRep)->bytes:data=obj["data"]ifisinstance(data,str):returnbase64.b64decode(data)elifisinstance(data,Buffer):buffer=data# in case of decode(encode(obj))else:id=data["id"]ifidinself._buffers:buffer=self._buffers[id]else:self.error(f"can't resolve buffer '{id}'")returnbuffer.datadef_decode_slice(self,obj:SliceRep)->slice:start=self._decode(obj["start"])stop=self._decode(obj["stop"])step=self._decode(obj["step"])returnslice(start,stop,step)def_decode_typed_array(self,obj:TypedArrayRep)->TypedArray[Any]:array=obj["array"]order=obj["order"]dtype=obj["dtype"]data=self._decode(array)dtype_to_typecode=dict(uint8="B",int8="b",uint16="H",int16="h",uint32="I",int32="i",#uint64="Q",#int64="q",float32="f",float64="d",)typecode=dtype_to_typecode.get(dtype)iftypecodeisNone:self.error(f"unsupported dtype '{dtype}'")typed_array:TypedArray[Any]=TypedArray(typecode,data)iforder!=sys.byteorder:typed_array.byteswap()returntyped_arraydef_decode_ndarray(self,obj:NDArrayRep)->npt.NDArray[Any]:array=obj["array"]order=obj["order"]dtype=obj["dtype"]shape=obj["shape"]decoded=self._decode(array)ndarray:npt.NDArray[Any]ifisinstance(decoded,bytes):ndarray=np.copy(np.frombuffer(decoded,dtype=dtype))iforder!=sys.byteorder:ndarray.byteswap(inplace=True)else:ndarray=np.array(decoded,dtype=dtype)iflen(shape)>1:ndarray=ndarray.reshape(shape)returnndarraydef_decode_object(self,obj:ObjectRep)->object:raiseNotImplementedError()def_decode_object_ref(self,obj:ObjectRefRep)->Model:id=obj["id"]instance=self._references.get(id)ifinstanceisnotNone:warn(f"reference already known '{id}'",BokehUserWarning)returninstancename=obj["name"]attributes=obj.get("attributes")cls=self._resolve_type(name)instance=cls.__new__(cls,id=id)ifinstanceisNone:self.error(f"can't instantiate {name}(id={id})")self._references[instance.id]=instance# We want to avoid any Model specific initialization that happens with# Slider(...) when reconstituting from JSON, but we do need to perform# general HasProps machinery that sets properties, so call it explicitlyifnotinstance._initialized:from.has_propsimportHasPropsHasProps.__init__(instance)ifattributesisnotNone:decoded_attributes={key:self._decode(val)forkey,valinattributes.items()}forkey,valindecoded_attributes.items():instance.set_from_json(key,val,setter=self._setter)returninstancedef_resolve_type(self,type:str)->type[Model]:from..modelimportModelcls=Model.model_class_reverse_map.get(type)ifclsisnotNone:ifissubclass(cls,Model):returnclselse:self.error(f"object of type '{type}' is not a subclass of 'Model'")else:iftype=="Figure":from..plottingimportfigurereturnfigure# XXX: helps with push_session(); this needs a better resolution schemeelse:self.error(f"can't resolve type '{type}'")deferror(self,error:str|DeserializationError)->NoReturn:ifisinstance(error,str):raiseDeserializationError(error)else:raiseerror
#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Private API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Code#-----------------------------------------------------------------------------