From 941a7cf6e6e285e49c4163899828d1377e842e5d Mon Sep 17 00:00:00 2001 From: Maxim Date: Wed, 20 May 2026 11:30:45 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D1=80=D0=B0=D0=B1=D1=82=D0=BA?= =?UTF-8?q?=D0=B0=20c=20HackRF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/array.txt | 1 + scripts/emulatohackrf.sh | 15 ++ scripts/emulator.py | 230 ++++++++++++++++++++++++++++ scripts/go2monitor.service | 29 ++++ scripts/imi_wire.py | 302 ++++++++++++++++++------------------- 5 files changed, 419 insertions(+), 158 deletions(-) create mode 100644 scripts/array.txt create mode 100644 scripts/emulatohackrf.sh create mode 100644 scripts/emulator.py create mode 100644 scripts/go2monitor.service diff --git a/scripts/array.txt b/scripts/array.txt new file mode 100644 index 0000000..bc14934 --- /dev/null +++ b/scripts/array.txt @@ -0,0 +1 @@ +[0xE0, 0x01, 0x4C, 0x88] diff --git a/scripts/emulatohackrf.sh b/scripts/emulatohackrf.sh new file mode 100644 index 0000000..1655a7d --- /dev/null +++ b/scripts/emulatohackrf.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash + +# Эмулятор GPIO -> stdout (1 мс на байт) +python3 /home/user/temp/golang/scripts/emulator.py \ + array.txt \ + - \ + 0.001 \ +| pv | # Передача через HackRF +/home/user/hackrf-liquid-dsp/hackrf-frame64-tx \ + -x 40 \ + -a 0 \ + -d a32868dc347c6e47 \ + -s 2000000 \ + -b 1750000 \ + -f 800000000 diff --git a/scripts/emulator.py b/scripts/emulator.py new file mode 100644 index 0000000..2aba748 --- /dev/null +++ b/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 [задержка] [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() diff --git a/scripts/go2monitor.service b/scripts/go2monitor.service new file mode 100644 index 0000000..18577c3 --- /dev/null +++ b/scripts/go2monitor.service @@ -0,0 +1,29 @@ +[Unit] +Description=GPIO Monitor Server +After=network.target +Wants=network.target + +[Service] +Type=simple +User=user +Group=user +WorkingDirectory=/home/user/temp/golang + +# Запуск сервера +ExecStart=/home/user/temp/golang/build/gpio-monitor-server -pipe /tmp/gpio_pipe + +# Перезапуск при падении +Restart=on-failure +RestartSec=10s + +# Лимиты и безопасность +LimitNOFILE=4096 +NoNewPrivileges=yes + +# Логирование +StandardOutput=journal +StandardError=journal +SyslogIdentifier=gpio-monitor + +[Install] +WantedBy=multi-user.target diff --git a/scripts/imi_wire.py b/scripts/imi_wire.py index 51206c2..1eb496b 100644 --- a/scripts/imi_wire.py +++ b/scripts/imi_wire.py @@ -1,191 +1,177 @@ -import sys +#!/usr/bin/env python3 +""" +Имитатор параллельной шины для gpio-interrupt (Raspberry Pi 4) +""" + +import gpiod import time -import ast +import argparse +import sys import signal -import os - -OUT_BUFFER_LEN = 32 -RECONNECT_DELAY = 0.5 -REPEAT_COUNT = 50 # повтор -DEFAULT_SEND_DELAY = 0.1 # по умолчанию (10 байт/с) - -stop_flag = False +from gpiod.line import Direction, Value -# ================= SIGNALS ================= +class ParallelBusImitator: + """Имитатор 7-битной параллельной шины данных""" -def handle_sigint(signum, frame): - global stop_flag - stop_flag = True + # Временные константы (секунды) + DATA_SETUP_TIME = 0.000005 + STROBE_WIDTH = 0.000005 + HOLD_TIME = 0.000005 + WR_PIN = 12 # BCM12 -signal.signal(signal.SIGINT, handle_sigint) -signal.signal(signal.SIGTERM, handle_sigint) + DATA_PINS_MAPPING = [ + (21, 18, "D0 (LSB)"), + (7, 27, "D1"), + (6, 22, "D2"), + (5, 23, "D3"), + (25, 17, "D4"), + (24, 20, "D5"), + (23, 21, "D6"), + (22, None, "D7 (не подключён)"), + ] + def __init__(self, use_mask=True, chip_path='/dev/gpiochip0'): + self.use_mask = use_mask + self.counter = 0 + self.active_pins = [] -# ================= UI ================= + for wpi, bcm, desc in self.DATA_PINS_MAPPING: + if bcm is not None: + self.active_pins.append((wpi, bcm, desc)) -last_status = "" + self.data_pins_bcm = [bcm for _, bcm, _ in self.active_pins] + self.all_pins = self.data_pins_bcm + [self.WR_PIN] - -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 - + self.chip = gpiod.Chip(chip_path) + self.lines = self.chip.request_lines( + { + p: gpiod.LineSettings( + direction=Direction.OUTPUT, + output_value=Value.INACTIVE + ) + for p in self.all_pins + }, + consumer="par-bus-imitator" + ) + except PermissionError: + print("Нет прав доступа к GPIO. Используйте sudo.") + sys.exit(1) except Exception as e: - print(f"[emu] Ошибка чтения: {e}", file=sys.stderr) - time.sleep(1) + print(f"Ошибка GPIO: {e}") + sys.exit(1) + bit_width = 4 if use_mask else len(self.data_pins_bcm) -# ================= PIPE ================= + print(f"✓ Имитатор шины ({bit_width}-бит режим)") + print("Нажмите Ctrl+C для остановки\n") -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 /tmp/gpio_pipe [задержка] [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 # сдвигаемся, значит logfile будет 4-м аргументом - 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 + def send_byte(self, value): + """Отправка одного байта""" + if self.use_mask: + value &= 0x0F else: - log_file = open(log_path, "a") + value &= 0x7F - return array_path, pipe_path, send_delay, log_file + out = {} + + for bit_pos, bcm_pin in enumerate(self.data_pins_bcm): + bit = (value >> bit_pos) & 1 + out[bcm_pin] = Value.ACTIVE if bit else Value.INACTIVE + + self.lines.set_values(out) + time.sleep(self.DATA_SETUP_TIME) + + self.lines.set_value(self.WR_PIN, Value.ACTIVE) + time.sleep(self.STROBE_WIDTH) + self.lines.set_value(self.WR_PIN, Value.INACTIVE) + time.sleep(self.HOLD_TIME) + + self.counter += 1 + self.show_status(value) + + def show_status(self, value): + """Красивый статус без спама""" + bits = 4 if self.use_mask else 7 + bin_val = format(value, f'0{bits}b') + + msg = ( + f"\r▶ Передача: " + f"HEX=0x{value:02X} | " + f"BIN={bin_val} | " + f"Пакетов: {self.counter}" + ) + + print(msg, end='', flush=True) + + def send_sequence(self, values, delay=0.2): + for val in values: + self.send_byte(val) + time.sleep(delay) + + def cleanup(self): + if hasattr(self, 'lines'): + self.lines.release() + if hasattr(self, 'chip'): + self.chip.close() -# ================= MAIN ================= +def handle_signal(signum, frame): + print("\nЗавершение работы...") + sys.exit(0) + def main(): - array_path, pipe_path, send_delay, log_file = parse_args() + signal.signal(signal.SIGINT, handle_signal) + signal.signal(signal.SIGTERM, handle_signal) - data = load_array(array_path) + parser = argparse.ArgumentParser( + description='Имитатор параллельной шины' + ) - log_buffer = [] - total_bytes = 0 - packet_id = 0 - i = 0 - repeat = 0 + parser.add_argument('--no-mask', action='store_true', + help='7-битный режим') + parser.add_argument('--test', + choices=['default', 'counter', 'pattern'], + default='default') + parser.add_argument('--delay', type=float, default=0.2) + parser.add_argument('--single', type=str) + parser.add_argument('--chip', default='/dev/gpiochip0') - pipe = None + args = parser.parse_args() - while not stop_flag: + imitator = ParallelBusImitator( + use_mask=not args.no_mask, + chip_path=args.chip + ) - if pipe is None: - pipe = open_pipe(pipe_path) - if pipe is None: - break + try: + if args.single: + val = int(args.single, 16) + imitator.send_byte(val) + print() + return - try: - byte = data[i] + if args.test == 'default': + test_data = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40] + elif args.test == 'counter': + test_data = list(range(128)) + else: + test_data = [0x00, 0x01, 0x03, 0x07, 0x0F, 0x55, 0x2A] - pipe.write(bytes([byte])) + while True: + imitator.send_sequence(test_data, args.delay) - 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) + except KeyboardInterrupt: + print("\nОстановка...") + except Exception as e: + print(f"\nОшибка: {e}") + finally: + imitator.cleanup() + print("✓ GPIO освобождены") if __name__ == "__main__": - main() \ No newline at end of file + main()