"""Layer 2 — the PACK codec (LZW with LRU slot recycling; spec §6.2, §8.1).
This is *not* the classic growing-code-width LZW found in generic
compression libraries: codes are always 12 bits, the dictionary starts
full (pre-seeded so every phrase slot expands to EBCDIC spaces) and slots
are recycled with a least-recently-used policy instead of resetting.
Existing LZW packages therefore cannot be reused.
The decoder follows spec §6.2 exactly. The encoder (spec §8.1) drives an
identical copy of the decoder's dictionary machinery with each code it
emits, so its model stays bit-identical to what the decoder will rebuild.
"""
from __future__ import annotations
import logging
from typing import Callable, Iterable, List
from .constants import CODESIZE, LZW_SEED, TREESIZE
from .exceptions import TerseCorruptError
logger = logging.getLogger(__name__)
def _init_arrays() -> "tuple[list[int], list[int], list[int], list[int]]":
"""Build the initial ``Father``/``CharExt``/``Forward``/``Backward``
arrays with the dictionary pre-seeded full (spec §6.2)."""
father = [0] * TREESIZE
charext = [0] * TREESIZE
fwd = [0] * TREESIZE
bwd = [0] * TREESIZE
h2 = LZW_SEED
for c in range(CODESIZE + 1, TREESIZE):
father[c] = h2
charext[c] = LZW_SEED
h2 = c
for c in range(CODESIZE + 1, TREESIZE - 1):
bwd[c + 1] = c
fwd[c] = c + 1
bwd[0] = TREESIZE - 1
fwd[0] = CODESIZE + 1
bwd[CODESIZE + 1] = 0
fwd[TREESIZE - 1] = 0
return father, charext, fwd, bwd
[docs]
def decode(get_code: Callable[[], int],
putchar: Callable[[int], None]) -> None:
"""Decode a PACK (LZW) code stream (spec §6.2).
:param get_code: Zero-argument callable returning successive 12-bit
codes; ``0`` terminates (see
:func:`terselib.bitpack.make_unpacker`).
:param putchar: Callable receiving each output codepoint ``0..257``
(see :class:`terselib.records.RecordSink`).
:raises TerseCorruptError: If a dictionary chain is longer than the
dictionary itself (a cycle caused by corrupt input).
"""
father, charext, fwd, bwd = _init_arrays()
x = 0
d = get_code()
while d != 0:
# (1) detach the LRU victim y to become this step's new phrase slot
y = bwd[0]
q = bwd[y]
bwd[0] = q
fwd[q] = 0
h = y
p = 0
# (2) decompose code d, reversing its Father chain into p
steps = 0
while d > CODESIZE:
q = fwd[d]
r = bwd[d]
fwd[r] = q
bwd[q] = r
fwd[d] = h
bwd[h] = d
h = d
e = father[d]
father[d] = p
p = d
d = e
steps += 1
if steps > TREESIZE:
raise TerseCorruptError(
"dictionary chain cycle: corrupt PACK stream")
# (3) re-insert y at the front of the LRU list
q = fwd[0]
fwd[y] = q
bwd[q] = y
fwd[0] = h
bwd[h] = 0
# (4) complete the previous new phrase, then output the leading atom
charext[x] = d
putchar(d)
x = y
# (5) output the rest of phrase d, restoring Father links
while p != 0:
e = father[p]
putchar(charext[p])
father[p] = d
d = p
p = e
# (6) the new phrase slot's prefix is the whole current code
father[y] = d
d = get_code()
[docs]
def encode(codepoints: Iterable[int]) -> List[int]:
"""Encode a codepoint stream with the PACK (LZW) method (spec §8.1).
Greedy longest-match selection over a dictionary model advanced in
lockstep with the decoder. Two rules keep the emitted stream safe for
the reference decoder:
* a match is never extended into the slot that is currently the LRU
victim (emitting the victim would make the decoder corrupt its LRU
list while recycling that same slot), and
* a newly learned phrase only becomes matchable once its extension
codepoint is fixed, i.e. from the following emission on.
:param codepoints: The Layer-1 codepoint stream (values ``1..257``).
:returns: The list of 12-bit codes **including** the terminating ``0``.
"""
father, charext, fwd, bwd = _init_arrays()
find_child: "dict[int, int]" = {} # (code << 9) | ext -> slot
registered: "dict[int, int]" = {} # slot -> its key in find_child
pending_slot = -1 # slot awaiting its extension
pending_code = 0 # code emitted when it was created
out: List[int] = []
emit = out.append
def model_step(d: int) -> None:
"""Advance the shared dictionary model exactly as the decoder
does upon reading code *d* (spec §6.2 loop body, sans output)."""
nonlocal pending_slot, pending_code
victim = bwd[0]
key = registered.pop(victim, None)
if key is not None and find_child.get(key) == victim:
del find_child[key]
# -- steps (1)..(6) of the decode loop --
y = victim
q = bwd[y]
bwd[0] = q
fwd[q] = 0
h = y
p = 0
dd = d
while dd > CODESIZE:
q = fwd[dd]
r = bwd[dd]
fwd[r] = q
bwd[q] = r
fwd[dd] = h
bwd[h] = dd
h = dd
e = father[dd]
father[dd] = p
p = dd
dd = e
q = fwd[0]
fwd[y] = q
bwd[q] = y
fwd[0] = h
bwd[h] = 0
# leading atom of d is dd; it completes the pending phrase
if pending_slot >= 0:
charext[pending_slot] = dd
k2 = (pending_code << 9) | dd
find_child[k2] = pending_slot
registered[pending_slot] = k2
else:
charext[0] = dd # decoder writes to sentinel slot 0; mirror it
while p != 0:
e = father[p]
father[p] = dd
dd = p
p = e
father[y] = dd
pending_slot = y
pending_code = d
emit(d)
it = iter(codepoints)
w = next(it, None)
if w is None:
emit(0)
return out
for k in it:
child = find_child.get((w << 9) | k)
if child is not None and child != bwd[0]:
w = child
continue
model_step(w)
w = k
model_step(w)
emit(0)
logger.debug("LZW encode: %d codes emitted", len(out))
return out