"""High-level compression / decompression API.
These are the functions most applications should use:
* :func:`compress` / :func:`decompress` operate on in-memory bytes.
* :func:`compress_file` / :func:`decompress_file` operate on paths and add
conveniences such as output-name selection and (for binary output)
file-type detection via libmagic.
* :func:`read_header` inspects a TERSE file without decompressing it.
Example::
import terselib
data = terselib.compress(b"HELLO WORLD\\n", text=True)
result = terselib.decompress(data, text=True)
assert result.data == b"HELLO WORLD\\n"
"""
from __future__ import annotations
import dataclasses
import logging
import os
from pathlib import Path
from typing import Optional, Union
from . import bitpack, filetype, lzmw, lzw, records, translate
from . import constants as C
from .exceptions import TerseParameterError
from .header import TerseHeader, build_header, parse_header
logger = logging.getLogger(__name__)
PathLike = Union[str, os.PathLike]
[docs]
@dataclasses.dataclass
class DecompressResult:
"""Outcome of a decompression call.
:ivar data: The decompressed bytes.
:ivar header: The parsed :class:`~terselib.header.TerseHeader`.
:ivar text: Whether text-mode rendering was applied.
"""
data: bytes
header: TerseHeader
text: bool
[docs]
def enable_verbose_logging(level: int = logging.DEBUG) -> None:
"""Attach a stderr handler to the ``terselib`` logger.
The library is silent by default (a :class:`logging.NullHandler` is
installed). Call this — or configure the ``terselib`` logger
yourself — to see progress and diagnostic messages.
:param level: Logging level to enable, default :data:`logging.DEBUG`.
"""
root = logging.getLogger("terselib")
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter("%(name)s %(levelname)s: %(message)s"))
root.addHandler(handler)
root.setLevel(level)
[docs]
def decompress(data: bytes, text: Optional[bool] = None,
codepage: str = translate.INTERNAL_CODEPAGE,
rdw: bool = True) -> DecompressResult:
"""Decompress a TERSE file image.
:param data: The complete TERSE file bytes.
:param text: Rendering mode. ``True`` converts EBCDIC to ASCII and
renders logical records as LF-terminated lines; ``False`` yields
the exact original bytes (with a 4-byte RDW before each record of
a variable-format file). ``None`` (the default) auto-detects:
the file is decoded in binary mode and, if the record payloads
look like EBCDIC text, re-rendered as text.
:param codepage: Code page for text-mode conversion — ``"internal"``
(the AMATERSE-compatible spec tables) or any EBCDIC codec name
such as ``"cp037"`` or ``"cp1047"``.
:param rdw: Emit RDW record prefixes in binary-variable mode (spec
behaviour). Set to ``False`` to get bare concatenated records.
:returns: A :class:`DecompressResult` with the output bytes and the
parsed header.
:raises TerseFormatError: If the header is invalid.
:raises TerseCorruptError: If the compressed payload is corrupt.
"""
header = parse_header(data)
auto = text is None
if auto:
text = False
if text and not header.host:
raise TerseParameterError(
"native TERSE files are binary only; text rendering "
"is not available")
e2a, _ = translate.get_tables(codepage)
# In auto mode decode binary with RDWs so record boundaries survive
# for a possible text re-rendering below.
sink = records.RecordSink(host=header.host, text=text,
variable=header.variable,
record_length=header.record_length,
ebc_to_asc=e2a,
emit_rdw=rdw or (auto and header.variable))
get_code = bitpack.make_unpacker(data, header.size)
decoder = lzw.decode if header.method == C.METHOD_PACK else lzmw.decode
logger.info("decompressing %s %s file (%s, %s mode)",
"host" if header.host else "native", header.method,
"variable" if header.variable else "fixed",
"auto" if auto else "text" if text else "binary")
decoder(get_code, sink.put)
out = sink.close()
if auto and header.host and _looks_like_ebcdic_text(out, header, e2a):
logger.info("record payloads look like EBCDIC text; "
"rendering as text")
out = _render_records_as_text(out, header, e2a)
text = True
elif auto and header.variable and not rdw:
out = _strip_rdws(out)
logger.info("decompressed %d bytes -> %d bytes", len(data), len(out))
return DecompressResult(data=out, header=header, text=text)
def _iter_binary_records(raw: bytes, header: TerseHeader):
"""Yield the payload of each record from binary-mode output *raw*
(RDW-framed for variable format, fixed-size chunks otherwise)."""
if header.variable:
i, n = 0, len(raw)
while i + 4 <= n:
length = (raw[i] << 8) | raw[i + 1]
if length < 4 or i + length > n:
raise TerseParameterError("malformed RDW during re-render")
yield raw[i + 4:i + length]
i += length
else:
lrecl = max(header.record_length, 1)
for i in range(0, len(raw), lrecl):
yield raw[i:i + lrecl]
def _looks_like_ebcdic_text(raw: bytes, header: TerseHeader,
e2a: bytes) -> bool:
"""Heuristic used by auto mode: do the record payloads of binary
output *raw* translate to printable ASCII text?
:param raw: Binary-mode decompressed output (with RDWs if variable).
:param header: The file's parsed header.
:param e2a: EBCDIC→ASCII table to test with.
:returns: True when a sample of the payload bytes is at least 95%
printable after translation and contains no NUL bytes.
"""
if not raw or not header.host:
return False
sample = bytearray()
try:
for payload in _iter_binary_records(raw, header):
sample += payload
if len(sample) >= 65536:
break
except TerseParameterError:
return False
if not sample:
return False
translated = bytes(sample[:65536]).translate(e2a)
if b"\x00" in translated:
return False
printable = sum(1 for b in translated if 0x20 <= b < 0x7F or b == 0x09)
return printable / len(translated) > 0.95
def _render_records_as_text(raw: bytes, header: TerseHeader,
e2a: bytes) -> bytes:
"""Re-render binary-mode output *raw* in text mode (EBCDIC→ASCII,
one LF-terminated line per record). Exactly equivalent to having
decoded with ``text=True``."""
out = bytearray()
for payload in _iter_binary_records(raw, header):
out += payload.translate(e2a)
out.append(0x0A)
return bytes(out)
def _strip_rdws(raw: bytes) -> bytes:
"""Remove the 4-byte RDW prefixes from binary variable-mode output."""
out = bytearray()
i, n = 0, len(raw)
while i + 4 <= n:
length = (raw[i] << 8) | raw[i + 1]
out += raw[i + 4:i + length]
i += length
return bytes(out)
[docs]
def compress(data: bytes, method: str = C.METHOD_SPACK, host: bool = True,
text: bool = False, variable: bool = True,
record_length: Optional[int] = None,
codepage: str = translate.INTERNAL_CODEPAGE,
rdw_input: bool = False, mvs_header: bool = True) -> bytes:
"""Compress bytes into a TERSE file image.
:param data: The source bytes.
:param method: ``"PACK"`` (LZW) or ``"SPACK"`` (LZMW, the default —
it generally compresses better).
:param host: ``True`` (default) writes a host file (12-byte header,
interoperable with AMATERSE on z/OS); ``False`` writes a native
SPACK file (4-byte header, raw byte stream, no record model).
:param text: Treat *data* as ASCII text: split it into lines,
translate ASCII→EBCDIC and store each line as a logical record.
Binary mode (default) stores the bytes untranslated.
:param variable: Use variable (V/VB) record format — the default —
rather than fixed (F/FB). Ignored for native files.
:param record_length: Fixed LRECL, variable-mode segment size, or
(text-variable) maximum line length recorded in the header.
Sensible defaults are chosen when omitted; see
:func:`terselib.records.build_codepoints`.
:param codepage: Code page for text-mode ASCII→EBCDIC conversion.
:param rdw_input: In binary-variable mode, treat *data* as RDW-framed
records (as produced by FTP ``SITE RDW`` transfers or by binary
decompression) instead of segmenting it every *record_length*
bytes.
:param mvs_header: Write the MVS-annotated header form AMATERSE
produces (default); ``False`` writes the minimal all-zero form.
:returns: The complete TERSE file bytes.
:raises TerseParameterError: On invalid parameter combinations.
"""
method = method.upper()
if method not in (C.METHOD_PACK, C.METHOD_SPACK):
raise TerseParameterError(f"unknown method {method!r}")
if not host:
if method != C.METHOD_SPACK:
raise TerseParameterError("native files always use SPACK")
text = False
variable = False
_, a2e = translate.get_tables(codepage)
cps, hdr_len = records.build_codepoints(
data, host=host, text=text, variable=variable,
record_length=record_length, asc_to_ebc=a2e, rdw_input=rdw_input,
pad_byte=0x40 if text else 0x00)
logger.info("compressing %d bytes as %s %s (%s, %s mode): "
"%d codepoints", len(data), "host" if host else "native",
method, "variable" if variable else "fixed",
"text" if text else "binary", len(cps))
encoder = lzw.encode if method == C.METHOD_PACK else lzmw.encode
codes = encoder(cps)
out = bytearray()
out += build_header(method, host, variable, max(hdr_len, 1) if host
else 0, mvs_style=mvs_header)
out += bitpack.pack_codes(codes)
if host:
bitpack.pad_file_to_block(out)
logger.info("compressed %d bytes -> %d bytes (%.1f%%)", len(data),
len(out), 100.0 * len(out) / len(data) if data else 0.0)
return bytes(out)
[docs]
def decompress_file(source: PathLike, dest: Optional[PathLike] = None,
text: Optional[bool] = None,
codepage: str = translate.INTERNAL_CODEPAGE,
rdw: bool = True,
detect_type: bool = True) -> "tuple[Path, DecompressResult]":
"""Decompress a TERSE file on disk.
:param source: Path of the TERSE file to read.
:param dest: Output path. When omitted, the name is derived from
*source* (its ``.trs``-style suffix stripped) and, for binary
output with *detect_type* enabled, an extension chosen by
sniffing the decompressed bytes with libmagic.
:param text: Rendering mode, as for :func:`decompress`.
:param codepage: Code page for text-mode conversion.
:param rdw: Emit RDWs in binary-variable mode (spec behaviour).
:param detect_type: Use file-magic content detection when choosing a
default output extension for binary output.
:returns: ``(output_path, result)``.
"""
source = Path(source)
result = decompress(source.read_bytes(), text=text, codepage=codepage,
rdw=rdw)
if dest is None:
stem = source.stem if source.suffix.lower() in (
".trs", ".ters", ".terse", ".pack", ".spack") else source.name
if result.text:
ext = ".txt"
elif detect_type:
ext = filetype.detect_extension(result.data)
logger.info("detected content: %s", filetype.describe(result.data))
else:
ext = ".bin"
dest = source.with_name(stem + ext)
if Path(dest) == source:
dest = source.with_name(stem + ".out")
dest = Path(dest)
dest.write_bytes(result.data)
logger.info("wrote %s (%d bytes)", dest, len(result.data))
return dest, result
[docs]
def compress_file(source: PathLike, dest: Optional[PathLike] = None,
**kwargs) -> Path:
"""Compress a file on disk into a TERSE file.
:param source: Path of the file to compress.
:param dest: Output path; defaults to *source* + ``".trs"``.
:param kwargs: Passed through to :func:`compress`.
:returns: The output path.
"""
source = Path(source)
data = compress(source.read_bytes(), **kwargs)
dest = Path(dest) if dest is not None else \
source.with_name(source.name + ".trs")
dest.write_bytes(data)
logger.info("wrote %s (%d bytes)", dest, len(data))
return dest