61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from PySide6.QtCore import QObject, QThread, Signal, Slot
|
|
|
|
|
|
class CommandPlayer(QThread):
|
|
send_raw = Signal(bytes)
|
|
command_started = Signal(str)
|
|
finished_file = Signal()
|
|
error = Signal(str)
|
|
|
|
def __init__(
|
|
self,
|
|
filepath: Path,
|
|
per_cmd_delay_ms: int,
|
|
per_char_delay_ms: int,
|
|
parent: Optional[QObject] = None,
|
|
) -> None:
|
|
super().__init__(parent)
|
|
self._filepath = filepath
|
|
self._per_cmd_delay = max(0, per_cmd_delay_ms) / 1000.0
|
|
self._per_char_delay = max(0, per_char_delay_ms) / 1000.0
|
|
self._running = True
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
lines = self._filepath.read_text(encoding="utf-8").splitlines()
|
|
except Exception as exc: # noqa: BLE001
|
|
self.error.emit(str(exc))
|
|
return
|
|
|
|
try:
|
|
for raw in lines:
|
|
if not self._running:
|
|
break
|
|
stripped = raw.strip("\r\n")
|
|
if not stripped:
|
|
continue
|
|
self.command_started.emit(stripped)
|
|
if self._per_char_delay > 0:
|
|
for ch in stripped:
|
|
if not self._running:
|
|
break
|
|
self.send_raw.emit(ch.encode("utf-8", errors="ignore"))
|
|
time.sleep(self._per_char_delay)
|
|
self.send_raw.emit(b"\r\n")
|
|
else:
|
|
self.send_raw.emit((stripped + "\r\n").encode("utf-8", errors="ignore"))
|
|
if self._per_cmd_delay:
|
|
time.sleep(self._per_cmd_delay)
|
|
finally:
|
|
self.finished_file.emit()
|
|
|
|
@Slot()
|
|
def stop(self) -> None:
|
|
self._running = False
|