kuba/TEF6686_Driver
1
0
Fork 0
Code Issues Activity

xrd improvements

Browse Source
This commit is contained in:
kuba
2026-03-20 18:16:37 +01:00
parent 06f9fdf83c
commit 5ca98b993e
2 changed files with 106 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libtimer2.py
+14
View File
@@ -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)
xrd.py
+92 -99
View File
@@ -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
except Exception as e: traceback.print_exception(e)