"""Layer 1 — the record / codepoint model (spec §4, §6.1, §7).
Decompression side
:class:`RecordSink` implements the spec's ``PutChar`` /
``endRecord`` / ``close`` machinery: it receives codepoints
(``0..257``) from a Layer-2 decoder and reconstructs the output byte
stream (lines, RDW-framed records or raw bytes depending on the mode).
Compression side
:func:`build_codepoints` converts source bytes into the stream of
codepoints (``1..257``) consumed by the Layer-2 encoders, applying
text-mode line parsing, ASCII→EBCDIC translation, record segmentation
and fixed-record padding.
"""
from __future__ import annotations
import logging
from array import array
from typing import Optional, Tuple
from .constants import RECORDMARK
from .exceptions import TerseFormatError, TerseParameterError
logger = logging.getLogger(__name__)
_LF = 0x0A
_CR = 0x0D
_SUB = 0x1A
[docs]
class RecordSink:
"""Reconstructs decompressed output from a stream of codepoints.
Implements ``PutChar``/``endRecord``/``close`` from spec §6.1. The
behaviour matrix (spec §9):
========= ========= ======== ==============================================
host text variable rendering
========= ========= ======== ==============================================
host text fixed EBCDIC→ASCII, line break every *record_length*
host text variable EBCDIC→ASCII, line break at each RECORDMARK
host binary fixed raw bytes, record marks discarded
host binary variable RDW-framed records (4-byte RDW + payload)
native binary fixed raw byte stream
========= ========= ======== ==============================================
:param host: True for host files.
:param text: True to render in text mode (EBCDIC→ASCII + lines).
:param variable: True for variable record format.
:param record_length: Fixed record length (used in host-text-fixed
mode only).
:param ebc_to_asc: 256-byte EBCDIC→ASCII table used in text mode.
:param emit_rdw: When False, suppress the 4-byte RDW prefix normally
written for each record in binary-variable mode (a convenience
deviation from the spec; the default True follows spec §6.1).
"""
def __init__(self, host: bool, text: bool, variable: bool,
record_length: int, ebc_to_asc: bytes,
emit_rdw: bool = True) -> None:
if text and not host:
raise TerseParameterError("native files cannot use text mode")
self.host = host
self.text = text
self.variable = variable
self.record_length = record_length
self.e2a = ebc_to_asc
self.emit_rdw = emit_rdw
self.out = bytearray()
self._record = bytearray()
# -- spec §6.1: PutChar ------------------------------------------------
[docs]
def put(self, x: int) -> None:
"""Process one codepoint ``X`` (``0..257``) — spec ``PutChar``."""
if x == 0:
if self.host and self.text and self.variable:
self.end_record()
return
if self.host and self.text:
if self.variable:
if x == RECORDMARK:
self.end_record()
else:
self._record.append(self.e2a[x - 1])
else:
self._record.append(self.e2a[x - 1])
if len(self._record) == self.record_length:
self.end_record()
else:
if x == RECORDMARK:
if self.variable:
self.end_record()
else:
self._record.append(x - 1)
# -- spec §6.1: endRecord ----------------------------------------------
[docs]
def end_record(self) -> None:
"""Flush the current record to the output — spec ``endRecord``."""
if self.variable and not self.text and self.emit_rdw:
length = len(self._record) + 4
self.out.append((length >> 8) & 0xFF)
self.out.append(length & 0xFF)
self.out.append(0)
self.out.append(0)
self.out += self._record
self._record.clear()
if self.text:
self.out.append(_LF)
# -- spec §6.1: close --------------------------------------------------
[docs]
def close(self) -> bytes:
"""Flush any trailing partial record and return the output bytes.
Only a non-empty trailing record is flushed. (Spec §6.1 as
published also forced a final empty ``endRecord`` in
text + variable mode; testing against AMATERSE-produced data shows
that this adds a spurious trailing line, so it is not done — see
the specification's erratum note.)
"""
if self._record:
self.end_record()
return bytes(self.out)
def _parse_text_records(data: bytes) -> "tuple[list[bytes], int]":
"""Split an ASCII text stream into logical records (spec §7.2).
Rules: CR bytes are skipped (swallowing CRLF pairs); LF ends the
current line; CR/SUB bytes immediately following a LF are skipped; a
SUB as the very last input byte is treated as end-of-input; a final
line without a terminator still forms a record.
:param data: Raw source bytes (ASCII side).
:returns: ``(records, max_length)`` where *records* excludes any line
terminators and *max_length* is the longest record (0 if none).
"""
records: "list[bytes]" = []
line = bytearray()
i, n = 0, len(data)
maxlen = 0
while i < n:
b = data[i]
if b == _CR:
i += 1
continue
if b == _LF:
records.append(bytes(line))
maxlen = max(maxlen, len(line))
line.clear()
i += 1
while i < n and data[i] in (_CR, _SUB):
i += 1
continue
if b == _SUB and i == n - 1:
break # trailing SUB = end of input
line.append(b)
i += 1
if line:
records.append(bytes(line))
maxlen = max(maxlen, len(line))
return records, maxlen
def _parse_rdw_records(data: bytes) -> "list[bytes]":
"""Split an RDW-framed byte stream into record payloads (spec §7.2).
Each record starts with a 2-byte big-endian length ``L`` (which
includes the 4-byte RDW itself) followed by two zero bytes and
``L - 4`` payload bytes.
:param data: RDW-framed input bytes.
:returns: The record payloads, in order.
:raises TerseFormatError: On a malformed or truncated RDW.
"""
records: "list[bytes]" = []
i, n = 0, len(data)
while i < n:
if i + 4 > n:
raise TerseFormatError("truncated RDW at end of input")
length = (data[i] << 8) | data[i + 1]
if length < 4 or i + length > n:
raise TerseFormatError(
f"invalid RDW length {length} at offset {i}")
records.append(bytes(data[i + 4:i + length]))
i += length
return records
[docs]
def build_codepoints(data: bytes, host: bool, text: bool, variable: bool,
record_length: Optional[int], asc_to_ebc: bytes,
rdw_input: bool = False,
pad_byte: int = 0x00) -> Tuple[array, int]:
"""Convert source bytes into the Layer-1 codepoint stream (spec §7).
:param data: The source file bytes.
:param host: True for host output, False for native.
:param text: True to treat *data* as ASCII text (line-split it and
translate ASCII→EBCDIC); False for binary.
:param variable: True for variable record format (host only; native is
always fixed/raw).
:param record_length: For fixed format, the LRECL; for variable
format, the record data length used to segment binary input (and
the upper bound recorded for text input). When ``None``, defaults
are chosen: text-variable uses the longest line, text-fixed uses
80, binary-fixed uses 1 and binary-variable segments at 4096.
:param asc_to_ebc: 256-byte ASCII→EBCDIC table (text mode only).
:param rdw_input: In binary-variable mode, parse *data* as RDW-framed
records instead of segmenting it at *record_length*.
:param pad_byte: Byte used to pad the final short fixed binary record.
:returns: ``(codepoints, header_record_length)`` — the codepoint
stream as an ``array('H')`` and the record length to place in the
header (LRECL for fixed, max data length for variable).
:raises TerseFormatError: If RDW parsing of the input fails.
:raises TerseParameterError: On invalid parameter combinations.
"""
cps = array("H")
if not host:
# Native: raw byte stream, no marks, no padding, no translation.
if text or variable:
raise TerseParameterError(
"native mode is always binary with no record structure")
cps.extend(b + 1 for b in data)
return cps, 0
if text:
records, maxlen = _parse_text_records(data)
translated = [r.translate(asc_to_ebc) for r in records]
if variable:
hdr_len = record_length if record_length else max(maxlen, 1)
for rec in translated:
if len(rec) > hdr_len:
logger.warning(
"record of %d bytes exceeds record length %d",
len(rec), hdr_len)
hdr_len = len(rec)
cps.extend(b + 1 for b in rec)
cps.append(RECORDMARK)
return cps, hdr_len
lrecl = record_length or 80
ebcdic_space = asc_to_ebc[0x20]
for j, rec in enumerate(translated):
if len(rec) > lrecl:
logger.warning(
"line %d truncated from %d to LRECL %d bytes",
j + 1, len(rec), lrecl)
rec = rec[:lrecl]
elif len(rec) < lrecl:
rec = rec + bytes([ebcdic_space]) * (lrecl - len(rec))
cps.extend(b + 1 for b in rec)
return cps, lrecl
if variable:
if rdw_input:
records = _parse_rdw_records(data)
hdr_len = max((len(r) for r in records), default=1)
hdr_len = max(hdr_len, 1)
if record_length:
hdr_len = max(hdr_len, record_length)
for rec in records:
cps.extend(b + 1 for b in rec)
cps.append(RECORDMARK)
return cps, hdr_len
seg = record_length or 4096
if seg <= 0:
raise TerseParameterError("record length must be positive")
for i in range(0, len(data), seg):
chunk = data[i:i + seg]
cps.extend(b + 1 for b in chunk)
cps.append(RECORDMARK)
return cps, seg
lrecl = record_length or 1
if lrecl <= 0:
raise TerseParameterError("record length must be positive")
cps.extend(b + 1 for b in data)
remainder = len(data) % lrecl
if remainder:
logger.debug("padding final fixed record with %d bytes of 0x%02X",
lrecl - remainder, pad_byte)
cps.extend([pad_byte + 1] * (lrecl - remainder))
return cps, lrecl