hat.sbs

Simple binary serializer

This implementation of SBS encoder/decoder translates between SBS types and Python types according to following translation table::

+----------+------------------+
| SBS type | Python type      |
+==========+==================+
| None     | NoneType         |
+----------+------------------+
| Boolean  | bool             |
+----------+------------------+
| Integer  | int              |
+----------+------------------+
| Float    | float            |
+----------+------------------+
| String   | str              |
+----------+------------------+
| Bytes    | bytes            |
+----------+------------------+
| Array    | list[Data]       |
+----------+------------------+
| Record   | dict[str, Data]  |
+----------+------------------+
| Choice   | tuple[str, Data] |
+----------+------------------+

Example usage of SBS serializer::

import hat.sbs

repo = hat.sbs.Repository('''
    module Module

    Entry(K, V) = Record {
        key: K
        value: V
    }

    T = Array(Optional(Entry(String, Integer)))
''')
data = [
    ('none', None),
    ('value', {
        'key': 'abc',
        'value': 123
    })
]
encoded_data = repo.encode('Module.T', data)
decoded_data = repo.decode('Module.T', encoded_data)
assert data == decoded_data

View Source

 1"""Simple binary serializer
 2
 3This implementation of SBS encoder/decoder translates between SBS types and
 4Python types according to following translation table::
 5
 6    +----------+------------------+
 7    | SBS type | Python type      |
 8    +==========+==================+
 9    | None     | NoneType         |
10    +----------+------------------+
11    | Boolean  | bool             |
12    +----------+------------------+
13    | Integer  | int              |
14    +----------+------------------+
15    | Float    | float            |
16    +----------+------------------+
17    | String   | str              |
18    +----------+------------------+
19    | Bytes    | bytes            |
20    +----------+------------------+
21    | Array    | list[Data]       |
22    +----------+------------------+
23    | Record   | dict[str, Data]  |
24    +----------+------------------+
25    | Choice   | tuple[str, Data] |
26    +----------+------------------+
27
28Example usage of SBS serializer::
29
30    import hat.sbs
31
32    repo = hat.sbs.Repository('''
33        module Module
34
35        Entry(K, V) = Record {
36            key: K
37            value: V
38        }
39
40        T = Array(Optional(Entry(String, Integer)))
41    ''')
42    data = [
43        ('none', None),
44        ('value', {
45            'key': 'abc',
46            'value': 123
47        })
48    ]
49    encoded_data = repo.encode('Module.T', data)
50    decoded_data = repo.decode('Module.T', encoded_data)
51    assert data == decoded_data
52
53"""
54
55from hat.sbs.common import Data
56from hat.sbs.repository import Repository
57from hat.sbs.serializer import (Serializer,
58                                CSerializer,
59                                PySerializer,
60                                DefaultSerializer)
61
62
63__all__ = ['Repository',
64           'Data',
65           'Serializer',
66           'CSerializer',
67           'PySerializer',
68           'DefaultSerializer']

class Repository: View Source

14class Repository:
15    """SBS schema repository.
16
17    Supported initialization arguments:
18        * string containing sbs schema
19        * file path to .sbs file
20        * path to direcory recursivly searched for .sbs files
21        * other repository
22
23    """
24
25    def __init__(self,
26                 *args: typing.Union['Repository', pathlib.Path, str]):
27        self._modules = list(_parse_args(args))
28        self._refs = evaluator.evaluate_modules(self._modules)
29
30    def encode(self,
31               name: str,
32               value: common.Data, *,
33               serializer: type[Serializer] = DefaultSerializer
34               ) -> util.Bytes:
35        """Encode value."""
36        ref = _parse_name(name)
37        return serializer.encode(self._refs, ref, value)
38
39    def decode(self,
40               name: str,
41               data: util.Bytes, *,
42               serializer: type[Serializer] = DefaultSerializer
43               ) -> common.Data:
44        """Decode data."""
45        ref = _parse_name(name)
46        return serializer.decode(self._refs, ref, data)
47
48    def to_json(self) -> json.Data:
49        """Export repository content as json serializable data.
50
51        Entire repository content is exported as json serializable data.
52        New repository can be created from the exported content by using
53        :meth:`Repository.from_json`.
54
55        """
56        return [parser.module_to_json(module) for module in self._modules]
57
58    @staticmethod
59    def from_json(data: pathlib.PurePath | json.Data) -> 'Repository':
60        """Create new repository from content exported as json serializable
61        data.
62
63        Creates a new repository from content of another repository that was
64        exported by using :meth:`Repository.to_json`.
65
66        """
67        if isinstance(data, pathlib.PurePath):
68            data = json.decode_file(data)
69
70        repo = Repository()
71        repo._modules = [parser.module_from_json(i) for i in data]
72        repo._refs = evaluator.evaluate_modules(repo._modules)
73        return repo

SBS schema repository.

Supported initialization arguments:

string containing sbs schema

file path to .sbs file

path to direcory recursivly searched for .sbs files

other repository

Repository(*args: Union[Repository, pathlib.Path, str]) View Source

25    def __init__(self,
26                 *args: typing.Union['Repository', pathlib.Path, str]):
27        self._modules = list(_parse_args(args))
28        self._refs = evaluator.evaluate_modules(self._modules)

def encode( self, name: str, value: Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]], *, serializer: type[Serializer] = <class 'CSerializer'>) -> bytes | bytearray | memoryview: View Source

30    def encode(self,
31               name: str,
32               value: common.Data, *,
33               serializer: type[Serializer] = DefaultSerializer
34               ) -> util.Bytes:
35        """Encode value."""
36        ref = _parse_name(name)
37        return serializer.encode(self._refs, ref, value)

Encode value.

def decode( self, name: str, data: bytes | bytearray | memoryview, *, serializer: type[Serializer] = <class 'CSerializer'>) -> Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]: View Source

39    def decode(self,
40               name: str,
41               data: util.Bytes, *,
42               serializer: type[Serializer] = DefaultSerializer
43               ) -> common.Data:
44        """Decode data."""
45        ref = _parse_name(name)
46        return serializer.decode(self._refs, ref, data)

Decode data.

48    def to_json(self) -> json.Data:
49        """Export repository content as json serializable data.
50
51        Entire repository content is exported as json serializable data.
52        New repository can be created from the exported content by using
53        :meth:`Repository.from_json`.
54
55        """
56        return [parser.module_to_json(module) for module in self._modules]

Export repository content as json serializable data.

Entire repository content is exported as json serializable data. New repository can be created from the exported content by using Repository.from_json().

@staticmethod

58    @staticmethod
59    def from_json(data: pathlib.PurePath | json.Data) -> 'Repository':
60        """Create new repository from content exported as json serializable
61        data.
62
63        Creates a new repository from content of another repository that was
64        exported by using :meth:`Repository.to_json`.
65
66        """
67        if isinstance(data, pathlib.PurePath):
68            data = json.decode_file(data)
69
70        repo = Repository()
71        repo._modules = [parser.module_from_json(i) for i in data]
72        repo._refs = evaluator.evaluate_modules(repo._modules)
73        return repo

Create new repository from content exported as json serializable data.

Creates a new repository from content of another repository that was exported by using Repository.to_json().

Data = typing.Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, typing.List[ForwardRef('Data')], typing.Dict[str, ForwardRef('Data')], typing.Tuple[str, ForwardRef('Data')]]

class Serializer(abc.ABC): View Source

11class Serializer(abc.ABC):
12
13    @staticmethod
14    @abc.abstractmethod
15    def encode(refs: dict[Ref, Type],
16               t: Type,
17               value: Data
18               ) -> util.Bytes:
19        """Encode value"""
20
21    @staticmethod
22    @abc.abstractmethod
23    def decode(refs: dict[Ref, Type],
24               t: Type,
25               data: util.Bytes
26               ) -> Data:
27        """Decode data"""

Helper class that provides a standard way to create an ABC using inheritance.

@staticmethod

@abc.abstractmethod

def encode( refs: dict[hat.sbs.common.Ref, hat.sbs.common.Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType], t: hat.sbs.common.Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType, value: Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Dict[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Tuple[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]]]) -> bytes | bytearray | memoryview: View Source

13    @staticmethod
14    @abc.abstractmethod
15    def encode(refs: dict[Ref, Type],
16               t: Type,
17               value: Data
18               ) -> util.Bytes:
19        """Encode value"""

Encode value

@staticmethod

@abc.abstractmethod

def decode( refs: dict[hat.sbs.common.Ref, hat.sbs.common.Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType], t: hat.sbs.common.Ref | hat.sbs.common.NoneType | hat.sbs.common.BooleanType | hat.sbs.common.IntegerType | hat.sbs.common.FloatType | hat.sbs.common.StringType | hat.sbs.common.BytesType | hat.sbs.common.ArrayType | hat.sbs.common.RecordType | hat.sbs.common.ChoiceType, data: bytes | bytearray | memoryview) -> Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Dict[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]], Tuple[str, Union[NoneType, bool, int, float, str, bytes, bytearray, memoryview, List[ForwardRef('Data')], Dict[str, ForwardRef('Data')], Tuple[str, ForwardRef('Data')]]]]: View Source

21    @staticmethod
22    @abc.abstractmethod
23    def decode(refs: dict[Ref, Type],
24               t: Type,
25               data: util.Bytes
26               ) -> Data:
27        """Decode data"""

Decode data

class CSerializer(hat.sbs.Serializer): View Source

11class CSerializer(common.Serializer):
12    """Serializer implementation in C"""
13
14    def encode(refs, t, value):
15        if not _cserializer:
16            raise Exception('implementation not available')
17
18        return _cserializer.encode(refs, t, value)
19
20    def decode(refs, t, data):
21        if not _cserializer:
22            raise Exception('implementation not available')
23
24        return _cserializer.decode(refs, t, memoryview(data))

Serializer implementation in C

def encode(refs, t, value): View Source

14    def encode(refs, t, value):
15        if not _cserializer:
16            raise Exception('implementation not available')
17
18        return _cserializer.encode(refs, t, value)

Encode value

def decode(refs, t, data): View Source

20    def decode(refs, t, data):
21        if not _cserializer:
22            raise Exception('implementation not available')
23
24        return _cserializer.decode(refs, t, memoryview(data))

Decode data

class PySerializer(hat.sbs.Serializer): View Source

 8class PySerializer(common.Serializer):
 9    """Serializer implementation in Python"""
10
11    def encode(refs, t, value):
12        return bytes(_encode_generic(refs, t, value))
13
14    def decode(refs, t, data):
15        value, _ = _decode_generic(refs, t, memoryview(data))
16        return value

Serializer implementation in Python

def encode(refs, t, value): View Source

11    def encode(refs, t, value):
12        return bytes(_encode_generic(refs, t, value))

Encode value

def decode(refs, t, data): View Source

14    def decode(refs, t, data):
15        value, _ = _decode_generic(refs, t, memoryview(data))
16        return value

Decode data

DefaultSerializer = <class 'CSerializer'>