доработка имитатора

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:
+# ================= SIGNALS =================
-    """Имитатор 7-битной параллельной шины данных"""
 
-    # Временные константы (секунды)
+def handle_sigint(signum, frame):
-    DATA_SETUP_TIME = 0.000005
+    global stop_flag
-    STROBE_WIDTH = 0.000005
+    stop_flag = True
-    HOLD_TIME = 0.000005
 
-    WR_PIN = 12  # BCM12
 
-    DATA_PINS_MAPPING = [
+signal.signal(signal.SIGINT, handle_sigint)
-        (21, 18, "D0 (LSB)"),
+signal.signal(signal.SIGTERM, handle_sigint)
-        (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 = []
 
-        for wpi, bcm, desc in self.DATA_PINS_MAPPING:
+# ================= UI =================
-            if bcm is not None:
-                self.active_pins.append((wpi, bcm, desc))
 
-        self.data_pins_bcm = [bcm for _, bcm, _ in self.active_pins]
+last_status = ""
-        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:
-            self.chip = gpiod.Chip(chip_path)
+            with open(path, "r") as f:
-            self.lines = self.chip.request_lines(
+                content = f.read().strip()
-                {
+
-                    p: gpiod.LineSettings(
+            data = ast.literal_eval(content)
-                        direction=Direction.OUTPUT,
+
-                        output_value=Value.INACTIVE
+            result = []
-                    )
+            for x in data:
-                    for p in self.all_pins
+                if isinstance(x, str):
-                },
+                    x = int(x, 0)
-                consumer="par-bus-imitator"
+                result.append(x & 0xFF)
-            )
+
-        except PermissionError:
+            print(f"[emu] Загружено {len(result)} байт", file=sys.stderr)
-            print("Нет прав доступа к GPIO. Используйте sudo.")
+            return result
-            sys.exit(1)
+
         except Exception as e:
-            print(f"Ошибка GPIO: {e}")
+            print(f"[emu] Ошибка чтения: {e}", file=sys.stderr)
-            sys.exit(1)
+            time.sleep(1)
 
-        bit_width = 4 if use_mask else len(self.data_pins_bcm)
 
-        print(f"✓ Имитатор шины ({bit_width}-бит режим)")
+# ================= PIPE =================
-        print("Нажмите Ctrl+C для остановки\n")
 
-    def send_byte(self, value):
+def open_pipe(path):
-        """Отправка одного байта"""
+    while not stop_flag:
-        if self.use_mask:
+        try:
-            value &= 0x0F
+            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 = {}
+    return array_path, pipe_path, send_delay, log_file
-
-        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()
 
 
-def handle_signal(signum, frame):
+# ================= MAIN =================
-    print("\nЗавершение работы...")
-    sys.exit(0)
-
 
 def main():
-    signal.signal(signal.SIGINT, handle_signal)
+    array_path, pipe_path, send_delay, log_file = parse_args()
-    signal.signal(signal.SIGTERM, handle_signal)
 
-    parser = argparse.ArgumentParser(
+    data = load_array(array_path)
-        description='Имитатор параллельной шины'
-    )
 
-    parser.add_argument('--no-mask', action='store_true',
+    log_buffer = []
-                        help='7-битный режим')
+    total_bytes = 0
-    parser.add_argument('--test',
+    packet_id = 0
-                        choices=['default', 'counter', 'pattern'],
+    i = 0
-                        default='default')
+    repeat = 0
-    parser.add_argument('--delay', type=float, default=0.2)
-    parser.add_argument('--single', type=str)
-    parser.add_argument('--chip', default='/dev/gpiochip0')
 
-    args = parser.parse_args()
+    pipe = None
 
-    imitator = ParallelBusImitator(
+    while not stop_flag:
-        use_mask=not args.no_mask,
-        chip_path=args.chip
-    )
 
-    try:
+        if pipe is None:
-        if args.single:
+            pipe = open_pipe(pipe_path)
-            val = int(args.single, 16)
+            if pipe is None:
-            imitator.send_byte(val)
+                break
-            print()
-            return
 
-        if args.test == 'default':
+        try:
-            test_data = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40]
+            byte = data[i]
-        elif args.test == 'counter':
-            test_data = list(range(128))
-        else:
-            test_data = [0x00, 0x01, 0x03, 0x07, 0x0F, 0x55, 0x2A]
 
-        while True:
+            pipe.write(bytes([byte]))
-            imitator.send_sequence(test_data, args.delay)
 
-    except KeyboardInterrupt:
+            log_buffer.append(byte)
-        print("\nОстановка...")
+            total_bytes += 1
-    except Exception as e:
+
-        print(f"\nОшибка: {e}")
+            if len(log_buffer) >= OUT_BUFFER_LEN and log_file:
-    finally:
+                packet_id += 1
-        imitator.cleanup()
+
-        print("✓ GPIO освобождены")
+                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()