"""Layer 2 — the SPACK codec (LZMW; spec §6.3, §8.2).
LZMW grows its dictionary by concatenating the two most recently emitted
phrases, so phrases can double in length each step. Phrases are stored as
binary-tree nodes (``Left``/``Right``); reference counting keeps a phrase
alive while it is a child of another phrase, and unreferenced phrases are
recycled least-recently-used once all 3838 slots are in use.
No reusable LZMW implementation exists on PyPI, and the slot / LRU /
refcount discipline must match the reference decoder bit-for-bit anyway,
so the codec is implemented here from the specification.
The encoder (spec §8.2) drives an identical copy of the decoder's
dictionary machinery and keeps a phrase trie for greedy longest-match
lookup.
"""
from __future__ import annotations
import logging
from typing import Callable, List, Sequence
from .constants import BASE, CODESIZE, NONE_NODE, STACKSIZE, TREESIZE
from .exceptions import TerseCorruptError
logger = logging.getLogger(__name__)
class _Tree:
"""The shared SPACK dictionary machinery (spec §6.3 ``TreeInit``,
``GetTreeNode``, ``LruAdd``, ``BumpRef``, ``DeleteRef``, ``LruKill``).
Used directly by the decoder and, unchanged, as the encoder's model of
the decoder — which guarantees the synchronization invariant.
"""
def __init__(self) -> None:
size = TREESIZE + 1
self.left = [NONE_NODE] * size
self.right = [NONE_NODE] * size
self.back = [0] * size
self.next = [0] * size
for i in range(CODESIZE + 1): # atoms 0..257
self.right[i] = i
for i in range(CODESIZE + 1, TREESIZE): # free list 258..4095
self.next[i] = i + 1
self.next[TREESIZE] = NONE_NODE
self.next[BASE] = BASE # empty LRU list
self.back[BASE] = BASE
for i in range(1, CODESIZE + 1):
self.next[i] = NONE_NODE
self.avail = CODESIZE + 1
self.next[TREESIZE - 1] = NONE_NODE
def get_node(self) -> int:
"""Pop and return a slot from the free list."""
node = self.avail
self.avail = self.next[node]
return node
def lru_add(self, n: int) -> None:
"""Append slot *n* at the most-recently-used end of the LRU list."""
b = self.back[BASE]
self.next[n] = BASE
self.back[BASE] = n
self.back[n] = b
self.next[b] = n
def bump_ref(self, c: int) -> None:
"""Add one reference to child *c* (removing it from the LRU list
on its first reference)."""
if self.next[c] < 0:
self.next[c] -= 1
else:
f = self.next[c]
p = self.back[c]
self.next[p] = f
self.back[f] = p
self.next[c] = -1
def delete_ref(self, c: int) -> None:
"""Drop one reference from child *c* (returning it to the LRU list
when the last reference goes away)."""
if self.next[c] == -1:
self.lru_add(c)
else:
self.next[c] += 1
def lru_kill(self) -> int:
"""Evict the least-recently-used unreferenced phrase and return
the freed slot number."""
p = self.next[BASE]
q = self.next[p]
r = self.back[p]
self.back[q] = r
self.next[r] = q
self.delete_ref(self.left[p])
self.delete_ref(self.right[p])
self.next[p] = self.avail
self.avail = p
return p
[docs]
def decode(get_code: Callable[[], int],
putchar: Callable[[int], None]) -> None:
"""Decode a SPACK (LZMW) code stream (spec §6.3).
:param get_code: Zero-argument callable returning successive 12-bit
codes; ``0`` terminates.
:param putchar: Callable receiving each output codepoint ``0..257``.
:raises TerseCorruptError: If expansion meets a negative child or the
expansion stack exceeds :data:`~terselib.constants.STACKSIZE`.
"""
tree = _Tree()
left = tree.left
right = tree.right
stack: List[int] = []
def put_chars(x: int) -> None:
# Emit all leaves of x's subtree, left to right (spec PutChars).
while True:
while x > CODESIZE:
stack.append(right[x])
if len(stack) > STACKSIZE:
raise TerseCorruptError(
"expansion stack overflow: corrupt SPACK stream")
x = left[x]
if x < 0:
raise TerseCorruptError(
"negative codepoint: corrupt SPACK stream")
putchar(x)
if stack:
x = stack.pop()
else:
return
h = get_code()
if h == 0:
return # empty payload -> empty output
put_chars(h)
g = get_code()
while g != 0:
if tree.avail == NONE_NODE:
tree.lru_kill()
put_chars(g)
n = tree.get_node()
left[n] = h
right[n] = g
tree.bump_ref(h)
tree.bump_ref(g)
tree.lru_add(n)
h = g
g = get_code()
[docs]
def encode(codepoints: Sequence[int]) -> List[int]:
"""Encode a codepoint stream with the SPACK (LZMW) method (spec §8.2).
Greedy longest-match selection using a trie of the phrases currently
live in the shared dictionary model. When the dictionary is full, the
phrase that is the current LRU victim is never emitted: the decoder
evicts that slot *before* expanding the incoming code, so emitting it
would make the decoder recycle the very slot it is about to expand.
:param codepoints: The Layer-1 codepoint stream (values ``1..257``),
indexable (e.g. ``array('H')``).
:returns: The list of 12-bit codes **including** the terminating ``0``.
"""
tree = _Tree()
seqs: "dict[int, tuple]" = {} # slot -> its codepoint sequence
trie: dict = {} # nested {cp: node}; node[0] = set(slots)
def seq_of(code: int) -> tuple:
return (code,) if code <= CODESIZE else seqs[code]
def trie_add(seq: tuple, code: int) -> None:
node = trie
for cp in seq:
node = node.setdefault(cp, {})
node.setdefault(0, set()).add(code)
def trie_remove(seq: tuple, code: int) -> None:
path = []
node = trie
for cp in seq:
path.append((node, cp))
node = node[cp]
terminals = node[0]
terminals.discard(code)
if not terminals:
del node[0]
for parent, cp in reversed(path):
if parent[cp]:
break
del parent[cp]
n = len(codepoints)
out: List[int] = []
emit = out.append
if n == 0:
emit(0)
return out
def longest_match(pos: int, victim: int) -> "tuple[int, int]":
"""Longest live phrase matching ``codepoints[pos:]``; falls back
to the single-atom match. *victim* (or -1) is never returned."""
best_code = codepoints[pos]
best_len = 1
node = trie
i = pos
while i < n:
node = node.get(codepoints[i])
if node is None:
break
i += 1
terminals = node.get(0)
if terminals:
for c in terminals:
if c != victim:
best_code = c
best_len = i - pos
break
return best_code, best_len
h, mlen = longest_match(0, -1)
emit(h)
pos = mlen
left = tree.left
right = tree.right
while pos < n:
full = tree.avail == NONE_NODE
g, mlen = longest_match(pos, tree.next[BASE] if full else -1)
emit(g)
pos += mlen
if full:
evicted = tree.lru_kill()
trie_remove(seqs.pop(evicted), evicted)
node = tree.get_node()
left[node] = h
right[node] = g
tree.bump_ref(h)
tree.bump_ref(g)
tree.lru_add(node)
new_seq = seq_of(h) + seq_of(g)
seqs[node] = new_seq
trie_add(new_seq, node)
h = g
emit(0)
logger.debug("LZMW encode: %d codes emitted", len(out))
return out