You've already forked TEF6686_Driver
some changes
This commit is contained in:
41
xrd.py
41
xrd.py
@@ -14,6 +14,10 @@ import os
|
||||
from functools import wraps
|
||||
from typing import Callable
|
||||
|
||||
INITIAL_FREQ = 9500
|
||||
INITIAL_EQ = False
|
||||
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"
|
||||
@@ -38,12 +42,13 @@ def init_tef():
|
||||
tef = TEF6686(p)
|
||||
tef.init(clock=CLOCK)
|
||||
tef.AUDIO_Set_Mute(False)
|
||||
tef.AUDIO_Set_Volume(40)
|
||||
tef.FM_Tune_To(1, 9500)
|
||||
tef.AUDIO_Set_Volume(45)
|
||||
tef.FM_Tune_To(1, INITIAL_FREQ)
|
||||
tef.FM_Set_RDS(1)
|
||||
tef.FM_Set_ChannelEqualizer(True)
|
||||
tef.FM_Set_MphSuppression(True)
|
||||
tef.APPL_Set_OperationMode(True)
|
||||
tef.FM_Set_ChannelEqualizer(INITIAL_EQ)
|
||||
tef.FM_Set_MphSuppression(INITIAL_IMS)
|
||||
tef.FM_Set_Stereo_Max(False) # Disables stereo blend??!!??!?
|
||||
tef.APPL_Set_OperationMode(True) # Turn off
|
||||
return tef
|
||||
|
||||
def authenticate(conn: socket.socket):
|
||||
@@ -62,36 +67,36 @@ def process_command(tef: TEF6686, data: bytes, state: dict, conn: socket.socket)
|
||||
if cmd.startswith(b"T"):
|
||||
freq = int(cmd.decode().removeprefix("T").strip()) // 10
|
||||
if freq < 6500 or freq > 10800: continue
|
||||
state['last_tune'] = freq
|
||||
tef.FM_Tune_To(1, freq)
|
||||
if not freq_allowed(freq): tef.AUDIO_Set_Mute(True)
|
||||
else: tef.AUDIO_Set_Mute(False)
|
||||
state['last_tune'] = freq
|
||||
out += f"T{freq*10}\n".encode()
|
||||
elif cmd.startswith(b"G"):
|
||||
eqims = int(cmd.decode().removeprefix("G").strip(), 2)
|
||||
state['last_eqims'] = eqims
|
||||
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"):
|
||||
mono = bool(int(cmd.decode().removeprefix("B").strip(), 2))
|
||||
state['forced_mono'] = mono
|
||||
tef.FM_Set_Stereo_Min(2 if mono else 0)
|
||||
out += f"B{int(mono)}\n".encode()
|
||||
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'] = deemp
|
||||
tef.FM_Set_Deemphasis(dtime)
|
||||
out += f"D{deemp}\n".encode()
|
||||
state['deemp'] = deemp
|
||||
elif cmd.startswith(b"x"):
|
||||
out += b"OK\n"
|
||||
tef.APPL_Set_OperationMode(False)
|
||||
tef.APPL_Set_OperationMode(False) # Enable
|
||||
tef.FM_Tune_To(1, 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)
|
||||
tef.APPL_Set_OperationMode(True) # turn off
|
||||
elif cmd.startswith(b"W"):
|
||||
bw = int(cmd.decode().removeprefix("W").strip())
|
||||
auto = (bw == 0)
|
||||
@@ -110,17 +115,22 @@ def process_command(tef: TEF6686, data: bytes, state: dict, conn: socket.socket)
|
||||
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"])
|
||||
conn.sendall(b"U")
|
||||
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(2, freq) # Auto mutes, less commands sent
|
||||
time.sleep(0.0064)
|
||||
if not freq_allowed(freq):
|
||||
conn.sendall(f"{freq * 10} = 11.25, ".encode())
|
||||
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() + b", ")
|
||||
conn.sendall(str(freq * 10).encode() + b" = " + str((level / 10) + 11.25).encode())
|
||||
conn.sendall(b"\n")
|
||||
|
||||
tef.FM_Tune_To(1, state["last_tune"])
|
||||
@@ -177,7 +187,6 @@ def send_rds_data(tef: TEF6686, conn: socket.socket, state: dict):
|
||||
res = tef.FM_Get_RDS_Data__decoder()
|
||||
if res is None: return
|
||||
status, A, B, C, D, dec_error = res
|
||||
if status is None or A is None or B is None or C is None or D is None or dec_error is None: return # Fucking hate pyright
|
||||
|
||||
if (status & (1 << 9) == 0) or (status & (1 << 15) == 0): return
|
||||
|
||||
@@ -220,8 +229,8 @@ def run_server():
|
||||
print(f"{variant_str} (V{hw_major}{hw_minor}{sw_major})")
|
||||
|
||||
state = {
|
||||
'last_tune': 9500,
|
||||
'last_eqims': 0b11,
|
||||
'last_tune': INITIAL_FREQ,
|
||||
'last_eqims': (INITIAL_EQ << 1) | INITIAL_IMS,
|
||||
'forced_mono': False,
|
||||
'deemp': 0,
|
||||
'bw': 0,
|
||||
|
||||
Reference in New Issue
Block a user