Merge pull request #12 from quillcraftsman/time

add TimeSequence capturer and replayr

main.py

@@ -1,13 +1,10 @@
 """
 Manual run module
 """
-import time
-from quickstart.capturing import capture_actions
-from quickstart.replay import replay_actions
 from replay_wizard.__main__ import main
 
 if __name__ == '__main__':
-    capture_actions()
-    time.sleep(5)
-    replay_actions()
+    # capture_actions()
+    # time.sleep(5)
+    # replay_actions()
     main()

replay_wizard/capturing/capture.py

@@ -3,18 +3,21 @@
 """
 import json
 import pynput
-from replay_wizard.models import Sequence
+from replay_wizard.models import get_sequence
 from .keyboard import on_press, on_release
 
 
-def capture(name):
+def capture(name, true_time=False):
     """
     capture user actions
 
     :param name: sequence name
+    :param true_time: save or not sequence with true time. default = False
     """
+    Sequence = get_sequence(true_time=true_time)
     sequence = Sequence(
-        name=name
+        name=name,
+        true_time=true_time,
     )
 
     def on_press_handler(key):

replay_wizard/capturing/keyboard.py

@@ -4,7 +4,7 @@
 from pynput.keyboard import Key, KeyCode
 
 from replay_wizard.capturing.errors import UnknownKeyError
-from replay_wizard.models import Sequence, ActionEnum, Action, Subtypes
+from replay_wizard.models import ActionEnum, Action, Subtypes
 
 
 def key_to_value(key):
@@ -23,7 +23,7 @@ def key_to_value(key):
     raise ValueError(key)
 
 
-def on_key_input(sequence: Sequence, key, action_type: ActionEnum):
+def on_key_input(sequence, key, action_type: ActionEnum):
     """
     Key input
 
@@ -45,7 +45,7 @@ def on_key_input(sequence: Sequence, key, action_type: ActionEnum):
     return True
 
 
-def on_press(sequence: Sequence, key):
+def on_press(sequence, key):
     """
     Key was pressed
 
@@ -55,39 +55,11 @@ def on_press(sequence: Sequence, key):
     return on_key_input(sequence, key, ActionEnum.PRESS)
 
 
-def on_release(sequence: Sequence, key):
+def on_release(sequence, key):
     """
     Key was release
 
     :param sequence: current sequence
     :param key: pressed key
     """
     return on_key_input(sequence, key, ActionEnum.RELEASE)
-
-#
-#
-# def on_release(key, f):
-#     if key == keyboard.Key.esc:
-#         # Stop listener
-#         return False
-#     print('{0} released'.format(
-#         key))
-#     f.write(f'{key.char} released\n')
-#
-#
-# def capture(name):
-#     file_name = f'{name}.sequence'
-#     with open(file_name, 'w', encoding='utf-8') as f:
-#         on_press_handler = lambda key: on_press(key, f)
-#         on_release_handler = lambda key: on_release(key, f)
-#
-#         with keyboard.Listener(
-#                 on_press=on_press_handler,
-#                 on_release=on_release_handler) as listener:
-#             listener.join()
-#
-#     # ...or, in a non-blocking fashion:
-#     # listener = keyboard.Listener(
-#     #     on_press=on_press,
-#     #     on_release=on_release)
-#     # listener.start()

replay_wizard/cli/app.py

@@ -28,11 +28,13 @@ def run_cli():
     parser.add_argument('mode', choices=[CAPTURE, REPLAY])  # positional argument
     parser.add_argument('sequence')
     parser.add_argument('-d', '--delay', default=0, type=int)
+    parser.add_argument('-t', '--timedelta', default=True, type=bool)
     args = parser.parse_args()
 
     sequence = args.sequence
     mode = args.mode
     delay = args.delay
+    timedelta = args.timedelta
 
     modes = {
         CAPTURE: capture,
@@ -42,4 +44,4 @@ def run_cli():
     run = modes[mode]
 
     time.sleep(delay)
-    run(sequence)
+    run(sequence, true_time=timedelta)

replay_wizard/models/__init__.py

@@ -2,4 +2,4 @@
 Models package
 """
 from .action import Action, Subtypes, ActionEnum
-from .sequence import Sequence
+from .sequence import get_sequence

replay_wizard/models/sequence.py

@@ -1,17 +1,21 @@
 """
 Sequence module
 """
+import time
 from typing import List
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 from .action import Action
 
 
 class Sequence(BaseModel):
     """
     Action sequence
     """
+    model_config = ConfigDict(frozen=True)
+
     name: str
     actions: List[Action] = []
+    # true_time: bool = False
 
     def __len__(self):
         return len(self.actions)
@@ -33,3 +37,37 @@ def __contains__(self, item):
         in method
         """
         return item in self.actions
+
+
+class TimeSequence(Sequence):
+    """
+    Sequence with time
+    """
+    timestamp_list: list = []
+
+    @staticmethod
+    def get_current_timestamp():
+        """
+        Get current timestamp
+        """
+        return time.time()
+
+    def add(self, new_action: Action):
+        super().add(new_action)
+        timestamp = self.get_current_timestamp()
+        self.timestamp_list.append(timestamp)
+
+    # def is_valid_timestamps(self):
+    #     """
+    #     Correct timestamp list
+    #     """
+    #     return len(self.timestamp_list) == len(self.actions)
+
+
+def get_sequence(true_time=False) -> Sequence:
+    """
+    Fabric method to get sequence object
+
+    :param true_time: use true time
+    """
+    return TimeSequence if true_time else Sequence

replay_wizard/replay/core.py

@@ -1,23 +1,27 @@
 """
 Replay core module
 """
+# import itertools
+import time
 import json
-from replay_wizard.models import Sequence, Action
+from replay_wizard.models import Action, get_sequence
 from .keyboard import push_button
 
 
-def replay(name, extension='sequence'):
+def replay(name, extension='sequence', true_time=False):
     """
     Replay sequence from file
 
     :param name: sequence name, without extension
     :param extension: sequence file extension
+    :param true_time: replay or not sequence with true time. default = False
     """
     filename = f'{name}.{extension}'
     with open(filename, 'r', encoding='utf-8') as f:
         sequence_dict = json.load(f)
+        Sequence = get_sequence(true_time=true_time)
         sequence = Sequence.model_validate(sequence_dict)
-        replay_sequence(sequence)
+        replay_sequence(sequence, true_time=true_time)
 
 
 def replay_action(action: Action):
@@ -27,11 +31,50 @@ def replay_action(action: Action):
     push_button(action)
 
 
-def replay_sequence(sequence: Sequence):
+def replay_simple_sequence(sequence):
     """
-    Replay sequence
+    Just repeat all actions
 
-    :param sequence: sequence to replay
+    :param sequence: current sequence
     """
     for action in sequence:
         replay_action(action)
+
+
+def create_timedelta_list(timestamp_list):
+    """
+    Create timedelta list
+
+    :param timestamp_list: List with timedelta
+    """
+    timedelta_list = [0]
+    for i in range(1, len(timestamp_list)):
+        value = timestamp_list[i] - timestamp_list[i-1]
+        timedelta_list.append(value)
+
+    return timedelta_list
+
+
+def replay_time_sequence(sequence):
+    """
+    Replay sequence with time
+
+    :param sequence: current sequence
+    """
+    timedelta_list = create_timedelta_list(sequence.timestamp_list)
+    for action, delay in zip(sequence.actions, timedelta_list):
+        time.sleep(delay)
+        replay_action(action)
+
+
+def replay_sequence(sequence, true_time=False):
+    """
+    Replay sequence
+
+    :param sequence: sequence to replay
+    :param true_time: replay or not sequence with true time. default = False
+    """
+    if true_time:
+        replay_time_sequence(sequence)
+    else:
+        replay_simple_sequence(sequence)

testing/test_replay_wizard/conftest.py

@@ -2,7 +2,7 @@
 Pytest fixtures
 """
 from pytest import fixture
-from replay_wizard.models import Action, Subtypes, Sequence, ActionEnum
+from replay_wizard.models import Action, Subtypes, ActionEnum, get_sequence
 
 
 @fixture
@@ -36,12 +36,26 @@ def empty_sequence():
     """
     Empty sequence fixture
     """
+    Sequence = get_sequence()
     return Sequence(
         name='open youtube',
         actions=[]
     )
 
 
+@fixture
+def true_time_sequence():
+    """
+    Empty sequence with true time fixture
+    """
+    TimeSequence = get_sequence(true_time=True)
+    return TimeSequence(
+        name='open youtube',
+        actions=[],
+        true_time=True,
+    )
+
+
 @fixture
 def one_action_sequence(empty_sequence, put_a_action):
     """

testing/test_replay_wizard/test_models/test_sequence.py

@@ -1,22 +1,10 @@
 """
 Test sequence module
 """
-from replay_wizard.models import Action, Sequence, Subtypes
+import pytest
+from pydantic import ValidationError
 
-
-def test_full_sequence(put_a_action):
-    """
-    Test full sequence data
-    """
-    put_b = Action(
-        subtype=Subtypes.KEYBOARD,
-        value='b',
-        timedelta=0.1,
-    )
-    Sequence(
-        name='input a then b',
-        actions=[put_a_action, put_b]
-    )
+from replay_wizard.models import Action
 
 
 def test_len(empty_sequence):
@@ -35,13 +23,29 @@ def test_add(empty_sequence, put_a_action):
     assert len(empty_sequence) == 1
 
 
+def test_add_true_time(true_time_sequence, put_a_action):
+    """
+    Test add method
+    """
+    true_time_sequence.add(put_a_action)
+
+
+def test_sequence_is_frozen(empty_sequence):
+    """
+    Test that sequence is frozen (immutable)
+    """
+    with pytest.raises(ValidationError):
+        empty_sequence.true_time = True
+
+
 def test_to_dict(empty_sequence, put_a_action, put_a_action_dict):
     """
     Test sequence to dict
     """
     result = {
         'name': 'open youtube',
         'actions': [],
+        # 'true_time': False,
     }
     assert result == empty_sequence.model_dump()
     empty_sequence.add(put_a_action)