Skip to content

Implement synchronization for rainbow effects #64

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Sae3sy wants to merge 1 commit into from
Closed

Implement synchronization for rainbow effects #64

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
233 changes: 180 additions & 53 deletions hPyT/hPyT.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,103 @@ class NCCALCSIZE_PARAMS(ctypes.Structure):

WINDOWS_VERSION = float(platform.version().split(".")[0])

# --- Synchronization Globals ---
sync_color_thread: Optional[threading.Thread] = None
sync_is_running: bool = False
sync_current_color: Optional[int] = None
sync_lock = threading.Lock() # Lock for thread-safe access to sync_current_color

def stop_sync_if_last():
"""Checks if any element is still running in sync mode and stops the master if none are."""

# Check if any window is still listed in either sync dictionary
is_title_bar_sync_running = bool(rainbow_title_bar._sync_timer_threads)
is_border_sync_running = bool(rainbow_border._sync_timer_threads)

if not is_title_bar_sync_running and not is_border_sync_running:
synchronized_rainbow.stop()

# *New Synchronization Class (Invisible Master)*

class synchronized_rainbow:
"""Manages the single, module-level thread for synchronized color cycling."""

@classmethod
def _color_changer_task(cls, interval: int, color_stops: int) -> None:
"""The continuous color cycle task for synchronization"""

global sync_current_color, sync_is_running
r, g, b = 200, 0, 0

while sync_is_running:

if r < 255 and g == 0 and b == 0:
r = min(255, r + color_stops)
elif r == 255 and g < 255 and b == 0:
g = min(255, g + color_stops)
elif r > 0 and g == 255 and b == 0:
r = max(0, r - color_stops)
elif g == 255 and b < 255 and r == 0:
b = min(255, b + color_stops)
elif g > 0 and b == 255 and r == 0:
g = max(0, g - color_stops)
elif b == 255 and r < 255 and g == 0:
r = min(255, r + color_stops)
elif b > 0 and r == 255 and g == 0:
b = max(0, b - color_stops)

with sync_lock:
sync_current_color = (r << 16) | (g << 8) | b

ctypes.windll.kernel32.Sleep(interval)

# When stopping, reset the global color
with sync_lock:
sync_current_color = None

@classmethod
def get_current_color(cls) -> Tuple[int, int, int]:
"""Gets the current RGB color from the synchronized master thread."""
global sync_current_color
with sync_lock:
color = sync_current_color

if color is None:
return (0, 0, 0)

# Convert integer color (BGR format) back to RGB tuple
Copy link
Copy Markdown
Contributor

@sourcery-ai sourcery-ai Bot Oct 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): Color conversion appears to use BGR order, but the packing uses RGB.

The unpacking in get_current_color expects BGR, but packing uses RGB. Please ensure both use the same channel order to avoid incorrect color values.

b = (color >> 16) & 0xFF
g = (color >> 8) & 0xFF
r = color & 0xFF
return (r, g, b)

@classmethod
def start(cls, interval: int, color_stops: int) -> None:
"""Starts the synchronized color thread (Invisible Master)."""
global sync_color_thread, sync_is_running

# Only start if not running
if not sync_is_running:
sync_is_running = True

# Use a dummy hwnd (0) as DWM calls are handled externally in sync mode
sync_color_thread = threading.Thread(
target=cls._color_changer_task,
args=(interval, color_stops),
daemon=True
)
sync_color_thread.start()

@classmethod
def stop(cls) -> None:
"""Stops the synchronized color thread."""
global sync_is_running, sync_color_thread
if sync_is_running:
sync_is_running = False
# Wait for the thread to finish cleanly
if threading.current_thread() is not sync_color_thread:
Copy link
Copy Markdown
Contributor

@sourcery-ai sourcery-ai Bot Oct 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): Joining the sync_color_thread with a timeout may not guarantee clean thread termination.

Consider increasing the timeout or implementing a loop to wait for thread termination, and ensure the thread responds promptly when sync_is_running is set to False.

Suggested implementation:

            # Wait for the thread to finish cleanly
            if threading.current_thread() is not sync_color_thread:
                max_wait = 2.0  # seconds
                waited = 0.0
                interval = 0.1
                while sync_color_thread.is_alive() and waited < max_wait:
                    sync_color_thread.join(timeout=interval)
                    waited += interval
            sync_color_thread = None
  • Ensure that the thread running in sync_color_thread checks the sync_is_running flag frequently and exits promptly when it is set to False. If this is not already implemented in the thread's target function, you should add such a check in its loop.

sync_color_thread.join(timeout=0.1)
sync_color_thread = None

class title_bar:
"""Hide or unhide the title bar of a window."""
Expand Down Expand Up @@ -783,26 +880,13 @@ class rainbow_title_bar:
"""Add a rainbow effect to the title bar of a window."""

current_color = None
_sync_timer_threads: Dict[int, threading.Thread] = {} # New: Tracks sync mode threads

@classmethod
def start(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
"""
Start the rainbow effect on the title bar of the specified window.

Args:
window (object): The window object to modify (e.g., a Tk instance in Tkinter).
interval (int): The interval between each color change in milliseconds. Default is 5.
color_stops (int): The number of color stops between each RGB value. Default is 5.

Example:
>>> rainbow_title_bar.start(window, interval=10, color_stops=10)

Notes:
The `interval` parameter controls the speed of the rainbow effect, and the `color_stops` parameter controls the number of color stops between each RGB value.
Higher values for `interval` will result in a slower rainbow effect.
Higher values for `color_stops` might skip most of the colors defying the purpose of the rainbow effect.
Start the standard, non-synchronized rainbow effect on the title bar.
"""

def color_changer(hwnd: int, interval: int) -> None:
r, g, b = 200, 0, 0
while hwnd in rnbtbs:
Expand Down Expand Up @@ -855,13 +939,6 @@ def color_changer(hwnd: int, interval: int) -> None:
def get_current_color(cls) -> Tuple[int, int, int]:
"""
Get the current RGB color value of the title bar, which is being displayed by the rainbow effect.
Can be useful if you want to synchronize the title bar's rainbow effect with other elements of the window.

Returns:
tuple: A tuple containing the RGB color values of the title bar.

Notes:
Example implementation of this feature available at [examples/rainbow-synchronization-example.py](https://github.com/Zingzy/hPyT/blob/main/examples/rainbow-synchronization-example.py).
"""

color = cls.current_color
Expand All @@ -875,10 +952,7 @@ def get_current_color(cls) -> Tuple[int, int, int]:
@classmethod
def stop(cls, window: Any) -> None:
"""
Stop the rainbow effect on the title bar of the specified window.

Args:
window (object): The window object to modify (e.g., a Tk instance in Tkinter).
Stop the standard rainbow effect on the title bar of the specified window.
"""

hwnd: int = module_find(window)
Expand All @@ -887,31 +961,56 @@ def stop(cls, window: Any) -> None:
else:
raise ValueError("Rainbow title bar is not running on this window.")

# New Synchronization Feature
@classmethod
def start_sync(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
"""
Starts the Title Bar in synchronization mode (as a slave).
It uses the synchronized_rainbow master thread for color data.
"""
hwnd: int = module_find(window)
if hwnd in cls._sync_timer_threads:
Copy link
Copy Markdown
Contributor

@sourcery-ai sourcery-ai Bot Oct 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): No explicit cleanup for threads in _sync_timer_threads after stop_sync.

Ensure that threads are properly joined or signaled to terminate when removed from _sync_timer_threads to prevent orphaned threads.

Suggested implementation:

    # Dictionary to store stop events for each thread
    _sync_timer_stop_events = {}

    @classmethod
    def start_sync(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
        """
        Starts the Title Bar in synchronization mode (as a slave).
        It uses the synchronized_rainbow master thread for color data.
        """
        hwnd: int = module_find(window)
WINDOWS_VERSION = float(platform.version().split(".")[0])

def _sync_timer_thread_func(hwnd, stop_event, interval, color_stops):
    while not stop_event.is_set():
        # Thread logic here, e.g. update color from master
        # ...
        time.sleep(interval)
    # Cleanup if needed

def stop_sync_if_last():
    """Checks if any element is still running in sync mode and stops the master if none are."""

def stop_sync(cls, window: Any) -> None:
    """
    Stops the Title Bar synchronization mode for the given window.
    Ensures thread is signaled to terminate and joined.
    """
    hwnd: int = module_find(window)
    thread = cls._sync_timer_threads.pop(hwnd, None)
    stop_event = cls._sync_timer_stop_events.pop(hwnd, None)
    if stop_event is not None:
        stop_event.set()
    if thread is not None:
        thread.join()
  • You will need to update the code where threads are created for sync mode to use _sync_timer_thread_func and pass a threading.Event as the stop signal.
  • Ensure that _sync_timer_threads and _sync_timer_stop_events are properly managed (added/removed) in both start_sync and stop_sync.
  • If your thread logic is elsewhere, adapt _sync_timer_thread_func accordingly.
  • You may need to add the stop_sync method to your class if it does not already exist, and ensure it is called when synchronization is stopped.

raise RuntimeError("Title bar is already running in synchronization mode.")

# 1. Start the invisible color manager master thread (if not already running)
synchronized_rainbow.start(interval, color_stops)

# 2. Start a timer thread for this window to periodically apply the master color
def sync_task(h: int, i: int):
while h in cls._sync_timer_threads:
rgb = synchronized_rainbow.get_current_color()
# Apply color using standard setter
title_bar_color.set(h, rgb)
ctypes.windll.kernel32.Sleep(i)

cls._sync_timer_threads[hwnd] = threading.Thread(target=sync_task, args=(hwnd, interval), daemon=True)
cls._sync_timer_threads[hwnd].start()

@classmethod
def stop_sync(cls, window: Any) -> None:
"""
Stops the Title Bar synchronization mode.
"""
hwnd: int = module_find(window)
if hwnd in cls._sync_timer_threads:
del cls._sync_timer_threads[hwnd]
title_bar_color.reset(window)
stop_sync_if_last() # Check if master thread should stop
else:
raise ValueError("Synchronized rainbow title bar is not running on this window.")


class rainbow_border:
"""Add a rainbow effect to the border of a window."""

current_color = None
_sync_timer_threads: Dict[int, threading.Thread] = {} # New: Tracks sync mode threads

@classmethod
def start(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
"""
Start the rainbow effect on the border of the specified window.

Args:
window (object): The window object to modify (e.g., a Tk instance in Tkinter).
interval (int): The interval between each color change in milliseconds. Default is 5.
color_stops (int): The number of color stops between each RGB value. Default is 5.

Example:
>>> rainbow_border.start(window, interval=10, color_stops=10)

Notes:
The `interval` parameter controls the speed of the rainbow effect, and the `color_stops` parameter controls the number of color stops between each RGB value.
Higher values for `interval` will result in a slower rainbow effect.
Higher values for `color_stops` might skip most of the colors defying the purpose of the rainbow effect.
Start the standard, non-synchronized rainbow effect on the border.
"""

def color_changer(hwnd, interval: int):
r, g, b = 200, 0, 0
while hwnd in rnbbcs:
Expand Down Expand Up @@ -964,13 +1063,6 @@ def color_changer(hwnd, interval: int):
def get_current_color(cls) -> Tuple[int, int, int]:
"""
Get the current RGB color value of the border, which is being displayed by the rainbow effect.
Can be useful if you want to synchronize the border's rainbow effect with other elements of the window.

Returns:
tuple: A tuple containing the RGB color values of the border.

Notes:
Example implementation of this feature available at [examples/rainbow-synchronization-example.py](https://github.com/Zingzy/hPyT/blob/main/examples/rainbow-synchronization-example.py
"""

color = cls.current_color
Expand All @@ -984,10 +1076,7 @@ def get_current_color(cls) -> Tuple[int, int, int]:
@classmethod
def stop(cls, window: Any) -> None:
"""
Stop the rainbow effect on the border of the specified window.

Args:
window (object): The window object to modify (e.g., a Tk instance in Tkinter).
Stop the standard rainbow effect on the border of the specified window.
"""

hwnd: int = module_find(window)
Expand All @@ -996,6 +1085,44 @@ def stop(cls, window: Any) -> None:
else:
raise ValueError("Rainbow border is not running on this window.")

# New Synchronization Feature
@classmethod
def start_sync(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
"""
Starts the Border in synchronization mode (as a slave).
It uses the synchronized_rainbow master thread for color data.
"""
hwnd: int = module_find(window)
if hwnd in cls._sync_timer_threads:
raise RuntimeError("Rainbow border is already running in synchronization mode.")

# 1. Start the invisible color manager master thread (if not already running)
synchronized_rainbow.start(interval, color_stops)

# 2. Start a timer thread for this window to periodically apply the master color
def sync_task(h: int, i: int):
while h in cls._sync_timer_threads:
rgb = synchronized_rainbow.get_current_color()
# Apply color using standard setter
border_color.set(h, rgb)
ctypes.windll.kernel32.Sleep(i)

cls._sync_timer_threads[hwnd] = threading.Thread(target=sync_task, args=(hwnd, interval), daemon=True)
cls._sync_timer_threads[hwnd].start()

@classmethod
def stop_sync(cls, window: Any) -> None:
"""
Stops the Border synchronization mode.
"""
hwnd: int = module_find(window)
if hwnd in cls._sync_timer_threads:
del cls._sync_timer_threads[hwnd]
border_color.reset(window)
stop_sync_if_last() # Check if master thread should stop
else:
raise ValueError("Synchronized rainbow border is not running on this window.")


class window_frame:
"""Change the position, size, and state of a window."""
Expand Down Expand Up @@ -1517,7 +1644,7 @@ def stylize_text(text: str, style: int) -> str:
"🅰🅱🅲🅳🅴🅵🅶🅷🅸🅹🅺🅻🅼🅽🅾🅿🆀🆁🆂🆃🆄🆅🆆🆇🆈🆉🅰🅱🅲🅳🅴🅵🅶🅷🅸🅹🅺🅻🅼🅽🅾🅿🆀🆁🆂🆃🆄🆅🆆🆇🆈🆉1234567890",
"ⓐⓑⓒⓓⓔⓕⓖⓗⓘⓙⓚⓛⓜⓝⓞⓟⓠⓡⓢⓣⓤⓥⓦⓧⓨⓩⒶⒷⒸⒹⒺⒻⒼⒽⒾⒿⓀⓁⓂⓃⓄⓅⓆⓇⓈⓉⓊⓋⓌⓍⓎⓏ①②③④⑤⑥⑦⑧⑨⓪",
"ᗩᗷᑕᗪEᖴGᕼIᒍKᒪᗰᑎOᑭᑫᖇᔕTᑌᐯᗯ᙭YᘔᗩᗷᑕᗪEᖴGᕼIᒍKᒪᗰᑎOᑭᑫᖇᔕTᑌᐯᗯ᙭Yᘔ1234567890",
"𝕒𝕓𝕔𝕕𝕖𝕗𝕘𝕙𝕚𝕛𝕜𝕝𝕞𝕟𝕠𝕡𝕢𝕣𝕤𝕥𝕦𝕧𝕨𝕩𝕪𝕫𝔸𝔹ℂ𝔻𝔼𝔽𝔾ℍ𝕀𝕁𝕂𝕃𝕄ℕ𝕆ℙℚℝ𝕊𝕋𝕌𝕍𝕎𝕏𝕐ℤ𝟙𝟚𝟛𝟜𝟝𝟞𝟟𝟠𝟡𝟘",
"𝕒𝕓𝕔𝕕𝕖𝕗𝕘𝕙𝕚𝕛𝕜𝕝𝕞𝕟𝕠𝕡𝕢𝕣𝕤𝕥𝕦𝕧𝕨𝕩𝕪𝕫𝔸𝔹ℂ𝔻𝔼𝔽𝔾ℍ𝕀𝕁𝕂𝕃𝕄ℕ𝕆ℙℚℝ𝕊𝕋𝕌𝕍𝕎𝕏𝕐ℤ𝟙𝚠𝟛𝟜𝟝𝟞𝟟𝟠𝟡𝟘",
"₳฿₵ĐɆ₣₲ⱧłJ₭Ⱡ₥₦Ø₱QⱤ₴₮ɄV₩ӾɎⱫ₳฿₵ĐɆ₣₲ⱧłJ₭Ⱡ₥₦Ø₱QⱤ₴₮ɄV₩ӾɎⱫ1234567890",
"卂乃匚ᗪ乇千Ꮆ卄丨ﾌҜㄥ爪几ㄖ卩Ɋ尺丂ㄒㄩᐯ山乂ㄚ乙卂乃匚ᗪ乇千Ꮆ卄丨ﾌҜㄥ爪几ㄖ卩Ɋ尺丂ㄒㄩᐯ山乂ㄚ乙1234567890",
]
Expand Down
Loading