remove verbose; add noise

botable/__main__.py

@@ -50,7 +50,11 @@
         default=0,
         help="in 'play' mode: how many events the first loop will skip, default is 0 event skipped",
     )
-    arg_parser.add_argument("--playback-verbose", required=False, type=bool)
+    arg_parser.add_argument(
+        "--playback-noise",
+        action="store_true",
+        help="in 'play' mode: to add noise to the time intervals between events",
+    )
     args = arg_parser.parse_args()
 
     mode = args.mode
@@ -64,7 +68,7 @@
             rate=args.playback_rate,
             delay=args.playback_delay,
             offset=args.playback_offset,
-            verbose=args.playback_verbose,
+            noise=args.playback_noise,
         ):
             print(tuple(button_event), flush=True)
     elif mode == "record":

botable/botable.py

@@ -1,5 +1,6 @@
 from collections import deque
 from contextlib import suppress
+import random
 import time
 from typing import Deque, Iterable, Iterator, List, Optional, Union, NamedTuple, Tuple
 from pynput import keyboard, mouse  # type: ignore
@@ -10,7 +11,7 @@
 class ButtonEvent(NamedTuple):
     button: str
     pressed: bool
-    seconds_since_last_event: float
+    pre_sleep: float
     coordinates: Optional[Tuple[int, int]]
 
 
@@ -101,6 +102,10 @@ def on_click(x: int, y: int, button: mouse.Button, pressed: bool):
         RECORDING = False
 
 
+def add_noise(x: float) -> float:
+    return x * (1 + random.betavariate(2, 5) / 2)
+
+
 def play(
     button_events: Iterable[ButtonEvent],
     exit_key: str = "f1",
@@ -109,13 +114,13 @@ def play(
     rate: float = 1.0,
     delay: float = 1.0,
     offset: int = 0,
-    verbose: bool = False,
+    noise: bool = False,
 ) -> Iterator[ButtonEvent]:
     """
     Waits `delay` and then iterates on `events` to play them,
       optionally playing them at a modified speed `rate`,
       optionally skipping the first `offset` events,
-      optionally logging the played events if `verbose`.
+      optionally adding noise to the time intervals between events (the original interval remains the minimum).
     Once `button_events` is exhausted the entire collected set of event will optionally be replayed additional times if `loops` > 1.
     Pressing the `exit_key` will terminate the recording.
     Pressing the `pause_key` will pause/resume the recording.
@@ -157,9 +162,6 @@ def on_press(key: keyboard.Key):
                 while continue_ and paused_at:
                     time.sleep(0.5)
 
-                if verbose:
-                    print(button_event)
-
                 if loop_index == 0 and offset > event_index:
                     continue
 
@@ -176,7 +178,15 @@ def on_press(key: keyboard.Key):
                 else:
                     evaluated_button = eval(button_event.button)
 
-                time.sleep(button_event.seconds_since_last_event / rate)
+                if noise:
+                    button_event = ButtonEvent(
+                        button=button_event.button,
+                        pressed=button_event.pressed,
+                        pre_sleep=add_noise(button_event.pre_sleep),
+                        coordinates=button_event.coordinates,
+                    )
+
+                time.sleep(button_event.pre_sleep / rate)
 
                 if button_event.pressed:
                     ctrl.press(evaluated_button)

build/lib/botable/__init__.py

@@ -0,0 +1 @@
+from botable.botable import record, play, ButtonEvent

build/lib/botable/__main__.py

@@ -0,0 +1,82 @@
+import argparse
+
+from botable.botable import play, record, input_button_events
+
+
+if __name__ == "__main__":
+    arg_parser = argparse.ArgumentParser(
+        description="Record mouse and keyboard keys pressures/releases."
+    )
+    arg_parser.add_argument("mode", help="either record or play")
+    arg_parser.add_argument(
+        "--exit-key",
+        required=False,
+        type=str,
+        default="f1",
+        help="the key to press to end the ongoing recording or playback, default is f1",
+    )
+    arg_parser.add_argument(
+        "--pause-key",
+        required=False,
+        type=str,
+        default="f2",
+        help="the key to press to pause/resume the ongoing recording or playback, default is f2",
+    )
+    arg_parser.add_argument(
+        "--playback-loops",
+        required=False,
+        type=int,
+        default=1,
+        help="in 'play' mode: number of times to loop through recorded events, default is 1 single loop",
+    )
+    arg_parser.add_argument(
+        "--playback-rate",
+        required=False,
+        type=float,
+        default=1.0,
+        help="in 'play' mode: speed coefficient to apply to the recording, default is x1.0",
+    )
+    arg_parser.add_argument(
+        "--playback-delay",
+        required=False,
+        type=float,
+        default=1.0,
+        help="in 'play' mode: number of seconds to sleep before playing the recording, default is 1.0 second",
+    )
+    arg_parser.add_argument(
+        "--playback-offset",
+        required=False,
+        type=int,
+        default=0,
+        help="in 'play' mode: how many events the first loop will skip, default is 0 event skipped",
+    )
+    arg_parser.add_argument(
+        "--playback-noise",
+        required=False,
+        type=bool,
+        help="in 'play' mode: to add noise to the time intervals between events",
+    )
+    args = arg_parser.parse_args()
+
+    mode = args.mode
+
+    if mode == "play":
+        for button_event in play(
+            button_events=input_button_events(),
+            exit_key=args.exit_key,
+            pause_key=args.pause_key,
+            loops=args.playback_loops,
+            rate=args.playback_rate,
+            delay=args.playback_delay,
+            offset=args.playback_offset,
+            verbose=args.playback_verbose,
+        ):
+            print(tuple(button_event), flush=True)
+    elif mode == "record":
+        for button_event in record(
+            exit_key=args.exit_key,
+            pause_key=args.pause_key,
+        ):
+            print(tuple(button_event), flush=True)
+    else:
+        raise ValueError("unsupported mode")

build/lib/botable/botable.py

@@ -0,0 +1,196 @@
+from collections import deque
+from contextlib import suppress
+import random
+import time
+from typing import Deque, Iterable, Iterator, List, Optional, Union, NamedTuple, Tuple
+from pynput import keyboard, mouse  # type: ignore
+from pynput.mouse import Button  # type: ignore
+from pynput.keyboard import Key, KeyCode  # type: ignore
+
+
+class ButtonEvent(NamedTuple):
+    button: str
+    pressed: bool
+    seconds_since_last_event: float
+    coordinates: Optional[Tuple[int, int]]
+
+
+def input_button_events() -> Iterator[ButtonEvent]:
+    with suppress(EOFError):
+        while event := input():
+            yield ButtonEvent(*eval(event))
+
+
+def _to_key(key: str) -> Union[Key, KeyCode, str]:
+    try:
+        return eval(f"Key.{key}")
+    except Exception:
+        pass
+    try:
+        return eval(KeyCode(key[1:-1]))
+    except Exception:
+        pass
+    return KeyCode.from_char(key)
+
+
+RECORDING = False
+PLAYING = False
+
+
+def record(exit_key: str = "f1", pause_key: str = "f2") -> Iterator[ButtonEvent]:
+    """
+    Launch the recording, yielding the keyboard and mouse click events as they occur.
+    Pressing the `exit_key` will terminate the recording.
+    Pressing the `pause_key` will pause/resume the recording.
+    """
+    global PLAYING, RECORDING
+    if PLAYING:
+        raise RuntimeError("Attempt to record while playing")
+    RECORDING = True
+    try:
+        paused_at: Optional[float] = None
+        last_event_at = time.time()
+        click_timestamps: List[float] = [time.time()]
+        button_events: Deque[ButtonEvent] = deque()
+        continue_ = True
+
+        exit_key_ = _to_key(exit_key)
+        pause_key_ = _to_key(pause_key)
+
+        def save_button_events(button: str, pressed: bool, position):
+            nonlocal paused_at, last_event_at
+            if paused_at:
+                return
+            current_time = time.time()
+            button_events.append(
+                ButtonEvent(button, pressed, current_time - last_event_at, position)
+            )
+            last_event_at = current_time
+
+        def on_press(key: keyboard.Key):
+            nonlocal paused_at, last_event_at, continue_
+            if key == exit_key_:
+                continue_ = False
+            elif key == pause_key_:
+                if paused_at:
+                    if click_timestamps:
+                        pause_time = time.time() - paused_at
+                        last_event_at += pause_time
+                        paused_at = None
+                else:
+                    paused_at = time.time()
+                return
+            save_button_events(str(key), True, None)
+
+        def on_release(key: keyboard.Key):
+            if key == pause_key_:
+                return
+            save_button_events(str(key), False, None)
+
+        def on_click(x: int, y: int, button: mouse.Button, pressed: bool):
+            save_button_events(str(button), pressed, (int(x), int(y)))
+
+        keyboard.Listener(on_press=on_press, on_release=on_release).start()
+
+        mouse.Listener(on_click=on_click).start()
+
+        while continue_:
+            while button_events:
+                yield button_events.popleft()
+            time.sleep(0.1)
+    finally:
+        RECORDING = False
+
+
+def play(
+    button_events: Iterable[ButtonEvent],
+    exit_key: str = "f1",
+    pause_key: str = "f2",
+    loops: int = 1,
+    rate: float = 1.0,
+    delay: float = 1.0,
+    offset: int = 0,
+    noise: bool = False,
+) -> Iterator[ButtonEvent]:
+    """
+    Waits `delay` and then iterates on `events` to play them,
+      optionally playing them at a modified speed `rate`,
+      optionally skipping the first `offset` events,
+      optionally adding noise to the time intervals between events (the original interval remains the minimum).
+    Once `button_events` is exhausted the entire collected set of event will optionally be replayed additional times if `loops` > 1.
+    Pressing the `exit_key` will terminate the recording.
+    Pressing the `pause_key` will pause/resume the recording.
+    """
+    global PLAYING, RECORDING
+    if RECORDING:
+        raise RuntimeError("Attempt to play while recording")
+    PLAYING = True
+    try:
+        time.sleep(delay)
+
+        continue_ = True
+        paused_at: Optional[float] = None
+        loop_index, event_index = 0, 0
+        mouse_ctrl = mouse.Controller()
+        keyboard_ctrl = keyboard.Controller()
+
+        exit_key_ = _to_key(exit_key)
+        pause_key_ = _to_key(pause_key)
+
+        def on_press(key: keyboard.Key):
+            nonlocal paused_at, continue_
+            if key == exit_key_:
+                continue_ = False
+            elif key == pause_key_:
+                paused_at = None if paused_at else time.time()
+
+        keyboard.Listener(on_press=on_press).start()
+
+        collected_button_events: List[ButtonEvent] = []
+
+        button_events_: Iterable[ButtonEvent] = button_events
+
+        for loop_index in range(loops):
+            for event_index, button_event in enumerate(button_events_):
+                if loops > 1 and not loop_index:
+                    collected_button_events.append(button_event)
+
+                while continue_ and paused_at:
+                    time.sleep(0.5)
+
+                if loop_index == 0 and offset > event_index:
+                    continue
+
+                if button_event.coordinates is None:
+                    ctrl = keyboard_ctrl
+                else:
+                    mouse_ctrl.position = button_event.coordinates
+                    ctrl = mouse_ctrl
+
+                if button_event.button.startswith("<") and button_event.button.endswith(
+                    ">"
+                ):
+                    evaluated_button = KeyCode(int(button_event.button[1:-1]))
+                else:
+                    evaluated_button = eval(button_event.button)
+
+                sleep_duration = button_event.seconds_since_last_event / rate
+                if noise:
+                    sleep_duration *= (1 + random.betavariate(2, 5) / 2)
+                time.sleep(sleep_duration)
+
+                if button_event.pressed:
+                    ctrl.press(evaluated_button)
+                else:
+                    ctrl.release(evaluated_button)
+
+                yield button_event
+
+                if not continue_:
+                    return
+
+            button_events_ = collected_button_events
+
+        time.sleep(0.2)
+    finally:
+        PLAYING = False

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name="botable",
-    version="0.0.5",
+    version="0.0.5-rc0",
     packages=["botable"],
     url="http://github.com/ebonnal/botable",
     license="Apache 2.",