From 28010e61daf0e807f4de276a1aef4a9e4eb90bd4 Mon Sep 17 00:00:00 2001 From: KubaPro010 Date: Sat, 25 Apr 2026 18:49:03 +0200 Subject: [PATCH] rds codec --- modules/rds.py | 23 +++- pyproject.toml | 2 +- rds_codec.py | 364 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 387 insertions(+), 2 deletions(-) create mode 100644 rds_codec.py diff --git a/modules/rds.py b/modules/rds.py index 5fe51e4..bfe962b 100644 --- a/modules/rds.py +++ b/modules/rds.py @@ -6,6 +6,7 @@ import socket # https://github.com/chrko/python-uecp import uecp.frame import uecp.commands +import rds_codec @uecp.commands.UECPCommand.register_type class ASCII(uecp.commands.UECPCommand): @@ -18,6 +19,26 @@ class ASCII(uecp.commands.UECPCommand): self.ELEMENT_CODE, 2+len(self.data)] + list(b"95") + list(self.data) +class RT_Set(uecp.commands.RadioTextSetCommand): + def encode(self) -> list[int]: + data = [self.ELEMENT_CODE, self.data_set_number, self.programme_service_number] + if ( + len(self._radiotext.text) == 0 + and self._buffer_configuration + is uecp.commands.RadioTextBufferConfiguration.TRUNCATE_BEFORE + ): + data.append(0) + else: + mel = 1 + len(self._radiotext.text) + flags = ( + self._buffer_configuration << 5 + | self._radiotext.number_of_transmissions << 1 + | self._radiotext.a_b_toggle + ) + data += [mel, flags] + data += list(self._radiotext.text.encode("rp-rds", "replace")) + return data + DEBUG = False name_table_path = "/home/user/mixes/.playlist/name_table.txt" @@ -90,7 +111,7 @@ def update_rds(track_name: str): f.settimeout(1.0) uecp_frame = uecp.frame.UECPFrame() if 0 < len(prt) < 61: prt += "\r" # makes the warning go away - uecp_frame.add_command(uecp.commands.RadioTextSetCommand(prt, 4, True)) + uecp_frame.add_command(RT_Set(prt, 0, True)) uecp_frame.add_command(ASCII(f"RTP={rtp}".encode())) data = uecp_frame.encode() diff --git a/pyproject.toml b/pyproject.toml index bf944c6..2fbb572 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,7 @@ version = "0.1" dependencies = ["log95", "unidecode", "libcache"] [tool.setuptools] -py-modules = ["radioPlaylist", "radioPlayer", "tinytag"] +py-modules = ["radioPlaylist", "radioPlayer", "tinytag", "rds_codec"] packages = ["modules"] [project.scripts] diff --git a/rds_codec.py b/rds_codec.py new file mode 100644 index 0000000..d41f27f --- /dev/null +++ b/rds_codec.py @@ -0,0 +1,364 @@ +""" +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('rp-rds') + + # Encode str to RDS bytes + data = 'Héllo'.encode('rp-rds') + + # With error handlers + text = rds_bytes.decode('rp-rds', errors='replace') + text = rds_bytes.decode('rp-rds', 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 ('rp-rds', '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