mirror of
https://github.com/radio95-rnt/RadioPlayer.git
synced 2026-07-29 07:09:15 +02:00
365 lines
7.9 KiB
Python
365 lines
7.9 KiB
Python
"""
|
|
rds_codec.py - Python codec for the RDS character set (IEC 62106 / R22_039_1 standard)
|
|
|
|
Usage:
|
|
import rds_codec # registers the codec
|
|
|
|
# Decode RDS bytes to str
|
|
text = b'\\x48\\x65\\x8A\\x6C\\x6F'.decode('radiodatasystem')
|
|
|
|
# Encode str to RDS bytes
|
|
data = 'Héllo'.encode('radiodatasystem')
|
|
|
|
# With error handlers
|
|
text = rds_bytes.decode('radiodatasystem', errors='replace')
|
|
text = rds_bytes.decode('radiodatasystem', errors='ignore')
|
|
"""
|
|
|
|
import codecs
|
|
from typing import Tuple
|
|
import unidecode
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mapping tables (from rdscharset.pdf, R22_039_1 standard)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# RDS code point -> Unicode code point
|
|
_RDS_TO_UCS2: dict[int, int] = {
|
|
0x0A: 0x0A,
|
|
0x0D: 0x0D,
|
|
0x1F: 0x1F,
|
|
0x20: 0x20,
|
|
0x21: 0x21,
|
|
0x22: 0x22,
|
|
0x23: 0x23,
|
|
0x24: 0xA4,
|
|
0x25: 0x25,
|
|
0x26: 0x26,
|
|
0x27: 0x27,
|
|
0x28: 0x28,
|
|
0x29: 0x29,
|
|
0x2A: 0x2A,
|
|
0x2B: 0x2B,
|
|
0x2C: 0x2C,
|
|
0x2D: 0x2D,
|
|
0x2E: 0x2E,
|
|
0x2F: 0x2F,
|
|
0x30: 0x30,
|
|
0x31: 0x31,
|
|
0x32: 0x32,
|
|
0x33: 0x33,
|
|
0x34: 0x34,
|
|
0x35: 0x35,
|
|
0x36: 0x36,
|
|
0x37: 0x37,
|
|
0x38: 0x38,
|
|
0x39: 0x39,
|
|
0x3A: 0x3A,
|
|
0x3B: 0x3B,
|
|
0x3C: 0x3C,
|
|
0x3D: 0x3D,
|
|
0x3E: 0x3E,
|
|
0x3F: 0x3F,
|
|
0x40: 0x40,
|
|
0x41: 0x41,
|
|
0x42: 0x42,
|
|
0x43: 0x43,
|
|
0x44: 0x44,
|
|
0x45: 0x45,
|
|
0x46: 0x46,
|
|
0x47: 0x47,
|
|
0x48: 0x48,
|
|
0x49: 0x49,
|
|
0x4A: 0x4A,
|
|
0x4B: 0x4B,
|
|
0x4C: 0x4C,
|
|
0x4D: 0x4D,
|
|
0x4E: 0x4E,
|
|
0x4F: 0x4F,
|
|
0x50: 0x50,
|
|
0x51: 0x51,
|
|
0x52: 0x52,
|
|
0x53: 0x53,
|
|
0x54: 0x54,
|
|
0x55: 0x55,
|
|
0x56: 0x56,
|
|
0x57: 0x57,
|
|
0x58: 0x58,
|
|
0x59: 0x59,
|
|
0x5A: 0x5A,
|
|
0x5B: 0x5B,
|
|
0x5C: 0x5C,
|
|
0x5D: 0x5D,
|
|
0x5E: 0x2015,
|
|
0x5F: 0x005F,
|
|
0x60: 0x2016,
|
|
0x61: 0x61,
|
|
0x62: 0x62,
|
|
0x63: 0x63,
|
|
0x64: 0x64,
|
|
0x65: 0x65,
|
|
0x66: 0x66,
|
|
0x67: 0x67,
|
|
0x68: 0x68,
|
|
0x69: 0x69,
|
|
0x6A: 0x6A,
|
|
0x6B: 0x6B,
|
|
0x6C: 0x6C,
|
|
0x6D: 0x6D,
|
|
0x6E: 0x6E,
|
|
0x6F: 0x6F,
|
|
0x70: 0x70,
|
|
0x71: 0x71,
|
|
0x72: 0x72,
|
|
0x73: 0x73,
|
|
0x74: 0x74,
|
|
0x75: 0x75,
|
|
0x76: 0x76,
|
|
0x77: 0x77,
|
|
0x78: 0x78,
|
|
0x79: 0x79,
|
|
0x7A: 0x7A,
|
|
0x7B: 0x7B,
|
|
0x7C: 0x7C,
|
|
0x7D: 0x7D,
|
|
0x7E: 0x203E,
|
|
0x80: 0xE1,
|
|
0x81: 0xE0,
|
|
0x82: 0xE9,
|
|
0x83: 0xE8,
|
|
0x84: 0xED,
|
|
0x85: 0xEC,
|
|
0x86: 0xF3,
|
|
0x87: 0xF2,
|
|
0x88: 0xFA,
|
|
0x89: 0xF9,
|
|
0x8A: 0xD1,
|
|
0x8B: 0xC7,
|
|
0x8C: 0x15E,
|
|
0x8D: 0xDF,
|
|
0x8E: 0xA1,
|
|
0x8F: 0x132,
|
|
0x90: 0xE2,
|
|
0x91: 0xE4,
|
|
0x92: 0xEA,
|
|
0x93: 0xEB,
|
|
0x94: 0xEE,
|
|
0x95: 0xEF,
|
|
0x96: 0xF4,
|
|
0x97: 0xF6,
|
|
0x98: 0xFB,
|
|
0x99: 0xFC,
|
|
0x9A: 0xF1,
|
|
0x9B: 0xE7,
|
|
0x9C: 0x15F,
|
|
0x9D: 0x11F,
|
|
0x9E: 0x131,
|
|
0x9F: 0x133,
|
|
0xA0: 0xAA,
|
|
0xA1: 0x3B1,
|
|
0xA2: 0xA9,
|
|
0xA3: 0x2030,
|
|
0xA4: 0x11E,
|
|
0xA5: 0x11B,
|
|
0xA6: 0x148,
|
|
0xA7: 0x151,
|
|
0xA8: 0x3C0,
|
|
0xA9: 0x20AC,
|
|
0xAA: 0xA3,
|
|
0xAB: 0x24,
|
|
0xAC: 0x2190,
|
|
0xAD: 0x2191,
|
|
0xAE: 0x2192,
|
|
0xAF: 0x2193,
|
|
0xB0: 0xBA,
|
|
0xB1: 0xB9,
|
|
0xB2: 0xB2,
|
|
0xB3: 0xB3,
|
|
0xB4: 0xB1,
|
|
0xB5: 0x130,
|
|
0xB6: 0x144,
|
|
0xB7: 0x171,
|
|
0xB8: 0xB5,
|
|
0xB9: 0xBF,
|
|
0xBA: 0xF7,
|
|
0xBB: 0xBB,
|
|
0xBC: 0xBC,
|
|
0xBD: 0xBD,
|
|
0xBE: 0xBE,
|
|
0xBF: 0xA7,
|
|
0xC0: 0xC1,
|
|
0xC1: 0xC0,
|
|
0xC2: 0xC9,
|
|
0xC3: 0xC8,
|
|
0xC4: 0xCD,
|
|
0xC5: 0xCC,
|
|
0xC6: 0xD3,
|
|
0xC7: 0xD2,
|
|
0xC8: 0xDA,
|
|
0xC9: 0xD9,
|
|
0xCA: 0x158,
|
|
0xCB: 0x10C,
|
|
0xCC: 0x160,
|
|
0xCD: 0x17D,
|
|
0xCE: 0xD0,
|
|
0xCF: 0x13F,
|
|
0xD0: 0xC2,
|
|
0xD1: 0xC4,
|
|
0xD2: 0xCA,
|
|
0xD3: 0xCB,
|
|
0xD4: 0xCE,
|
|
0xD5: 0xCF,
|
|
0xD6: 0xD4,
|
|
0xD7: 0xD6,
|
|
0xD8: 0xDB,
|
|
0xD9: 0xDC,
|
|
0xDA: 0x159,
|
|
0xDB: 0x10D,
|
|
0xDC: 0x161,
|
|
0xDD: 0x17E,
|
|
0xDE: 0x111,
|
|
0xDF: 0x140,
|
|
0xE0: 0xC3,
|
|
0xE1: 0xC5,
|
|
0xE2: 0xC6,
|
|
0xE3: 0x152,
|
|
0xE4: 0x177,
|
|
0xE5: 0xDD,
|
|
0xE6: 0xD5,
|
|
0xE7: 0xD8,
|
|
0xE8: 0xDE,
|
|
0xE9: 0x14A,
|
|
0xEA: 0x154,
|
|
0xEB: 0x106,
|
|
0xEC: 0x15A,
|
|
0xED: 0x179,
|
|
0xEE: 0x166,
|
|
0xEF: 0xF0,
|
|
0xF0: 0xE3,
|
|
0xF1: 0xE5,
|
|
0xF2: 0xE6,
|
|
0xF3: 0x153,
|
|
0xF4: 0x175,
|
|
0xF5: 0xFD,
|
|
0xF6: 0xF5,
|
|
0xF7: 0xF8,
|
|
0xF8: 0x00FE,
|
|
0xF9: 0x14B,
|
|
0xFA: 0x155,
|
|
0xFB: 0x107,
|
|
0xFC: 0x15B,
|
|
0xFD: 0x17A,
|
|
0xFE: 0x167,
|
|
}
|
|
|
|
# Reverse map: Unicode code point -> RDS byte
|
|
_UCS2_TO_RDS: dict[int, int] = {v: k for k, v in _RDS_TO_UCS2.items()}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Codec implementation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _rds_decode(data: bytes, errors: str = "strict") -> Tuple[str, int]:
|
|
"""Decode RDS-encoded bytes to a Unicode string."""
|
|
chars = []
|
|
for i, byte in enumerate(data):
|
|
ucs = _RDS_TO_UCS2.get(byte)
|
|
if ucs is not None:
|
|
chars.append(chr(ucs))
|
|
else:
|
|
if errors == 'strict':
|
|
raise UnicodeDecodeError(
|
|
'rds', data, i, i + 1,
|
|
f'RDS byte 0x{byte:02X} has no Unicode mapping'
|
|
)
|
|
elif errors == 'replace':
|
|
chars.append('\uFFFD')
|
|
elif errors == 'ignore':
|
|
pass
|
|
else:
|
|
raise LookupError(f"unknown error handler '{errors}'")
|
|
return ''.join(chars), len(data)
|
|
|
|
|
|
def _rds_encode(text: str, errors: str = 'strict') -> Tuple[bytes, int]:
|
|
"""Encode a Unicode string to RDS bytes."""
|
|
out = []
|
|
for i, ch in enumerate(text):
|
|
rds = _UCS2_TO_RDS.get(ord(ch))
|
|
if rds is not None:
|
|
out.append(rds)
|
|
else:
|
|
if errors == 'strict':
|
|
raise UnicodeEncodeError(
|
|
'rds', text, i, i + 1,
|
|
f'U+{ord(ch):04X} ({ch!r}) has no RDS mapping'
|
|
)
|
|
elif errors == 'replace':
|
|
rds2 = _UCS2_TO_RDS.get(ord(unidecode.unidecode(ch, "replace", " ")))
|
|
if rds2 is not None: out.append(rds2)
|
|
else: out.append(0x20) # substitute with space
|
|
elif errors == 'ignore':
|
|
pass
|
|
else:
|
|
raise LookupError(f"unknown error handler '{errors}'")
|
|
return bytes(out), len(text)
|
|
|
|
|
|
class RDSIncrementalDecoder(codecs.BufferedIncrementalDecoder):
|
|
def _buffer_decode(self, data: bytes, errors: str, final: bool):
|
|
return _rds_decode(data, errors)
|
|
|
|
|
|
class RDSIncrementalEncoder(codecs.IncrementalEncoder):
|
|
def encode(self, text: str, final: bool = False) -> bytes:
|
|
result, _ = _rds_encode(text, self.errors)
|
|
return result
|
|
|
|
|
|
class RDSStreamReader(codecs.StreamReader):
|
|
def decode(self, data: bytes, errors: str = 'strict'):
|
|
return _rds_decode(data, errors)
|
|
|
|
|
|
class RDSStreamWriter(codecs.StreamWriter):
|
|
def encode(self, text: str, errors: str = 'strict'):
|
|
return _rds_encode(text, errors)
|
|
|
|
|
|
_CODEC_INFO = codecs.CodecInfo(
|
|
name='rds',
|
|
encode=_rds_encode,
|
|
decode=_rds_decode, # pyright: ignore[reportArgumentType]
|
|
incrementalencoder=RDSIncrementalEncoder,
|
|
incrementaldecoder=RDSIncrementalDecoder,
|
|
streamreader=RDSStreamReader,
|
|
streamwriter=RDSStreamWriter,
|
|
)
|
|
|
|
|
|
def _rds_lookup(name: str):
|
|
if name in ('radiodatasystem', 'rds-charset', 'rds_charset'):
|
|
return _CODEC_INFO
|
|
return None
|
|
|
|
|
|
# Register the codec when this module is imported
|
|
codecs.register(_rds_lookup)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Convenience helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def decode(data: bytes, errors: str = 'strict') -> str:
|
|
"""Decode RDS bytes to a Python str."""
|
|
text, _ = _rds_decode(data, errors)
|
|
return text
|
|
|
|
|
|
def encode(text: str, errors: str = 'strict') -> bytes:
|
|
"""Encode a Python str to RDS bytes."""
|
|
data, _ = _rds_encode(text, errors)
|
|
return data
|