Skip to content

Add synchronized rainbow effects for title bar and border #65

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

Open
Sae3sy wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Add synchronized rainbow effects for title bar and border #65

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
675bcb5
Add synchronized rainbow effects for title bar and border
Sae3sy Oct 18, 2025
35f8397
ruff format hPyT.py
Sae3sy Oct 22, 2025
529da6b
fix: Resolve mypy union-attr error in synchronized_rainbow.stop
Sae3sy Oct 22, 2025
38901ff
forgot to re-run 'uvx ruff format'
Sae3sy Oct 22, 2025
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
187 changes: 187 additions & 0 deletions hPyT/hPyT.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,107 @@ 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()


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
Copy link
Copy Markdown
Contributor

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

Choose a reason for hiding this comment

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

issue (complexity): Consider refactoring synchronization to use a single master thread with subscriber callbacks instead of per-window threads and global color bit-packing.

Suggested change
# New Synchronization Class
Rather than spawning a thread+Sleep loop for each window and bit-packing colors into an int, you can:
1. Store `sync_current_color` as an `(r,g,b)` tuple.
2. Keep a single master thread in `synchronized_rainbow` that
– updates the color,
– iterates over a `{hwnd: setter}` subscriber dict to push the new color,
– sleeps once per interval.
3. Expose `subscribe(hwnd, setter)`/`unsubscribe(hwnd)` so `title_bar` and `border` just register/reset callbacks.
Example refactoring of `synchronized_rainbow`:
```python
class synchronized_rainbow:
_thread: Optional[threading.Thread] = None
_running = False
_lock = threading.Lock()
_current_color: Tuple[int,int,int] = (0,0,0)
_subscribers: Dict[int, Callable[[int, Tuple[int,int,int]], None]] = {}
@classmethod
def _task(cls, interval: int, color_stops: int):
r, g, b = 200, 0, 0
while cls._running:
# ... same RGB‐cycle logic, but store tuple
with cls._lock:
cls._current_color = (r, g, b)
for hwnd, setter in cls._subscribers.items():
setter(hwnd, cls._current_color)
time.sleep(interval / 1000.0)
with cls._lock:
cls._current_color = (0,0,0)
@classmethod
def start(cls, interval: int, color_stops: int):
if not cls._running:
cls._running = True
cls._thread = threading.Thread(
target=cls._task, args=(interval, color_stops), daemon=True
)
cls._thread.start()
@classmethod
def stop(cls):
cls._running = False
if cls._thread:
cls._thread.join(0.1)
cls._thread = None
@classmethod
def subscribe(cls, hwnd: int, setter: Callable[[int, Tuple[int,int,int]], None]):
cls._subscribers[hwnd] = setter
@classmethod
def unsubscribe(cls, hwnd: int):
cls._subscribers.pop(hwnd, None)
if not cls._subscribers:
cls.stop()

Then in rainbow_title_bar and rainbow_border:

@classmethod
def start_sync(cls, window, interval=5, color_stops=5):
    hwnd = module_find(window)
    synchronized_rainbow.start(interval, color_stops)
    synchronized_rainbow.subscribe(hwnd, title_bar_color.set)

@classmethod
def stop_sync(cls, window):
    hwnd = module_find(window)
    synchronized_rainbow.unsubscribe(hwnd)
    title_bar_color.reset(window)

This removes the per-window thread loops, global bit‐packing, and duplicated sync logic.

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
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 (
sync_color_thread is not None
and threading.current_thread() is not sync_color_thread
):
sync_color_thread.join(timeout=0.1)
sync_color_thread = None
Comment on lines +192 to +200
Copy link
Copy Markdown
Contributor

@sourcery-ai sourcery-ai Bot Oct 18, 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 thread with a short timeout may not guarantee clean shutdown.

A short timeout may not be sufficient if the thread is sleeping. Consider increasing the timeout or adding logic to handle cases where the thread does not terminate, to prevent resource leaks.

Suggested change
if sync_is_running:
sync_is_running = False
# Wait for the thread to finish cleanly
if threading.current_thread() is not sync_color_thread:
sync_color_thread.join(timeout=0.1)
sync_color_thread = None
if sync_is_running:
sync_is_running = False
# Wait for the thread to finish cleanly
if threading.current_thread() is not sync_color_thread:
sync_color_thread.join(timeout=2.0)
if sync_color_thread.is_alive():
import logging
logging.warning("sync_color_thread did not terminate after timeout. Possible resource leak.")
sync_color_thread = None



class title_bar:
"""Hide or unhide the title bar of a window."""
Expand Down Expand Up @@ -783,6 +884,7 @@ 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] = {} # Tracks sync mode threads

@classmethod
def start(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
Expand Down Expand Up @@ -887,11 +989,53 @@ def stop(cls, window: Any) -> None:
else:
raise ValueError("Rainbow title bar is not running on this window.")

@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:
raise RuntimeError("Title bar is already running in synchronization mode.")

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

# 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)
Comment on lines +1006 to +1011
Copy link
Copy Markdown
Contributor

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

Choose a reason for hiding this comment

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

suggestion: Using kernel32.Sleep in Python threads may block signals and affect responsiveness.

Using ctypes.windll.kernel32.Sleep can interfere with Python's threading and signal handling. Prefer time.sleep(i / 1000) for compatibility and responsiveness.

Suggested implementation:

                time.sleep(i / 1000)
import time

        hwnd: int = module_find(window)


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] = {} # Tracks sync mode threads

@classmethod
def start(cls, window: Any, interval: int = 5, color_stops: int = 5) -> None:
Expand Down Expand Up @@ -996,6 +1140,49 @@ def stop(cls, window: Any) -> None:
else:
raise ValueError("Rainbow border is not running on this window.")

@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."
)

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

# 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