diff --git a/diff.py b/diff.py index 5d2a50a..5e932b1 100755 --- a/diff.py +++ b/diff.py @@ -432,6 +432,7 @@ def complete_symbol( import os import queue import re +import signal import string import struct import subprocess @@ -649,12 +650,18 @@ def get_arch(arch_str: str) -> "ArchSettings": BUFFER_CMD: List[str] = ["tail", "-c", str(10**9)] -# -S truncates long lines instead of wrapping them -# -R interprets color escape sequences -# -i ignores case when searching -# -c something about how the screen gets redrawn; I don't remember the purpose -# -#6 makes left/right arrow keys scroll by 6 characters -LESS_CMD: List[str] = ["less", "-SRic", "-+F", "-+X", "-#6"] + +def get_less_cmd(use_alternate_screen: bool) -> List[str]: + # -S truncates long lines instead of wrapping them + # -R interprets color escape sequences + # -i ignores case when searching + # -c something about how the screen gets redrawn; I don't remember the purpose + # -#6 makes left/right arrow keys scroll by 6 characters + + # -X sets up the terminal, and -+X does the same terminal setup plus entering alternate + # screen. + return ["less", "-SRic", "-+F", "-+X" if use_alternate_screen else "-X", "-#6"] + DEBOUNCE_DELAY: float = 0.1 @@ -4348,6 +4355,7 @@ def __init__(self, basedump: str, mydump: str, config: Config) -> None: self.emsg = None self.last_refresh_key = None self.last_diff_output = None + self.less_proc = None def run_diff(self) -> Tuple[str, object]: if self.emsg is not None: @@ -4377,7 +4385,7 @@ def run_diff(self) -> Tuple[str, object]: return (output, refresh_key) def run_less( - self, output: str + self, output: str, use_alternate_screen: bool = True ) -> "Tuple[subprocess.Popen[bytes], subprocess.Popen[bytes]]": # Pipe the output through 'tail' and only then to less, to ensure the # write call doesn't block. ('tail' has to buffer all its input before @@ -4386,7 +4394,9 @@ def run_less( buffer_proc = subprocess.Popen( BUFFER_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE ) - less_proc = subprocess.Popen(LESS_CMD, stdin=buffer_proc.stdout) + less_proc = subprocess.Popen( + get_less_cmd(use_alternate_screen), stdin=buffer_proc.stdout + ) assert buffer_proc.stdin assert buffer_proc.stdout buffer_proc.stdin.write(output.encode()) @@ -4397,8 +4407,16 @@ def run_less( def run_sync(self) -> None: output, _ = self.run_diff() proca, procb = self.run_less(output) - procb.wait() - proca.wait() + try: + procb.wait() + except KeyboardInterrupt: + # Send SIGTERM to less so it cleans up the terminal on Ctrl-C. + # otherwise the terminal will be stuck in echo off + procb.terminate() + procb.wait() + raise + finally: + proca.wait() def run_async(self, watch_queue: "queue.Queue[Optional[float]]") -> None: self.watch_queue = watch_queue @@ -4411,21 +4429,21 @@ def run_async(self, watch_queue: "queue.Queue[Optional[float]]") -> None: self.ready_queue.get() def display_thread(self, initial_output: str) -> None: - proca, procb = self.run_less(initial_output) + # when rendering in display thread, we already handled + # alternate screen ourselves to prevent flickering, + # so let less run without using alternate screen + proca, procb = self.run_less(initial_output, use_alternate_screen=False) self.less_proc = procb self.ready_queue.put(None) while True: ret = procb.wait() proca.wait() self.less_proc = None - if ret != 0: - # fix the terminal - os.system("tput reset") if ret != 0 and self.pending_update is not None: - # killed by program with the intent to refresh + # terminated by program with the intent to refresh output = self.pending_update self.pending_update = None - proca, procb = self.run_less(output) + proca, procb = self.run_less(output, use_alternate_screen=False) self.less_proc = procb self.ready_queue.put(None) else: @@ -4456,16 +4474,133 @@ def update(self, text: str, error: bool) -> None: self.pending_update = output if not self.less_proc: return - self.less_proc.kill() + # use SIGTERM instead of SIGKILL so less exits cleanly. + self.less_proc.terminate() self.ready_queue.get() def terminate(self) -> None: + # don't let display_thread restart less with an update that was + # interrupted half-way (e.g. Ctrl-C during update()) + self.pending_update = None if not self.less_proc: return - self.less_proc.kill() + # use SIGTERM instead of SIGKILL so less exits cleanly. + self.less_proc.terminate() self.ready_queue.get() +class WatchTerminal: + """ + Object that owns the alternate screen for a watch mode session + + In watch mode, we set up alternate screen once before running less, + and tears it down once before exiting. This lets us run less without + using alternate screen, so re-running it doesn't cause flickering. + + One extra thing we need to handle is suspend/resume, since less + will now suspend without restoring alternate screen. + """ + + ENTER_SEQ = "\x1b[?1049h" # switch to alternate screen + # reset colors, keypad off (in case a less died without cleaning up), + # switch back to the normal screen + LEAVE_SEQ = "\x1b[0m\x1b[?1l\x1b>\x1b[?1049l" + + def __init__(self, display: Display) -> None: + self.display = display + self.fd = -1 + self.saved_attrs: Optional[List[Any]] = None + self.prev_tstp_handler: Any = None + + def __enter__(self) -> "WatchTerminal": + if not (sys.stdin.isatty() and sys.stdout.isatty()): + return self + try: + import termios + except ImportError: + return self + + # save terminal attrs and enter alternate screen + self.fd = sys.stdin.fileno() + self.saved_attrs = termios.tcgetattr(self.fd) + self._write(self.ENTER_SEQ) + attrs = termios.tcgetattr(self.fd) + # keep ISIG so Ctrl-C and Ctrl-Z still work + attrs[3] &= ~(termios.ECHO | termios.ICANON) + termios.tcsetattr(self.fd, termios.TCSANOW, attrs) + + # register our tstp hook + if hasattr(signal, "SIGTSTP"): + self.prev_tstp_handler = signal.signal(signal.SIGTSTP, self._on_suspend) + return self + + def __exit__(self, *exc_info: object) -> None: + if self.saved_attrs is None: + return + if hasattr(signal, "SIGTSTP"): + signal.signal(signal.SIGTSTP, self.prev_tstp_handler) + self._restore() + + def _write(self, seq: str) -> None: + sys.stdout.write(seq) + sys.stdout.flush() + + def _restore(self) -> None: + import termios + + assert self.saved_attrs is not None + self._write(self.LEAVE_SEQ) + termios.tcsetattr(self.fd, termios.TCSADRAIN, self.saved_attrs) + + def _on_suspend(self, signum: int, frame: object) -> None: + # Ctrl-Z goes to the whole process group, so less is stopping itself + # in parallel. While doing so it restores the tty settings it saved + # at startup (our no-echo ones), so wait for that to finish before + # restoring the original ones. + less_proc = self.display.less_proc + if less_proc is not None: + _wait_until_stopped(less_proc.pid) + self._restore() + # restores the tstp hook and suspend + signal.signal(signal.SIGTSTP, signal.SIG_DFL) + os.kill(os.getpid(), signal.SIGTSTP) + + # resumed (fg), less will restore the settings itself in parallel + + # install our signal hook again and enter alternate screen + signal.signal(signal.SIGTSTP, self._on_suspend) + self._write(self.ENTER_SEQ) + # less may have repainted before we switched back to the alternate + # screen; make it repaint again + less_proc = self.display.less_proc + if less_proc is not None: + try: + less_proc.send_signal(signal.SIGWINCH) + except OSError: + pass + + +def _wait_until_stopped(pid: int, timeout: float = 0.2) -> None: + """Wait for a process to be stopped by a job-control signal.""" + stat_path = f"/proc/{pid}/stat" + if not os.path.exists(stat_path): + # on system without /proc, just wait for a little bit + time.sleep(0.05) + return + # poll for stop, use a timeout to make sure we don't get stuck + deadline = time.time() + timeout + while time.time() < deadline: + try: + with open(stat_path) as f: + # format: "pid (comm) state ...", comm may contain spaces + state = f.read().rsplit(")", 1)[1].split()[0] + except (OSError, IndexError): + return + if state in ("T", "t"): + return + time.sleep(0.005) + + def main_late() -> None: assert parser is not None, "set by main_early" args = parser.parse_args() @@ -4560,30 +4695,35 @@ def main_late() -> None: watch_sources = [make_target] q: "queue.Queue[Optional[float]]" = queue.Queue() debounced_fs_watch(watch_sources, q, config, project) - display.run_async(q) last_build = 0.0 - try: - while True: - t = q.get() - if t is None: - break - if t < last_build: - continue - last_build = time.time() - if args.make: - display.progress("Building...") - ret = run_make_capture_output(make_target, project) - if ret.returncode != 0: - display.update( - ret.stderr.decode("utf-8-sig", "replace") - or ret.stdout.decode("utf-8-sig", "replace"), - error=True, - ) + with WatchTerminal(display): + try: + display.run_async(q) + while True: + t = q.get() + if t is None: + break + if t < last_build: continue - mydump = run_objdump(mycmd, config, project) - display.update(mydump, error=False) - except KeyboardInterrupt: - display.terminate() + last_build = time.time() + if args.make: + display.progress("Building...") + ret = run_make_capture_output(make_target, project) + if ret.returncode != 0: + display.update( + ret.stderr.decode("utf-8-sig", "replace") + or ret.stdout.decode("utf-8-sig", "replace"), + error=True, + ) + continue + mydump = run_objdump(mycmd, config, project) + display.update(mydump, error=False) + except KeyboardInterrupt: + pass + finally: + # make sure less is gone before WatchTerminal restores the + # terminal, on any exit (e.g. fail() raising SystemExit) + display.terminate() def main() -> None: diff --git a/test.py b/test.py index 192da1a..62e14eb 100644 --- a/test.py +++ b/test.py @@ -1,6 +1,25 @@ import unittest import diff import json +import os +import platform +import random +import re +import select +import shlex +import shutil +import signal +import struct +import subprocess +import sys +import tempfile +import time +from typing import List, Optional + +if sys.platform != "win32": + import fcntl + import pty + import termios class TestSh2(unittest.TestCase): @@ -409,5 +428,339 @@ def test_sh4_switch(self) -> None: i += 1 +class TestFixture: + """ + A directory for running diff.py on generated objects. + """ + + __test__ = False # not a test case, despite the name + + FUNC = "bench_fn" + REPO_DIR = os.path.dirname(os.path.abspath(__file__)) + + def __init__(self, workdir: str) -> None: + if not os.path.exists(workdir): + os.makedirs(workdir) + self.workdir = workdir + + @staticmethod + def host_arch() -> Optional[str]: + """diff.py arch for objects compiled on this machine, None if unsupported.""" + # platform.machine() (lowercased) -> diff.py arch of the objects gcc + # builds natively on that machine + HOST_ARCHES = { + "x86_64": "x86", + "amd64": "x86", + "aarch64": "aarch64", + "arm64": "aarch64", + } + return HOST_ARCHES.get(platform.machine().lower()) + + def setup(self, lines: int, num_variants: int) -> List[bytes]: + """ + Populate the directory and return the contents of the my.o variants. + my.o starts out as the first variant. + """ + arch = self.host_arch() + if arch is None: + raise RuntimeError(f"unsupported host architecture: {platform.machine()}") + with open(os.path.join(self.workdir, "diff_settings.py"), "w") as f: + f.write( + "def apply(config, args):\n" + f" config['arch'] = {arch!r}\n" + " config['objdump_executable'] = 'objdump'\n" + ) + # diff.py imports diff_settings from its own directory first, which is + # the repo's example settings. Run the repo's diff.py through a launcher + # here so this dir's diff_settings.py is picked up instead. + with open(os.path.join(self.workdir, "run_diff.py"), "w") as f: + f.write( + "import runpy\n" + f"runpy.run_path({os.path.join(self.REPO_DIR, 'diff.py')!r}, " + "run_name='__main__')\n" + ) + self._compile_obj("ref", seed=0, lines=lines) + variants: List[bytes] = [] + for i in range(num_variants): + path = self._compile_obj(f"var{i}", seed=i + 1, lines=lines) + with open(path, "rb") as f: + variants.append(f.read()) + self.write_obj(variants[0]) + return variants + + def write_obj(self, data: bytes) -> None: + """Replace my.o, rewriting it in place like a compiler/linker would.""" + with open(os.path.join(self.workdir, "my.o"), "wb") as f: + f.write(data) + + def command(self, python: str, watch: bool = True) -> str: + """Shell command that runs diff.py on this fixture, in watch mode by default.""" + watch_flags = "-w -y " if watch else "" + return ( + f"cd {shlex.quote(self.workdir)} && {shlex.quote(python)} run_diff.py " + f"-o -f my.o --ref-file ref.o {watch_flags}{self.FUNC}" + ) + + def _compile_obj(self, name: str, seed: int, lines: int) -> str: + c_path = os.path.join(self.workdir, f"{name}.c") + o_path = os.path.join(self.workdir, f"{name}.o") + with open(c_path, "w") as f: + f.write(self._gen_source(seed, lines)) + subprocess.run(["gcc", "-O1", "-c", c_path, "-o", o_path], check=True) + return o_path + + def _gen_source(self, seed: int, lines: int) -> str: + rng = random.Random(seed) + body = "\n".join( + ( + f" sink = {rng.randrange(1 << 16)};" + if rng.random() < 0.3 + else f" sink = {i * 7};" + ) + for i in range(lines) + ) + return f"volatile int sink;\nvoid {self.FUNC}(void) {{\n{body}\n}}\n" + + +class TestTerminal(unittest.TestCase): + """ + Test various terminal UI behavior + """ + + ENTER = b"\x1b[?1049h" # switch to alternate screen + LEAVE = b"\x1b[?1049l" # switch back to normal screen + RESET = b"\x1bc" # full terminal reset (tput reset) + + class _PtyShell: + """An interactive bash (with job control) running in a pseudo-terminal.""" + + PROMPT = b"PROMPT> " + + def __init__(self) -> None: + self.pid, self.fd = pty.fork() + if self.pid == 0: + os.environ.update(TERM="xterm-256color", PS1=self.PROMPT.decode()) + os.environ.pop("LESS", None) + os.execvp("bash", ["bash", "--norc", "--noprofile", "-i"]) + winsize = struct.pack("HHHH", 40, 200, 0, 0) + fcntl.ioctl(self.fd, termios.TIOCSWINSZ, winsize) + self.buf = b"" + self.read_until(self.PROMPT, 0, timeout=5) + + def mark(self) -> int: + """Position in the output, to wait for output that comes after it.""" + return len(self.buf) + + def _read(self, timeout: float) -> bool: + """Read available output into buf. False if the pty is closed.""" + r, _, _ = select.select([self.fd], [], [], timeout) + if not r: + return True + try: + data = os.read(self.fd, 65536) + except OSError: + return False + self.buf += data + return bool(data) + + def wait_for(self, pattern: bytes, start: int, timeout: float) -> bool: + """Wait until pattern shows up in the output after start.""" + end = time.time() + timeout + while pattern not in self.buf[start:]: + remaining = end - time.time() + if remaining <= 0 or not self._read(min(remaining, 0.05)): + return pattern in self.buf[start:] + return True + + def read_until(self, pattern: bytes, start: int, timeout: float) -> bytes: + """Like wait_for(), but fails on timeout. Returns the output after start.""" + if not self.wait_for(pattern, start, timeout): + raise AssertionError( + f"timed out waiting for {pattern!r}; got {self.buf[-300:]!r}" + ) + return self.buf[start:] + + def send(self, data: bytes) -> None: + os.write(self.fd, data) + + def tty_flags(self) -> List[bytes]: + """echo/icanon flags of the tty as seen by a command run from the shell.""" + start = self.mark() + # E''ND so the echoed command line doesn't match the marker + self.send( + b"stty -a | tr ' ;' '\\n\\n' | grep -xE -- '-?(echo|icanon)'" + b" | tr '\\n' ' '; echo E''ND\n" + ) + out = self.read_until(b"END\r\n", start, timeout=5) + m = re.search(rb"((?:-?\w+ )+)END\r\n", out) + return m.group(1).split() if m else [] + + def close(self) -> None: + self.send(b"exit\n") + end = time.time() + 5 + while time.time() < end: + if os.waitpid(self.pid, os.WNOHANG)[0] != 0: + break + # keep reading so bash never blocks on writing output + self._read(0.05) + else: + # e.g. a failed test left a stopped job, so bash refuses to exit + os.kill(self.pid, signal.SIGKILL) + os.waitpid(self.pid, 0) + os.close(self.fd) + + fixture: TestFixture + variants: List[bytes] + + @classmethod + def setUpClass(cls) -> None: + reason = cls._skip_reason() + if reason: + raise unittest.SkipTest(reason) + cls.fixture = TestFixture(tempfile.mkdtemp(prefix="asm-differ-watch-test-")) + cls.variants = cls.fixture.setup(lines=200, num_variants=3) + + @classmethod + def tearDownClass(cls) -> None: + shutil.rmtree(cls.fixture.workdir, ignore_errors=True) + + @staticmethod + def _skip_reason() -> str: + if sys.platform == "win32": + return "needs a POSIX pty" + if TestFixture.host_arch() is None: + return f"no diff.py arch for host machine {platform.machine()}" + for tool in ["bash", "gcc", "objdump", "less", "stty"]: + if not shutil.which(tool): + return f"needs {tool}" + return "" + + def _refresh(self, shell: "TestTerminal._PtyShell", data: bytes) -> bytes: + """ + Replace my.o and wait for the new less to draw. Returns the output from + the write on. + """ + start = shell.mark() + # retry in case the write happened before the file watcher was ready + for _ in range(3): + self.fixture.write_obj(data) + if shell.wait_for(b"TARGET", start, timeout=3): + return shell.buf[start:] + self.fail("watch mode did not refresh after my.o changed") + + # === no cases === + + def _run_case(self, exit_key: bytes, suspend: bool) -> None: + self.fixture.write_obj(self.variants[0]) + shell = self._PtyShell() + self.addCleanup(shell.close) + start = shell.mark() + shell.send(self.fixture.command(sys.executable).encode() + b"\n") + shell.read_until(b"TARGET", start, timeout=15) + for i in range(3): + self._refresh(shell, self.variants[1 + i % 2]) + # the old less leaves before the new one draws, so a leave would + # already be in the output by now + out = shell.buf[start:] + self.assertEqual(out.count(self.ENTER), 1, "enters alt screen once") + self.assertNotIn(self.LEAVE, out, "no alt screen leave during refreshes") + self.assertNotIn(self.RESET, out, "no terminal reset during refreshes") + + if suspend: + mark = shell.mark() + shell.send(b"\x1a") + out = shell.read_until(shell.PROMPT, mark, timeout=5) + leave_at = out.find(self.LEAVE) + self.assertTrue( + 0 <= leave_at < out.find(shell.PROMPT), + "Ctrl-Z leaves alt screen before the shell prompt", + ) + self.assertIn(b"Stopped", out) + # al''ive so the echoed command line doesn't match + mark = shell.mark() + shell.send(b"echo al''ive\n") + shell.read_until(b"alive\r\n", mark, timeout=5) + + mark = shell.mark() + shell.send(b"fg\n") + out = shell.read_until(b"TARGET", mark, timeout=5) + enter_at = out.find(self.ENTER) + self.assertGreaterEqual(enter_at, 0, "fg re-enters alt screen") + self.assertIn(b"TARGET", out[enter_at:], "less repaints after fg") + + out = self._refresh(shell, self.variants[0]) + self.assertNotIn(self.LEAVE, out, "refresh after fg stays on alt screen") + + mark = shell.mark() + shell.send(exit_key) + out = shell.read_until(shell.PROMPT, mark, timeout=10) + self.assertIn(self.LEAVE, out, "leaves alt screen on exit") + flags = shell.tty_flags() + self.assertIn(b"echo", flags) + self.assertIn(b"icanon", flags) + + def test_quit(self) -> None: + """ + Refreshes stay on the alternate screen without a terminal reset, and + quitting with q leaves the alternate screen and restores echo and line + mode for the shell. + """ + self._run_case(b"q", suspend=False) + + def test_ctrl_c(self) -> None: + """ + Refreshes stay on the alternate screen without a terminal reset, and + Ctrl-C leaves the alternate screen and restores echo and line mode for + the shell. + """ + self._run_case(b"\x03", suspend=False) + + def test_suspend_then_quit(self) -> None: + """ + Like test_quit, plus Ctrl-Z in between: Ctrl-Z leaves the alternate + screen before the shell prompt shows up and the shell is usable; fg + switches back to the alternate screen, less repaints, and refreshes + keep working; then q exits cleanly. + """ + self._run_case(b"q", suspend=True) + + def test_suspend_then_ctrl_c(self) -> None: + """Like test_suspend_then_quit, but exits with Ctrl-C after fg.""" + self._run_case(b"\x03", suspend=True) + + # === no watch cases === + + def _run_no_watch_case(self, exit_key: bytes) -> None: + self.fixture.write_obj(self.variants[0]) + shell = self._PtyShell() + self.addCleanup(shell.close) + start = shell.mark() + shell.send(self.fixture.command(sys.executable, watch=False).encode() + b"\n") + out = shell.read_until(b"TARGET", start, timeout=15) + self.assertIn(self.ENTER, out[: out.find(b"TARGET")], "less enters alt screen") + + mark = shell.mark() + shell.send(exit_key) + out = shell.read_until(shell.PROMPT, mark, timeout=10) + self.assertIn(self.LEAVE, out, "leaves alt screen on exit") + flags = shell.tty_flags() + self.assertIn(b"echo", flags) + self.assertIn(b"icanon", flags) + + def test_no_watch_quit(self) -> None: + """ + Without -w, less switches to the alternate screen itself, and quitting + with q leaves it and restores echo and line mode for the shell. + """ + self._run_no_watch_case(b"q") + + def test_no_watch_ctrl_c(self) -> None: + """ + Without -w, Ctrl-C makes diff.py terminate less, which leaves the + alternate screen and restores echo and line mode for the shell. + """ + self._run_no_watch_case(b"\x03") + + if __name__ == "__main__": unittest.main() diff --git a/tests/.gitignore b/tests/.gitignore new file mode 100644 index 0000000..0b07dda --- /dev/null +++ b/tests/.gitignore @@ -0,0 +1 @@ +*_fixture/ diff --git a/tests/flicker_test.py b/tests/flicker_test.py new file mode 100644 index 0000000..8c68101 --- /dev/null +++ b/tests/flicker_test.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +""" +Manual flicker test for watch mode (-w). + +Builds a reference object and a set of variants of the watched object, prints +the command to run diff.py in watch mode on them, then keeps overwriting the +watched object with a random variant at random short intervals until Ctrl-C. +Run the printed command in another terminal and watch for flickering. + +All generated files go in flicker_test_fixture/ next to this script +(gitignored), which is recreated on each run. The fixture is test.py's +TestFixture, shared with the automated watch mode tests there. + +Usage: tests/flicker_test.py +""" + +import argparse +import os +import random +import shutil +import sys +import time + +TESTS_DIR = os.path.dirname(os.path.abspath(__file__)) +REPO_DIR = os.path.dirname(TESTS_DIR) +FIXTURE_DIR = os.path.join(TESTS_DIR, "flicker_test_fixture") + +# the repo's test.py, not the standard library's test package +sys.path.insert(0, REPO_DIR) +from test import TestFixture # noqa: E402 + + +def main() -> None: + parser = argparse.ArgumentParser(description=str(__doc__).split("\n\n")[1]) + parser.add_argument( + "--min", type=float, default=0.15, help="min seconds between edits" + ) + parser.add_argument( + "--max", type=float, default=0.6, help="max seconds between edits" + ) + parser.add_argument( + "--lines", type=int, default=300, help="instructions in the function" + ) + parser.add_argument( + "--variants", type=int, default=8, help="number of object variants" + ) + args = parser.parse_args() + if args.min <= 0.1: + print( + "warning: edits closer than the 100ms debounce get merged", + file=sys.stderr, + ) + + for tool in ["gcc", "objdump"]: + if not shutil.which(tool): + sys.exit(f"missing required tool: {tool}") + + workdir = FIXTURE_DIR + shutil.rmtree(workdir, ignore_errors=True) + os.makedirs(workdir) + try: + fixture = TestFixture(workdir) + variants = fixture.setup(args.lines, args.variants) + + PYTHON = "python3" + print( + "Run this in another terminal (using your python3 that has the dependencies installed):\n" + ) + print(f" {fixture.command(PYTHON)}\n") + _ = input("Press Enter to start editing (Ctrl-C to stop)...") + + current = 0 + count = 0 + while True: + nxt = random.choice([i for i in range(len(variants)) if i != current]) + fixture.write_obj(variants[nxt]) + current = nxt + count += 1 + print(f"\redits: {count} (variant {nxt}) ", end="", flush=True) + time.sleep(random.uniform(args.min, args.max)) + except KeyboardInterrupt: + print("\nstopped.") + + +if __name__ == "__main__": + main()