go-service/scripts/emulator.py

"""
Эмулятор чтения из файла и отправки в pipe.

Создание сокета: mkfifo /tmp/gpio_pipe
Формат файла: [0xE0, 0x01, 0x4C, 0x88]
Запуск: python3 emulator.py array.txt /tmp/gpio_pipe
"""

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):
    """Открывает pipe для записи."""
    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()