Run installation concurrently (Unexpected behavior), class for managing console lines

scripts/install/theme_installer.py

@@ -1,4 +1,5 @@
 import argparse
+import concurrent.futures
 from abc import ABC, abstractmethod
 
 from scripts.install.colors_definer import ColorsDefiner
@@ -75,17 +76,24 @@ class ThemeInstaller(ABC):
     def _apply_default_color(self) -> bool:
         colors = self.colors.colors
         args = self.args
-        installed_any = False
 
+        colors_to_install = []
         for color, values in colors.items():
-            if self.args.all or getattr(self.args, color, False):
+            if args.all or getattr(args, color, False):
                 hue = values.get('h')
-                sat = values.get('s', args.sat)  # if saturation already defined in color (gray)
-
-                self._install_theme(hue, color, sat)
-                installed_any = True
+                sat = values.get('s', args.sat)
+                colors_to_install.append((hue, color, sat))
 
                 if self.stop_after_first_installed_color:
                     break
 
-        return installed_any
+        if not colors_to_install:
+            return False
+        self._run_concurrent_installation(colors_to_install)
+        return True
+
+    def _run_concurrent_installation(self, colors_to_install):
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = [executor.submit(self._install_theme, hue, color, sat)
+                       for hue, color, sat in colors_to_install]
+            concurrent.futures.wait(futures)

scripts/theme.py

@@ -8,6 +8,7 @@ from .utils import (
     copy_files,          # copy files from source to destination
     destination_return,  # copied/modified theme location
     generate_file)       # combine files from folder to one file
+from .utils.console import Console, Color, Format
 
 
 class Theme:
@@ -28,7 +29,7 @@ class Theme:
         self.temp_folder = f"{temp_folder}/{theme_type}"
         self.theme_folder = theme_folder
         self.theme_type = theme_type
-        self.mode = [mode] if mode else ['light', 'dark']
+        self.modes = [mode] if mode else ['light', 'dark']
         self.destination_folder = destination_folder
         self.main_styles = f"{self.temp_folder}/{theme_type}.css"
         self.is_filled = is_filled
@@ -58,9 +59,81 @@ class Theme:
 
         return self
 
-    def __del__(self):
-        # delete temp folder
-        shutil.rmtree(self.temp_folder, ignore_errors=True)
+    def prepare(self):
+        # move files to temp folder
+        copy_files(self.theme_folder, self.temp_folder)
+        generate_file(f"{self.theme_folder}", self.temp_folder, self.main_styles)
+        # after generating main styles, remove .css and .versions folders
+        shutil.rmtree(f"{self.temp_folder}/.css/", ignore_errors=True)
+        shutil.rmtree(f"{self.temp_folder}/.versions/", ignore_errors=True)
+
+        # if theme is filled
+        if self.is_filled:
+            for apply_file in os.listdir(f"{self.temp_folder}/"):
+                replace_keywords(f"{self.temp_folder}/{apply_file}",
+                                 ("BUTTON-COLOR", "ACCENT-FILLED-COLOR"),
+                                 ("BUTTON_HOVER", "ACCENT-FILLED_HOVER"),
+                                 ("BUTTON_ACTIVE", "ACCENT-FILLED_ACTIVE"),
+                                 ("BUTTON_INSENSITIVE", "ACCENT-FILLED_INSENSITIVE"),
+                                 ("BUTTON-TEXT-COLOR", "TEXT-BLACK-COLOR"),
+                                 ("BUTTON-TEXT_SECONDARY", "TEXT-BLACK_SECONDARY"))
+
+    def add_to_start(self, content):
+        """
+        Add content to the start of main styles
+        :param content: content to add
+        """
+
+        with open(self.main_styles, 'r') as main_styles:
+            main_content = main_styles.read()
+
+        with open(self.main_styles, 'w') as main_styles:
+            main_styles.write(content + '\n' + main_content)
+
+    def install(self, hue, name: str, sat=None, destination=None):
+        """
+        Copy files and generate theme with specified accent color
+        :param hue
+        :param name: theme name
+        :param sat
+        :param destination: folder where theme will be installed
+        """
+
+        joint_modes = f"({', '.join(self.modes)})"
+
+        line = Console.Line(name)
+        formatted_name = Console.format(name.capitalize(), color=Color.get(name), format_type=Format.BOLD)
+        formatted_mode = Console.format(joint_modes, color=Color.GRAY)
+        line.update(f"Creating {formatted_name} {formatted_mode} theme...")
+
+        try:
+            self._install_and_apply_theme(hue, name, sat=sat, destination=destination)
+            line.success(f"{formatted_name} {formatted_mode} theme created successfully.")
+
+        except Exception as err:
+            line.error(f"Error installing {formatted_name} theme: {str(err)}")
+
+    def _install_and_apply_theme(self, hue, name, sat=None, destination=None):
+        is_dest = bool(destination)
+        for mode in self.modes:
+            if not is_dest:
+                destination = destination_return(self.destination_folder, name, mode, self.theme_type)
+
+            copy_files(self.temp_folder + '/', destination)
+            self.__apply_theme(hue, self.temp_folder, destination, mode, sat=sat)
+
+    def __apply_theme(self, hue, source, destination, theme_mode, sat=None):
+        """
+        Apply theme to all files in directory
+        :param hue
+        :param source
+        :param destination: file directory
+        :param theme_mode: theme name (light or dark)
+        :param sat: color saturation (optional)
+        """
+
+        for apply_file in os.listdir(f"{source}/"):
+            self.__apply_colors(hue, destination, theme_mode, apply_file, sat=sat)
 
     def __apply_colors(self, hue, destination, theme_mode, apply_file, sat=None):
         """
@@ -97,74 +170,3 @@ class Theme:
 
         # replace colors
         replace_keywords(os.path.expanduser(f"{destination}/{apply_file}"), *replaced_colors)
-
-    def __apply_theme(self, hue, source, destination, theme_mode, sat=None):
-        """
-        Apply theme to all files in directory
-        :param hue
-        :param source
-        :param destination: file directory
-        :param theme_mode: theme name (light or dark)
-        :param sat: color saturation (optional)
-        """
-
-        for apply_file in os.listdir(f"{source}/"):
-            self.__apply_colors(hue, destination, theme_mode, apply_file, sat=sat)
-
-    def prepare(self):
-        # move files to temp folder
-        copy_files(self.theme_folder, self.temp_folder)
-        generate_file(f"{self.theme_folder}", self.temp_folder, self.main_styles)
-        # after generating main styles, remove .css and .versions folders
-        shutil.rmtree(f"{self.temp_folder}/.css/", ignore_errors=True)
-        shutil.rmtree(f"{self.temp_folder}/.versions/", ignore_errors=True)
-
-        # if theme is filled
-        if self.is_filled:
-            for apply_file in os.listdir(f"{self.temp_folder}/"):
-                replace_keywords(f"{self.temp_folder}/{apply_file}",
-                                 ("BUTTON-COLOR", "ACCENT-FILLED-COLOR"),
-                                 ("BUTTON_HOVER", "ACCENT-FILLED_HOVER"),
-                                 ("BUTTON_ACTIVE", "ACCENT-FILLED_ACTIVE"),
-                                 ("BUTTON_INSENSITIVE", "ACCENT-FILLED_INSENSITIVE"),
-                                 ("BUTTON-TEXT-COLOR", "TEXT-BLACK-COLOR"),
-                                 ("BUTTON-TEXT_SECONDARY", "TEXT-BLACK_SECONDARY"))
-
-    def install(self, hue, name, sat=None, destination=None):
-        """
-        Copy files and generate theme with different accent color
-        :param hue
-        :param name: theme name
-        :param sat: color saturation (optional)
-        :param destination: folder where theme will be installed
-        """
-
-        is_dest = bool(destination)
-
-        print(f"Creating {name} {', '.join(self.mode)} theme...", end=" ")
-
-        try:
-            for mode in self.mode:
-                if not is_dest:
-                    destination = destination_return(self.destination_folder, name, mode, self.theme_type)
-
-                copy_files(self.temp_folder + '/', destination)
-                self.__apply_theme(hue, self.temp_folder, destination, mode, sat=sat)
-
-        except Exception as err:
-            print("\nError: " + str(err))
-
-        else:
-            print("Done.")
-
-    def add_to_start(self, content):
-        """
-        Add content to the start of main styles
-        :param content: content to add
-        """
-
-        with open(self.main_styles, 'r') as main_styles:
-            main_content = main_styles.read()
-
-        with open(self.main_styles, 'w') as main_styles:
-            main_styles.write(content + '\n' + main_content)

scripts/utils/console.py

@@ -0,0 +1,92 @@
+import sys
+import threading
+from enum import Enum
+from typing import Optional
+
+
+class Console:
+    """Manages console output for concurrent processes with line tracking"""
+    _print_lock = threading.Lock()
+    _line_mapping = {}
+    _next_line = 0
+
+    class Line:
+        def __init__(self, name):
+            """Initialize a new managed line"""
+            self.name = name
+            self._reserve_line()
+
+        def update(self, message, icon="⏳"):
+            """Update the status message for the line"""
+            with Console._print_lock:
+                # Calculate how many lines to move up
+                lines_up = Console._next_line - Console._line_mapping[self.name]
+                # Move the cursor to correct line
+                if lines_up > 0:
+                    sys.stdout.write(f"\033[{lines_up}F")
+                # Clear line and write status
+                sys.stdout.write(f"\r\033[K{icon} {message}")
+                # Move the cursor back down
+                if lines_up > 0:
+                    sys.stdout.write(f"\033[{lines_up}E")
+                sys.stdout.flush()
+
+        def success(self, message):
+            self.update(message, "✅")
+
+        def error(self, message):
+            self.update(message, "❌")
+
+        def warn(self, message):
+            self.update(message, "⚠️")
+
+        def _reserve_line(self):
+            """Reserve a line for future updates"""
+            with Console._print_lock:
+                line_number = Console._next_line
+                Console._next_line += 1
+                Console._line_mapping[self.name] = line_number
+                sys.stdout.write(f"\n")  # Ensure we have a new line
+                sys.stdout.flush()
+                return line_number
+
+    @staticmethod
+    def format(text: str, color: Optional['Color']=None, format_type: Optional['Format']=None):
+        """Apply color and formatting to text"""
+        formatted_text = text
+
+        if color:
+            formatted_text = color.value + formatted_text + Color.NORMAL.value
+        if format_type:
+            formatted_text = format_type.value + formatted_text + Format.NORMAL.value
+
+        return formatted_text
+
+
+class Color(Enum):
+    """ANSI color codes for terminal output"""
+    NORMAL = '\033[0m'  # Reset color
+    BLACK = '\033[30m'
+    RED = '\033[31m'
+    GREEN = '\033[32m'
+    YELLOW = '\033[33m'
+    BLUE = '\033[34m'
+    MAGENTA = '\033[35m'
+    PURPLE = '\033[35m'
+    CYAN = '\033[36m'
+    WHITE = '\033[37m'
+    GRAY = '\033[90m'
+
+    @classmethod
+    def get(cls, color: str) -> Optional['Color']:
+        return getattr(cls, color.upper(), None)
+
+
+class Format(Enum):
+    """ANSI formatting codes for terminal output"""
+    NORMAL = '\033[0m'  # Reset formatting
+    BOLD = '\033[1m'
+    ITALIC = '\033[3m'
+    UNDERLINE = '\033[4m'
+    BLINK = '\033[5m'
+    REVERSE = '\033[7m'

scripts/utils/copy_files.py

@@ -1,4 +1,6 @@
 import os
+import shutil
+
 
 def copy_files(source, destination):
     """
@@ -7,6 +9,7 @@ def copy_files(source, destination):
     :param destination: where files will be pasted
     """
 
-    destination = os.path.expanduser(destination)  # expand ~ to /home/user
+    destination = os.path.expanduser(destination)
     os.makedirs(destination, exist_ok=True)
-    os.system(f"cp -aT {source} {destination}")
+
+    shutil.copytree(source, destination, dirs_exist_ok=True)