dmr-server/dmrsrv/bptc.py

"""
RadioStar — DMR BPTC(196,96) codec (pure Python).

Ported from dmr_utils3 (N0MJS, GPL) + ETSI TS 102 361-1. Used to decode/encode
the 96-bit payloads carried by DATA_HEADER and CSBK bursts (data headers, CSBK
preambles, etc.). Lets the server READ and BUILD those control frames.

A burst's 196 info bits live at burst-bit positions [0:98] + [166:264]; the
middle 68 bits (slot type + sync) are spliced from a real burst of the same type.
"""
from __future__ import annotations

# interleaver index (on-air[INDEX_181[i]] = logical[i])
INDEX_181 = (
    0, 181, 166, 151, 136, 121, 106, 91, 76, 61, 46, 31, 16, 1, 182, 167, 152, 137,
    122, 107, 92, 77, 62, 47, 32, 17, 2, 183, 168, 153, 138, 123, 108, 93, 78, 63,
    48, 33, 18, 3, 184, 169, 154, 139, 124, 109, 94, 79, 64, 49, 34, 19, 4, 185, 170,
    155, 140, 125, 110, 95, 80, 65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111, 96,
    81, 66, 51, 36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82, 67, 52, 37, 22, 7,
    188, 173, 158, 143, 128, 113, 98, 83, 68, 53, 38, 23, 8, 189, 174, 159, 144, 129,
    114, 99, 84, 69, 54, 39, 24, 9, 190, 175, 160, 145, 130, 115, 100, 85, 70, 55, 40,
    25, 10, 191, 176, 161, 146, 131, 116, 101, 86, 71, 56, 41, 26, 11, 192, 177, 162,
    147, 132, 117, 102, 87, 72, 57, 42, 27, 12, 193, 178, 163, 148, 133, 118, 103, 88,
    73, 58, 43, 28, 13, 194, 179, 164, 149, 134, 119, 104, 89, 74, 59, 44, 29, 14,
    195, 180, 165, 150, 135, 120, 105, 90, 75, 60, 45, 30, 15)

# the 96 data-bit positions in the deinterleaved matrix (72 LC/header + 24 RS/CRC)
DATA_POS = [
    136, 121, 106, 91, 76, 61, 46, 31,
    152, 137, 122, 107, 92, 77, 62, 47, 32, 17, 2,
    123, 108, 93, 78, 63, 48, 33, 18, 3, 184, 169,
    94, 79, 64, 49, 34, 19, 4, 185, 170, 155, 140,
    65, 50, 35, 20, 5, 186, 171, 156, 141, 126, 111,
    36, 21, 6, 187, 172, 157, 142, 127, 112, 97, 82,
    7, 188, 173, 158, 143, 128, 113, 98, 83,
    68, 53, 174, 159, 144, 129, 114, 99, 84, 69, 54, 39,
    24, 145, 130, 115, 100, 85, 70, 55, 40, 25, 10, 191,
]


def _enc_15113(d):
    return [
        d[0] ^ d[1] ^ d[2] ^ d[3] ^ d[5] ^ d[7] ^ d[8],
        d[1] ^ d[2] ^ d[3] ^ d[4] ^ d[6] ^ d[8] ^ d[9],
        d[2] ^ d[3] ^ d[4] ^ d[5] ^ d[7] ^ d[9] ^ d[10],
        d[0] ^ d[1] ^ d[2] ^ d[4] ^ d[6] ^ d[7] ^ d[10],
    ]


def _enc_1393(d):
    return [
        d[0] ^ d[1] ^ d[3] ^ d[5] ^ d[6],
        d[0] ^ d[1] ^ d[2] ^ d[4] ^ d[6] ^ d[7],
        d[0] ^ d[1] ^ d[2] ^ d[3] ^ d[5] ^ d[7] ^ d[8],
        d[0] ^ d[2] ^ d[4] ^ d[5] ^ d[8],
    ]


def _bytes_to_bits(data):
    out = []
    for byte in data:
        for i in range(7, -1, -1):
            out.append((byte >> i) & 1)
    return out


def _bits_to_bytes(bits):
    out = bytearray()
    for i in range(0, len(bits), 8):
        b = 0
        for j in range(8):
            b = (b << 1) | bits[i + j]
        out.append(b)
    return bytes(out)


def _burst_info_bits(burst33):
    """Extract the 196 on-air info bits from a 33-byte burst ([0:98]+[166:264])."""
    allbits = _bytes_to_bits(burst33)
    return allbits[0:98] + allbits[166:264]


def decode(burst33: bytes) -> bytes:
    """Decode a BPTC(196,96) burst -> 12 bytes (96 bits). No error correction.
    DATA_POS index the on-air (interleaved) info bits directly (per dmr_utils)."""
    onair = _burst_info_bits(burst33)              # 196 on-air bits
    bits = [onair[p] for p in DATA_POS]            # extract 96 data bits
    return _bits_to_bytes(bits)


def encode(data12: bytes) -> list:
    """Encode 12 bytes -> 196 on-air info bits (list)."""
    bits = _bytes_to_bits(data12)                  # 96 bits
    for _ in range(4):
        bits.insert(0, 0)                          # R0-R3
    for index in range(9):
        spos = index * 15 + 1
        epos = spos + 11
        par = _enc_15113(bits[spos:epos])
        for pbit in range(4):
            bits.insert(epos + pbit, par[pbit])
    for _ in range(60):
        bits.append(0)
    for col in range(15):
        spos = col + 1
        column = []
        for _ in range(9):
            column.append(bits[spos])
            spos += 15
        colp = _enc_1393(column)
        cpar = 136 + col
        for pbit in range(4):
            bits[cpar] = colp[pbit]
            cpar += 15
    onair = [0] * 196
    for i in range(196):
        onair[INDEX_181[i]] = bits[i]              # interleave
    return onair


def splice_into_burst(template33: bytes, onair196: list) -> bytes:
    """Build a burst: info bits from onair196, sync+slot-type from a template burst
    of the same data type (positions [98:166])."""
    allbits = _bytes_to_bits(template33)
    allbits[0:98] = onair196[0:98]
    allbits[166:264] = onair196[98:196]
    return _bits_to_bytes(allbits)


if __name__ == "__main__":
    # decode the captured RT-4D DATA_HEADER (TEST123, 1400001 -> 1234567)
    hdr = bytes.fromhex("157a07fb5a6681a8b2828e4b058d5d7f77fd7573388475f5803549e54a87813418")
    dec = decode(hdr)
    print("DATA_HEADER 12 bytes:", dec.hex())
    # round-trip: encode(decode) info bits must match the burst's info bits
    rt = splice_into_burst(hdr, encode(dec))
    print("round-trip burst == original:", rt == hdr)
    # decode a captured CSBK preamble
    csbk = bytes.fromhex("47e7160d92e007ac4e398c7244cd5d7f77fd757ac996047099099220839a4a9b4e")
    print("CSBK 12 bytes:", decode(csbk).hex())