edi/buzzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync

Browse Source
This commit is contained in:
Eduard Iten
2026-02-25 10:23:34 +01:00
parent f12d9c5c2f
commit 9ef73a2832
4 changed files with 120 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
buzzer_tool/core/commands/put.py Normal file
View File

@@ -0,0 +1,67 @@
# core/commands/put.py
import os
import zlib
import glob
from core.connection import BuzzerError
def get_file_crc32(filepath: str) -> int:
"""Berechnet die IEEE CRC32-Prüfsumme einer Datei in Chunks."""
crc = 0
with open(filepath, 'rb') as f:
while chunk := f.read(4096):
crc = zlib.crc32(chunk, crc)
return crc & 0xFFFFFFFF
def execute(conn, sources: list, target: str):
"""
Lädt Dateien auf den Mikrocontroller hoch.
"""
# 1. Globbing auflösen (Notwendig für Windows, da die CMD Wildcards nicht auflöst)
resolved_files = []
for src in sources:
matches = glob.glob(src)
if matches:
resolved_files.extend(matches)
else:
print(f"Warnung: Datei '{src}' nicht gefunden.")
if not resolved_files:
print("Keine gültigen Dateien zum Übertragen gefunden.")
return
is_target_dir = target.endswith('/')
# Wenn mehrere Dateien angegeben sind, muss das Ziel zwingend ein Verzeichnis sein
if len(resolved_files) > 1 and not is_target_dir:
print("Fehler: Bei mehreren Quelldateien muss das Ziel ein Verzeichnis sein (mit '/' enden).")
return
# 2. Transfer-Schleife
for filepath in resolved_files:
if not os.path.isfile(filepath):
print(f"Überspringe '{filepath}' (ist keine Datei)")
continue
filename = os.path.basename(filepath)
filesize = os.path.getsize(filepath)
crc32 = get_file_crc32(filepath)
dest_path = f"{target}{filename}" if is_target_dir else target
size_kb = filesize / 1024
print(f"Sende 📄 {filename} \033[90m({size_kb:.1f} KB)\033[0m nach {dest_path} ... ", end="", flush=True)
try:
# PUT-Befehl ohne Warten auf 'OK' senden
cmd = f"put {dest_path};{filesize};{crc32}\n"
conn.serial.write(cmd.encode('utf-8'))
conn.serial.flush()
# Warten auf READY, Binärdaten senden und auf abschließendes OK warten
conn.send_binary(filepath)
print("\033[32mErfolgreich\033[0m")
except BuzzerError as e:
print(f"\n ❌ \033[31mFehler vom Controller: {e}\033[0m")
except Exception as e:
print(f"\n ❌ \033[31mÜbertragungsfehler: {e}\033[0m")

82
buzzer_tool/core/connection.py
View File

@@ -63,49 +63,49 @@ class BuzzerConnection:
raise TimeoutError(f"Lese-Timeout ({eff_timeout}s) beim Warten auf Antwort für: '{command}'")
def send_binary(self, filepath: str, chunk_size: int = 512, timeout: float = 10.0):
"""
Ü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)
def send_binary(self, filepath: str, chunk_size: int = 512, timeout: float = 10.0):
"""
Ü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.")
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
# 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)
# Flush blockiert, bis die Daten an den OS-USB-Treiber übergeben wurden
self.serial.flush()
bytes_sent += len(chunk)
with open(filepath, 'rb') as f:
while bytes_sent < file_size:
chunk = f.read(chunk_size)
if not chunk:
break
self.serial.write(chunk)
# Flush blockiert, bis die Daten an den OS-USB-Treiber übergeben wurden
self.serial.flush()
bytes_sent += 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)
# 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).")
raise TimeoutError("Zeitüberschreitung nach Binärtransfer (kein OK empfangen).")