import struct import librds.charset from librds.comfort import SubstituteCharacterAtPosition from cliserver_Countries import rds_ecc_lookup, iso, countries def decodeRds(wsdata: dict, status: int, A: int, B: int, C: int, D: int, dec_error: int): a_error = dec_error >> 6 b_error = (dec_error >> 4) & 0b11 c_error = (dec_error >> 2) & 0b11 d_error = dec_error & 0b11 pi = A pi_error = a_error if (status >> 12) & 1 and c_error < a_error: pi = C pi_error = c_error if pi_error < 3: wsdata["pi"] = f"{pi:04X}" + ("?" * pi_error) if b_error > 3: return pi_error group = B >> 11 print(group) wsdata["tp"] = (B >> 10) & 1 wsdata["pty"] = (B >> 5) & 0b11111 match group: case 0 | 1: # 0A or 0B ps_errors: list[str] = wsdata["ps_errors"].split(",") wsdata["ta"] = (B >> 4) & 1 wsdata["ms"] = (B >> 3) & 1 if d_error > 2: return pi_error segment = B & 0b11 old_chars = wsdata["ps"][2*segment:(2*segment)+2] chars = "".join([librds.charset.RDSCharsetDecode.translate(i) for i in struct.pack(">H", D)]) err = int(d_error * (10/3)) if (old_chars == chars and int(ps_errors[2*segment]) > err) or old_chars != chars: for offset, char in enumerate(chars): wsdata["ps"] = SubstituteCharacterAtPosition(wsdata["ps"], char, (2 * segment) + offset) ps_errors[2*segment] = str(err) ps_errors[2*segment+1] = str(err) wsdata["ps_errors"] = ",".join(ps_errors) # TODO: AF case 2: # 1A if c_error > 2: return pi_error variant_code = (C >> 12) & 0b111 match variant_code: case 0: # ECC wsdata["ecc"] = f"{(C & 0xff):02X}" wsdata["country_name"] = rds_ecc_lookup(pi, C & 0xff) wsdata["country_iso"] = "UN" if wsdata["country_name"] == 0 else ( iso[countries.index(wsdata["country_name"])] ) case 4 | 5: # 2A or 2B segment = B & 0b1111 segment_pointer = (4 if group == 4 else 2) * segment d_offset = 2 if group == 4 else 0 ab = bool((B >> 4) & 1) if wsdata["rt_flag"] != int(ab): wsdata["rt1" if ab else "rt0"] = "".ljust(64) wsdata["rt1_errors" if ab else "rt0_errors"] = ",".join((["0"]*64)) wsdata["rt_flag"] = int(ab) chars_c = "".join([librds.charset.RDSCharsetDecode.translate(i) for i in struct.pack(">H", C)]).replace("\r", "") chars_d = "".join([librds.charset.RDSCharsetDecode.translate(i) for i in struct.pack(">H", D)]).replace("\r", "") rt_errors: list[str] = wsdata["rt1_errors" if ab else "rt0_errors"].split(",") old_rt: str = wsdata["rt1" if ab else "rt0"] old_c = old_rt[segment_pointer:segment_pointer+2] old_c_error = int(rt_errors[segment_pointer]) _c_error = int(c_error * (10/3)) old_d = old_rt[segment_pointer+d_offset:segment_pointer+d_offset+2] old_d_error = int(rt_errors[segment_pointer+d_offset]) _d_error = int(d_error * (10/3)) if group == 4 and _c_error < 3 and (chars_c != old_c or _c_error < old_c_error): for offset, char in enumerate(chars_c): old_rt = SubstituteCharacterAtPosition(old_rt, char, segment_pointer + offset) rt_errors[segment_pointer] = str(_c_error) rt_errors[segment_pointer+1] = str(_c_error) if _d_error < 3 and (chars_d != old_d or _d_error < old_d_error): for offset, char in enumerate(chars_d): old_rt = SubstituteCharacterAtPosition(old_rt, char, segment_pointer + offset + d_offset) rt_errors[segment_pointer+d_offset] = str(_d_error) rt_errors[segment_pointer+d_offset+1] = str(_d_error) wsdata["rt1_errors" if ab else "rt0_errors"] = ",".join(rt_errors) wsdata["rt1" if ab else "rt0"] = old_rt return pi_error