diff --git a/driver/base_tef.py b/driver/base_tef.py index eb2fdc3..5c3a780 100644 --- a/driver/base_tef.py +++ b/driver/base_tef.py @@ -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() diff --git a/driver/protocol.py b/driver/protocol.py index e2768d7..f3391c6 100644 --- a/driver/protocol.py +++ b/driver/protocol.py @@ -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 diff --git a/driver/tef.py b/driver/tef.py index 104607c..0694437 100644 --- a/driver/tef.py +++ b/driver/tef.py @@ -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 diff --git a/xrd.py b/xrd.py index dcb8d6f..7fd0510 100644 --- a/xrd.py +++ b/xrd.py @@ -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