From 5ca98b993ef30cb5384dd60c8f1bb4adc2c2e90d Mon Sep 17 00:00:00 2001 From: KubaPro010 Date: Fri, 20 Mar 2026 18:16:37 +0100 Subject: [PATCH] xrd improvements --- libtimer2.py | 14 ++++ xrd.py | 191 +++++++++++++++++++++++++-------------------------- 2 files changed, 106 insertions(+), 99 deletions(-) create mode 100644 libtimer2.py diff --git a/libtimer2.py b/libtimer2.py new file mode 100644 index 0000000..5b01540 --- /dev/null +++ b/libtimer2.py @@ -0,0 +1,14 @@ +import time + +class Timer: + def __init__(self) -> None: + self.ctime = 0.0 + self.reset() + def reset(self) -> float: + self.ctime = time.monotonic() + return self.ctime + def subtract(self, value: float) -> float: + self.ctime -= value + return self.ctime + def get_time(self) -> float: + return (time.monotonic()-self.ctime) \ No newline at end of file diff --git a/xrd.py b/xrd.py index 8284454..ddbef59 100644 --- a/xrd.py +++ b/xrd.py @@ -1,18 +1,15 @@ # This is a compatible server for the TEF6686 firmware, but instead using the i2c control mode -# This fully works with FM-DX-Webserver (no plugins) +# This fully works with FM-DX-Webserver (no plugins) + Spectrum plugin # Released under the Unlicense (however both FM-DX-Webserver and the firmware are under the tyranny of GPL-3) -import socket -import hashlib -import secrets -import string -import libtimer -import time +import socket, os, time +import hashlib, secrets, string +from libtimer2 import Timer from driver.tef import TEF6686 from driver.protocol import I2CPCClient -import os from functools import wraps from typing import Callable +from dataclasses import dataclass INITIAL_FREQ = 9500 INITIAL_EQ = True @@ -20,37 +17,47 @@ INITIAL_IMS = True HOST = os.getenv("HOST") or '0.0.0.0' PORT = int(os.getenv("PORT") or 0) or 7373 -DEVICE = os.getenv("DEV") or "COM6" +DEVICE = os.getenv("DEV", "COM6") -FREQ_NOT_ALLOWED_RANGE = [] +PASSWORD = os.getenv("PW", "test") +if PASSWORD == "test": print("Password left at default. Please change if neccesary") -PASSWORD = os.getenv("PW") or "test" - -CLOCK_ENV = os.getenv("CLK", "").strip().lower() or "dp" +CLOCK_ENV = os.getenv("CLK", "dp").strip().lower() if CLOCK_ENV == "dp": CLOCK = 12000000 # DP-666 elif CLOCK_ENV == "x": CLOCK = 9216000 else: CLOCK = int(CLOCK_ENV) SALT_LENGTH = 16 -SS_UPDATE_INTERVAL = 0.095 -RDS_UPDATE_INTERVAL = 0.086 +SS_UPDATE_INTERVAL = 0.098 +RDS_UPDATE_INTERVAL = 0.0856 -def freq_allowed(freq: int): - if not FREQ_NOT_ALLOWED_RANGE: return True - for (lower, upper) in FREQ_NOT_ALLOWED_RANGE: - if freq >= lower and freq <= upper: return False - return True +@dataclass +class State: + last_tune: int = INITIAL_FREQ + last_eqims: int = (INITIAL_EQ << 1) | INITIAL_IMS + forced_mono: bool = False + deemp: int = 0 + bw: int = 0 + scan_start: int = 8750 + scan_stop: int = 10800 + scan_step: int = 10 + scan_bw: int = 0 def init_tef(): - p = I2CPCClient(DEVICE, int(os.getenv("BAUD") or 0) or 115200) - tef = TEF6686(p) + tef = TEF6686(I2CPCClient(DEVICE, int(os.getenv("BAUD") or 0) or 115200)) tef.init(clock=CLOCK) tef.AUDIO_Set_Mute(False) - tef.AUDIO_Set_Volume(20) tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, INITIAL_FREQ) tef.FM_Set_RDS(1) tef.FM_Set_ChannelEqualizer(INITIAL_EQ) tef.FM_Set_MphSuppression(INITIAL_IMS) + + tef.FM_Set_LevelStep(-1, -1, -1, -1, -4, -8, 0) + tef.FM_Set_Highcut_Mph(0) + tef.FM_Set_Highcut_Max(False) + tef.FM_Set_Lowcut_Max(True, 60) + tef.FM_Set_Stereo_Time(60, 120, 100, 200) + tef.FM_Set_Stereo_Max(False) # Disables stereo blend??!!??!? tef.APPL_Set_OperationMode(True) # Turn off return tef @@ -59,100 +66,96 @@ def authenticate(conn: socket.socket): salt = "".join(secrets.choice(string.ascii_lowercase) for _ in range(SALT_LENGTH)) conn.sendall(salt.encode() + b"\n") expected_hash = hashlib.sha1((salt + PASSWORD).encode()).hexdigest().encode() - start = time.monotonic() + tries = 0 while True: try: data = conn.recv(len(expected_hash)) - if not data: return False - if data.strip() == expected_hash.strip(): return True - except BlockingIOError: - if (time.monotonic() - start) > 3: conn.close() # Close connection if they can't fucking bruteforce the password in 3 seconds + if not data: break -def process_command(tef: TEF6686, data: bytes, state: dict, conn: socket.socket): + if data.strip() == expected_hash.strip(): + conn.sendall(b"o1,0\n") + return True + else: + tries += 1 + conn.sendall(b"a0\n") # You failed + except BlockingIOError: + if tries > 2: + conn.close() + break + return False + +def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket): out = b"" for cmd in data.splitlines(): if cmd.startswith(b"T"): freq = int(cmd.decode().removeprefix("T").strip()) // 10 if freq < 6500 or freq > 10800: continue tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, freq) - if not freq_allowed(freq): tef.AUDIO_Set_Mute(True) - else: tef.AUDIO_Set_Mute(False) - state['last_tune'] = freq + state.last_tune = freq out += f"T{freq*10}\n".encode() elif cmd.startswith(b"G"): eqims = int(cmd.decode().removeprefix("G").strip(), 2) tef.FM_Set_ChannelEqualizer((eqims & 1) == 1) tef.FM_Set_MphSuppression((eqims & 2) == 2) out += f"G{bin(eqims).removeprefix('0b').zfill(2)}\n".encode() - state['last_eqims'] = eqims + state.last_eqims = eqims elif cmd.startswith(b"B"): mono = bool(int(cmd.decode().removeprefix("B").strip(), 2)) tef.FM_Set_Stereo_Min(2 if mono else 0) out += f"B{int(mono)}\n".encode() - state['forced_mono'] = mono + state.forced_mono = mono elif cmd.startswith(b"D"): - deemp = int(cmd.decode().removeprefix("D").strip()) - dtime = 500 if deemp == 0 else (750 if deemp == 1 else 0) + state.deemp = int(cmd.decode().removeprefix("D").strip()) + dtime = 500 if state.deemp == 0 else (750 if state.deemp == 1 else 0) tef.FM_Set_Deemphasis(dtime) - out += f"D{deemp}\n".encode() - state['deemp'] = deemp + out += f"D{state.deemp}\n".encode() elif cmd.startswith(b"x"): out += b"OK\n" tef.APPL_Set_OperationMode(False) # Enable - tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state["last_tune"]) - if not freq_allowed(state["last_tune"]): tef.AUDIO_Set_Mute(True) - else: tef.AUDIO_Set_Mute(False) - elif cmd.startswith(b"X"): - tef.APPL_Set_OperationMode(True) # turn off + tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune) + elif cmd.startswith(b"X"): tef.APPL_Set_OperationMode(True) # turn off elif cmd.startswith(b"W"): - bw = int(cmd.decode().removeprefix("W").strip()) - auto = (bw == 0) - tef.FM_Set_Bandwidth(auto, 2360 if auto else (bw // 100)) - state['bw'] = bw - out += f"W{bw}\n".encode() - elif cmd.startswith(b"?"): - out += b"XRD Python driver\n" + state.bw = int(cmd.decode().removeprefix("W").strip()) + auto = (state.bw == 0) + tef.FM_Set_Bandwidth(auto, 2360 if auto else (state.bw // 100)) + out += f"W{state.bw}\n".encode() + elif cmd.startswith(b"?"): out += b".XRD Python driver\n" elif cmd.startswith(b"S"): cmd = cmd[1:] if cmd != b"": arg = int(cmd.decode()[1:].strip()) match cmd[0]: - case 97: state['scan_start'] = (arg + 5) // 10 - case 98: state['scan_stop'] = (arg + 5) // 10 - case 99: state['scan_step'] = (arg + 5) // 10 - case 119: state['scan_bw'] = arg + case 97: state.scan_start = (arg + 5) // 10 + case 98: state.scan_stop = (arg + 5) // 10 + case 99: state.scan_step = (arg + 5) // 10 + case 119: state.scan_bw = arg else: start = True - tef.FM_Set_Bandwidth((state["scan_bw"] == 0), state["scan_bw"]) + tef.FM_Set_Bandwidth((state.scan_bw == 0), state.scan_bw) conn.sendall(b"U") - for freq in range(state["scan_start"], state["scan_stop"] + state["scan_step"], state["scan_step"]): + for freq in range(state.scan_start, state.scan_stop + state.scan_step, state.scan_step): if not start: conn.sendall(b", ") # Prevent trailing comma, because the FM-DX-Webserver spectrum plugin treats us as actual firmware and throws as harder api, without it we're treated as a module start = False tef.FM_Tune_To(TEF6686.TuneTo_Mode.Search, freq) # Auto mutes, less commands sent time.sleep(0.0067) # sick seven - if not freq_allowed(freq): - conn.sendall(f"{freq * 10} = 11.25".encode()) - continue _, level, *_ = d if (d := tef.FM_Get_Quality_Data()) else (None, None) if level is None: continue conn.sendall(str(freq * 10).encode() + b" = " + str((level / 10) + 11.25).encode()) conn.sendall(b"\n") - tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state["last_tune"]) - if not freq_allowed(state["last_tune"]): tef.AUDIO_Set_Mute(True) - else: tef.AUDIO_Set_Mute(False) - tef.FM_Set_Bandwidth((state["bw"] == 0), 2360 if (state["bw"] == 0) else (state["bw"] // 100)) + tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune) + tef.FM_Set_Bandwidth((state.bw == 0), 2360 if (state.bw == 0) else (state.bw // 100)) return out -PERIODIC_FUNCTIONS: list[tuple[Callable, float, libtimer.Timer]] = [] +PERIODIC_FUNCTIONS: list[tuple[Callable, float, Timer]] = [] def periodic(t: float): def decorator(func): - PERIODIC_FUNCTIONS.append((func, t, libtimer.Timer())) + PERIODIC_FUNCTIONS.append((func, t, Timer())) @wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper @@ -163,15 +166,15 @@ def reset_periodic(): def run_periodic(*args, **kwargs): for (func, t, timer) in PERIODIC_FUNCTIONS: - if timer.get_time() > t: + max_catchup = 10 + count = 0 + while timer.get_time() > t and count < max_catchup: func(*args, **kwargs) - timer.reset() + timer.subtract(t) + count += 1 @periodic(SS_UPDATE_INTERVAL) -def send_signal_status(tef: TEF6686, conn: socket.socket, state: dict): - if not freq_allowed(state["last_tune"]): - conn.sendall(b"Sm11.25,0,0,0\n\n") - return +def send_signal_status(tef: TEF6686, conn: socket.socket, state: State): stereo_pilot, _ = d if (d := tef.FM_Get_Signal_Status()) else (None, None) res = tef.FM_Get_Quality_Data() @@ -181,17 +184,15 @@ def send_signal_status(tef: TEF6686, conn: socket.socket, state: dict): level = level / 10 data = b"S" - if state['forced_mono']: data += b"M" + + if state.forced_mono: data += b"M" elif stereo_pilot: data += b"s" else: data += b"m" - data += f"{level + 11.25},{wam//10},{usn//10},{bandwidth}\n\n".encode() - conn.sendall(data) + conn.sendall(data + f"{level + 11.25},{wam//10},{usn//10},{bandwidth}\n\n".encode()) @periodic(RDS_UPDATE_INTERVAL) -def send_rds_data(tef: TEF6686, conn: socket.socket, state: dict): - if not freq_allowed(state["last_tune"]): return - +def send_rds_data(tef: TEF6686, conn: socket.socket, state: State): res = tef.FM_Get_RDS_Data__decoder() if res is None: return status, A, B, C, D, dec_error = res @@ -209,7 +210,7 @@ def send_rds_data(tef: TEF6686, conn: socket.socket, state: dict): data = b"R" data += f"{B:04X}{C:04X}{D:04X}{err:02X}\n".encode() - pi_error = (dec_error >> 8) & 0b11 + pi_error = (dec_error >> 6) & 0b11 if pi_error < 3: data += b"P" + f"{A:04X}".encode() data += b"?" * pi_error + b"\n" @@ -218,7 +219,6 @@ def send_rds_data(tef: TEF6686, conn: socket.socket, state: dict): if pi_error < 3: data += b"P" + f"{C:04X}".encode() data += b"?" * pi_error + b"\n" - if not data.strip(): return conn.sendall(data) def run_server(): @@ -229,25 +229,17 @@ def run_server(): hw_major = hw_version >> 8 hw_minor = hw_version & 127 sw_major = sw_version >> 8 + sw_minor = sw_version & 127 variant_str = "TEF6686" if variant == 1: variant_str = "TEF6687" elif variant == 9: variant_str = "TEF6688" elif variant == 3: variant_str = "TEF6689" - print(f"{variant_str} (V{hw_major}{hw_minor}{sw_major})") - - state = { - 'last_tune': INITIAL_FREQ, - 'last_eqims': (INITIAL_EQ << 1) | INITIAL_IMS, - 'forced_mono': False, - 'deemp': 0, - 'bw': 0, - 'scan_start': 87500, - 'scan_stop': 108000, - 'scan_step': 100, - 'scan_bw': 0, - } + print(f"{variant_str} (V{hw_major}{hw_minor}{sw_major}.{sw_minor})") + else: print("Could not fetch variant") + + state = State() with socket.socket() as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) @@ -266,11 +258,11 @@ def run_server(): continue # Send initial state - conn.sendall(f"T{state['last_tune']*10}\n".encode()) - conn.sendall(f"G{bin(state['last_eqims']).removeprefix('0b').zfill(2)}\n".encode()) - conn.sendall(f"B{int(state['forced_mono'])}\n".encode()) - conn.sendall(f"D{state['deemp']}\n".encode()) - conn.sendall(f"W{state['bw']}\n".encode()) + conn.sendall(f"T{state.last_tune*10}\n".encode()) + conn.sendall(f"G{bin(state.last_eqims).removeprefix('0b').zfill(2)}\n".encode()) + conn.sendall(f"B{int(state.forced_mono)}\n".encode()) + conn.sendall(f"D{state.deemp}\n".encode()) + conn.sendall(f"W{state.bw}\n".encode()) conn.sendall(f"M0\n".encode()) while True: @@ -282,8 +274,9 @@ def run_server(): except ConnectionAbortedError: break except BlockingIOError: run_periodic(tef, conn, state) - time.sleep(0.025) + time.sleep(0.03) +import traceback if __name__ == "__main__": try: run_server() - except Exception: pass \ No newline at end of file + except Exception as e: traceback.print_exception(e) \ No newline at end of file