communication.communication module

This module contains the serialization logic used in sending and receiving arbitrary objects.

exception communication.communication.AnnotationError

Bases: Exception

Raised when an improperly function is incorrectly annotated

class communication.communication.Communication

Bases: object

Virtual class that provides packing and unpacking functions used for communications. The outline is as follows: - serialization functions for different classes - packing function that handles metadata and determines which serialization needs to happen

• deserialization functions for different classes

• unpacking function that handles metadata and determines which deserialization needs to happen

StandardT = ~StandardT

static bytes_deserialize(json_obj)

Function for deserializing bytes

Parameters:

json_obj (Dict[str, str]) – object to deserialize

Return type:

bytes

Returns:

deserialized bytes object

static bytes_serialize(obj)

Function for serializing bytes

Parameters:

obj (bytes) – bytes object to serialize

Return type:

Dict[str, str]

Returns:

serialized object

static clear_new_serialization_logic()

Clear all custom serialization (and deserialization) logic that was added to this class.

Return type:

None

static collection_deserialize(json_obj)

Function for deserializing collections

The class tree is recursed in the same order as the original collection was processed during serialization. Every time a ‘leaf’ is encountered, the keys are identified for the respective type and for each key, the first element of the list is extracted. These values are then used to deserialize the type.

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Raises:

ValueError – raised when (nested) value cannot be deserialized

Return type:

Union[Dict[Any, Any], List[Any], Any]

Returns:

deserialized collection

static collection_serialize(collection, use_pickle=True)

function to serialize lists and dictionaries

The structure of the collection is saved in a class tree, that has the same structure as the original collection, but instead of values it contains the class types. All ‘leaf’ values inside the collection are added to a one-dimensional list for each key of the serialization of all the leaves

Parameters:

• collection (Union[Dict[Any, Any], List[Any]]) – collection to serialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback serializer

Return type:

Dict[str, Any]

Returns:

serialized collection

static default_deserialize(json_obj)

Fall-back function is case no specific deserialization function is available. This function uses the pickle library

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

Any

Returns:

deserialized object

static default_serialize(obj, use_pickle=True)

Fall-back function is case no specific serialization function is available. This function uses the pickle library

Parameters:

• obj (Any) – object to serialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback serializer

Raises:

NotImplementedError – raised when no serialization function is defined for object

Return type:

Dict[str, Dict[str, str]]

Returns:

serialized object

default_use_pickle = True

default_verify_obfuscation = True

static deserialize(json_obj)

Function that detects which deserialization function should be run and calls it

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

Any

Returns:

deserialized object

static dict_deserialize(json_obj)

Function for deserializing a dictionary.

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

Dict[Any, Any]

Returns:

deserialized dictionary of which the contents have also been deserialized

static extract_keys(result, serialized_element)

function that unpacks low level results for collection serialization

Parameters:

• result (Dict[str, Any]) – dictionary to store the extracted results in

• serialized_element (Any) – elements to extract

Return type:

None

static list_deserialize(json_obj)

Function for deserializing lists

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

List[Any]

Returns:

deserialized list object

static list_deserialize_dif_type(json_obj)

Function for deserializing a list containing objects of more than one type.

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

List[Any]

Returns:

deserialized list of which the contents have also been deserialized

static list_deserialize_same_type(json_obj)

Function for deserializing a list containing objects of only one type.

Parameters:

json_obj (Dict[str, Any]) – object to deserialize

Return type:

List[Any]

Returns:

deserialized list of which the contents have also been deserialized

new_deserialization_funcs: ClassVar[Dict[str, Callable[[Dict[str, Any]], Any]]] = {}

new_serialization_funcs: ClassVar[Dict[str, Callable[[Any], Dict[str, Any]]]] = {}

static pack(obj, msg_id, use_pickle=True)

Function that adds metadata and serializes the object for transmission.

Parameters:

• obj (Any) – object to pack

• msg_id (Union[str, int]) – message identifier associated to the message

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback packer

Return type:

Dict[str, Any]

Returns:

packed object (serialized and annotated)

static serialize(obj, use_pickle=True)

Function that detects with serialization function should be used and applies it

Parameters:

• obj (Any) – object to serialize

• use_pickle (bool) – set to true if one wishes to use pickle as a fallback serializer

Return type:

Dict[str, Any]

Returns:

serialized object

static set_serialization_logic(obj_class, check_annotations=True)

Function for storing serialization logic for classes that have not been specified here or need to be overridden

Parameters:

• obj_class (Type[SupportsSerialization]) – object class to set serialization logic for

• check_annotations (bool) – validate return annotation of the serialization logic

Raises:

• RepetitionError – raised when serialization function is already defined for object class

• TypeError – raised when provided object class has no (de)serialization function

• AnnotationError – raised when the return annotation is inconsistent

Return type:

None

static standard_deserialize(json_obj)

Function for the most basic way of deserialization that works for basic data types, such as int, float, string

Parameters:

json_obj (Dict[str, TypeVar(StandardT, int, float, str)]) – object to deserialize

Return type:

TypeVar(StandardT, int, float, str)

Returns:

deserialized object

static standard_serialize(obj)

Function for the most basic way of serialization that works for basic data types, such as int, float, string

Parameters:

obj (TypeVar(StandardT, int, float, str)) – object to serialize

Return type:

Dict[str, TypeVar(StandardT, int, float, str)]

Returns:

serialized object

static unpack(json_obj)

Function that handles metadata and turns the json object into a python object

Parameters:

json_obj (Dict[str, Any]) – json object to unpack

Return type:

Tuple[str, Any]

Returns:

unpacked object

class communication.communication.MultiDimensionalArrayEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

Class that represents a JSON encoder that can additionally handle collections like tuples, lists and dictionaries

encode(o)

Function that encodes a python object

Parameters:

o (Any) – python object

Return type:

str

Returns:

the object encoded in a string

exception communication.communication.RepetitionError

Bases: Exception

Raised when the action has already been performed and should not be repeated

class communication.communication.SupportsSerialization(*args, **kwargs)

Bases: Protocol

Type placeholder for classes supporting custom serialization.

static deserialize(obj)

Deserialize the given JSON dict into an object of this class.

Parameters:

obj (Dict[str, Any]) – JSON dict to be deserialized

Return type:

SupportsSerialization

Returns:

deserialized object

serialize()

Serialize this object into a JSON dict.

Return type:

Dict[str, Any]

Returns:

serialization of this instance as a JSON dict

communication.communication.hinted_tuple_hook(obj)

Object hook for tuples and dictionaries which is used in json.loads

Parameters:

obj (Any) – json object

Return type:

Any

Returns:

tuple containing elements to recreate a collection