Added codecs: huffman, zstd

Author: dhondta

README.md

@@ -338,12 +338,14 @@ This category also contains `ascii85`, `adobe`, `[x]btoa`, `zeromq` with the `ba
 #### [Compression](https://python-codext.readthedocs.io/en/latest/enc/compressions.html)
 
 - [X] `gzip`: standard Gzip compression/decompression
+- [X] `huffman`: canonical Huffman coding of the bytes, with a self-describing header
 - [X] `lz77`: compresses the given data with the algorithm of Lempel and Ziv of 1977
 - [X] `lz78`: compresses the given data with the algorithm of Lempel and Ziv of 1978
 - [X] `pkzip_deflate`: standard Zip-deflate compression/decompression
 - [X] `pkzip_bzip2`: standard BZip2 compression/decompression
 - [X] `pkzip_lzma`: standard LZMA compression/decompression
 - [X] `rle`: compresses runs of identical characters with Run-Length Encoding (count before or after each character, compact, bytes or repeated-byte escape variants)
+- [X] `zstd`: Zstandard compression/decompression (relies on `compression.zstd` from Python 3.14, on `backports.zstd` before)
 
 > :warning: Compression functions are of course definitely **NOT** encoding functions ; they are implemented for leveraging the `.encode(...)` API from `codecs`.
 

docs/pages/enc/compressions.md

@@ -17,6 +17,23 @@
 
 -----
 
+### Huffman
+
+[Huffman coding](https://en.wikipedia.org/wiki/Huffman_coding) gives the frequent bytes shorter codes than the rare ones, with an optimal prefix code. This codec writes the code as a [canonical Huffman code](https://en.wikipedia.org/wiki/Canonical_Huffman_code) (given by the lengths of the codes only), in a self-describing format: the number of distinct bytes minus 1 (1 byte), each byte with the length of its code (2 bytes each), the number of padding bits at the end (1 byte), then the codes of the bytes of the text, from the most significant bit.
+
+**Codec** | **Conversions** | **Aliases** | **Comment**
+:---: | :---: | --- | ---
+`huffman` | data <-> Huffman-coded data |  | malformed data replaced, dropped or left with a non-strict errors handling ; as the length of the text is not stored, data truncated at the end of a code gives a truncated text
+
+```python
+>>> codext.encode(b"abracadabra", "huffman")
+b'\x04a\x01b\x03c\x03d\x03r\x03\x01N\xac\x9c'
+>>> codext.decode(b"\x04a\x01b\x03c\x03d\x03r\x03\x01N\xac\x9c", "huffman")
+b'abracadabra'
+```
+
+-----
+
 ### Lempel-Ziv
 
 This implements the algorithm of Lempel and Ziv of 1977 and 1978.
@@ -106,3 +123,21 @@ This replaces each run of identical consecutive characters by its length followe
 >>> codext.decode("AA\x03B", "rle-escape")
 'AAAAAB'
 ```
+
+-----
+
+### Zstandard
+
+[Zstandard](https://datatracker.ietf.org/doc/html/rfc8878) (zstd) combines LZ77-style matching with entropy coding, in frames starting with a magic number. This codec relies on the `compression.zstd` module of the standard library (Python 3.14 and later) or on its backport, the [`backports.zstd`](https://pypi.org/project/backports.zstd/) package, installed as a dependency for the previous versions of Python ; it is not available without them.
+
+**Codec** | **Conversions** | **Aliases** | **Comment**
+:---: | :---: | --- | ---
+`zstd` | data <-> Zstandard-compressed data | `zstandard`, `zstd-1`, `zstd_19`, ... | parametrized with the compression level, from 1 to 22 (default: 3) ; decoding handles concatenated frames
+
+```python
+>>> data = codext.encode(b"This is a test " * 10, "zstd-19")
+>>> data[:4]
+b'(\xb5/\xfd'
+>>> codext.decode(data, "zstd")
+b'This is a test This is a test This is a test This is a test This is a test This is a test This is a test This is a test This is a test This is a test '
+```

pyproject.toml

@@ -57,6 +57,7 @@ classifiers = [
   "Topic :: Software Development :: Libraries :: Python Modules",
 ]
 dependencies = [
+  "backports.zstd; python_version < '3.14'",
   "legacycrypt; python_version >= '3.13'",
   "markdown2>=2.5.4",
 ]

requirements.txt

@@ -1,2 +1,3 @@
+backports.zstd; python_version < '3.14'
 legacycrypt; python_version >= '3.13'
 markdown2>=2.5.4

src/codext/compressions/__init__.py

@@ -1,9 +1,11 @@
 # -*- coding: UTF-8 -*-
 from .gzipp import *
+from .huffman import *
 from .lz77 import *
 from .lz78 import *
 from .pkzip import *
 from .rle import *
+from .zstd import *
 
 
 for e in list_encodings("compression"):

src/codext/compressions/huffman.py

@@ -0,0 +1,137 @@
+# -*- coding: UTF-8 -*-
+"""Huffman Codec - Huffman coding content compression.
+
+Huffman coding, devised by David A. Huffman in 1952, gives the frequent symbols shorter codes than the rare ones, with
+ an optimal prefix code (no code being the beginning of another) built by merging the two least frequent groups of
+ symbols until a single one remains. This codec builds the code of the bytes of the text and writes it as a canonical
+ Huffman code (the codes being assigned in order of length, then of symbol, hence given by their lengths only), in a
+ self-describing format:
+- the number of distinct bytes minus 1 (1 byte)
+- for each of them, the byte and the length of its code (2 bytes)
+- the number of padding bits at the end of the last byte (1 byte)
+- the codes of the bytes of the text, from the most significant bit (a single distinct byte having the code "0")
+
+An empty text gives an empty output. As with the other compression codecs, malformed data is reported (or, with a
+ non-strict errors handling, replaced, dropped or left as is), but, as the length of the text is not stored, data
+ truncated at the end of a code gives a truncated text.
+
+NB: Not an encoding properly speaking.
+
+This codec:
+- en/decodes strings from str to str
+- en/decodes strings from bytes to bytes
+- decodes file content to str (read)
+- encodes file content from str to bytes (write)
+
+Reference: https://en.wikipedia.org/wiki/Canonical_Huffman_code
+"""
+import heapq
+from collections import Counter
+
+from ..__common__ import *
+
+
+__examples__ = {
+    'enc(huffman)': {'': "", 'a': "\x00a\x01\x07\x00", 'aab': "\x01a\x01b\x01\x05 ",
+                     'abracadabra': "\x04a\x01b\x03c\x03d\x03r\x03\x01N\xac\x9c"},
+    'dec(huffman)': {'\x00a\x01\x07\x00': "a", '\x01a\x01b\x01\x05 ': "aab", '\x01a\x01a\x01\x05 ': None,
+                     '\x01a\x01b\x02\x05 ': None, '\x01a\x01b\x01\x08 ': None, '\x01a\x01b\x01': None,
+                     '\x05': None, '\x01a\x01b\x00\x05 ': None, '\x00a\x02\x07\x00': None},
+    'enc-dec(huffman)': ["", "aaaa", "This is a test", "@random{1,256,2048}"],
+}
+
+
+def _lengths(data):
+    """ Lengths of the Huffman codes of the bytes of the data, merging the least frequent groups (the ties being broken
+         by the order of creation, the bytes being taken in increasing order). """
+    counts = Counter(data)
+    if len(counts) == 1:
+        return {next(iter(counts)): 1}
+    heap, lengths = [(n, i, [s]) for i, (s, n) in enumerate(sorted(counts.items()))], dict.fromkeys(counts, 0)
+    heapq.heapify(heap)
+    i = len(heap)
+    while len(heap) > 1:
+        n1, _, g1 = heapq.heappop(heap)
+        n2, _, g2 = heapq.heappop(heap)
+        for s in g1 + g2:
+            lengths[s] += 1
+        heapq.heappush(heap, (n1 + n2, i, g1 + g2))
+        i += 1
+    return lengths
+
+
+def _canonical(lengths):
+    """ Canonical codes (as (code, length)) of the bytes, assigned in order of length, then of byte. """
+    codes, code, prev = {}, 0, 0
+    for s, n in sorted(lengths.items(), key=lambda x: (x[1], x[0])):
+        code <<= n - prev
+        codes[s], code, prev = (code, n), code + 1, n
+    return codes
+
+
+def huffman_compress(text, errors="strict"):
+    data = b(ensure_str(text))
+    if not data:
+        return b"", len(text)
+    lengths = _lengths(data)
+    codes, header = _canonical(lengths), bytes([len(lengths) - 1])
+    for s in sorted(lengths):
+        header += bytes([s, lengths[s]])
+    acc, nbits = 0, 0
+    for x in data:
+        c, n = codes[x]
+        acc, nbits = acc << n | c, nbits + n
+    pad = -nbits % 8
+    # the compressed data is given as latin-1 text (or bytes), so that it is read back as is
+    return (header + bytes([pad]) + (acc << pad).to_bytes((nbits + pad) // 8, "big")).decode("latin-1"), len(text)
+
+
+def _error(msg):
+    raise HuffmanDecodeError(f"'huffman' codec can't decode: {msg}")
+
+
+def _decode(data):
+    """ Decode Huffman-coded data, raising an error with a precise message if it is malformed. """
+    if not data:
+        return b""
+    n = data[0] + 1
+    if len(data) < 2 + 2 * n:
+        _error("truncated header")
+    lengths = {}
+    for i in range(n):
+        s, length = data[1 + 2 * i], data[2 + 2 * i]
+        if s in lengths or length == 0:
+            _error("bad code length" if length == 0 else f"duplicate byte {s}")
+        lengths[s] = length
+    # the code lengths must give a complete prefix code (Kraft's equality), a single byte having a code of 1 bit
+    m = max(lengths.values())
+    if (sum(1 << (m - length) for length in lengths.values()) != 1 << m) if n > 1 else m != 1:
+        _error("code lengths not giving a complete prefix code")
+    pad, payload = data[1 + 2 * n], data[2 + 2 * n:]
+    if pad > 7 or not payload or 8 * len(payload) - pad < min(lengths.values()):
+        _error("bad padding")
+    table = {(c, l): s for s, (c, l) in _canonical(lengths).items()}
+    bits, r, code, length = int.from_bytes(payload, "big") >> pad, bytearray(), 0, 0
+    # as the code is complete, any sequence of bits is a sequence of codes, possibly followed by a truncated one
+    for i in range(8 * len(payload) - pad - 1, -1, -1):
+        code, length = code << 1 | bits >> i & 1, length + 1
+        if (code, length) in table:
+            r.append(table[code, length])
+            code, length = 0, 0
+    if length:
+        _error("truncated code at the end")
+    return bytes(r)
+
+
+def huffman_decompress(text, errors="strict"):
+    data = b(ensure_str(text))
+    try:
+        return _decode(data), len(text)
+    except HuffmanDecodeError:
+        # as with gzip and zstd, malformed data is reported (with a precise message), or replaced, dropped or left as is
+        if errors == "strict":
+            raise
+        return handle_error("huffman", errors, decode=True)(data[:1].decode("latin-1"), 0), len(text)
+
+
+add("huffman", huffman_compress, huffman_decompress, guess=None)

src/codext/compressions/zstd.py

@@ -0,0 +1,60 @@
+# -*- coding: UTF-8 -*-
+"""Zstandard Codec - zstd content compression.
+
+Zstandard (zstd), created by Yann Collet at Facebook in 2015 and specified in RFC 8878, combines LZ77-style matching
+ with entropy coding (Huffman coding for the literals and finite-state entropy for the sequences), in frames with a
+ magic number (0xFD2FB528). This codec relies on the compression.zstd module of the standard library (Python 3.14 and
+ later) or on its backport (the backports.zstd package, installed as a dependency for the previous versions of Python),
+ and is not available without them. The compression level is given with the codec name (from 1 to 22, e.g. "zstd-19")
+ and defaults to 3 ; decoding handles concatenated frames.
+
+NB: Not an encoding properly speaking.
+
+This codec:
+- en/decodes strings from str to str
+- en/decodes strings from bytes to bytes
+- decodes file content to str (read)
+- encodes file content from str to bytes (write)
+
+Reference: https://datatracker.ietf.org/doc/html/rfc8878
+"""
+from ..__common__ import *
+
+try:
+    from compression import zstd
+except ImportError:
+    try:
+        from backports import zstd
+    except ImportError:
+        zstd = None
+
+
+__examples__ = {
+    'dec(zstd)': {'(\xb5/\xfd \x04!\x00\x00test': "test", '(\xb5/\xfd \x04!\x00\x00tes': None, 'test': None},
+    'enc-dec(zstd|zstandard|zstd-1|zstd_19|zstd-22)': ["", "test", "This is a test", "@random{512,1024,2048}"],
+}
+
+
+def zstd_compress(level=None):
+    def compress(text, errors="strict"):
+        lvl = level or 3
+        if lvl not in range(1, 23):
+            raise LookupError("Bad parameter for encoding 'zstd': compression level must be between 1 and 22")
+        # the compressed data is given as latin-1 text (or bytes), so that it is read back as is
+        return zstd.compress(b(ensure_str(text)), level=lvl).decode("latin-1"), len(text)
+    return compress
+
+
+def zstd_decompress(level=None):
+    def decompress(text, errors="strict"):
+        data = b(ensure_str(text))
+        try:
+            r = zstd.decompress(data)
+        except zstd.ZstdError:
+            return handle_error("zstd", errors, decode=True)(data[:1].decode("latin-1"), 0), len(text)
+        return r, len(text)
+    return decompress
+
+
+if zstd is not None:
+    add("zstd", zstd_compress, zstd_decompress, r"^(?:zstd|zstandard)(?:[-_](\d{1,2}))?$", examples=__examples__)

tests/test_manual.py

@@ -768,6 +768,50 @@ def test_codec_dummy_str_manips(self):
         self.assertEqual(codecs.encode(STR, "tokenize-2"), "th is  i s  a  te st")
         self.assertRaises(LookupError, codecs.encode, STR, "tokenize-200")
 
+    def test_codec_huffman(self):
+        import heapq
+        from collections import Counter
+        # optimal codes: the length of the codes is the sum of the weights of the merged groups
+        data = b"The quick brown fox jumps over the lazy dog " * 20
+        enc = codecs.encode(data, "huffman")
+        weights, total = list(Counter(data).values()), 0
+        heapq.heapify(weights)
+        while len(weights) > 1:
+            w = heapq.heappop(weights) + heapq.heappop(weights)
+            total += w
+            heapq.heappush(weights, w)
+        n = enc[0] + 1
+        self.assertEqual(8 * (len(enc) - 2 - 2 * n) - enc[1 + 2 * n], total)
+        # text with characters outside latin-1, and malformed data (truncated header, null code length, extra bits,
+        #  missing payload) reported with a precise message, or replaced, dropped or left as is (as with gzip and zstd)
+        self.assertEqual(codecs.decode(codecs.encode("caf\u00e9 \u20ac", "huffman"), "huffman"), "caf\u00e9 \u20ac")
+        for bad, msg in [(b"\x01a", "truncated header"), (b"\x01a\x00b\x01\x05 ", "bad code length"),
+                         (enc + b"\xff", "truncated code at the end"), (enc[:2 + 2 * n], "bad padding")]:
+            self.assertRaisesRegex(ValueError, msg, codecs.decode, bad, "huffman")
+            self.assertEqual(codecs.decode(bad, "huffman", "replace"), b"?")
+            self.assertEqual(codecs.decode(bad, "huffman", "ignore"), b"")
+            self.assertEqual(codecs.decode(bad, "huffman", "leave"), bad[:1])
+        self.assertRaises(ValueError, codecs.decode, b"\x01a", "huffman", "BAD_ERRORS")
+    
+    def test_codec_zstd(self):
+        try:
+            from compression import zstd
+        except ImportError:
+            try:
+                from backports import zstd
+            except ImportError:
+                self.skipTest("neither compression.zstd (Python 3.14+) nor backports.zstd is available")
+        # a frame with a raw block (magic number, frame header with the content size, block header), levels and errors
+        self.assertEqual(codecs.decode(b"\x28\xb5\x2f\xfd\x20\x04\x21\x00\x00test", "zstd"), b"test")
+        data = b"This is a test " * 100
+        self.assertLess(len(codecs.encode(data, "zstd-22")), len(data) // 10)
+        self.assertEqual(zstd.decompress(codecs.encode(data, "zstd-19")), data)
+        for e in ["zstd-0", "zstd-23"]:
+            self.assertRaises(LookupError, codecs.encode, data, e)
+        self.assertRaises(ValueError, codecs.decode, b"not zstd", "zstd")
+        self.assertEqual(codecs.decode(b"not zstd", "zstd", "ignore"), b"")
+        self.assertEqual(codecs.decode(b"not zstd", "zstd", "replace"), b"?")
+    
     def test_codec_hash_functions(self):
         STR = b"This is a test string!"
         for h in ["adler32", "md2", "md5", "sha1", "sha224", "sha256", "sha384", "sha512"]: