Source code for terselib.header

"""Layer 0 — parsing and building TERSE file headers (spec §2).

Two header layouts exist:

* **Native** (version ``0x01`` / ``0x07``): 4 bytes — the version byte plus
  three fixed validation bytes.  Always SPACK, binary, fixed-record.
* **Host** (version ``0x02`` = PACK, ``0x05`` = SPACK): 12 bytes carrying
  the record format (F/V), the record length, and informational MVS fields.
"""

from __future__ import annotations

import dataclasses
import logging
import struct

from . import constants as C
from .exceptions import TerseFormatError, TerseParameterError

logger = logging.getLogger(__name__)

#: BlockSize used by the MVS-annotated header form: the usable half-track
#: capacity of a 3390 DASD volume, as written by AMATERSE.
MVS_BLOCK_CAPACITY = 27998




@dataclasses.dataclass
class TerseHeader:
    """Parsed representation of a TERSE file header.

    :ivar version: The version byte (``0x01``, ``0x02``, ``0x05`` or
        ``0x07``).
    :ivar method: Compression method, ``"PACK"`` or ``"SPACK"``.
    :ivar host: True for host files (12-byte header), False for native.
    :ivar variable: True when the record format is variable (V/VB).
    :ivar record_length: Effective record length.  For fixed files this is
        the LRECL; for variable files it is the maximum record *data*
        length (LRECL − 4).  ``0`` for native files, which carry none.
    :ivar flags: Raw flags byte (host only).
    :ivar ratio: Compression-ratio hint (informational, host only).
    :ivar block_size: Block size hint (informational, host only).
    :ivar size: Header size in bytes (4 or 12).
    """

    version: int
    method: str
    host: bool
    variable: bool
    record_length: int
    flags: int = 0
    ratio: int = 0
    block_size: int = 0
    size: int = 12



    def describe(self) -> str:
        """Return a human-readable multi-line summary of the header."""
        lines = [
            f"version      : 0x{self.version:02X} "
            f"({'host' if self.host else 'native'} {self.method})",
            f"method       : {self.method} "
            f"({'LZW' if self.method == C.METHOD_PACK else 'LZMW'})",
            f"record format: {'V (variable)' if self.variable else 'F (fixed)'}"
            if self.host else "record format: n/a (native raw stream)",
        ]
        if self.host:
            lrecl = (self.record_length + 4) if self.variable \
                else self.record_length
            lines.append(f"record length: {self.record_length} "
                         f"(LRECL {lrecl})")
            lines.append(f"flags        : 0x{self.flags:02X}"
                         + (" [MVS]" if self.flags & C.FLAGMVS else ""))
            lines.append(f"ratio hint   : {self.ratio}")
            lines.append(f"block size   : {self.block_size}")
        return "\n".join(lines)






def parse_header(data: bytes) -> TerseHeader:
    """Parse and validate the header at the start of *data* (spec §2, §12).

    :param data: The file image (only the first 12 bytes are examined).
    :returns: A validated :class:`TerseHeader`; its :attr:`~TerseHeader.size`
        gives the offset at which the 12-bit code stream begins.
    :raises TerseFormatError: If the header is not a valid TERSE header.
    """
    if len(data) < 4:
        raise TerseFormatError("file too short to be a TERSE file")

    version = data[0]
    if version in (C.VERSION_NATIVE_SPACK, C.VERSION_NATIVE_SPACK_ALT):
        if data[1:4] != C.NATIVE_VALIDATION:
            raise TerseFormatError(
                "bad native-header validation bytes "
                f"{data[1:4].hex()} (expected {C.NATIVE_VALIDATION.hex()})")
        hdr = TerseHeader(version=version, method=C.METHOD_SPACK,
                          host=False, variable=False, record_length=0,
                          size=4)
        logger.debug("parsed native header: %s", hdr)
        return hdr

    if version not in (C.VERSION_HOST_PACK, C.VERSION_HOST_SPACK):
        raise TerseFormatError(
            f"unknown TERSE version byte 0x{version:02X}")
    if len(data) < 12:
        raise TerseFormatError("file too short for a host TERSE header")

    variable_flag = data[1]
    if variable_flag not in (0x00, 0x01):
        raise TerseFormatError(
            f"invalid VariableFlag 0x{variable_flag:02X}")
    reclen1, flags, ratio, blocksize, reclen2 = struct.unpack(
        ">HBBHI", data[2:12])

    if reclen1 == 0 and reclen2 == 0:
        raise TerseFormatError("both record-length fields are zero")
    if reclen1 != 0 and reclen2 != 0 and reclen1 != reclen2:
        raise TerseFormatError(
            f"inconsistent record lengths {reclen1} != {reclen2}")
    record_length = reclen1 if reclen1 != 0 else reclen2
    if record_length >= 2 ** 31:
        raise TerseFormatError(f"record length {record_length} out of range")

    if not flags & C.FLAGMVS:
        if flags or ratio or blocksize:
            raise TerseFormatError(
                "FLAGMVS clear but Flags/Ratio/BlockSize are non-zero")

    method = C.METHOD_PACK if version == C.VERSION_HOST_PACK \
        else C.METHOD_SPACK
    hdr = TerseHeader(version=version, method=method, host=True,
                      variable=(variable_flag == 0x01),
                      record_length=record_length, flags=flags,
                      ratio=ratio, block_size=blocksize, size=12)
    logger.debug("parsed host header: %s", hdr)
    return hdr





def build_header(method: str, host: bool, variable: bool,
                 record_length: int, mvs_style: bool = True) -> bytes:
    """Build a header for a file being compressed (spec §2, §8.3).

    :param method: ``"PACK"`` or ``"SPACK"``.
    :param host: True for a host file (12-byte header), False for native
        (4-byte header; forces SPACK).
    :param variable: True for variable record format (host only).
    :param record_length: Record length to record in the header — the
        LRECL for fixed files, LRECL − 4 (the maximum record data length)
        for variable files.  Ignored for native files.
    :param mvs_style: When true (the default), write the MVS-annotated
        header form AMATERSE itself produces (``FLAGMVS | FLAGVS`` and a
        3390 half-track block size); when false, write the minimal form
        with ``Flags = Ratio = BlockSize = 0``.
    :returns: The encoded header bytes.
    :raises TerseParameterError: On an invalid method / mode combination.
    """
    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")
        return bytes((C.VERSION_NATIVE_SPACK,)) + C.NATIVE_VALIDATION

    if record_length <= 0:
        raise TerseParameterError(
            f"record length must be positive, got {record_length}")
    if record_length >= 2 ** 31:
        raise TerseParameterError(
            f"record length {record_length} out of range")

    version = C.VERSION_HOST_PACK if method == C.METHOD_PACK \
        else C.VERSION_HOST_SPACK
    if record_length > 0xFFFF:
        reclen1, reclen2 = 0, record_length
    else:
        reclen1, reclen2 = record_length, 0

    flags = ratio = blocksize = 0
    if mvs_style:
        flags = C.FLAGMVS | C.FLAGVS
        if variable:
            blocksize = MVS_BLOCK_CAPACITY
        else:
            blocksize = max(
                record_length,
                (MVS_BLOCK_CAPACITY // record_length) * record_length
                if record_length <= MVS_BLOCK_CAPACITY else record_length)
            blocksize = min(blocksize, 0xFFFF)

    return struct.pack(">BBHBBHI", version, 0x01 if variable else 0x00,
                       reclen1, flags, ratio, blocksize, reclen2)