chore: add time module and remove thrash

cereja/__init__.py

@@ -49,7 +49,7 @@
 from . import scraping
 from . import wcag
 
-VERSION = "2.0.3.final.0"
+VERSION = "2.0.4.final.0"
 
 __version__ = get_version_pep440_compliant(VERSION)
 

cereja/utils/__init__.py

@@ -26,3 +26,6 @@
 from ._utils import get_batch_strides as stride_values
 from . import colors
 from . import typography
+from . import decorators
+from . import time
+from .time import Timer, set_interval

cereja/utils/_utils.py

@@ -94,12 +94,11 @@
     "combinations_sizes",
     "value_from_memory",
     "str_gen",
-    "set_interval",
     "SourceCodeAnalyzer",
     "map_values",
     'decode_coordinates',
     'encode_coordinates',
-    'SingletonMeta'
+    'SingletonMeta',
 ]
 
 logger = logging.getLogger(__name__)
@@ -1582,16 +1581,6 @@ def str_gen(pattern: AnyStr) -> Sequence[AnyStr]:
     return regex.findall(string.printable)
 
 
-def set_interval(func: Callable, sec: float):
-    """
-    Call a function every sec seconds
-    @param func: function
-    @param sec: seconds
-    """
-    from .decorators import on_elapsed
-    on_elapsed(sec, loop=True, use_threading=True)(func)()
-
-
 def encode_coordinates(x: int, y: int):
     """
     Encode the coordinates (x, y) into a single lParam value.

cereja/utils/decorators.py

@@ -23,8 +23,8 @@
 import functools
 import threading
 import time
+import time as _time
 from abc import abstractmethod
-from typing import Callable, Any
 import abc
 import logging
 import warnings
@@ -33,15 +33,15 @@
 __all__ = [
     "depreciation",
     "synchronized",
-    "time_exec",
     "thread_safe_generator",
     "singleton",
     "on_except",
     "use_thread",
-    "on_elapsed",
     "retry_on_failure",
 ]
 
+from typing import Callable, Any
+
 from ..config.cj_types import PEP440
 
 logger = logging.getLogger(__name__)
@@ -89,25 +89,6 @@ def gen(*a, **kw):
     return gen
 
 
-def time_exec(func: Callable[[Any], Any]) -> Callable:
-    """
-    Decorator used to signal or perform a particular function.
-
-    :param func: Receives the function that will be executed as well as its parameters.
-    :return
-    """
-
-    @functools.wraps(func)
-    def wrapper(*args, **kwargs) -> Any:
-        from cereja import console
-        first_time = time.time()
-        result = func(*args, **kwargs)
-        console.log(f"[{func.__name__}] performed {time.time() - first_time}")
-        return result
-
-    return wrapper
-
-
 class Decorator(abc.ABC):
     def __init__(self):
         self.func = None
@@ -211,6 +192,7 @@ def wrapper(*args, **kwargs):
                         raise e
 
         return wrapper
+
     return register
 
 
@@ -226,6 +208,25 @@ def instance(*args, **kwargs):
     return instance
 
 
+def time_exec(func: Callable[[Any], Any]) -> Callable:
+    """
+    Decorator used to signal or perform a particular function.
+
+    :param func: Receives the function that will be executed as well as its parameters.
+    :return: Returns the function that will be executed.
+    """
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs) -> Any:
+        from cereja import console
+        first_time = time.time()
+        result = func(*args, **kwargs)
+        console.log(f"[{func.__name__}] performed {time.time() - first_time}")
+        return result
+
+    return wrapper
+
+
 def on_elapsed(interval: float = 1,
                loop: bool = False,
                use_threading: bool = False,
@@ -261,6 +262,7 @@ def run():
                     import threading
                     th = threading.Thread(target=run, daemon=is_daemon)
                     th.start()
+                    return th
                 else:
                     run()
             else:
@@ -277,6 +279,7 @@ def run():
                     import threading
                     th = threading.Thread(target=run, daemon=is_daemon)
                     th.start()
+                    return th
                 else:
                     return run()
 

cereja/utils/time.py

@@ -0,0 +1,204 @@
+"""
+Copyright (c) 2019 The Cereja Project
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+import threading
+import time as _time
+from typing import Callable
+
+__all__ = ["Timer", "set_interval"]
+
+
+class Timer:
+    """
+    A Timer class to manage time intervals and check if a specified interval has passed.
+
+    Attributes:
+        _interval (float): The time interval in seconds.
+        _start (float): The start time in seconds since the epoch.
+        _auto_reset (bool): Whether the timer should automatically reset after the interval has passed.
+    """
+
+    def __init__(self, interval, start=True, auto_reset=False):
+        """
+        Initialize the Timer.
+
+        Args:
+            interval (float): The time interval in seconds.
+            start (bool): Whether to start the timer immediately. Default is True.
+            auto_reset (bool): Whether the timer should automatically reset after the interval has passed.
+                               Default is False.
+        """
+        self._interval = interval
+        self._start = 0
+        if start:
+            self.start()
+        self._auto_reset = auto_reset
+
+    def __bool__(self):
+        """
+        Check if the specified time interval has passed.
+
+        Returns:
+            bool: True if the interval has passed, False otherwise.
+        """
+
+        if self.elapsed >= self._interval:
+            if self._auto_reset:
+                self.start()
+            return True
+        return False
+
+    def start(self):
+        """
+        Start the timer by setting the start time to the current time.
+        """
+        self._start = _time.time()
+
+    @property
+    def elapsed(self):
+        """
+        Get the elapsed time since the timer was started.
+
+        Returns:
+            float: The elapsed time in seconds.
+        """
+        return _time.time() - self._start
+
+    @property
+    def remaining(self):
+        """
+        Get the remaining time until the interval has passed.
+
+        Returns:
+            float: The remaining time in seconds.
+        """
+        return self._interval - self.elapsed
+
+    @property
+    def interval(self):
+        """
+        Get the time interval.
+
+        Returns:
+            float: The time interval in seconds.
+        """
+        return self._interval
+
+    @interval.setter
+    def interval(self, value):
+        """
+        Set the time interval.
+
+        Args:
+            value (float): The time interval in seconds.
+        """
+        self._interval = value
+
+    @property
+    def auto_reset(self):
+        """
+        Get the auto reset setting.
+
+        Returns:
+            bool: True if the timer automatically resets after the interval has passed, False otherwise.
+        """
+        return self._auto_reset
+
+    @auto_reset.setter
+    def auto_reset(self, value):
+        """
+        Set the auto reset setting.
+
+        Args:
+            value (bool): True to automatically reset the timer after the interval has passed, False otherwise.
+        """
+        self._auto_reset = value
+
+
+class IntervalScheduler:
+    """
+    A class to schedule the periodic execution of a function using a separate thread.
+
+    Attributes:
+        func (Callable): The function to be executed periodically.
+        interval (float): The time interval in seconds between each execution of the function.
+        is_daemon (bool): Whether the thread should be a daemon thread. Daemon threads are terminated when the main program exits.
+        thread (threading.Thread): The thread that runs the function periodically.
+        _stop_event (threading.Event): An event used to signal the thread to stop execution.
+    """
+
+    def __init__(self, func: Callable, interval: float, is_daemon: bool = False):
+        """
+        Initialize the IntervalScheduler.
+
+        Args:
+            func (Callable): The function to be executed periodically.
+            interval (float): The time interval in seconds between each execution of the function.
+            is_daemon (bool): Whether the thread should be a daemon thread. Default is False.
+        """
+        self.func = func
+        self.interval = interval
+        self.is_daemon = is_daemon
+        self.thread = None
+        self._stop_event = threading.Event()
+
+    def _run(self):
+        """
+        The internal method that runs the function periodically.
+        """
+        while not self._stop_event.is_set():
+            self.func()
+            _time.sleep(self.interval)
+
+    def start(self):
+        """
+        Start the periodic execution of the function in a new thread.
+        """
+        if self.thread is None or not self.thread.is_alive():
+            self._stop_event.clear()
+            self.thread = threading.Thread(target=self._run, daemon=self.is_daemon)
+            self.thread.start()
+
+    def stop(self):
+        """
+        Stop the periodic execution of the function and wait for the thread to finish.
+        """
+        if self.thread is not None:
+            self._stop_event.set()
+            self.thread.join()
+
+    def is_running(self):
+        """
+        Check if the thread is currently running.
+
+        Returns:
+            bool: True if the thread is running, False otherwise.
+        """
+        return self.thread is not None and self.thread.is_alive()
+
+
+def set_interval(func: Callable, sec: float, is_daemon=False) -> IntervalScheduler:
+    """
+    Call a function every sec seconds
+    @param func: function
+    @param is_daemon: If True, the thread will be a daemon
+    @param sec: seconds
+    """
+    scheduler = IntervalScheduler(func, sec, is_daemon)
+    scheduler.start()
+    return scheduler