"""Layer 3 — packing and unpacking of 12-bit codes (spec §3).
The compressed payload of a TERSE file is a stream of 12-bit codes packed
MSB-first into bytes; two codes occupy exactly three bytes. The stream is
terminated by the explicit code ``0``, after which host files are padded
with EBCDIC spaces to a 1024-byte block boundary.
"""
from __future__ import annotations
from typing import Callable, Iterable
from .constants import ENDOFFILE, HOST_PAD_BLOCK, HOST_PAD_BYTE
from .exceptions import TerseCorruptError
[docs]
class CodePacker:
"""Packs 12-bit codes into bytes, MSB-first (spec §3.1).
:param out: A growable :class:`bytearray` the packed bytes are
appended to.
"""
def __init__(self, out: bytearray) -> None:
self._out = out
self._pending = -1 # pending 4-bit nibble, -1 = aligned
[docs]
def put(self, code: int) -> None:
"""Append one 12-bit code (``0..4095``) to the output."""
if self._pending < 0:
self._out.append((code >> 4) & 0xFF)
self._pending = code & 0x0F
else:
self._out.append((self._pending << 4) | ((code >> 8) & 0x0F))
self._out.append(code & 0xFF)
self._pending = -1
[docs]
def flush(self) -> None:
"""Flush a trailing half-written code, if any.
If a 4-bit nibble is pending it is emitted in the high half of one
final byte with the low four bits zero.
"""
if self._pending >= 0:
self._out.append(self._pending << 4)
self._pending = -1
[docs]
def pack_codes(codes: Iterable[int]) -> bytearray:
"""Pack an iterable of 12-bit codes into a payload byte string.
The caller must include the terminating ``0`` code in *codes*, and —
for host files — apply the 1024-byte block padding to the complete
file (header included) afterwards, e.g. via :func:`pad_file_to_block`.
:param codes: 12-bit codes ``0..4095``, ending with the terminator.
:returns: The packed payload bytes (bit-flushed, unpadded).
"""
out = bytearray()
packer = CodePacker(out)
for c in codes:
packer.put(c)
packer.flush()
return out
[docs]
def pad_file_to_block(file_image: bytearray,
block: int = HOST_PAD_BLOCK,
pad_byte: int = HOST_PAD_BYTE) -> None:
"""Pad a complete host file image to a 1024-byte boundary (spec §3.2).
The padding counts the *whole* file, header included: AMATERSE writes
host files whose total size is a multiple of 1024 bytes.
:param file_image: The header + packed payload, modified in place.
:param block: Block boundary, default 1024.
:param pad_byte: Filler byte; AMATERSE writes ``0x00`` (the default).
"""
remainder = len(file_image) % block
if remainder:
file_image.extend(bytes([pad_byte]) * (block - remainder))
[docs]
def make_unpacker(data: bytes, start: int = 0) -> Callable[[], int]:
"""Build a ``GetBlok`` function over *data* (spec §3.3).
The returned zero-argument callable yields successive 12-bit codes and
returns :data:`~terselib.constants.ENDOFFILE` (``0``) once the input
bytes are exhausted.
:param data: The full file image (or payload buffer).
:param start: Offset of the first payload byte.
:returns: A callable ``get_code() -> int``.
:raises TerseCorruptError: If the stream ends after 8 of the 12 bits of
a fresh code (a truncated file).
"""
pos = start
end = len(data)
saved = 0
have_nibble = False
def get_code() -> int:
nonlocal pos, saved, have_nibble
if not have_nibble:
if pos >= end:
return ENDOFFILE # clean end at a byte boundary
if pos + 1 >= end:
raise TerseCorruptError(
"truncated input: EOF after 8 of 12 bits of a code")
b1 = data[pos]
b2 = data[pos + 1]
pos += 2
saved = b2 & 0x0F
have_nibble = True
return (b1 << 4) | (b2 >> 4)
if pos >= end:
return ENDOFFILE # leftover nibble was padding
b2 = data[pos]
pos += 1
have_nibble = False
return (saved << 8) | b2
return get_code