msrtc-rans-python 0.2.3

Python extension module for msrtc_rans (published as wheel, not crate)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.
# Modified: import _msrtc_rans from top-level (maturin creates a package, not a bare .so)

import sys

import _msrtc_rans
from .types import RansVariant

__all__ = ["RansVariant", "RansEncoderStream", "RansDecoderStream", "EntropyEncoder", "EntropyDecoder"]

RansEncoderStream = _msrtc_rans.RansEncoderStream
RansDecoderStream = _msrtc_rans.RansDecoderStream


class EntropyEncoder:
    """
    Entropy encoder
    """

    __slots__ = "_impl", "_variant"

    def __init__(
        self,
        *,
        pmfLengths,
        pmfOffsets,
        pmfTable,
        variant: RansVariant = RansVariant.RansByte,
        symbolBits=16,
        bypassBits=4,
    ):
        """
        Initialize entropy encoder

        :param pmfLengths: PMF table lengths (int32 1d-array)
        :param pmfOffsets: PMF table zero element offsets (int32 1d-array)
        :param pmfTable: concatenated PMF tables (int32 1d-array)
        :param variant: rANS variant to use (default RansByte)
        :param symbolBits: symbol bits (default 16)
        :param bypassBits: bypass bits (default 4)
        """
        if symbolBits <= 1 or symbolBits > sys.maxsize:
            raise ValueError("invalid symbolBits value")
        if bypassBits <= 1 or bypassBits > sys.maxsize:
            raise ValueError("invalid bypassBits value")

        self._impl = _msrtc_rans.EntropyEncoder(
            pmfLengths=pmfLengths,
            pmfOffsets=pmfOffsets,
            pmfTable=pmfTable,
            variant=variant,
            symbolBits=symbolBits,
            bypassBits=bypassBits,
        )
        self._variant = variant

    @property
    def variant(self):
        return self._variant

    def encode(self, indices, values):
        """
        Encode message

        :param indices: PMF table indices (int32 1d-array)
        :param values: values (int32 1d-array)
        :return: entropy encoded message
        """
        stream = RansEncoderStream(self._variant)
        self._impl.encode(stream, indices, values)
        return stream.flush()

    def push(self, stream: RansEncoderStream, indices, values) -> None:
        """
        Push message into encoder stream

        :param stream: rANS encoder stream
        :param indices: PMF table indices (int32 1d-array)
        :param values: values (int32 1d-array)
        """
        return self._impl.encode(stream, indices, values)

    def __copy__(self):
        return self

    def __deepcopy__(self, memo):
        _ = memo
        return self.__copy__()


class EntropyDecoder:
    """
    Entropy Decoder
    """

    def __init__(
        self,
        *,
        pmfLengths,
        pmfOffsets,
        pmfTable,
        variant: RansVariant = RansVariant.RansByte,
        symbolBits=16,
        bypassBits=4,
    ):
        """
        Initialize entropy decoder

        :param pmfLengths: PMF table lengths (int32 1d-array)
        :param pmfOffsets: PMF table zero element offsets (int32 1d-array)
        :param pmfTable: concatenated PMF tables (int32 1d-array)
        :param variant: rANS variant to use (default RansByte)
        :param symbolBits: symbol bits (default 16)
        :param bypassBits: bypass bits (default 4)
        """
        if symbolBits <= 1 or symbolBits > sys.maxsize:
            raise ValueError("invalid symbolBits value")
        if bypassBits <= 1 or bypassBits > sys.maxsize:
            raise ValueError("invalid bypassBits value")

        self._impl = _msrtc_rans.EntropyDecoder(
            pmfLengths=pmfLengths,
            pmfOffsets=pmfOffsets,
            pmfTable=pmfTable,
            variant=variant,
            symbolBits=symbolBits,
            bypassBits=bypassBits,
        )

    def decode(self, values, indices, stream) -> None:
        """
        Decode message

        :param values: decoded values (output int32 1d-array)
        :param indices: PMF table indices (int32 1d-array)
        :param stream: message data (any buffer) or RansDecoderStream
        """
        return self._impl.decode(values, indices, stream)

    def __copy__(self):
        return self

    def __deepcopy__(self, memo):
        _ = memo
        return self.__copy__()