From 306eda66c61a199f131309cae4caade6e3f6cc38 Mon Sep 17 00:00:00 2001 From: Tot Maxim Date: Sat, 9 May 2026 11:46:21 +0300 Subject: [PATCH] =?UTF-8?q?=D0=B4=D0=BE=D1=80=D0=B0=D0=B1=D0=BE=D1=82?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=B8=D0=BC=D0=B8=D1=82=D0=B0=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/imi_wire.py | 302 +++++++++++++++++++++++--------------------- 1 file changed, 158 insertions(+), 144 deletions(-) diff --git a/scripts/imi_wire.py b/scripts/imi_wire.py index 1eb496b..51206c2 100644 --- a/scripts/imi_wire.py +++ b/scripts/imi_wire.py @@ -1,177 +1,191 @@ -#!/usr/bin/env python3 -""" -Имитатор параллельной шины для gpio-interrupt (Raspberry Pi 4) -""" - -import gpiod -import time -import argparse import sys +import time +import ast import signal -from gpiod.line import Direction, Value +import os + +OUT_BUFFER_LEN = 32 +RECONNECT_DELAY = 0.5 +REPEAT_COUNT = 50 # повтор +DEFAULT_SEND_DELAY = 0.1 # по умолчанию (10 байт/с) + +stop_flag = False -class ParallelBusImitator: - """Имитатор 7-битной параллельной шины данных""" +# ================= SIGNALS ================= - # Временные константы (секунды) - DATA_SETUP_TIME = 0.000005 - STROBE_WIDTH = 0.000005 - HOLD_TIME = 0.000005 +def handle_sigint(signum, frame): + global stop_flag + stop_flag = True - WR_PIN = 12 # BCM12 - 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 (не подключён)"), - ] +signal.signal(signal.SIGINT, handle_sigint) +signal.signal(signal.SIGTERM, handle_sigint) - def __init__(self, use_mask=True, chip_path='/dev/gpiochip0'): - self.use_mask = use_mask - self.counter = 0 - self.active_pins = [] - for wpi, bcm, desc in self.DATA_PINS_MAPPING: - if bcm is not None: - self.active_pins.append((wpi, bcm, desc)) +# ================= UI ================= - self.data_pins_bcm = [bcm for _, bcm, _ in self.active_pins] - self.all_pins = self.data_pins_bcm + [self.WR_PIN] +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: - 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) + 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"Ошибка GPIO: {e}") - sys.exit(1) + print(f"[emu] Ошибка чтения: {e}", file=sys.stderr) + time.sleep(1) - bit_width = 4 if use_mask else len(self.data_pins_bcm) - print(f"✓ Имитатор шины ({bit_width}-бит режим)") - print("Нажмите Ctrl+C для остановки\n") +# ================= PIPE ================= - def send_byte(self, value): - """Отправка одного байта""" - if self.use_mask: - value &= 0x0F +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 else: - value &= 0x7F + log_file = open(log_path, "a") - 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() + return array_path, pipe_path, send_delay, log_file -def handle_signal(signum, frame): - print("\nЗавершение работы...") - sys.exit(0) - +# ================= MAIN ================= def main(): - signal.signal(signal.SIGINT, handle_signal) - signal.signal(signal.SIGTERM, handle_signal) + array_path, pipe_path, send_delay, log_file = parse_args() - parser = argparse.ArgumentParser( - description='Имитатор параллельной шины' - ) + data = load_array(array_path) - 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') + log_buffer = [] + total_bytes = 0 + packet_id = 0 + i = 0 + repeat = 0 - args = parser.parse_args() + pipe = None - imitator = ParallelBusImitator( - use_mask=not args.no_mask, - chip_path=args.chip - ) + while not stop_flag: - try: - if args.single: - val = int(args.single, 16) - imitator.send_byte(val) - print() - return + if pipe is None: + pipe = open_pipe(pipe_path) + if pipe is None: + break - 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] + try: + byte = data[i] - while True: - imitator.send_sequence(test_data, args.delay) + pipe.write(bytes([byte])) - except KeyboardInterrupt: - print("\nОстановка...") - except Exception as e: - print(f"\nОшибка: {e}") - finally: - imitator.cleanup() - print("✓ GPIO освобождены") + 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() + main() \ No newline at end of file