V2.0.4 (#220)

* fix: fixed win32 send event param

* chore: add decorator retry_on_failure

* release: v2.0.3

* chore: add time module and remove thrash

* chore: add PoolMeta

* improves: improve on_elapsed func.

* improves: improve on_elapsed func.

* improves: win32

.gitignore

@@ -110,4 +110,5 @@ venv.bak/
 
 CMakeLists.txt
 MANIFEST.in
-*.png
+*.png
+/local

README.md

@@ -288,7 +288,7 @@ cj.mathtools.theta_angle((2, 2), (0, -2))  # 135.0
 ### 🧰 Utils
 
 ```python
-
+import cereja.utils.time
 import cereja as cj
 
 data = {"key1": 'value1', "key2": 'value2', "key3": 'value3', "key4": 'value4'}
@@ -322,7 +322,7 @@ cj.utils.list_methods(cj.Path)
 cj.utils.string_to_literal('[1,2,3,4]')
 # Output -> [1, 2, 3, 4]
 
-cj.utils.time_format(3600)
+cereja.utils.time.time_format(3600)
 # Output -> '01:00:00'
 
 cj.utils.truncate("Cereja is fun.", k=3)

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/display/_display.py

@@ -35,7 +35,8 @@
 from cereja.utils import is_iterable
 from cereja.config.cj_types import Number
 from cereja.system.unicode import Unicode
-from cereja.utils import fill, time_format, get_instances_of, import_string
+from cereja.utils import fill, get_instances_of, import_string
+from cereja.utils.time import time_format
 from cereja.mathtools import proportional, estimate, percent
 
 __all__ = ["Progress", "State", "console"]

cereja/system/_win32.py

@@ -176,7 +176,7 @@ class Keyboard:
 
     __WM_KEYDOWN = 0x0100
     __WM_KEYUP = 0x0101
-    MAX_TIME_SIMULE_KEY_PRESS = 0.05
+    MAX_TIME_SIMULE_KEY_PRESS = 0.1
 
     def __init__(self, hwnd=None, is_async=False):
         self._is_async = is_async
@@ -258,8 +258,8 @@ def _press_and_wait(self, key, secs):
     def _press_n_times(self, key, n_times=1):
         for _ in range(n_times):
             self._key_down(key)
-            time.sleep(0.05 + (self._max_time_simule_key_press - (random.random() * self._max_time_simule_key_press)))
-        self._key_up(key)
+            time.sleep((random.random() * self._max_time_simule_key_press) + 0.01)
+            self._key_up(key)
 
     def _key_press(self, key_code, n_times=1, secs=None):
         key_code = self._parse_key(key_code)

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, time_format

cereja/utils/_utils.py

@@ -23,7 +23,6 @@
 import re
 import string
 import threading
-import time
 from collections import OrderedDict, defaultdict
 from importlib import import_module
 import importlib
@@ -58,7 +57,6 @@
     "logger_level",
     "module_references",
     "set_log_level",
-    "time_format",
     "string_to_literal",
     "rescale_values",
     "Source",
@@ -94,12 +92,12 @@
     "combinations_sizes",
     "value_from_memory",
     "str_gen",
-    "set_interval",
     "SourceCodeAnalyzer",
     "map_values",
     'decode_coordinates',
     'encode_coordinates',
-    'SingletonMeta'
+    'SingletonMeta',
+    'PoolMeta'
 ]
 
 logger = logging.getLogger(__name__)
@@ -165,7 +163,6 @@ def split_sequence(seq: List[Any], is_break_fn: Callable) -> List[List[Any]]:
 
 def map_values(obj: Union[dict, list, tuple, Iterator], fn: Callable) -> Union[dict, list, tuple, Iterator]:
     fn_arg_count = SourceCodeAnalyzer(fn).argument_count
-    is_dict = isinstance(obj, dict)
     if isinstance(obj, dict):
         obj = obj.items()
     _iter = iter(obj)
@@ -610,33 +607,6 @@ def get_attr_if_exists(obj: Any, attr: str) -> Union[object, None]:
     return None
 
 
-def time_format(seconds: float, format_="%H:%M:%S") -> Union[str, float]:
-    """
-    Default format is '%H:%M:%S'
-    If the time exceeds 24 hours, it will return a format like 'X days HH:MM:SS'
-
-    >>> time_format(3600)
-    '01:00:00'
-    >>> time_format(90000)
-    '1 days 01:00:00'
-
-    """
-    # Check if seconds is a valid number
-    if seconds >= 0 or seconds < 0:
-        # Calculate the absolute value of days
-        days = int(seconds // 86400)
-        # Format the time
-        time_ = time.strftime(format_, time.gmtime(abs(seconds) % 86400))
-        # Return with days if more than 24 hours
-        if days > 0:
-            return f"{days} days {time_}"
-        # Return the formatted time
-        if seconds < 0:
-            return f"-{time_}"
-        return time_
-    return seconds  # Return NaN or any invalid input as it is
-
-
 def fill(value: Union[list, str, tuple], max_size, with_=" ") -> Any:
     """
     Calculates and adds value
@@ -1582,16 +1552,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.
@@ -1627,6 +1587,37 @@ def decode_coordinates(lparam: int):
     return x, y
 
 
+class PoolMeta(type):
+    """A thread-safe implementation to control the maximum number of instances."""
+    _instances = []
+    _lock: threading.Lock = threading.Lock()  # Class-level lock
+    _max_instances = 3  # Define the maximum number of instances allowed
+    _available_instances = threading.Condition(_lock)
+
+    def __call__(cls, *args, **kwargs):
+        with cls._available_instances:
+            while len(cls._instances) >= cls._max_instances:
+                cls._available_instances.wait()
+            instance = super(PoolMeta, cls).__call__(*args, **kwargs)
+            cls._instances.append(instance)
+            return instance
+
+    def release_instance(cls, instance):
+        with cls._available_instances:
+            if instance in cls._instances:
+                cls._instances.remove(instance)
+                cls._available_instances.notify()
+
+    def set_max_instances(cls, max_instances):
+        cls._max_instances = max_instances
+
+    def get_max_instances(cls):
+        return cls._max_instances
+
+    def get_instances(cls):
+        return cls._instances
+
+
 class SingletonMeta(type):
     """A thread-safe implementation of Singleton."""
     _instances = {}

cereja/utils/decorators.py

@@ -24,7 +24,6 @@
 import threading
 import time
 from abc import abstractmethod
-from typing import Callable, Any
 import abc
 import logging
 import warnings
@@ -33,15 +32,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 +88,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 +191,7 @@ def wrapper(*args, **kwargs):
                         raise e
 
         return wrapper
+
     return register
 
 
@@ -226,11 +207,32 @@ 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,
                verbose: bool = False,
-               is_daemon: bool = False):
+               is_daemon: bool = False,
+               take_last=False,
+               default=None):
     """
     Run a function if the interval has elapsed
 
@@ -239,13 +241,19 @@ def on_elapsed(interval: float = 1,
     @param verbose: If True, the function name will be printed
     @param use_threading: If True, the function will be executed in a thread
     @param is_daemon: If True, the thread will be a daemon
+    @param take_last: If the time has not run out, it stores and returns the last result of the function execution,
+                      if the function returns anything.
+    @param default: return it if the time has not run out
+
     """
 
     def decorator(func: Callable):
         last_time = 0.0
+        last_result = None
 
         def wrapper(*args, **kwargs):
             nonlocal last_time
+            nonlocal last_result
             if loop:
                 def run():
                     nonlocal last_time
@@ -261,22 +269,29 @@ def run():
                     import threading
                     th = threading.Thread(target=run, daemon=is_daemon)
                     th.start()
+                    return th
                 else:
                     run()
             else:
                 def run():
                     nonlocal last_time
+                    nonlocal last_result
                     current_time = time.time()
                     if current_time - last_time >= interval:
                         if verbose:
                             print(f"Running {func.__name__}")
                         last_time = current_time
-                        return func(*args, **kwargs)
+                        if take_last:
+                            last_result = func(*args, **kwargs)
+                        else:
+                            return func(*args, **kwargs)
+                    return default or last_result
 
                 if use_threading:
                     import threading
                     th = threading.Thread(target=run, daemon=is_daemon)
                     th.start()
+                    return th
                 else:
                     return run()