Merge pull request #14 from quillcraftsman/time

Time

Makefile

@@ -37,4 +37,19 @@ sphinx-help:
 	make help -f Sphinxfile
 
 package_docs:
-	sphinx-apidoc -o docs/package replay_wizard/
+	sphinx-apidoc -o docs/package replay_wizard/
+
+capture:
+	python main.py capture debug
+
+time_capture:
+	python main.py capture debug -t true
+
+replay:
+	python main.py replay debug -d 5
+
+time_replay:
+	python main.py replay debug -t true -d 5
+
+monitor_replay:
+	python main.py replay debug -t true -d 5 -m true

main.py

@@ -2,9 +2,14 @@
 Manual run module
 """
 from replay_wizard.__main__ import main
+# from replay_wizard.replay.replay import compare
+# from replay_wizard.storage import load_from_file
 
 if __name__ == '__main__':
     # capture_actions()
     # time.sleep(5)
     # replay_actions()
     main()
+    # sequence = load_from_file('debug', true_time=True)
+    # other = load_from_file('monitoring', true_time=True)
+    # compare(sequence, other)

replay_wizard/capturing/capture.py

@@ -1,23 +1,22 @@
 """
 Capture process module
 """
-import json
 import pynput
+from replay_wizard.capturing.keyboard import on_press, on_release
 from replay_wizard.models import get_sequence
-from .keyboard import on_press, on_release
 
 
-def capture(name, true_time=False):
+def capture(name, true_time=False, non_blocking_mode=False):
     """
     capture user actions
 
     :param name: sequence name
     :param true_time: save or not sequence with true time. default = False
+    :param non_blocking_mode: use non-blocking threading mode. Default = false
     """
     Sequence = get_sequence(true_time=true_time)
     sequence = Sequence(
         name=name,
-        true_time=true_time,
     )
 
     def on_press_handler(key):
@@ -26,13 +25,15 @@ def on_press_handler(key):
     def on_release_handler(key):
         return on_release(sequence, key)
 
-    with pynput.keyboard.Listener(
+    if non_blocking_mode:
+        listener = pynput.keyboard.Listener(
             on_press=on_press_handler,
-            on_release=on_release_handler) as listener:
-        listener.join()
+            on_release=on_release_handler)
+        listener.start()
+    else:
+        with pynput.keyboard.Listener(
+                on_press=on_press_handler,
+                on_release=on_release_handler) as listener:
+            listener.join()
 
-    # save result
-    result_dict = sequence.model_dump()
-    file_name = f'{name}.sequence'
-    with open(file_name, 'w', encoding='utf-8') as f:
-        json.dump(result_dict, f)
+    return sequence

replay_wizard/cli/app.py

@@ -4,6 +4,7 @@
 import time
 import argparse
 from replay_wizard import capture, replay
+from replay_wizard.storage import save_to_file, load_from_file
 
 CAPTURE = 'capture'
 REPLAY = 'replay'
@@ -28,20 +29,28 @@ 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)
+    parser.add_argument('-t', '--timedelta', default=False, type=bool)
+    parser.add_argument('-m', '--monitoring', default=False, type=bool)
     args = parser.parse_args()
 
-    sequence = args.sequence
+    sequence_name = args.sequence
     mode = args.mode
     delay = args.delay
     timedelta = args.timedelta
-
-    modes = {
-        CAPTURE: capture,
-        REPLAY: replay,
-    }
-
-    run = modes[mode]
+    is_monitoring = args.monitoring
 
     time.sleep(delay)
-    run(sequence, true_time=timedelta)
+
+    if mode == CAPTURE:
+        sequence = capture(sequence_name, timedelta)
+        save_to_file(sequence)
+    else:
+        sequence = load_from_file(sequence_name, true_time=timedelta)
+        # create duplicated monitoring sequence in monitoring mode
+        # to check how sequence will be replayed
+        if is_monitoring:
+            duplicated_sequence = capture('monitoring', timedelta, non_blocking_mode=True)
+        replay(sequence, timedelta)
+
+        if is_monitoring:
+            save_to_file(duplicated_sequence)

replay_wizard/models/__init__.py

@@ -2,4 +2,14 @@
 Models package
 """
 from .action import Action, Subtypes, ActionEnum
-from .sequence import get_sequence
+from .sequence import Sequence
+from .time_sequence import TimeSequence
+
+
+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/models/sequence.py

@@ -1,7 +1,6 @@
 """
 Sequence module
 """
-import time
 from typing import List
 from pydantic import BaseModel, ConfigDict
 from .action import Action
@@ -15,7 +14,6 @@ class Sequence(BaseModel):
 
     name: str
     actions: List[Action] = []
-    # true_time: bool = False
 
     def __len__(self):
         return len(self.actions)
@@ -37,37 +35,3 @@ 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/models/time_sequence.py

@@ -0,0 +1,31 @@
+"""
+Time sequence model
+"""
+import time
+from .sequence import Sequence
+from .action import Action
+
+
+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)

replay_wizard/package.py

@@ -2,5 +2,5 @@
 Package info
 """
 name = 'replay-wizard'
-version = '0.3.0'
+version = '0.4.0'
 status = '3 - Alpha'

replay_wizard/replay/__init__.py

@@ -1,4 +1,4 @@
 """
 Replay package
 """
-from .core import replay
+from .replay import replay