From 98f77500411a587d7e9ee76f0f2ab86a1fab830f Mon Sep 17 00:00:00 2001 From: KubaPro010 Date: Sat, 21 Mar 2026 19:02:38 +0100 Subject: [PATCH] am? --- xrd.py | 86 +++++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 61 insertions(+), 25 deletions(-) diff --git a/xrd.py b/xrd.py index 2f61390..cd18aec 100644 --- a/xrd.py +++ b/xrd.py @@ -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: