Дорабтка c HackRF

scripts/emulator.py

@@ -0,0 +1,230 @@
+import sys
+import time
+import ast
+import signal
+import os
+
+OUT_BUFFER_LEN = 32
+RECONNECT_DELAY = 0.5
+REPEAT_COUNT = 100         # повтор
+DEFAULT_SEND_DELAY = 0.1   # по умолчанию (10 байт/с)
+
+stop_flag = False
+
+
+# ================= SIGNALS =================
+
+def handle_sigint(signum, frame):
+    global stop_flag
+    stop_flag = True
+
+
+signal.signal(signal.SIGINT, handle_sigint)
+signal.signal(signal.SIGTERM, handle_sigint)
+
+
+# ================= UI =================
+
+last_status = ""
+
+
+def status(msg):
+    global last_status
+    if msg != last_status:
+        sys.stderr.write("\r" + msg + " " * 20)
+        sys.stderr.flush()
+        last_status = msg
+
+
+def status_clear():
+    sys.stderr.write("\r" + " " * 80 + "\r")
+    sys.stderr.flush()
+
+
+# ================= DATA =================
+
+def load_array(path):
+    while True:
+        try:
+            with open(path, "r") as f:
+                content = f.read().strip()
+
+            data = ast.literal_eval(content)
+
+            result = []
+            for x in data:
+                if isinstance(x, str):
+                    x = int(x, 0)
+                result.append(x & 0xFF)
+
+            print(f"[emu] Загружено {len(result)} байт", file=sys.stderr)
+            return result
+
+        except Exception as e:
+            print(f"[emu] Ошибка чтения: {e}", file=sys.stderr)
+            time.sleep(1)
+
+
+# ================= PIPE =================
+
+def open_pipe(path):
+    while not stop_flag:
+        try:
+            fd = os.open(path, os.O_WRONLY | os.O_NONBLOCK)
+            status("[emu] pipe: подключен")
+            return os.fdopen(fd, "wb", buffering=0)
+
+        except OSError:
+            status("[emu] pipe: нет читателя")
+            time.sleep(RECONNECT_DELAY)
+
+    return None
+
+
+# ================= ARGS PARSING =================
+
+def parse_args():
+    """
+    Разбирает аргументы командной строки.
+    Использование:
+        python3 emulator.py array.txt /tmp/gpio_pipe [задержка] [logfile|-]
+    
+    Где:
+        задержка  — float, секунд между байтами (по умолчанию DEFAULT_SEND_DELAY)
+        logfile   — путь к файлу лога или "-" для stderr
+    """
+    if len(sys.argv) < 3:
+        print("Использование: python emulator.py array.txt <pipe_path|-> [задержка] [logfile|-]", file=sys.stderr)
+        sys.exit(1)
+
+    array_path = sys.argv[1]
+    pipe_path = sys.argv[2]
+
+    send_delay = DEFAULT_SEND_DELAY
+    log_file = None
+    arg_offset = 0
+
+    if len(sys.argv) > 3:
+        try:
+            send_delay = float(sys.argv[3])
+            arg_offset = 1
+            print(f"[emu] Задержка: {send_delay} с (~{1/send_delay:.0f} байт/с)", file=sys.stderr)
+        except ValueError:
+            arg_offset = 0
+
+    log_idx = 3 + arg_offset
+    if len(sys.argv) > log_idx:
+        log_path = sys.argv[log_idx]
+        if log_path == "-":
+            log_file = sys.stderr
+        else:
+            log_file = open(log_path, "a")
+
+    return array_path, pipe_path, send_delay, log_file
+
+# ================= MAIN =================
+
+def main():
+    array_path, pipe_path, send_delay, log_file = parse_args()
+
+    data = load_array(array_path)
+
+    log_buffer = []
+    total_bytes = 0
+    packet_id = 0
+    i = 0
+    repeat = 0
+
+    # ===== STDOUT MODE =====
+    if pipe_path == "-":
+        pipe = sys.stdout.buffer  # бинарный stdout
+        print(f"[emu] Режим stdout, {len(data)} байт в цикле", file=sys.stderr)
+        try:
+            while not stop_flag:
+                byte = data[i]
+                pipe.write(bytes([byte]))
+                pipe.flush()
+                
+                log_buffer.append(byte)
+                total_bytes += 1
+                
+                if len(log_buffer) >= OUT_BUFFER_LEN and log_file:
+                    packet_id += 1
+                    log_file.write(f"--- Пакет #{packet_id} ({OUT_BUFFER_LEN} байт) ---\n")
+                    for idx, b in enumerate(log_buffer):
+                        log_file.write(f" [{idx:02d}] = 0x{b:02x} ({b:3d})\n")
+                    log_file.flush()
+                    log_buffer.clear()
+                
+                repeat += 1
+                if repeat >= REPEAT_COUNT:
+                    repeat = 0
+                    i += 1
+                    if i >= len(data):
+                        i = 0
+                
+                time.sleep(send_delay)
+                
+        except KeyboardInterrupt:
+            pass
+        except BrokenPipeError:
+            pass  # нормально для pipe
+        finally:
+            status_clear()
+            print("[emu] завершён", file=sys.stderr)
+        return
+
+    # ===== FIFO MODE =====
+
+    pipe = None
+
+    while not stop_flag:
+
+        if pipe is None:
+            pipe = open_pipe(pipe_path)
+            if pipe is None:
+                break
+
+        try:
+            byte = data[i]
+
+            pipe.write(bytes([byte]))
+
+            log_buffer.append(byte)
+            total_bytes += 1
+
+            if len(log_buffer) >= OUT_BUFFER_LEN and log_file:
+                packet_id += 1
+
+                log_file.write(f"--- Пакет #{packet_id} ({OUT_BUFFER_LEN} байт) ---\n")
+                for idx, b in enumerate(log_buffer):
+                    log_file.write(f" [{idx:02d}] = 0x{b:02x} ({b:3d})\n")
+
+                log_file.flush()
+                log_buffer.clear()
+
+            repeat += 1
+
+            if repeat >= REPEAT_COUNT:
+                repeat = 0
+                i += 1
+                if i >= len(data):
+                    i = 0
+
+            time.sleep(send_delay)
+
+        except BrokenPipeError:
+            pipe.close()
+            pipe = None
+            status("[emu] pipe: отключён")
+
+        except Exception as e:
+            status(f"[emu] ошибка: {e}")
+            time.sleep(1)
+
+    status_clear()
+    print("[emu] завершён", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()