# 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) + Spectrum plugin # Released under the Unlicense (however both FM-DX-Webserver and the firmware are under the tyranny of GPL-3) import socket, os, time import hashlib, secrets, string from libtimer2 import Timer from driver.tef import TEF6686 from driver.protocol import I2CPCClient from functools import wraps from typing import Callable from dataclasses import dataclass from enum import IntEnum class Modulation(IntEnum): FM = 0 AM = 1 class AutoFMAM(TEF6686): def __init__(self, p: I2CPCClient) -> None: self.modulation = Modulation.FM self.last_tune = INITIAL_FREQ super().__init__(p, 0x64) def set_modulation(self, frequency: int): if frequency < 2700: self.modulation = Modulation.AM else: self.modulation = Modulation.FM def Tune_To(self, mode, frequency: int | None): if frequency is not None: self.set_modulation(frequency) self.last_tune = frequency if self.modulation == Modulation.FM: self.FM_Tune_To(mode, frequency) elif self.modulation == Modulation.AM: self.AM_Tune_To(mode, frequency) def Get_Quality_Data(self): if self.modulation == Modulation.FM: return self.FM_Get_Quality_Data() elif self.modulation == Modulation.AM: return self.AM_Get_Quality_Data() def Set_Bandwidth(self, bw_client: int): bw_client //= 100 if self.modulation == Modulation.FM: auto = (bw_client == 0) self.FM_Set_Bandwidth(auto, 2360 if auto else bw_client) elif self.modulation == Modulation.AM: self.AM_Set_Bandwidth(bw_client) STEREO_BLENDING = False INITIAL_FREQ_FM = 9500 INITIAL_FREQ_AM = 225 INITIAL_FREQ = INITIAL_FREQ_FM INITIAL_EQ = True 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", "COM6") DEFAULT_PW = "test" PASSWORD = os.getenv("PW", DEFAULT_PW) if PASSWORD == DEFAULT_PW: print("Password left at default. Please change if neccesary") 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.175 RDS_UPDATE_INTERVAL = 0.087 @dataclass class State: last_tune_fm: int = INITIAL_FREQ_FM last_tune_am: int = INITIAL_FREQ_AM last_eqims: int = (INITIAL_EQ << 1) | INITIAL_IMS forced_mono: bool = False deemp: int = 0 bw_fm: int = 0 bw_am: int = 0 scan_start: int = 8750 scan_stop: int = 10800 scan_step: int = 10 scan_bw: int = 0 last_bw_extend: bool = False def init_tef(): tef = AutoFMAM(I2CPCClient(DEVICE, int(os.getenv("BAUD") or 0) or 115200)) tef.init(clock=CLOCK) tef.AUDIO_Set_Mute(False) tef.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(STEREO_BLENDING) tef.APPL_Set_OperationMode(True) # Turn off return tef 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() while True: data = conn.recv(len(expected_hash)) if not data: break if data.strip() == expected_hash.strip(): conn.sendall(b"o1,0\nOK\n") return True else: conn.sendall(b"a0\n") # You failed break conn.close() return False def process_command(tef: AutoFMAM, 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()) if freq < 144 or freq > 108000: continue mode = TEF6686.TuneTo_Mode.FM_Jump if tef.modulation == Modulation.AM: mode = TEF6686.TuneTo_Mode.Preset tef.set_modulation(freq) match tef.modulation: case Modulation.FM: tef.FM_Tune_To(mode, freq // 10) state.last_tune_fm = tef.last_tune = freq // 10 case Modulation.AM: tef.AM_Tune_To(TEF6686.TuneTo_Mode.Preset, freq) state.last_tune_am = tef.last_tune = freq out += f"T{freq}\nM{tef.modulation}\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 elif cmd.startswith(b"B"): state.forced_mono = bool(int(cmd.decode().removeprefix("B").strip(), 2)) tef.FM_Set_Stereo_Min(2 if state.forced_mono else 0) out += f"B{int(state.forced_mono)}\n".encode() elif cmd.startswith(b"D"): 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{state.deemp}\n".encode() elif cmd.startswith(b"x"): tef.Tune_To(TEF6686.TuneTo_Mode.Preset, tef.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()) if tef.modulation == Modulation.FM: state.bw_fm = bw else: state.bw_am = bw tef.Set_Bandwidth(bw) out += f"W{bw}\n".encode() elif cmd.startswith(b"?"): out += b">XRD Python driver\n" elif cmd.startswith(b"S"): cmd = cmd[1:] if cmd.strip(): 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 else: saved_last_tune = tef.last_tune tef.Set_Bandwidth(state.scan_bw) conn.sendall(b"U") start = True for freq in range(state.scan_start, state.scan_stop + state.scan_step, state.scan_step): if not start: conn.sendall(b", ") start = False tef.Tune_To(TEF6686.TuneTo_Mode.Search, freq) time.sleep(0.0067) # sick seven _, level, *_ = d if (d := tef.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.Tune_To(TEF6686.TuneTo_Mode.Preset, saved_last_tune) tef.Set_Bandwidth(state.bw_fm if tef.modulation == Modulation.FM else state.bw_am) elif cmd.startswith(b"M"): mod = int(cmd.decode().removeprefix("M").strip()) if mod == 1: tef.Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune_am) else: tef.Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune_fm) return out PERIODIC_FUNCTIONS: list[tuple[Callable, float, Timer]] = [] def periodic(t: float): def decorator(func): PERIODIC_FUNCTIONS.append((func, t, Timer())) @wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper return decorator def reset_periodic(): for (_, _, timer) in PERIODIC_FUNCTIONS: timer.reset() def run_periodic(*args, **kwargs): for (func, t, timer) in PERIODIC_FUNCTIONS: max_catchup = 10 count = 0 while timer.get_time() > t and count < max_catchup: func(*args, **kwargs) timer.subtract(t) count += 1 @periodic(SS_UPDATE_INTERVAL) def send_signal_status(tef: AutoFMAM, conn: socket.socket, state: State): res = tef.Get_Quality_Data() if res is None: return status, level, usn, wam, _, bandwidth, *_ = res ms_since_tune = (status & 0x3ff) / 10 if ms_since_tune < 36: return # Give only "quality data" level = level / 10 if level > 120 or level < -20: return level += 11.25 # Convert to dbf data = b"S" if tef.modulation == Modulation.FM: stereo_pilot, _ = d if (d := tef.FM_Get_Signal_Status()) else (None, None) letter = b"s" if stereo_pilot else b"m" if state.forced_mono: letter = letter.upper() data += letter if not state.last_bw_extend and level > 38 and usn < 7: tef.FM_Set_Bandwidth_Options(400) state.last_bw_extend = True if state.last_bw_extend and (level < 33 or usn > 10): tef.FM_Set_Bandwidth_Options() state.last_bw_extend = False else: data += b"M" conn.sendall(data + f"{level},{wam//10},{usn//10},{bandwidth}\n\n".encode()) @periodic(RDS_UPDATE_INTERVAL) def send_rds_data(tef: AutoFMAM, conn: socket.socket, state: State): if tef.modulation != Modulation.FM: return res = tef.FM_Get_RDS_Data__decoder() if res is None: return status, A, B, C, D, dec_error = res dec_error >>= 8 if (status & (1 << 9) == 0) or (status & (1 << 15) == 0) or (status & (1 << 13) == (1 << 13)): return data = b"" if (status & (1 << 13) == 0): err = 0 err |= (dec_error & 0x30) >> 4 err |= (dec_error & 0xC) err |= (dec_error & 3) << 4 data = b"R" data += f"{B:04X}{C:04X}{D:04X}{err:02X}\n".encode() pi_error = (dec_error >> 6) & 0b11 if pi_error < 3: data += b"P" + f"{A:04X}".encode() data += b"?" * pi_error + b"\n" elif status & (1 << 12): pi_error = (dec_error >> 2) & 0b11 if pi_error < 3: data += b"P" + f"{C:04X}".encode() data += b"?" * pi_error + b"\n" conn.sendall(data) def run_server(): with init_tef() as tef: state = State() device, hw_version, sw_version = d if (d := tef.APPL_Get_Identification()) else (None, None, None) if device and hw_version and sw_version: variant = device & 127 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}.{sw_minor})") else: print("Could not fetch variant") with socket.socket() as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((HOST, PORT)) s.listen() print(f"Server listening on {HOST}:{PORT}") while True: conn, addr = s.accept() with conn: reset_periodic() print(f"Connected by {addr}") if not authenticate(conn): print("Authentication failed.") continue conn.setblocking(False) # Send initial state conn.sendall(f"T{state.last_tune_fm*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_fm}\n".encode()) conn.sendall(f"M{tef.modulation}\n".encode()) while True: try: if not (data := conn.recv(1024)): break try: resp = process_command(tef, data, state, conn) if resp: conn.sendall(resp) except Exception as e: try: conn.sendall(b"!" + str(e).encode() + b"\n") except Exception: pass except ConnectionResetError: break except ConnectionAbortedError: break except BlockingIOError: try: run_periodic(tef, conn, state) except Exception as e: try: conn.sendall(b"!" + str(e).encode() + b"\n") except Exception: pass time.sleep(0.03) import traceback if __name__ == "__main__": try: run_server() except Exception as e: traceback.print_exception(e)