kuba/TEF6686_Driver
1
0
Fork 0
Code Issues Activity

say ok on login, not x

Browse Source
This commit is contained in:
kuba
2026-03-21 15:26:45 +01:00
parent 6d1b600b0b
commit 90435b477a
4 changed files with 34 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
driver/base_tef.py
+6 -4
View File
@@ -1,6 +1,7 @@
import struct
from driver.protocol import I2CPCClient, time
from driver.patches import *
import time
from typing import TypeVar, ParamSpec
@@ -23,8 +24,8 @@ class BaseTEF668X:
self.p.write_i2c(self.address, b"\x1c\x00\x00")
self.p.write_i2c(self.address, b"\x1c\x00\x74")
def send_patch(_patch: bytes):
for i in range(0, len(_patch), 24):
data = _patch[i:i+24]
for i in range(0, len(_patch), 20):
data = _patch[i:i+20]
if self.p.write_i2c(self.address, b"\x1b" + data)[0] != 0: raise Exception
send_patch(bytes(patch))
self.p.write_i2c(self.address, b"\x1c\x00\x00")
@@ -35,15 +36,16 @@ class BaseTEF668X:
def APPL_Get_Operation_Status(self):
data = self.p.write_read_i2c(self.address, b"\x40\x80\x01", 2)
start = time.monotonic()
while data[0] != 0:
data = self.p.write_read_i2c(self.address, b"\x40\x80\x01", 2)
time.sleep(0.01)
if (time.monotonic() - start) > 1: raise TimeoutError("Could not get status")
return data[-1]
def APPL_Set_ReferenceClock(self, clock: int, type_clock: bool):
return self.p.write_i2c(self.address, b"\x40\x04\x01" + struct.pack(">IH", clock, type_clock))
def APPL_Activate(self):
return self.p.write_i2c(self.address, b"\x40\x05\x01\x00\x01")
def APPL_Activate(self): return self.p.write_i2c(self.address, b"\x40\x05\x01\x00\x01")
def init(self, patch = tef_102_patch, patch_lut = tef_102_patch_lut, clock: int = 9216000):
self._reset()
driver/protocol.py
+4 -2
View File
@@ -80,8 +80,8 @@ class I2CPCClient:
def write_eeprom(self, addr: int, data: bytes):
payload = bytes([7]) + struct.pack(">H", addr) + data
return self._send_packet(payload, False)
def read_eeprom(self, addr: int, len: int):
payload = bytes([8]) + struct.pack(">H", addr) + bytes([len])
def read_eeprom(self, addr: int, datalen: int):
payload = bytes([8]) + struct.pack(">H", addr) + bytes([datalen])
return self._send_packet(payload, False)
def version(self): return self._send_packet(bytes([4]))
@@ -95,6 +95,8 @@ class I2CPCClient:
def set_baudrate(self, baud: int):
payload = bytes([6]) + struct.pack(">I", baud)
out = self._send_packet(payload)
# Response is sent in old baudrate
self.ser.baudrate = baud
return out
driver/tef.py
+16 -27
View File
@@ -19,10 +19,10 @@ def _command_wrapper(func: Callable[Concatenate[TEF6686, P], tuple[bytes, int |
data, read_bytes, out_parser = func(self, *args, **kwargs)
data, read_bytes, out_parser = self._pre_command_(func, data, read_bytes, out_parser)
if read_bytes: data = self.p.write_read_i2c(self.address, data, read_bytes)
if read_bytes:
data = self.p.write_read_i2c(self.address, data, read_bytes)
else: data = self.p.write_i2c(self.address, data)
if out_parser: return out_parser(data)
return data
return out_parser(data) if out_parser else data
return inner
class SignedInt(int): ...
@@ -53,14 +53,10 @@ class TEF6686(BaseTEF668X):
return b"\x20\x02\x01" + pack(afu_bw_mode, afu_bandwidth, afu_mute_time, afu_sample_time), None, None
@_command_wrapper
@overload
def FM_Set_Bandwidth(self, mode: bool = True, bandwidth: int = 2360, control_sensitivity: int = 1000, low_level_sensitivity: int = 1000, min_bandwidth: int = 560, nominal_bandwidth: int = 2360, control_attack: int = 300):
"""Minimal bandwidth requires p2.13, while nominal_bandwidth and control_attack require p2.17 (we have p2.24)"""
return b"\x20\x0a\x01" + pack(mode, bandwidth, control_sensitivity, low_level_sensitivity, min_bandwidth, nominal_bandwidth, control_attack), None, None
@_command_wrapper
def FM_Set_Bandwidth(self, mode: bool = True, bandwidth: int = 2360, control_sensitivity: int = 1000, low_level_sensitivity: int = 1000):
return b"\x20\x0a\x01" + pack(mode, bandwidth, control_sensitivity, low_level_sensitivity), None, None
@_command_wrapper
def AM_Set_Bandwidth(self, bandwidth: int = 40): return b"\x21\x0a\x01\x00\x00" + pack(bandwidth), None, None
@_command_wrapper
@@ -81,11 +77,8 @@ class TEF6686(BaseTEF668X):
def FM_Set_ChannelEqualizer(self, mode: bool = False): return b"\x20\x16\x01" + pack(mode), None, None
@_command_wrapper
@overload
def FM_Set_NoiseBlanker(self, mode: bool = True, sensitivity: int = 1000, modulation: int = 900, offset: int = 1, attack: int = 140, decay: int = 2800):
return b"\x20\x17\x01" + pack(mode, sensitivity, 0, modulation, offset, attack, decay), None, None
@_command_wrapper
def FM_Set_NoiseBlanker(self, mode: bool = True, sensitivity: int = 1000): return b"\x20\x17\x01" + pack(mode, sensitivity), None, None
@_command_wrapper
def FM_Set_NoiseBlanker_Options(self, blank_time: int = 210, blank_time2: int = 210, blank_modulation: int = 250): return b"\x20\x18\x01" + pack(blank_time, blank_time2, blank_modulation), None, None
@@ -224,8 +217,7 @@ class TEF6686(BaseTEF668X):
@_command_wrapper
def AUDIO_Set_WaveGen(self, mode: int = 0, offset: int = 0, amplitude1: int = -200, frequency1: int = 400, amplitude2: int = -200, frequency2: int = 1000):
# Not sure if offset is signed or not
return b"\x30\x18\x01" + pack(mode, offset, SignedInt(amplitude1), frequency1, SignedInt(amplitude2), frequency2), None, None
return b"\x30\x18\x01" + pack(mode, SignedInt(offset), SignedInt(amplitude1), frequency1, SignedInt(amplitude2), frequency2), None, None
@_command_wrapper
def APPL_Set_OperationMode(self, mode: bool = True): return b"\x40\x01\x01" + pack(mode), None, None
@@ -238,7 +230,7 @@ class TEF6686(BaseTEF668X):
@staticmethod
def _get_quality_data(data) -> None | tuple[int, int, int, int, int, int, int]:
if data[0] != 0: return None
return struct.unpack(">HhHHhHH", data[1:])
return struct.unpack_from(">HhHHhHH", data, 1)
@_command_wrapper
def FM_Get_Quality_Status(self):
return b"\x20\x80\x01", 7*2, self._get_quality_data
@@ -254,15 +246,12 @@ class TEF6686(BaseTEF668X):
@staticmethod
def _get_rds_data_proc_decoder(data) -> None | tuple[int, int, int, int, int, int]:
if data[0] != 0: return None
return struct.unpack(">HHHHHH", data[1:])
if data[0] != 0: return None
return struct.unpack_from(">HHHHHH", data, 1)
@staticmethod
def _get_rds_data_proc_demodulator(data):
if data[0] != 0: return None
status = (data[1] << 8) | data[2]
raw_data_high = (data[3] << 8) | data[4]
raw_data_low = (data[5] << 8) | data[6]
return status, (raw_data_high << 16) | raw_data_low
if data[0] != 0: return None
return struct.unpack_from(">HI", data, 1)
@_command_wrapper
def FM_Get_RDS_Status__decoder(self):
return b"\x20\x82\x01", 2*6, self._get_rds_data_proc_decoder
@@ -283,7 +272,7 @@ class TEF6686(BaseTEF668X):
"""
def proc(data):
if data[0] != 0: return None
return struct.unpack(">HH", data[1:])
return struct.unpack_from(">HH", data, 1)
return b"\x20\x84\x01", 2*2, proc
@_command_wrapper
@@ -293,7 +282,7 @@ class TEF6686(BaseTEF668X):
"""
def proc(data):
if data[0] != 0: return None
value, *_ = struct.unpack(">H", data[1:])
value, *_ = struct.unpack_from(">H", data, 1)
return (value & (1 << 15)) != 0, (value & (1 << 14)) != 0
return b"\x20\x85\x01", 2, proc
@@ -301,7 +290,7 @@ class TEF6686(BaseTEF668X):
def FM_Get_Processing_Status(self):
def proc(data):
if data[0] != 0: return None
return struct.unpack(">HHHH", data[1:])
return struct.unpack_from(">HHHH", data, 1)
return b"\x20\x86\x01", 2*4, proc
# Get_Operation_Status is defined in the base
@@ -317,14 +306,14 @@ class TEF6686(BaseTEF668X):
def APPL_Get_Identification(self):
def proc(data):
if data[0] != 0: return None
return struct.unpack(">HHH", data[1:])
return struct.unpack_from(">HHH", data, 1)
return b"\x40\x82\x01", 2*3, proc
@_command_wrapper
def APPL_Get_LastWrite(self):
def proc(data):
if data[0] != 0: return None
return struct.unpack(">HHHHHHH", data[1:])
return struct.unpack_from(">HHHHHHH", data, 1)
return b"\x40\x83\x01", 2*7, proc
@overload
@@ -401,13 +390,13 @@ class TEF6688(TEF6686):
def FM_Get_Interface_Status(self):
def parse(data):
if data[0] != 0: return None
return struct.unpack(">H", data[1:])
return struct.unpack_from(">H", data, 1)
return b"\x20\x87\x01", 2, parse
@_8_command_wrapper
def AM_Get_Interface_Status(self):
def parse(data):
if data[0] != 0: return None
return struct.unpack(">H", data[1:])
return struct.unpack_from(">H", data, 1)
return b"\x21\x87\x01", 2, parse
# This is a combination of all of the features of those all
xrd.py
+8 -13
View File
@@ -67,7 +67,6 @@ 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()
tries = 0
while True:
try:
@@ -75,15 +74,13 @@ def authenticate(conn: socket.socket):
if not data: break
if data.strip() == expected_hash.strip():
conn.sendall(b"o1,0\n")
conn.sendall(b"o1,0\nOK\n")
return True
else:
tries += 1
conn.sendall(b"a0\n") # You failed
except BlockingIOError:
if tries > 2:
conn.close()
break
except BlockingIOError: pass
conn.close()
return False
def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket):
@@ -103,17 +100,15 @@ def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket
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))
tef.FM_Set_Stereo_Min(2 if mono else 0)
out += f"B{int(mono)}\n".encode()
state.forced_mono = mono
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"):
out += b"OK\n"
tef.APPL_Set_OperationMode(False) # Enable
tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, state.last_tune)
elif cmd.startswith(b"X"): tef.APPL_Set_OperationMode(True) # turn off
@@ -122,10 +117,10 @@ def process_command(tef: TEF6686, data: bytes, state: State, conn: socket.socket
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"?"): out += b">XRD Python driver\n"
elif cmd.startswith(b"S"):
cmd = cmd[1:]
if cmd != b"":
if cmd.strip():
arg = int(cmd.decode()[1:].strip())
match cmd[0]:
case 97: state.scan_start = (arg + 5) // 10