115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
# core/connection.py
|
|
import serial
|
|
import time
|
|
import os
|
|
|
|
class BuzzerError(Exception):
|
|
pass
|
|
|
|
class BuzzerConnection:
|
|
def __init__(self, config):
|
|
self.port = config.get("port")
|
|
self.baudrate = config.get("baudrate", 115200)
|
|
self.timeout = config.get("timeout", 5.0)
|
|
self.serial = None
|
|
|
|
def __enter__(self):
|
|
if not self.port:
|
|
raise ValueError("Kein serieller Port konfiguriert.")
|
|
|
|
# write_timeout verhindert endloses Blockieren auf inaktiven Ports
|
|
self.serial = serial.Serial(
|
|
port=self.port,
|
|
baudrate=self.baudrate,
|
|
timeout=self.timeout,
|
|
write_timeout=self.timeout
|
|
)
|
|
self.serial.reset_input_buffer()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self.serial and self.serial.is_open:
|
|
self.serial.close()
|
|
|
|
def send_command(self, command: str, custom_timeout: float = None) -> list:
|
|
eff_timeout = custom_timeout if custom_timeout is not None else self.timeout
|
|
self.serial.reset_input_buffer()
|
|
|
|
try:
|
|
self.serial.write(f"{command}\n".encode('utf-8'))
|
|
self.serial.flush()
|
|
except serial.SerialTimeoutException:
|
|
raise TimeoutError(f"Schreib-Timeout am Port {self.port}. Ist das Gerät blockiert?")
|
|
|
|
lines = []
|
|
start_time = time.monotonic()
|
|
|
|
while (time.monotonic() - start_time) < eff_timeout:
|
|
if self.serial.in_waiting > 0:
|
|
try:
|
|
line = self.serial.readline().decode('utf-8', errors='ignore').strip()
|
|
if not line:
|
|
continue
|
|
if line == "OK":
|
|
return lines
|
|
elif line.startswith("ERR"):
|
|
err_code = line.split(" ")[1] if " " in line else "UNKNOWN"
|
|
raise BuzzerError(f"Controller meldet Fehlercode: {err_code}")
|
|
else:
|
|
lines.append(line)
|
|
except Exception as e:
|
|
raise BuzzerError(f"Fehler beim Lesen der Antwort: {e}")
|
|
else:
|
|
time.sleep(0.01)
|
|
|
|
raise TimeoutError(f"Lese-Timeout ({eff_timeout}s) beim Warten auf Antwort für: '{command}'")
|
|
|
|
def send_binary(self, filepath: str, chunk_size: int = 4096, timeout: float = 10.0, progress_callback=None):
|
|
"""
|
|
Überträgt eine Binärdatei in Chunks, nachdem das READY-Signal empfangen wurde.
|
|
"""
|
|
# 1. Warte auf die READY-Bestätigung vom Controller
|
|
start_time = time.time()
|
|
ready = False
|
|
while (time.time() - start_time) < timeout:
|
|
if self.serial.in_waiting > 0:
|
|
line = self.serial.readline().decode('utf-8', errors='ignore').strip()
|
|
if line == "READY":
|
|
ready = True
|
|
break
|
|
elif line.startswith("ERR"):
|
|
raise BuzzerError(f"Fehler vor Binärtransfer: {line}")
|
|
time.sleep(0.01)
|
|
|
|
if not ready:
|
|
raise TimeoutError("Kein READY-Signal vom Controller empfangen.")
|
|
|
|
# 2. Sende die Datei in Blöcken
|
|
file_size = os.path.getsize(filepath)
|
|
bytes_sent = 0
|
|
|
|
with open(filepath, 'rb') as f:
|
|
while bytes_sent < file_size:
|
|
chunk = f.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
self.serial.write(chunk)
|
|
self.serial.flush()
|
|
bytes_sent += len(chunk)
|
|
|
|
# Callback aufrufen, falls vorhanden
|
|
if progress_callback:
|
|
progress_callback(len(chunk))
|
|
|
|
# 3. Warte auf das finale OK (oder ERR bei CRC/Schreib-Fehlern)
|
|
start_time = time.time()
|
|
while (time.time() - start_time) < timeout:
|
|
if self.serial.in_waiting > 0:
|
|
line = self.serial.readline().decode('utf-8', errors='ignore').strip()
|
|
if line == "OK":
|
|
return True
|
|
elif line.startswith("ERR"):
|
|
raise BuzzerError(f"Fehler beim Speichern der Binärdatei: {line}")
|
|
time.sleep(0.01)
|
|
|
|
raise TimeoutError("Zeitüberschreitung nach Binärtransfer (kein OK empfangen).") |