kuba/TEF6686_Driver
1
0
Fork 0
Code Issues Activity

am?

Browse Source
This commit is contained in:
kuba
2026-03-21 19:02:38 +01:00
parent 0829592755
commit 98f7750041
1 changed files with 61 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xrd.py
+61 -25
View File
@@ -10,10 +10,34 @@ 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.FM_Tune_To(mode, frequency)
STEREO_BLENDING = False
INITIAL_FREQ = 9500
INITIAL_FREQ_FM = 9500
INITIAL_FREQ_AM = 225
INITIAL_FREQ = INITIAL_FREQ_FM
INITIAL_EQ = True
INITIAL_IMS = True
@@ -36,7 +60,8 @@ RDS_UPDATE_INTERVAL = 0.086
@dataclass
class State:
last_tune: int = INITIAL_FREQ
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
@@ -48,10 +73,10 @@ class State:
last_bw_extend: bool = False
def init_tef():
tef = TEF6686(I2CPCClient(DEVICE, int(os.getenv("BAUD") or 0) or 115200))
tef = AutoFMAM(I2CPCClient(DEVICE, int(os.getenv("BAUD") or 0) or 115200))
tef.init(clock=CLOCK)
tef.AUDIO_Set_Mute(False)
tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, INITIAL_FREQ)
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)
@@ -84,15 +109,22 @@ def authenticate(conn: socket.socket):
conn.close()
return False
def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket):
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()) // 10
if freq < 6500 or freq > 10800: continue
tef.FM_Tune_To(TEF6686.TuneTo_Mode.FM_Jump, freq)
state.last_tune = freq
out += f"T{freq*10}\n".encode()
if freq < 153 or freq > 10800: continue
tef.set_modulation(freq)
match tef.modulation:
case Modulation.FM:
tef.Tune_To(TEF6686.TuneTo_Mode.FM_Jump, freq)
state.last_tune_fm = freq
case Modulation.AM:
tef.Tune_To(TEF6686.TuneTo_Mode.Preset, freq)
state.last_tune_am = freq
freq /= 10
out += f"T{freq*10}\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)
@@ -108,7 +140,7 @@ def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket
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.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune)
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"):
state.bw = int(cmd.decode().removeprefix("W").strip())
@@ -141,8 +173,8 @@ def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket
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)
tef.FM_Set_Bandwidth((state.bw == 0), 2360 if (state.bw == 0) else (state.bw // 100))
tef.Tune_To(TEF6686.TuneTo_Mode.Preset, tef.last_tune)
return out
@@ -169,7 +201,7 @@ def run_periodic(*args, **kwargs):
count += 1
@periodic(SS_UPDATE_INTERVAL)
def send_signal_status(tef: TEF6686, conn: socket.socket, state: State):
def send_signal_status(tef: AutoFMAM, conn: socket.socket, state: State):
res = tef.FM_Get_Quality_Data()
if res is None: return
@@ -182,22 +214,26 @@ def send_signal_status(tef: TEF6686, conn: socket.socket, state: State):
data = b"S"
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 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
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 + 11.25},{wam//10},{usn//10},{bandwidth}\n\n".encode())
@periodic(RDS_UPDATE_INTERVAL)
def send_rds_data(tef: TEF6686, conn: socket.socket, state: State):
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
@@ -263,12 +299,12 @@ def run_server():
conn.setblocking(False)
# Send initial state
conn.sendall(f"T{state.last_tune*10}\n".encode())
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}\n".encode())
conn.sendall(f"M0\n".encode())
conn.sendall(f"M{tef.modulation}\n".encode())
while True:
try: