191 lines
4.7 KiB
Python
191 lines
4.7 KiB
Python
import sys
|
||
import time
|
||
import ast
|
||
import signal
|
||
import os
|
||
|
||
OUT_BUFFER_LEN = 32
|
||
RECONNECT_DELAY = 0.5
|
||
REPEAT_COUNT = 50 # повтор
|
||
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 /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:
|
||
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
|
||
|
||
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() |