diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 352b50c..f4b64f6 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -53,6 +53,7 @@ jobs: scriptPath="${{ github.workspace }}/blocknet_aio_monitor.py" cmd=(pyinstaller --noconfirm --onefile \ + --hidden-import='PIL._tkinter_finder' \ --add-data "theme:theme" \ --add-data "img:img" \ --clean \ @@ -97,13 +98,19 @@ jobs: - name: Set up Python using Homebrew run: | - brew install python@3.10 - brew install python-tk@3.10 - python3.10 -m pip install --upgrade pip + brew update + brew upgrade || true + brew link --overwrite python@3.12 + brew install python + brew install python-tk + python3 -m pip install --upgrade pip + # brew install python@3.10 + # brew install python-tk@3.10 + # python3.10 -m pip install --upgrade pip - name: Install dependencies / Build executable on macOS run: | - python3.10 -m venv venv + python3 -m venv venv source venv/bin/activate pip install -r requirements.txt pip install pyinstaller diff --git a/blocknet_aio_monitor.py b/blocknet_aio_monitor.py index 80bdc59..677a810 100644 --- a/blocknet_aio_monitor.py +++ b/blocknet_aio_monitor.py @@ -3,10 +3,8 @@ import os import signal -import PIL._tkinter_finder import customtkinter as ctk from PIL import Image -from psutil import process_iter import widgets_strings from gui.binary_manager import BinaryManager @@ -27,7 +25,10 @@ class Blocknet_AIO_GUI(ctk.CTk): + """Main GUI class for Blocknet AIO application.""" + def __init__(self): + """Initialize the Blocknet AIO GUI application.""" super().__init__() self.install_greyed_img = None self.install_img = None @@ -40,12 +41,12 @@ def __init__(self): self.transparent_img = None self.theme_img = None - self.disable_daemons_conf_check = False + self.disable_daemons_conf_check: bool = False - self.cfg = utils.load_cfg_json() + self.cfg: dict = utils.load_cfg_json() self.adjust_theme() - self.custom_path = None - self.stored_password = None + self.custom_path: str = None + self.stored_password: str = None if self.cfg: if 'custom_path' in self.cfg: self.custom_path = self.cfg['custom_path'] @@ -56,26 +57,17 @@ def __init__(self): logging.error(f"Error decrypting XLite password: {e}") self.stored_password = None - self.binary_manager = None - self.blocknet_manager = None - self.blockdx_manager = None - self.xlite_manager = None - - self.time_disable_button = 3000 + self.time_disable_button: int = 3000 - # frames - self.bins_download_frame = None - self.bins_title_frame = None - self.blocknet_core_frame = None - self.blocknet_title_frame = None - self.blockdx_frame = None - self.blockdx_title_frame = None - self.xlite_frame = None - self.xlite_title_frame = None + self.tooltip_manager: TooltipManager = TooltipManager(self) - self.tooltip_manager = TooltipManager(self) + self.blocknet_manager: BlocknetManager = BlocknetManager(self) + self.binary_manager: BinaryManager = BinaryManager(self) + self.blockdx_manager: BlockDXManager = BlockDXManager(self) + self.xlite_manager: XliteManager = XliteManager(self) - async def setup_management_sections(self): + async def setup_management_sections(self) -> None: + """Initialize and setup all management sections asynchronously.""" await asyncio.gather( self.binary_manager.setup(), self.blocknet_manager.setup(), @@ -83,18 +75,11 @@ async def setup_management_sections(self): self.xlite_manager.setup() ) - def create_managers(self): - self.blocknet_manager = BlocknetManager(self, self.blocknet_core_frame, self.blocknet_title_frame) - self.binary_manager = BinaryManager(self, self.bins_download_frame, self.bins_title_frame) - self.blockdx_manager = BlockDXManager(self, self.blockdx_frame, self.blockdx_title_frame) - self.xlite_manager = XliteManager(self, self.xlite_frame, self.xlite_title_frame) - - def init_setup(self): + def init_setup(self) -> None: + """Initialize the GUI setup, including layout, images, and frame configuration.""" self.title(widgets_strings.app_title_string) self.resizable(False, False) self.setup_load_images() - self.init_frames() - self.create_managers() self.after(0, self.check_processes) asyncio.run(self.setup_management_sections()) self.setup_tooltips() @@ -104,218 +89,185 @@ def init_setup(self): signal.signal(signal.SIGINT, self.handle_signal) signal.signal(signal.SIGTERM, self.handle_signal) - def init_frames(self): - self.bins_download_frame = ctk.CTkFrame(master=self) - self.bins_title_frame = ctk.CTkFrame(self.bins_download_frame) - - self.blocknet_core_frame = ctk.CTkFrame(master=self) - self.blocknet_title_frame = ctk.CTkFrame(self.blocknet_core_frame) - - self.blockdx_frame = ctk.CTkFrame(master=self) - self.blockdx_title_frame = ctk.CTkFrame(self.blockdx_frame) - - self.xlite_frame = ctk.CTkFrame(master=self) - self.xlite_title_frame = ctk.CTkFrame(self.xlite_frame) - - def setup_load_images(self): + def setup_load_images(self) -> None: + """Load and set up images for use in the GUI.""" resize = (65, 30) self.theme_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "light.png")).resize(resize, - PIL.Image.LANCZOS), - dark_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "dark.png")).resize(resize, - PIL.Image.LANCZOS), - size=resize) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "light.png")).resize(resize, + Image.LANCZOS), + dark_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "dark.png")).resize(resize, + Image.LANCZOS), + size=resize + ) resize = (50, 50) self.transparent_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "transparent.png")).resize(resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "transparent.png")).resize(resize, + Image.LANCZOS) + ) self.start_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "start-50.png")).resize(resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "start-50.png")).resize(resize, + Image.LANCZOS) + ) self.start_greyed_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "start-50_greyed.png")).resize( - resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "start-50_greyed.png")).resize(resize, + Image.LANCZOS) + ) self.stop_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "stop-50.png")).resize(resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "stop-50.png")).resize(resize, + Image.LANCZOS) + ) self.stop_greyed_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "stop-50_greyed.png")).resize( - resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "stop-50_greyed.png")).resize(resize, + Image.LANCZOS) + ) self.delete_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "delete-50.png")).resize(resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "delete-50.png")).resize(resize, + Image.LANCZOS) + ) self.delete_greyed_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "delete-50_greyed.png")).resize( - resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "delete-50_greyed.png")).resize(resize, + Image.LANCZOS) + ) self.install_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "installer-50.png")).resize(resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "installer-50.png")).resize(resize, + Image.LANCZOS) + ) self.install_greyed_img = ctk.CTkImage( - light_image=PIL.Image.open(os.path.join(global_variables.DIRPATH, "img", "installer-50_greyed.png")).resize( - resize, - PIL.Image.LANCZOS)) + light_image=Image.open(os.path.join(global_variables.DIRPATH, "img", "installer-50_greyed.png")).resize( + resize, Image.LANCZOS) + ) - def setup_tooltips(self): - self.tooltip_manager.register_tooltip(self.blocknet_core_frame, - msg=widgets_strings.tooltip_howtouse, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.blockdx_frame, - msg=widgets_strings.tooltip_howtouse, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.xlite_frame, - msg=widgets_strings.tooltip_howtouse, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.bins_download_frame, - msg=widgets_strings.tooltip_howtouse, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.bins_title_frame, - msg=widgets_strings.tooltip_bins_title_msg, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") + def setup_tooltips(self) -> None: + """Set up tooltips for various GUI components.""" + self.tooltip_manager.register_tooltip(self.blocknet_manager.frame_manager.master_frame, + msg=widgets_strings.tooltip_howtouse, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.blockdx_manager.frame_manager.master_frame, + msg=widgets_strings.tooltip_howtouse, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.xlite_manager.frame_manager.master_frame, + msg=widgets_strings.tooltip_howtouse, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.master_frame, + msg=widgets_strings.tooltip_howtouse, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.title_frame, + msg=widgets_strings.tooltip_bins_title_msg, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.header_label, - msg=widgets_strings.tooltip_bins_title_msg, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") + msg=widgets_strings.tooltip_bins_title_msg, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.xlite_manager.frame_manager.xlite_label, - msg=widgets_strings.tooltip_xlite_label_msg, - delay=1.0, border_width=2, follow=True, bg_color=tooltip_bg_color) + msg=widgets_strings.tooltip_xlite_label_msg, delay=1.0, border_width=2, + follow=True, bg_color=tooltip_bg_color) self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blocknet_label, - msg=widgets_strings.tooltip_blocknet_core_label_msg, delay=1, - follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") + msg=widgets_strings.tooltip_blocknet_core_label_msg, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blockdx_label, msg=widgets_strings.tooltip_blockdx_label_msg, delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.xlite_label, - msg=widgets_strings.tooltip_xlite_label_msg, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.install_delete_blocknet_button, - msg='', delay=1, width=1, follow=True, bg_color=tooltip_bg_color, - border_width=2, justify="left") - self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.install_delete_blockdx_button, - msg=global_variables.blockdx_release_url, + msg=widgets_strings.tooltip_xlite_label_msg, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.install_delete_blocknet_button, msg='', delay=1, width=1, follow=True, bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.install_delete_blockdx_button, + msg=global_variables.blockdx_release_url, delay=1, width=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.install_delete_xlite_button, - msg=global_variables.xlite_release_url, - delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, - justify="left") - self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blocknet_start_close_button, - msg='', + msg=global_variables.xlite_release_url, delay=1, follow=True, + bg_color=tooltip_bg_color, border_width=2, justify="left") + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blocknet_start_close_button, msg='', delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, justify="left") - self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blockdx_start_close_button, - msg='', + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.blockdx_start_close_button, msg='', delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, justify="left") - self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.xlite_toggle_execution_button, - msg='', + self.tooltip_manager.register_tooltip(self.binary_manager.frame_manager.xlite_toggle_execution_button, msg='', delay=1, follow=True, bg_color=tooltip_bg_color, border_width=2, justify="left") self.tooltip_manager.register_tooltip(self.blocknet_manager.frame_manager.label, - msg=widgets_strings.tooltip_blocknet_core_label_msg, - delay=1.0, border_width=2, follow=True, bg_color=tooltip_bg_color) + msg=widgets_strings.tooltip_blocknet_core_label_msg, delay=1.0, + border_width=2, follow=True, bg_color=tooltip_bg_color) self.tooltip_manager.register_tooltip(self.blockdx_manager.frame_manager.label, - msg=widgets_strings.tooltip_blockdx_label_msg, - delay=1.0, border_width=2, follow=True, bg_color=tooltip_bg_color) - - def init_grid(self): - x = 0 - y = 0 - padx_main_frame = 10 - pady_main_frame = 5 + msg=widgets_strings.tooltip_blockdx_label_msg, delay=1.0, border_width=2, + follow=True, bg_color=tooltip_bg_color) + + def init_grid(self) -> None: + """Initialize the grid layout for GUI components.""" + x: int = 0 + y: int = 0 + padx_main_frame: int = 10 + pady_main_frame: int = 5 self.grid_frames(x, y, padx_main_frame, pady_main_frame) self.binary_manager.frame_manager.grid_widgets(x, y) self.blocknet_manager.frame_manager.grid_widgets(x, y) self.blockdx_manager.frame_manager.grid_widgets(x, y) self.xlite_manager.frame_manager.grid_widgets(x, y) - def grid_frames(self, x, y, padx_main_frame, pady_main_frame): - self.bins_download_frame.grid(row=x, column=y, padx=padx_main_frame, pady=pady_main_frame, - sticky=MAIN_FRAMES_STICKY) + def grid_frames(self, x: int, y: int, padx_main_frame: int, pady_main_frame: int) -> None: + """Grid layout for frames in the GUI.""" + self.binary_manager.frame_manager.master_frame.grid(row=x, column=y, padx=padx_main_frame, pady=pady_main_frame, + sticky=MAIN_FRAMES_STICKY) # bin panel have 5 buttons per row - self.bins_title_frame.grid(row=0, column=0, columnspan=5, padx=5, pady=5, sticky=TITLE_FRAMES_STICKY) - - self.blocknet_core_frame.grid(row=x + 1, column=y, padx=padx_main_frame, pady=pady_main_frame, - sticky=MAIN_FRAMES_STICKY) - self.blocknet_title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, sticky=TITLE_FRAMES_STICKY) - - self.blockdx_frame.grid(row=x + 2, column=y, padx=padx_main_frame, pady=pady_main_frame, - sticky=MAIN_FRAMES_STICKY) - self.blockdx_title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, sticky=TITLE_FRAMES_STICKY) - - self.xlite_frame.grid(row=x + 3, column=y, padx=padx_main_frame, pady=pady_main_frame, - sticky=MAIN_FRAMES_STICKY) - self.xlite_title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, sticky=TITLE_FRAMES_STICKY) - - def handle_signal(self, signum, frame): - print("Signal {} received.".format(signum)) + self.binary_manager.frame_manager.title_frame.grid(row=0, column=0, columnspan=5, padx=5, pady=5, + sticky=TITLE_FRAMES_STICKY) + + self.blocknet_manager.frame_manager.master_frame.grid(row=x + 1, column=y, padx=padx_main_frame, + pady=pady_main_frame, + sticky=MAIN_FRAMES_STICKY) + self.blocknet_manager.frame_manager.title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, + sticky=TITLE_FRAMES_STICKY) + + self.blockdx_manager.frame_manager.master_frame.grid(row=x + 2, column=y, padx=padx_main_frame, + pady=pady_main_frame, + sticky=MAIN_FRAMES_STICKY) + self.blockdx_manager.frame_manager.title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, + sticky=TITLE_FRAMES_STICKY) + + self.xlite_manager.frame_manager.master_frame.grid(row=x + 3, column=y, padx=padx_main_frame, + pady=pady_main_frame, + sticky=MAIN_FRAMES_STICKY) + self.xlite_manager.frame_manager.title_frame.grid(row=0, column=0, columnspan=2, padx=5, pady=5, + sticky=TITLE_FRAMES_STICKY) + + def handle_signal(self, signum: int, frame) -> None: + """Handle signals like SIGINT and SIGTERM.""" + print(f"Signal {signum} received.") self.on_close() - def on_close(self): + def on_close(self) -> None: + """Handle application close event.""" logging.info("Closing application...") utils.terminate_all_threads() logging.info("Threads terminated.") os._exit(0) - def adjust_theme(self): + def adjust_theme(self) -> None: + """Adjust the theme of the application based on the configuration.""" if self.cfg and 'theme' in self.cfg: - actual = ctk.get_appearance_mode() + actual: str = ctk.get_appearance_mode() if self.cfg['theme'] != actual: if actual == "Dark": - new_theme = "Light" + new_theme: str = "Light" else: - new_theme = "Dark" + new_theme: str = "Dark" ctk.set_appearance_mode(new_theme) - def switch_theme_command(self): - actual = ctk.get_appearance_mode() + def switch_theme_command(self) -> None: + """Switch the application theme to the opposite of the current theme.""" + actual: str = ctk.get_appearance_mode() if actual == "Dark": - new_theme = "Light" + new_theme: str = "Light" else: - new_theme = "Dark" + new_theme: str = "Dark" ctk.set_appearance_mode(new_theme) utils.save_cfg_json("theme", new_theme) - def check_processes(self): - blocknet_bin = global_variables.blocknet_bin - blockdx_bin = global_variables.blockdx_bin[-1] if global_variables.system == "Darwin" \ - else global_variables.blockdx_bin - xlite_bin = global_variables.xlite_bin[-1] if global_variables.system == "Darwin" \ - else global_variables.xlite_bin - xlite_daemon_bin = global_variables.xlite_daemon_bin - blocknet_processes = [] - blockdx_processes = [] - xlite_processes = [] - xlite_daemon_processes = [] - - try: - # Get all processes - for proc in process_iter(['pid', 'name']): - # Check if any process matches the Blocknet process name - if blocknet_bin == proc.info['name']: - blocknet_processes.append(proc.info['pid']) - # Check if any process matches the Block DX process name - if blockdx_bin == proc.info['name']: - blockdx_processes.append(proc.info['pid']) - # Check if any process matches the Xlite process name - if xlite_bin == proc.info['name']: - xlite_processes.append(proc.info['pid']) - # Check if any process matches the Xlite-daemon process name - if xlite_daemon_bin == proc.info['name']: - xlite_daemon_processes.append(proc.info['pid']) - except Exception as e: - logging.warning(f"Error while checking processes: {e}") - + def check_processes(self) -> None: + blocknet_processes, blockdx_processes, xlite_processes, xlite_daemon_processes = utils.processes_check() # Update Blocknet process status and store the PIDs self.blocknet_manager.blocknet_process_running = bool(blocknet_processes) self.blocknet_manager.utility.blocknet_pids = blocknet_processes @@ -332,10 +284,11 @@ def check_processes(self): self.xlite_manager.daemon_process_running = bool(xlite_daemon_processes) self.xlite_manager.utility.xlite_daemon_pids = xlite_daemon_processes - self.after(2000, self.check_processes) + self.after(5000, func=self.check_processes) -def run_gui(): +def run_gui() -> None: + """Run the Blocknet AIO GUI application.""" app = Blocknet_AIO_GUI() # try: app.init_setup() diff --git a/gui/binary_frame_manager.py b/gui/binary_frame_manager.py index c3fcc20..b38d7e6 100644 --- a/gui/binary_frame_manager.py +++ b/gui/binary_frame_manager.py @@ -7,23 +7,21 @@ class BinaryFrameManager: - def __init__(self, parent, master_frame, title_frame): + def __init__(self, parent): self.root_gui = parent.root_gui self.parent = parent - self.master_frame = master_frame - self.title_frame = title_frame - self.xbridge_bot_manager = XBridgeBotManager() + self.master_frame = ctk.CTkFrame(master=self.root_gui) + self.title_frame = ctk.CTkFrame(self.master_frame) + self.xbridge_bot_manager = XBridgeBotManager() self.header_label = ctk.CTkLabel(self.title_frame, text="Binaries Control panel:", anchor=HEADER_FRAMES_STICKY, width=BINS_FRAME_WIDTH) self.title_frame.columnconfigure(1, weight=1) - self.found_label = ctk.CTkLabel(self.title_frame, text="Found:", anchor='s') - self.button_switch_theme = ctk.CTkButton(self.title_frame, image=self.root_gui.theme_img, command=self.root_gui.switch_theme_command, @@ -167,6 +165,14 @@ def toggle_bots_execution_command(self): utilities.utils.disable_button(self.install_delete_bots_button, self.root_gui.install_greyed_img) utilities.utils.disable_button(self.bots_toggle_execution_button, self.root_gui.start_greyed_img) self.xbridge_bot_manager.toggle_execution(branch) + if not self.xbridge_bot_manager.repo_management.venv: + self.run_after_setup() + + def run_after_setup(self): + if self.xbridge_bot_manager.repo_management.venv: + self.xbridge_bot_manager.toggle_execution() + else: + self.root_gui.after(1000, self.run_after_setup) def grid_widgets(self, x, y): # bin diff --git a/gui/binary_manager.py b/gui/binary_manager.py index dd36d94..a080acb 100644 --- a/gui/binary_manager.py +++ b/gui/binary_manager.py @@ -1,18 +1,71 @@ import logging import os import shutil +import time from threading import Thread +from watchdog.events import FileSystemEvent, FileSystemEventHandler +from watchdog.observers import Observer + import widgets_strings from gui.binary_frame_manager import BinaryFrameManager from utilities import utils, global_variables +class BinaryFileHandler(FileSystemEventHandler): + """ + Handles file modification events with rate limiting for binary updates. + """ + + def __init__(self, binary_manager: 'BinaryManager'): + """ + Initializes the handler. + :param binary_manager: The manager responsible for binary updates. + """ + super().__init__() + self.binary_manager: 'BinaryManager' = binary_manager + self.max_delay: float = 5 # seconds + self.last_run: float = 0 + self.scheduled: bool = False + + def on_modified(self, event: 'FileSystemEvent') -> None: + """ + Called when a file is modified. Executes binary check/update with rate limiting. + """ + # logging.info("File modified detected: %s", event.src_path) + + if self.scheduled: + # logging.debug("Update already scheduled, skipping immediate execution.") + return + + time_since_last = time.time() - self.last_run + # logging.debug("Time since last run: %.2f seconds", time_since_last) + + if time_since_last >= self.max_delay: + # Execute immediately + # logging.info("Executing check_and_update_aio_folder immediately.") + self.binary_manager.check_and_update_aio_folder() + self.last_run = time.time() + else: + # Schedule for later + delay_ms = int((self.max_delay - time_since_last) * 1000) + # logging.info("Scheduling check_and_update_aio_folder in %d ms.", delay_ms) + self.scheduled = True + self.binary_manager.root_gui.after(delay_ms, self._execute_scheduled) + + def _execute_scheduled(self) -> None: + """ + Executes the scheduled update and resets the schedule flag. + """ + # logging.info("Executing scheduled check_and_update_aio_folder.") + self.binary_manager.check_and_update_aio_folder() + self.last_run = time.time() + self.scheduled = False + + class BinaryManager: - def __init__(self, root_gui, master_frame, title_frame): + def __init__(self, root_gui): self.root_gui = root_gui - self.title_frame = title_frame - self.master_frame = master_frame self.frame_manager = None self.disable_start_blocknet_button = False @@ -23,12 +76,17 @@ def __init__(self, root_gui, master_frame, title_frame): self.download_blockdx_thread = None self.download_xlite_thread = None + self.observer = Observer() + self.handler = BinaryFileHandler(self) + self.observer.schedule(self.handler, global_variables.aio_folder, recursive=False) + self.observer.start() + self.tooltip_manager = self.root_gui.tooltip_manager async def setup(self): - self.frame_manager = BinaryFrameManager(self, self.master_frame, self.title_frame) + self.frame_manager = BinaryFrameManager(self) - self.root_gui.after(0, self.bins_check_aio_folder) + self.root_gui.after(0, self.check_and_update_aio_folder) self.root_gui.after(0, self.update_blocknet_buttons) self.root_gui.after(0, self.update_blockdx_buttons) self.root_gui.after(0, self.update_xlite_buttons) @@ -71,11 +129,11 @@ def start_or_close_blockdx(self): ) def start_or_close_xlite(self): - if not self.root_gui.xlite_manager.process_running: - if self.root_gui.stored_password: - env_vars = [{"CC_WALLET_PASS": self.root_gui.stored_password}, {"CC_WALLET_AUTOLOGIN": 'true'}] - else: - env_vars = [] + if not self.root_gui.xlite_manager.process_running and self.root_gui.stored_password: + env_vars = [{"CC_WALLET_PASS": self.root_gui.stored_password}, {"CC_WALLET_AUTOLOGIN": 'true'}] + else: + env_vars = [] + self._start_or_close_binary( process_running=self.root_gui.xlite_manager.process_running, stop_func=self.root_gui.xlite_manager.utility.close_xlite, @@ -167,8 +225,8 @@ def delete_xlite_command(self): logging.info(f"deleting {item_path}") shutil.rmtree(item_path) - def bins_check_aio_folder(self): - # logging.info("bins_check_aio_folder") + def check_and_update_aio_folder(self): + # logging.info("check_and_update_aio_folder") # Get system information and versions is_darwin = global_variables.system == "Darwin" @@ -217,16 +275,13 @@ def bins_check_aio_folder(self): # logging.info(app_info) app_info["boolvar"].set(app_info["found"]) - # Schedule next check - self.root_gui.after(5000, self.bins_check_aio_folder) - def _prune_version(self, version): """Remove 'v' prefix from version string.""" return version[0].replace('v', '') def _log_incorrect_target(self, target): """Log incorrect version found.""" - # logging.info(f"incorrect version: {target}") + logging.info(f"incorrect version: {target}") # shutil.rmtree(target) if os.path.isdir(target) else os.remove(target) return @@ -245,69 +300,6 @@ def _check_app_version(self, app_info, item, full_path): else: self._log_incorrect_target(full_path) - def bins_check_aio_folder_original(self): - blocknet_pruned_version = self.root_gui.blocknet_manager.version[0].replace('v', '') - blockdx_pruned_version = self.root_gui.blockdx_manager.version[0].replace('v', '') - xlite_pruned_version = self.root_gui.xlite_manager.version[0].replace('v', '') - - blocknet_present = False - blockdx_present = False - xlite_present = False - - for item in os.listdir(global_variables.aio_folder): - if global_variables.system == "Darwin": - blockdx_filename = os.path.basename(global_variables.blockdx_release_url) - xlite_filename = os.path.basename(global_variables.xlite_release_url) - item_path = os.path.join(global_variables.aio_folder, item) - if os.path.isdir(item_path): - if 'blocknet-' in item: - if blocknet_pruned_version in item: - blocknet_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - shutil.rmtree(item_path) - elif os.path.isfile(item_path): - if 'BLOCK-DX-' in item: - if blockdx_filename in item: - blockdx_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - os.remove(item_path) - elif 'XLite-' in item: - if xlite_filename in item: - xlite_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - os.remove(item_path) - else: - item_path = os.path.join(global_variables.aio_folder, item) - if os.path.isdir(item_path): - # if a wrong version is found, delete it. - if 'blocknet-' in item: - if blocknet_pruned_version in item: - blocknet_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - shutil.rmtree(item_path) - elif 'BLOCK-DX-' in item: - if blockdx_pruned_version in item: - blockdx_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - shutil.rmtree(item_path) - elif 'XLite-' in item: - if xlite_pruned_version in item: - xlite_present = True - else: - logging.info(f"deleting outdated version: {item_path}") - shutil.rmtree(item_path) - - self.root_gui.binary_manager.frame_manager.blocknet_installed_boolvar.set(blocknet_present) - self.root_gui.binary_manager.frame_manager.blockdx_installed_boolvar.set(blockdx_present) - self.root_gui.binary_manager.frame_manager.xlite_installed_boolvar.set(xlite_present) - - self.root_gui.after(2000, self.bins_check_aio_folder_original) - def update_blocknet_buttons(self): # BLOCKNET self.update_blocknet_start_close_button() diff --git a/gui/blockdx_frame_manager.py b/gui/blockdx_frame_manager.py index 5bbad0a..11618cd 100644 --- a/gui/blockdx_frame_manager.py +++ b/gui/blockdx_frame_manager.py @@ -10,20 +10,18 @@ class BlockDxFrameManager: - def __init__(self, parent, master_frame, title_frame): + def __init__(self, parent): self.root_gui = parent.root_gui self.parent = parent - self.master_frame = master_frame - self.title_frame = title_frame - # Label for Block-dx frame + self.master_frame = ctk.CTkFrame(master=self.root_gui) + self.title_frame = ctk.CTkFrame(self.master_frame) + self.label = ctk.CTkLabel(self.title_frame, text=widgets_strings.blockdx_frame_title_string, anchor=HEADER_FRAMES_STICKY, width=BLOCKDX_FRAME_WIDTH) - # Checkboxes - # width_mod = 35 self.process_status_checkbox_state = ctk.BooleanVar() self.process_status_checkbox_string_var = ctk.StringVar(value='') self.process_status_checkbox = ctkCheckBoxMod.CTkCheckBox(self.master_frame, diff --git a/gui/blockdx_manager.py b/gui/blockdx_manager.py index 8a30405..2f4483d 100644 --- a/gui/blockdx_manager.py +++ b/gui/blockdx_manager.py @@ -7,18 +7,16 @@ class BlockDXManager: - def __init__(self, root_gui, master_frame, title_frame): + def __init__(self, root_gui): self.frame_manager = None self.root_gui = root_gui - self.title_frame = title_frame - self.master_frame = master_frame self.utility = BlockdxUtility() self.version = [global_variables.blockdx_release_url.split('/')[7]] self.process_running = False self.is_config_sync = None async def setup(self): - self.frame_manager = BlockDxFrameManager(self, self.master_frame, self.title_frame) + self.frame_manager = BlockDxFrameManager(self) self.root_gui.after(0, self.update_status_blockdx) def blockdx_check_config(self): diff --git a/gui/blocknet_frame_manager.py b/gui/blocknet_frame_manager.py index 4f2616c..7549109 100644 --- a/gui/blocknet_frame_manager.py +++ b/gui/blocknet_frame_manager.py @@ -11,18 +11,16 @@ class BlocknetCoreFrameManager: - def __init__(self, parent, master_frame, title_frame): + def __init__(self, parent): self.root_gui = parent.root_gui self.parent = parent - self.master_frame = master_frame - self.title_frame = title_frame + self.master_frame = ctk.CTkFrame(master=self.root_gui) + self.title_frame = ctk.CTkFrame(self.master_frame) - # Create all Blocknet Core widgets here self.label = ctk.CTkLabel(self.title_frame, text=widgets_strings.blocknet_frame_title_string, anchor=HEADER_FRAMES_STICKY) # , width=FRAME_WIDTH) - # Label for Data Path self.data_path_label = ctk.CTkLabel(self.title_frame, text="Data Path: ") self.data_path_entry_string_var = ctk.StringVar(value=self.parent.utility.data_folder) diff --git a/gui/blocknet_manager.py b/gui/blocknet_manager.py index c4642c9..a4e12f5 100644 --- a/gui/blocknet_manager.py +++ b/gui/blocknet_manager.py @@ -4,11 +4,9 @@ class BlocknetManager: - def __init__(self, root_gui, master_frame, title_frame): + def __init__(self, root_gui): self.frame_manager = None self.root_gui = root_gui - self.title_frame = title_frame - self.master_frame = master_frame self.version = [global_variables.blocknet_release_url.split('/')[7]] self.blocknet_process_running = False @@ -17,7 +15,7 @@ def __init__(self, root_gui, master_frame, title_frame): self.utility = BlocknetUtility(custom_path=self.root_gui.custom_path) async def setup(self): - self.frame_manager = BlocknetCoreFrameManager(self, self.master_frame, self.title_frame) + self.frame_manager = BlocknetCoreFrameManager(self) self.root_gui.after(0, self.update_status_blocknet_core) diff --git a/gui/xbridge_bot_manager.py b/gui/xbridge_bot_manager.py index fa460f9..1f71aea 100644 --- a/gui/xbridge_bot_manager.py +++ b/gui/xbridge_bot_manager.py @@ -3,15 +3,17 @@ import subprocess import threading import time +from datetime import datetime from utilities.git_repo_management import GitRepoManagement from utilities.global_variables import aio_folder class XBridgeBotManager: - def __init__(self, repo_url: str = "https://github.com/tryiou/xbridge_trading_bots", current_branch: str = "main"): - - self.repo_url = repo_url + def __init__(self, current_branch: str = "main"): + self.author = "tryiou" + self.repo_name = "xbridge_trading_bots" + self.repo_url = "https://github.com/" + self.author + "/" + self.repo_name self.target_dir = os.path.join(aio_folder, "xbridge_trading_bots") self.started = False self.current_branch = current_branch @@ -27,9 +29,10 @@ def get_available_branches(self) -> list: """Return list of available branches from remote repo""" try: if not self.repo_management: - self.repo_management = GitRepoManagement(self.repo_url, self.target_dir, branch=self.current_branch, - workdir=aio_folder) - return self.repo_management.get_remote_branches() + logging.error(f"GitRepoManagement not initialized ?") + return ["main"] + else: + return self.repo_management.get_remote_branches() except Exception as e: logging.error(f"Error fetching branches: {e}") return ["main"] @@ -69,11 +72,41 @@ def _do_install_update(self, branch: str) -> None: self.current_branch = branch logging.info(f"Successfully updated to branch {branch}") except Exception as e: - logging.error(f"Failed to update: {str(e)}", exc_info=True) + error_msg = str(e) + if "conflict prevents checkout" in error_msg or "conflicts prevent checkout" in error_msg: + self.handle_config_folder_rename() + else: + logging.error(f"Failed to update: {error_msg}") logging.debug(f"Repository URL: {self.repo_url}") logging.debug(f"Target directory: {self.target_dir}") logging.debug(f"Branch: {branch}") + def handle_config_folder_rename(self): + logging.info("Successfully updated after resolving config conflict") + config_path = os.path.join(self.target_dir, "config") + + if not os.path.exists(config_path): + logging.warning("Config folder not found, cannot rename") + return + + # Generate a unique backup folder name with timestamp + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + config_bak_path = os.path.join(self.target_dir, f"config_bak_{timestamp}") + + try: + os.rename(config_path, config_bak_path) + logging.info(f"Renamed config folder to {config_bak_path} to resolve conflict") + except Exception as e: + logging.error(f"Failed to rename config folder: {e}") + return + + # Retry setup once + try: + self.repo_management.setup() + logging.info("Successfully updated after resolving config conflict") + except Exception as retry_e: + logging.error(f"Failed to update even after config rename: {retry_e}") + def delete_local_repo(self) -> None: """Delete local repository""" if self.repo_exists(): @@ -101,23 +134,23 @@ def toggle_execution(self, branch=None) -> None: if not self.repo_exists() or not self.repo_management.venv or branch != self.current_branch: self.install_or_update(branch) - if not self.repo_management.venv: - logging.error("failed to set repo") - return - - if not self.started: - self._start_execution() - self.started = True + if self.repo_management.venv: + if not self.process or self.process and self.process.poll() is not None: + self._start_execution() + self.started = True + else: + self._stop_execution() + self.started = False else: - self._stop_execution() - self.started = False + logging.info("Venv setup in progress") def _start_execution(self) -> None: """Start script execution in background""" while self.thread and self.thread.is_alive(): # wait for update tread to close completely. + logging.info("waiting thread end") time.sleep(0.5) - + # self.thread = threading.Thread(target=self._run_script) self.thread.start() diff --git a/gui/xlite_frame_manager.py b/gui/xlite_frame_manager.py index 59e5a58..92753b8 100644 --- a/gui/xlite_frame_manager.py +++ b/gui/xlite_frame_manager.py @@ -12,10 +12,12 @@ class XliteFrameManager: - def __init__(self, parent, master_frame, title_frame): + def __init__(self, parent): + self.root_gui = parent.root_gui self.parent = parent - self.master_frame = master_frame - self.title_frame = title_frame + self.master_frame = ctk.CTkFrame(master=self.root_gui) + self.title_frame = ctk.CTkFrame(self.master_frame) + self.xlite_label = ctk.CTkLabel(self.title_frame, text=widgets_strings.xlite_frame_title_string, anchor=HEADER_FRAMES_STICKY, @@ -92,7 +94,7 @@ def xlite_store_password_button_mouse_click(self, event=None): # Prevent the right-click event from propagating further utils.remove_cfg_json_key("salt") utils.remove_cfg_json_key("xl_pass") - self.parent.stored_password = None + self.root_gui.stored_password = None # Delete CC_WALLET_PASS variable if "CC_WALLET_PASS" in os.environ: os.environ.pop("CC_WALLET_PASS") @@ -118,7 +120,7 @@ def xlite_store_password_button_mouse_click(self, event=None): utils.save_cfg_json(key="salt", data=encryption_key.decode()) utils.save_cfg_json(key="xl_pass", data=salted_pass) # Store the password in a variable - self.parent.stored_password = password + self.root_gui.stored_password = password else: logging.info("No password entered.") # Perform actions for left-click (if needed) @@ -143,7 +145,7 @@ def update_xlite_process_status_checkbox(self): def update_xlite_store_password_button(self): # xlite_store_password_button - var = widgets_strings.xlite_stored_password_string if self.parent.stored_password else widgets_strings.xlite_store_password_string + var = widgets_strings.xlite_stored_password_string if self.root_gui.stored_password else widgets_strings.xlite_store_password_string self.store_password_button_string_var.set(var) def update_xlite_daemon_process_status(self): diff --git a/gui/xlite_manager.py b/gui/xlite_manager.py index 6def602..9e66228 100644 --- a/gui/xlite_manager.py +++ b/gui/xlite_manager.py @@ -6,10 +6,8 @@ class XliteManager: - def __init__(self, root_gui, master_frame, title_frame): + def __init__(self, root_gui): self.root_gui = root_gui - self.title_frame = title_frame - self.master_frame = master_frame self.frame_manager = None self.utility = XliteUtility() @@ -17,10 +15,9 @@ def __init__(self, root_gui, master_frame, title_frame): self.version = [global_variables.xlite_release_url.split('/')[7]] self.process_running = False self.daemon_process_running = False - self.stored_password = None async def setup(self): - self.frame_manager = XliteFrameManager(self, self.master_frame, self.title_frame) + self.frame_manager = XliteFrameManager(self) self.root_gui.after(0, self.update_status_xlite) def refresh_xlite_confs(self): diff --git a/img/blocknet_aio_monitor.png b/img/blocknet_aio_monitor.png index 7f5f057..7f42c58 100644 Binary files a/img/blocknet_aio_monitor.png and b/img/blocknet_aio_monitor.png differ diff --git a/requirements.txt b/requirements.txt index d490863..aca68b1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,5 @@ cryptography~=44.0.1 customtkinter~=5.2.2 CTkToolTip~=0.8 pillow -pygit2==1.18.0 \ No newline at end of file +pygit2==1.18.0 +watchdog \ No newline at end of file diff --git a/utilities/blockdx_util.py b/utilities/blockdx_util.py index 98a747b..ca4b3b3 100644 --- a/utilities/blockdx_util.py +++ b/utilities/blockdx_util.py @@ -3,20 +3,17 @@ import logging import os import subprocess -import tarfile import time -import zipfile - -import psutil -import requests from utilities import global_variables +from utilities.helper_util import UtilityHelper logging.basicConfig(level=logging.DEBUG) class BlockdxUtility: def __init__(self): + self.helper = UtilityHelper() if global_variables.system == "Darwin": self.dmg_mount_path = f"/Volumes/{global_variables.blockdx_volume_name}" self.blockdx_exe = os.path.join(global_variables.aio_folder, os.path.basename(global_variables.blockdx_url)) @@ -98,14 +95,7 @@ def compare_and_update_local_conf(self, xbridgeconfpath, rpc_user, rpc_password) logging.info("No changes detected in Blockdx config.") def unmount_dmg(self): - if os.path.ismount(self.dmg_mount_path): - try: - subprocess.run(["hdiutil", "detach", self.dmg_mount_path], check=True) - logging.info("DMG unmounted successfully.") - except subprocess.CalledProcessError as e: - logging.error(f"Error: Failed to unmount DMG: {e}") - else: - logging.error("Error: DMG is not mounted.") + self.helper.handle_dmg(None, self.dmg_mount_path, "unmount") def start_blockdx(self): if not os.path.exists(self.blockdx_exe): @@ -118,13 +108,7 @@ def start_blockdx(self): # Start the BLOCK-DX process using subprocess if global_variables.system == "Darwin": # mac mod - - # Check if the volume is already mounted - if not os.path.ismount(self.dmg_mount_path): - # Mount the DMG file - os.system(f'hdiutil attach "{self.blockdx_exe}"') - else: - logging.info("Volume is already mounted.") + self.helper.handle_dmg(self.blockdx_exe, self.dmg_mount_path, "mount") full_path = os.path.join(self.dmg_mount_path, *global_variables.conf_data.blockdx_bin_name[global_variables.system]) logging.info( @@ -182,25 +166,7 @@ def kill_blockdx(self): logging.error(f"Error: {e}") def close_blockdx_pids(self): - # Close the blockdx processes using their PIDs - for pid in self.blockdx_pids: - try: - # Get the process object corresponding to the PID - proc = psutil.Process(pid) - proc.terminate() - logging.info(f"Initiated termination of blockdx process with PID {pid}.") - proc.wait(timeout=60) # Wait for the process to terminate with a timeout of 60 seconds - logging.info(f"blockdx process with PID {pid} has been terminated.") - except psutil.NoSuchProcess: - logging.warning(f"blockdx process with PID {pid} not found.") - except psutil.TimeoutExpired: - logging.warning(f"Force terminating blockdx process with PID {pid}.") - if proc: - proc.kill() - proc.wait() - logging.info(f"blockdx process with PID {pid} has been force terminated.") - except Exception as e: - logging.error(f"Error: {e}") + self.helper.terminate_processes(self.blockdx_pids, "BlockDX") def download_blockdx_bin(self): self.downloading_bin = True @@ -209,56 +175,14 @@ def download_blockdx_bin(self): if url is None: raise ValueError(f"Unsupported OS or architecture {global_variables.system} {global_variables.machine}") - # Set timeout values in seconds - connection_timeout = 10 - read_timeout = 30 - response = requests.get(url, stream=True, timeout=(connection_timeout, read_timeout)) - response.raise_for_status() # Raise an exception for 4xx and 5xx status codes - if response.status_code == 200: - file_name = os.path.basename(url) - tmp_file_path = os.path.join(global_variables.aio_folder, "tmp_dx_bin") - try: - remote_file_size = int(response.headers.get('Content-Length', 0)) - logging.info(f"Downloading {url} to {tmp_file_path}, remote size: {int(remote_file_size / 1024)} kb") - bytes_downloaded = 0 - total = remote_file_size - with open(tmp_file_path, "wb") as f: - for chunk in response.iter_content(chunk_size=8192): # Iterate over response content in chunks - if chunk: # Filter out keep-alive new chunks - f.write(chunk) - bytes_downloaded += len(chunk) - self.binary_percent_download = (bytes_downloaded / total) * 100 - except requests.exceptions.RequestException as e: - logging.error(f"Error occurred during download: {str(e)}") - - self.binary_percent_download = None - - if os.path.getsize(tmp_file_path) != remote_file_size: - os.remove(tmp_file_path) - raise ValueError( - f"Downloaded {os.path.basename(url)} size doesn't match the expected size. Deleting it") + tmp_path = os.path.join(global_variables.aio_folder, "tmp_dx_bin") + final_path = self.blockdx_exe # For DMG + extract_to = global_variables.aio_folder # For zip/tar.gz - logging.info(f"{os.path.basename(url)} downloaded successfully.") - - # Extract the archive - if url.endswith(".zip"): - with zipfile.ZipFile(tmp_file_path, "r") as zip_ref: - local_path = os.path.join(global_variables.aio_folder, - global_variables.conf_data.blockdx_bin_path[global_variables.system]) - zip_ref.extractall(local_path) - logging.info("Zip file extracted successfully.") - os.remove(tmp_file_path) - elif url.endswith(".tar.gz"): - with tarfile.open(tmp_file_path, "r:gz") as tar: - tar.extractall(global_variables.aio_folder) - logging.info("Tar.gz file extracted successfully.") - os.remove(tmp_file_path) - elif url.endswith(".dmg"): - file_path = os.path.join(global_variables.aio_folder, file_name) - os.rename(tmp_file_path, file_path) - logging.info("DMG file saved successfully.") - else: - logging.error("Failed to download the Blockdx binary.") + self.helper.download_file( + url, tmp_path, final_path, extract_to, + global_variables.system, "binary_percent_download", self + ) self.downloading_bin = False @@ -268,11 +192,3 @@ def get_blockdx_data_folder(): return os.path.expandvars(os.path.expanduser(path)) else: raise ValueError("Unsupported system") - -# async def main(): -# blockdx_utility = BlockdxUtility() -# # blockdx_utility.compare_and_update_local_conf() -# -# -# if __name__ == "__main__": -# asyncio.run(main()) diff --git a/utilities/blocknet_util.py b/utilities/blocknet_util.py index 968ee34..9e314f7 100644 --- a/utilities/blocknet_util.py +++ b/utilities/blocknet_util.py @@ -5,21 +5,15 @@ import shutil import string import subprocess -import tarfile import threading import time import zipfile -from subprocess import check_output -import psutil import requests -# from utilities.conf_data import (remote_blocknet_conf_url, blocknet_default_paths, base_xbridge_conf, blocknet_bin_path, -# blocknet_bootstrap_url, nodes_to_add, remote_xbridge_conf_url, remote_manifest_url, -# remote_blockchain_configuration_repo) from utilities import global_variables +from utilities.helper_util import UtilityHelper -# Configure logging logging.basicConfig(level=logging.DEBUG) # Disable log entries from the urllib3 module (used by requests) @@ -44,22 +38,16 @@ def send_rpc_request(self, method=None, params=None): "id": 1, } try: - # logging.debug( - # f"Sending RPC request to URL: {url}, Method: {data['method']}, Params: {data['params']}, Auth: {auth}") response = requests.post(url, json=data, headers=headers, auth=auth) - # Check status code explicitly if response.status_code != 200: - # logging.error(f"Error sending RPC request: HTTP status code {response.status_code}") return None json_answer = response.json() - # logging.debug(f"RPC request successful. Response: {json}") if 'result' in json_answer: return json_answer['result'] else: logging.error(f"No result in json: {json_answer}") except requests.RequestException as e: - # logging.error(f"Error sending RPC request: {e}") return None except Exception as ex: logging.exception(f"An unexpected error occurred while sending RPC request: {ex}") @@ -68,9 +56,10 @@ def send_rpc_request(self, method=None, params=None): class BlocknetUtility: def __init__(self, custom_path=None): - self.blocknet_exe = global_variables.os.path.join(global_variables.aio_folder, - *global_variables.conf_data.blocknet_bin_path, - global_variables.blocknet_bin) + self.helper = UtilityHelper() + self.blocknet_exe = os.path.join(global_variables.aio_folder, + *global_variables.conf_data.blocknet_bin_path, + global_variables.blocknet_bin) self.binary_percent_download = None self.parsed_wallet_confs = {} self.parsed_xbridge_confs = {} @@ -108,26 +97,22 @@ def check_blocknet_rpc(self): valid = True self.valid_rpc = valid - # logging.info(result) time.sleep(2) def init_blocknet_rpc(self): - # Retrieve RPC user, password, and port from blocknet_conf_local with error handling if 'global' in self.blocknet_conf_local: global_conf = self.blocknet_conf_local['global'] rpc_user = global_conf.get('rpcuser') rpc_password = global_conf.get('rpcpassword') - rpc_port = int(global_conf.get('rpcport', 0)) # Assuming default port is 0, change as per requirement + rpc_port = int(global_conf.get('rpcport', 0)) else: rpc_user = None rpc_password = None rpc_port = 0 - # Initialize BlocknetRPCClient if RPC user, password, and port are available if rpc_user is not None and rpc_password is not None and rpc_port != 0: self.blocknet_rpc = BlocknetRPCClient(rpc_user, rpc_password, rpc_port) else: - # Handle the case when RPC user, password, or port is missing logging.error("RPC user, password, or port not found in the configuration.") self.blocknet_rpc = None @@ -137,7 +122,6 @@ def start_blocknet(self): logging.info(f"Blocknet executable not found at {self.blocknet_exe}. Downloading...") self.download_blocknet_bin() try: - # Start the Blocknet process using subprocess with custom data folder argument self.blocknet_process = subprocess.Popen([self.blocknet_exe, f"-datadir={self.data_folder}"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, @@ -148,12 +132,10 @@ def start_blocknet(self): logging.error(f"Error: {e}") def close_blocknet(self): - # Close the Blocknet subprocess if it exists if self.blocknet_process: try: self.blocknet_process.terminate() - # logging.info(f"Terminating Blocknet subprocess.") - self.blocknet_process.wait(timeout=60) # Wait for the process to terminate with a timeout of 60 seconds + self.blocknet_process.wait(timeout=60) logging.info(f"Closed Blocknet subprocess.") self.blocknet_process = None return @@ -169,7 +151,6 @@ def close_blocknet(self): self.close_blocknet_pids() def kill_blocknet(self): - # Kill the Blocknet subprocess if it exists if self.blocknet_process: try: self.blocknet_process.kill() @@ -180,31 +161,14 @@ def kill_blocknet(self): logging.error(f"Error: {e}") def close_blocknet_pids(self): - # Close the Blocknet processes using their PIDs - for pid in self.blocknet_pids: - try: - # Get the process object corresponding to the PID - proc = psutil.Process(pid) - proc.terminate() - logging.info(f"Initiated termination of Blocknet process with PID {pid}.") - proc.wait(timeout=60) # Wait for the process to terminate with a timeout of 60 seconds - logging.info(f"Blocknet process with PID {pid} has been terminated.") - except psutil.NoSuchProcess: - logging.warning(f"Blocknet process with PID {pid} not found.") - except psutil.TimeoutExpired: - logging.warning(f"Force terminating Blocknet process with PID {pid}.") - proc.kill() - proc.wait() - logging.info(f"Blocknet process with PID {pid} has been force terminated.") - except Exception as e: - logging.error(f"Error: {e}") + self.helper.terminate_processes(self.blocknet_pids, "Blocknet") def check_data_folder_existence(self): return os.path.exists(self.data_folder) def set_custom_data_path(self, custom_path): if not os.path.exists(custom_path): - os.makedirs(custom_path) # Recursively create the folder if it doesn't exist + os.makedirs(custom_path) logging.info(f"Custom data path created: {custom_path}") self.data_folder = custom_path logging.debug(f"Custom data path set: {custom_path}") @@ -241,9 +205,6 @@ def save_xbridge_conf(self): def check_blocknet_conf(self): self.parse_blocknet_conf() - # logging.info(f"Current remote configuration:\n{self.blocknet_conf_remote}") - # logging.info(f"Current local configuration:\n{self.blocknet_conf_local}") - old_local_json = json.dumps(self.blocknet_conf_local, sort_keys=True) if self.blocknet_conf_remote is None: @@ -274,7 +235,6 @@ def check_blocknet_conf(self): if not isinstance(addnode_value, list): addnode_value = [addnode_value] - # Add the nodes to the end of the file if not already existing for node in global_variables.conf_data.nodes_to_add: if node not in addnode_value: addnode_value.append(node) @@ -283,35 +243,24 @@ def check_blocknet_conf(self): self.blocknet_conf_local[section_name]['addnode'] = addnode_value for section, options in self.blocknet_conf_remote.items(): - # if section not in self.blocknet_conf_local: - # self.blocknet_conf_local[section] = {} - for key, value in options.items(): if key == 'rpcuser' or key == 'rpcpassword': if key not in self.blocknet_conf_local[section]: self.blocknet_conf_local[section][key] = generate_random_string(32) - # logging.info(f"Generated {key} value: {self.blocknet_conf_local[section][key]}") else: if self.blocknet_conf_local[section][key] == '': self.blocknet_conf_local[section][key] = generate_random_string(32) - # logging.info( - # f"Value for {key} is empty. Generated new value: {self.blocknet_conf_local[section][key]}") - # CHECK IF VALUE IS NOT EMPTY STRING ELSE GENERATE NEW VALUE else: if key == "rpcallowip": self.blocknet_conf_local[section][key] = "127.0.0.1" elif key not in self.blocknet_conf_local[section] or self.blocknet_conf_local[section][ key] != value: self.blocknet_conf_local[section][key] = value - # logging.debug(f"Updated {key} value: {value}") logging.info("Local blocknet.conf updated successfully.") new_local_json = json.dumps(self.blocknet_conf_local, sort_keys=True) - # logging.info(f"Old local configuration:\n{old_local_json}") - # logging.info(f"Updated local configuration:\n{new_local_json}") - # logging.info(new_local_json) if old_local_json != new_local_json: logging.info("Local blocknet.conf has been updated. Saving...") self.save_blocknet_conf() @@ -337,26 +286,18 @@ def retrieve_coin_conf(self, coin): xbridge_url = f"{global_variables.conf_data.remote_blockchain_configuration_repo}/xbridge-confs/{xbridge_conf}" wallet_conf = latest_version['wallet_conf'] wallet_conf_url = f"{global_variables.conf_data.remote_blockchain_configuration_repo}/wallet-confs/{wallet_conf}" - # download_remote_conf() parsed_xbridge_conf = retrieve_remote_conf(xbridge_url, "xbridge-confs", xbridge_conf) parsed_wallet_conf = retrieve_remote_conf(wallet_conf_url, "wallet-confs", wallet_conf) self.parsed_xbridge_confs[coin] = parsed_xbridge_conf self.parsed_wallet_confs[coin] = parsed_wallet_conf - # logging.info(parsed_xbridge_conf) - # logging.info(parsed_wallet_conf) - # Do whatever you need to do with the highest version entry else: logging.error("No entries found in the manifest. " + coin) def check_xbridge_conf(self, xlite_daemon_conf): self.parse_xbridge_conf() - # logging.info(f"Current local configuration:\n{self.xbridge_conf_local}") - # logging.info(f"Current remote configuration:\n{self.xbridge_conf_remote}") - old_local_json = json.dumps(self.xbridge_conf_local, sort_keys=True) if 'Main' not in self.xbridge_conf_local: - # We want this on 'top' of file, add it if missing self.xbridge_conf_local['Main'] = global_variables.conf_data.base_xbridge_conf if self.blocknet_xbridge_conf_remote is None: @@ -366,9 +307,7 @@ def check_xbridge_conf(self, xlite_daemon_conf): if self.xbridge_conf_local is None: logging.error("Local xbridge.conf not available.") return False - # section = 'global' if xlite_daemon_conf: - # XLITE SESSION DETECTED, USE XLITE RPC PARAMS for coin in xlite_daemon_conf: if coin == "master": continue @@ -376,7 +315,6 @@ def check_xbridge_conf(self, xlite_daemon_conf): if coin in self.parsed_xbridge_confs: if coin not in self.xbridge_conf_local: self.xbridge_conf_local[coin] = {} - # logging.warning(self.parsed_xbridge_confs[coin]) for section, options in self.parsed_xbridge_confs[coin].items(): for key, value in options.items(): if key == 'Username': @@ -388,12 +326,9 @@ def check_xbridge_conf(self, xlite_daemon_conf): else: if key not in self.xbridge_conf_local[section] or self.xbridge_conf_local[section][ key] != value: - # logging.warning(f"value: {value}") - # exit() self.xbridge_conf_local[section][key] = str(value) if not (xlite_daemon_conf and "BLOCK" in xlite_daemon_conf): - # NO XLITE SESSION DETECTED, SET XBRIDGE TO USE BLOCKNET CORE RPC for section, options in self.blocknet_xbridge_conf_remote.items(): if section not in self.xbridge_conf_local: self.xbridge_conf_local[section] = {} @@ -410,10 +345,8 @@ def check_xbridge_conf(self, xlite_daemon_conf): key] != value: self.xbridge_conf_local[section][key] = str(value) - # Prepare the string of sections (excluding 'Main') sections_string = ','.join(section for section in self.xbridge_conf_local.keys() if section != 'Main') - # Update the 'ExchangeWallets' value with the sections string if 'Main' in self.xbridge_conf_local: self.xbridge_conf_local['Main']['ExchangeWallets'] = sections_string else: @@ -452,10 +385,8 @@ def download_bootstrap(self): filename = "Blocknet.zip" local_file_path = os.path.join(global_variables.aio_folder, filename) remote_file_size = get_remote_file_size(global_variables.conf_data.blocknet_bootstrap_url) - # Check if the file already exists on disk need_to_download = True if os.path.exists(local_file_path): - # Compare the size of the local file with the remote file local_file_size = os.path.getsize(local_file_path) if local_file_size == remote_file_size: @@ -463,28 +394,22 @@ def download_bootstrap(self): need_to_download = False else: logging.info("Local bootstrap file exists but does not match the remote file. Re-downloading...") - os.remove(local_file_path) # Remove the local file and proceed with download + os.remove(local_file_path) try: if need_to_download: with open(local_file_path, 'wb') as f: - # Set timeout values in seconds - connection_timeout = 10 - read_timeout = 30 response = requests.get(global_variables.conf_data.blocknet_bootstrap_url, stream=True, - timeout=(connection_timeout, read_timeout)) + timeout=(10, 30)) response.raise_for_status() if response.status_code == 200: - try: - logging.info( - f"Downloading {global_variables.conf_data.blocknet_bootstrap_url} to {local_file_path}, remote size: {int(remote_file_size / 1024)} kb") - bytes_downloaded = 0 - for chunk in response.iter_content(chunk_size=8192): - if chunk: - f.write(chunk) - bytes_downloaded += len(chunk) - self.bootstrap_percent_download = (bytes_downloaded / remote_file_size) * 100 - except requests.exceptions.RequestException as e: - logging.error(f"Error occurred during download: {str(e)}") + logging.info( + f"Downloading {global_variables.conf_data.blocknet_bootstrap_url} to {local_file_path}, remote size: {int(remote_file_size / 1024)} kb") + bytes_downloaded = 0 + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + bytes_downloaded += len(chunk) + self.bootstrap_percent_download = (bytes_downloaded / remote_file_size) * 100 else: logging.error("Failed to download the Blocknet Bootstrap.") @@ -527,83 +452,29 @@ def download_blocknet_bin(self): if url is None: raise ValueError(f"Unsupported OS or architecture {global_variables.system} {global_variables.machine}") - # Set timeout values in seconds - connection_timeout = 10 - read_timeout = 30 - response = requests.get(url, stream=True, timeout=(connection_timeout, read_timeout)) - response.raise_for_status() - if response.status_code == 200: - local_file_path = os.path.join(global_variables.aio_folder, os.path.basename(url)) - try: - remote_file_size = int(response.headers.get('Content-Length', 0)) - logging.info(f"Downloading {url} to {local_file_path}, remote size: {int(remote_file_size / 1024)} kb") - bytes_downloaded = 0 - with open(local_file_path, "wb") as f: - for chunk in response.iter_content(chunk_size=8192): # Iterate over response content in chunks - if chunk: # Filter out keep-alive new chunks - f.write(chunk) - bytes_downloaded += len(chunk) - self.binary_percent_download = (bytes_downloaded / remote_file_size) * 100 - except requests.exceptions.RequestException as e: - logging.error(f"Error occurred during download: {str(e)}") - - self.binary_percent_download = None - - if os.path.getsize(local_file_path) != remote_file_size: - os.remove(local_file_path) - raise ValueError( - f"Downloaded {os.path.basename(url)} size doesn't match the expected size. Deleting it") - - logging.info(f"{os.path.basename(url)} downloaded successfully.") + tmp_path = os.path.join(global_variables.aio_folder, os.path.basename(url)) + final_path = self.blocknet_exe # For DMG + extract_to = global_variables.aio_folder # For zip/tar.gz - if url.endswith(".zip"): - with zipfile.ZipFile(local_file_path, "r") as zip_ref: - zip_ref.extractall(global_variables.aio_folder) - logging.info("Zip file extracted successfully.") - os.remove(local_file_path) - elif url.endswith(".tar.gz"): - with tarfile.open(local_file_path, "r:gz") as tar: - tar.extractall(global_variables.aio_folder) - logging.info("Tar.gz file extracted successfully.") - os.remove(local_file_path) - else: - print("Failed to download the Blocknet binary.") + self.helper.download_file( + url, tmp_path, final_path, extract_to, + global_variables.system, "binary_percent_download", self + ) self.downloading_bin = False def get_remote_file_size(url): - """ - Fetches the size of a remote file specified by its URL. - """ r = requests.head(url) r.raise_for_status() return int(r.headers.get('content-length', 0)) -def get_pid(name): - return map(int, check_output(["pidof", name]).split()) - - -def get_blocknet_data_folder(custom_path=None): - if custom_path: - path = custom_path - else: - path = global_variables.conf_data.blocknet_default_paths.get(global_variables.system) - if path: - expanded_path = os.path.expandvars(os.path.expanduser(path)) - # logging.info(f"\n path {norm_path} \n") - return os.path.normpath(expanded_path) # Normalize path separators - else: - logging.error(f"invalid blocknet data folder path: {path}") - - def generate_random_string(length): return ''.join(random.choices(string.ascii_letters + string.digits, k=length)) def save_conf_to_file(conf_data, file_path): try: - # Create missing directories if needed os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, 'w') as f: for section, options in conf_data.items(): @@ -650,7 +521,6 @@ def download_remote_conf(url, filepath): conf_data = response.text parsed_conf = parse_conf_file(input_string=conf_data) if parsed_conf: - # Save the remote configuration to a local file save_conf_to_file(parsed_conf, filepath) logging.info(f"retrieved and parsed ok: [{filepath}]") return parsed_conf @@ -667,28 +537,17 @@ def download_remote_conf(url, filepath): def retrieve_xb_manifest(): - # remote_manifest_url folder = "xb_conf" filename = os.path.basename(global_variables.conf_data.remote_manifest_url) local_manifest_file = os.path.join(global_variables.aio_folder, folder, filename) - # if os.path.exists(local_manifest_file): - # try: - # with open(local_manifest_file, 'r') as f: - # json_data = f.read() - # parsed_json = json.loads(json_data) - # logging.info(f"REMOTE: Found and parsed successfully: {local_manifest_file}") - # return parsed_json - # except Exception as e: - # logging.error(f"{local_manifest_file} Error opening or parsing file: {e}") - try: response = requests.get(global_variables.conf_data.remote_manifest_url) if response.status_code == 200: parsed_json = response.json() os.makedirs(os.path.dirname(local_manifest_file), exist_ok=True) with open(local_manifest_file, 'w') as f: - f.write(json.dumps(parsed_json, indent=4)) # Save the JSON data to local file + f.write(json.dumps(parsed_json, indent=4)) logging.info(f"REMOTE: Retrieved and parsed ok: [{local_manifest_file}]") return parsed_json else: @@ -712,7 +571,7 @@ def retrieve_remote_blocknet_xbridge_conf(): def parse_conf_file(file_path=None, input_string=None): conf_data = {} - current_section = 'global' # Set a default section + current_section = 'global' if file_path: with open(file_path, 'r') as f: @@ -737,15 +596,22 @@ def parse_conf_file(file_path=None, input_string=None): if not line or line.startswith('#'): continue if '=' in line: - conf_data.setdefault(current_section.strip('[]'), {}) key, value = line.split('=', 1) - conf_data[current_section.strip('[]')][key.strip()] = value.strip() + conf_data.setdefault(current_section.strip('[]'), {})[key.strip()] = value.strip() else: current_section = line.strip() conf_data.setdefault(current_section.strip('[]'), {}) return conf_data -# if __name__ == "__main__": -# a = BlocknetUtility() -# a.retrieve_coin_conf('BTC') + +def get_blocknet_data_folder(custom_path=None): + if custom_path: + path = custom_path + else: + path = global_variables.conf_data.blocknet_default_paths.get(global_variables.system) + if path: + expanded_path = os.path.expandvars(os.path.expanduser(path)) + return os.path.normpath(expanded_path) + else: + logging.error(f"invalid blocknet data folder path: {path}") diff --git a/utilities/conf_data.py b/utilities/conf_data.py index 001c221..828842c 100644 --- a/utilities/conf_data.py +++ b/utilities/conf_data.py @@ -2,7 +2,7 @@ aio_blocknet_data_path = { "Windows": "%appdata%\\AIO_Blocknet", "Linux": "~/.AIO_Blocknet", - "Darwin": "~/Library/Application Support/AIO_Blocknet" + "Darwin": "~/Library/AIO_Blocknet" } blocknet_bootstrap_url = "https://utils.blocknet.org/Blocknet.zip" @@ -72,7 +72,6 @@ "Windows": "XLite.exe", "Linux": "xlite", "Darwin": ["XLite.app", "Contents", "MacOS", "XLite"] - # ["BLOCK-DX-1.9.5-mac", "BLOCK DX.app", "Contents", "MacOS"] # List of folders for Darwin } xlite_daemon_bin_name = { ("Linux", "x86_64"): "xlite-daemon-linux64", diff --git a/utilities/git_repo_management.py b/utilities/git_repo_management.py index 50c9561..183510b 100644 --- a/utilities/git_repo_management.py +++ b/utilities/git_repo_management.py @@ -264,6 +264,11 @@ def _update_repo(self): logging.error(f"Remote branch '{branch}' not found in '{remote_name}'") return + current_branch = self.repo.head.shorthand + logging.info(f"current_branch: {current_branch}, self.remote_branch: {self.remote_branch}") + if current_branch != self.remote_branch: + self._checkout_branch() + # Ensure local branch exists try: repo_branch = self.repo.lookup_reference(f"refs/heads/{branch}") @@ -361,7 +366,7 @@ def __init__(self, repo_url: str, target_dir: str, branch: str = "main", workdir self.git_repo = GitRepository(repo_url, self.target_dir, branch) self.venv = None - def setup(self) -> bool: + def setup(self) -> None: """ Clone/update the repository and set up the virtual environment. @@ -391,11 +396,9 @@ def setup(self) -> bool: self.venv.install_requirements(self.target_dir / "requirements.txt") logging.info(f"Repository setup complete") - return True except Exception as e: - logging.error(f"Repository setup failed: {e}") - return False + raise Exception(f"Repository setup failed: {e}") def run_script(self, script_path: str, script_args: Optional[List[str]] = None, timeout: Optional[int] = None) -> Optional[subprocess.Popen]: diff --git a/utilities/helper_util.py b/utilities/helper_util.py new file mode 100644 index 0000000..7bbad5a --- /dev/null +++ b/utilities/helper_util.py @@ -0,0 +1,81 @@ +import logging +import os +import subprocess +import tarfile +import zipfile + +import psutil +import requests + +logging.basicConfig(level=logging.DEBUG) + + +class UtilityHelper: + def __init__(self): + pass + + # Shared by all 3 utilities + def download_file(self, url, tmp_path, final_path, extract_to, system, progress_attr, instance): + logging.info(f"Starting download from {url}") + response = requests.get(url, stream=True, timeout=(10, 30)) + response.raise_for_status() + + remote_size = int(response.headers.get('Content-Length', 0)) + with open(tmp_path, 'wb') as f: + bytes_downloaded = 0 + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + bytes_downloaded += len(chunk) + if progress_attr and instance: + setattr(instance, progress_attr, (bytes_downloaded / remote_size) * 100) + logging.debug(f"Downloaded {bytes_downloaded}/{remote_size} bytes") + + if os.path.getsize(tmp_path) != remote_size: + os.remove(tmp_path) + raise ValueError("Download size mismatch") + logging.info(f"File downloaded successfully to {tmp_path}") + + if url.endswith(".zip"): + with zipfile.ZipFile(tmp_path, 'r') as zip_ref: + zip_ref.extractall(extract_to) + os.remove(tmp_path) + logging.info(f"Extracted ZIP file to {extract_to}") + elif url.endswith(".tar.gz"): + with tarfile.open(tmp_path, 'r:gz') as tar: + tar.extractall(extract_to) + os.remove(tmp_path) + logging.info(f"Extracted TAR.GZ file to {extract_to}") + elif url.endswith(".dmg") and system == "Darwin": + os.rename(tmp_path, final_path) + logging.info(f"Renamed DMG file to {final_path}") + + # Shared by all 3 utilities + def terminate_processes(self, pids, name): + for pid in pids: + try: + proc = psutil.Process(pid) + proc.terminate() + proc.wait(timeout=10) + logging.info(f"Process {name} PID {pid} terminated successfully") + except (psutil.NoSuchProcess, psutil.TimeoutExpired) as e: + if isinstance(e, psutil.TimeoutExpired): + proc.kill() + logging.warning(f"Process {name} PID {pid}: Timeout expired, killed process") + else: + logging.warning(f"Process {name} PID {pid}: {str(e)}") + + # Shared by BlockdxUtility and XliteUtility + def handle_dmg(self, dmg_path, mount_path, action): + if action == "mount": + if not os.path.ismount(mount_path): + subprocess.run(["hdiutil", "attach", dmg_path], check=True) + logging.info(f"Mounted DMG {dmg_path} to {mount_path}") + else: + logging.warning(f"{mount_path} is already mounted") + elif action == "unmount": + if os.path.ismount(mount_path): + subprocess.run(["hdiutil", "detach", mount_path], check=True) + logging.info(f"Unmounted DMG from {mount_path}") + else: + logging.warning(f"{mount_path} is not mounted") diff --git a/utilities/miniforge_portable.py b/utilities/miniforge_portable.py index 72c4b54..bb557cf 100644 --- a/utilities/miniforge_portable.py +++ b/utilities/miniforge_portable.py @@ -78,14 +78,10 @@ def install(self): # logging.info(f"🔹 pip: {python_bin} -m pip") # logging.info(f"🔹 venv: {python_bin} -m venv {self.venv}") # After successful installation - if self.system != "Windows": + if self.system == "Linux": + # fix for tkinter low quality display. linux only. try: - # Construct conda path based on OS - if self.system == "Windows": - conda_path = install_path / "Scripts" / "conda.exe" - else: - conda_path = install_path / "bin" / "conda" - + conda_path = install_path / "bin" / "conda" # Run conda install command, update tk packages for GUI app logging.info("Installing tk with xft_* support...") conda_cmd = [str(conda_path), "install", "-c", "conda-forge", "tk=*=xft_*", "-y"] diff --git a/utilities/utils.py b/utilities/utils.py index 10bc2af..74eeb66 100644 --- a/utilities/utils.py +++ b/utilities/utils.py @@ -4,6 +4,7 @@ from threading import enumerate, current_thread import customtkinter as ctk +import psutil from cryptography.fernet import Fernet from utilities import global_variables @@ -116,3 +117,54 @@ def disable_button(button, img=None): button.configure(state=ctk.DISABLED) if img: button.configure(image=img) + + +def processes_check(): + """Check for running processes related to Blocknet, BlockDX, and Xlite.""" + + blocknet_bin = global_variables.blocknet_bin + blockdx_bin = global_variables.blockdx_bin[-1] if global_variables.system == "Darwin" \ + else global_variables.blockdx_bin + xlite_bin = global_variables.xlite_bin[-1] if global_variables.system == "Darwin" \ + else global_variables.xlite_bin + xlite_daemon_bin = global_variables.xlite_daemon_bin + # Initialize process lists + process_lists: dict = { + blocknet_bin: [], + blockdx_bin: [], + xlite_bin: [], + xlite_daemon_bin: [] + } + + # Process all running processes + for proc in psutil.process_iter(['pid', 'name', 'status']): + pid = proc.info['pid'] + name = proc.info['name'] + status = proc.info['status'] + + # Check against each target process type + for target_name, process_list in process_lists.items(): + result_pid = handle_process(pid, name, status, target_name) + if result_pid is not None: + process_list.append(result_pid) + break # Process matched, no need to check other types + + return ( + process_lists[blocknet_bin], + process_lists[blockdx_bin], + process_lists[xlite_bin], + process_lists[xlite_daemon_bin] + ) + + +def handle_process(pid, name, status, target_name): + """Helper function to handle individual process logic.""" + if name == target_name: + if status == "zombie": + # the app was closed by user manually, clean zombie process + process = psutil.Process(pid) + process.wait() + return None # Don't add zombie processes to the list + else: + return pid + return None diff --git a/utilities/xlite_util.py b/utilities/xlite_util.py index e6d599f..14bd3da 100644 --- a/utilities/xlite_util.py +++ b/utilities/xlite_util.py @@ -2,16 +2,13 @@ import logging import os import subprocess -import tarfile import threading import time -import zipfile -import psutil import requests -# from utilities.conf_data import (xlite_bin_path, xlite_default_paths, xlite_daemon_default_paths, vc_redist_win_url) from utilities import global_variables +from utilities.helper_util import UtilityHelper logging.basicConfig(level=logging.DEBUG) @@ -20,7 +17,6 @@ def check_vc_redist_installed(): - # Define the base key path base_key_path = r"SOFTWARE\Classes\Installer\Dependencies\Microsoft.VS.VC_RuntimeMinimumVSU_amd64,v14" value_name = "DisplayName" @@ -28,7 +24,7 @@ def check_vc_redist_installed(): if display_name is not None: return True else: - logging.info("No vc_redist found. installing") + logging.info("No vc_redist found. Installing") install_vc_redist(global_variables.conf_data.vc_redist_win_url) @@ -40,32 +36,27 @@ def check_registry_value(key_path, value_name): except FileNotFoundError: return None except Exception as e: - print(f"Error: {e}") + logging.error(f"Error: {e}") return None def install_vc_redist(url): try: - # Parse filename from URL installer_name = os.path.basename(url) - # Download the installer with open(installer_name, 'wb') as file: response = requests.get(url) file.write(response.content) - # Command to run the installer silently command = f"{installer_name} /install /quiet /norestart" - # Run the installer silently subprocess.run(command, shell=True, check=True) - print("Visual C++ Redistributable installed successfully.") + logging.info("Visual C++ Redistributable installed successfully.") - # Remove the installer file after installation os.remove(installer_name) except Exception as e: - print(f"Error: {e}") + logging.error(f"Error: {e}") class XliteRPCClient: @@ -85,22 +76,16 @@ def send_rpc_request(self, method=None, params=None): "id": 1, } try: - # logging.debug( - # f"Sending RPC request to URL: {url}, Method: {data['method']}, Params: {data['params']}, Auth: {auth}") response = requests.post(url, json=data, headers=headers, auth=auth) - # Check status code explicitly if response.status_code != 200: - # logging.error(f"Error sending RPC request: HTTP status code {response.status_code}") return None json_answer = response.json() - # logging.debug(f"RPC request successful. Response: {json}") if 'result' in json_answer: return json_answer['result'] else: logging.error(f"No result in json: {json_answer}") except requests.RequestException as e: - # logging.error(f"Error sending RPC request: {e}") return None except Exception as ex: logging.exception(f"An unexpected error occurred while sending RPC request: {ex}") @@ -109,6 +94,7 @@ def send_rpc_request(self, method=None, params=None): class XliteUtility: def __init__(self): + self.helper = UtilityHelper() if global_variables.system == "Darwin": self.xlite_exe = os.path.join(global_variables.aio_folder, os.path.basename(global_variables.xlite_url)) self.dmg_mount_path = f"/Volumes/{global_variables.xlite_volume_name}" @@ -125,7 +111,6 @@ def __init__(self): self.xlite_process = None self.xlite_daemon_process = None self.xlite_conf_local = {} - self.xlite_daemon_confs_local = {} self.running = True # flag for async funcs self.xlite_pids = [] self.xlite_daemon_pids = [] @@ -150,18 +135,15 @@ def check_xlite_daemon_confs(self): def check_valid_xlite_coins_rpc(self, runonce=False): while self.running: - # logging.debug(f"valid_coins_rpc: {self.valid_coins_rpc}, runonce: {runonce}") valid = False if self.coins_rpc: for coin, rpc_server in self.coins_rpc.items(): if coin != "master" and coin != "TBLOCK": - # logging.info(self.xlite_daemon_confs_local[coin]['rpcEnabled']) if self.xlite_daemon_confs_local[coin]['rpcEnabled'] is True: res = rpc_server.send_rpc_request("getinfo") if res is not None: valid = True if not valid: - # logging.debug(f"coin {coin} not valid") break if valid: self.valid_coins_rpc = True @@ -189,41 +171,31 @@ def parse_xlite_conf(self): try: with open(file_path, 'r') as file: meta_data = json.load(file) - logging.info(f"XLITE: Loaded JSON data from [{file_path}]") #: {meta_data}") + logging.info(f"XLITE: Loaded JSON data from [{file_path}]") except Exception as e: logging.error(f"Error parsing {file_path}: {e}, repairing file") - else: - # logging.warning(f"{file_path} doesn't exist") - pass self.xlite_conf_local = meta_data def parse_xlite_daemon_conf(self, silent=False): - # Assuming daemon_data_path and confs_folder are defined earlier in your code daemon_data_path = os.path.expandvars( os.path.expanduser( global_variables.conf_data.xlite_daemon_default_paths.get(global_variables.system, None))) confs_folder = os.path.join(daemon_data_path, "settings") - # List all files in the confs_folder if not os.path.exists(confs_folder): - # logging.warning(f"{confs_folder} doesn't exist") self.xlite_daemon_confs_local = {} return files_in_folder = os.listdir(confs_folder) - # Filter out only JSON files json_files = [file for file in files_in_folder if file.endswith('.json')] - # Parse each JSON file for json_file in json_files: json_file_path = os.path.join(confs_folder, json_file) coin = str(json_file).split("-")[1].split(".")[0] try: with open(json_file_path, 'r') as file: data = json.load(file) - # Do something with the parsed JSON data - # logging.debug(f"Parsed data from {coin} {json_file}: {data}") self.xlite_daemon_confs_local[coin] = data except Exception as e: self.xlite_daemon_confs_local[coin] = "ERROR PARSING" @@ -234,13 +206,10 @@ def parse_xlite_daemon_conf(self, silent=False): def start_xlite(self, env_vars=[]): if global_variables.system == "Windows": - # check vcredist - # install_vc_redist(vc_redist_win_url) check_vc_redist_installed() for var_dict in env_vars: for var_name, var_value in var_dict.items(): - # logging.info(f"var_name: {var_name} var_value: {var_value}") os.environ[var_name] = var_value if not os.path.exists(self.xlite_exe): @@ -249,16 +218,7 @@ def start_xlite(self, env_vars=[]): try: if global_variables.system == "Darwin": - # mac mod - # https://github.com/blocknetdx/xlite/releases/download/v1.0.7/XLite-1.0.7-mac.dmg - # Path to the application inside the DMG file - - # Check if the volume is already mounted - if not os.path.ismount(self.dmg_mount_path): - # Mount the DMG file - os.system(f'hdiutil attach "{self.xlite_exe}"') - else: - logging.info("Volume is already mounted.") + self.helper.handle_dmg(self.xlite_exe, self.dmg_mount_path, "mount") full_path = os.path.join(self.dmg_mount_path, *global_variables.conf_data.xlite_bin_name[global_variables.system]) logging.info( @@ -269,15 +229,13 @@ def start_xlite(self, env_vars=[]): stdin=subprocess.PIPE, start_new_session=True) else: - # Start the Blocknet process using subprocess self.xlite_process = subprocess.Popen([self.xlite_exe], stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, start_new_session=True) - # Check if the process has started while self.xlite_process.pid is None: - time.sleep(1) # Wait for 1 second before checking again + time.sleep(1) pid = self.xlite_process.pid logging.info(f"Started Xlite process with PID {pid}: {self.xlite_exe}") @@ -285,11 +243,10 @@ def start_xlite(self, env_vars=[]): logging.error(f"Error: {e}") def close_xlite(self): - # Close the Xlite subprocess if it exists if self.xlite_process: try: self.xlite_process.terminate() - self.xlite_process.wait(timeout=10) # Wait for the process to terminate with a timeout of 60 seconds + self.xlite_process.wait(timeout=10) logging.info(f"Closed Xlite") self.xlite_process = None except subprocess.TimeoutExpired: @@ -304,7 +261,6 @@ def close_xlite(self): self.close_xlite_daemon_pids() def kill_xlite(self): - # Kill the Xlite subprocess if it exists if self.xlite_process: try: self.xlite_process.kill() @@ -315,48 +271,10 @@ def kill_xlite(self): logging.error(f"Error: {e}") def close_xlite_pids(self): - # Close the Xlite processes using their PIDs - for pid in self.xlite_pids: - try: - # Get the process object corresponding to the PID - proc = psutil.Process(pid) - proc.terminate() - logging.info(f"Initiated termination of Xlite process with PID {pid}.") - proc.wait(timeout=10) # Wait for the process to terminate with a timeout of 60 seconds - logging.info(f"Xlite process with PID {pid} has been terminated.") - except psutil.NoSuchProcess: - logging.warning(f"Xlite process with PID {pid} not found.") - except psutil.TimeoutExpired: - logging.warning(f"Force terminating Xlite process with PID {pid}.") - if proc: - proc.kill() - proc.wait() - logging.info(f"Xlite process with PID {pid} has been force terminated.") - except Exception as e: - logging.error(f"Error: {e}") + self.helper.terminate_processes(self.xlite_pids, "XLite") def close_xlite_daemon_pids(self): - - # Close the Xlite-daemon processes using their PIDs - for pid in self.xlite_daemon_pids: - try: - # Get the process object corresponding to the PID - proc = psutil.Process(pid) - proc.terminate() - logging.info(f"Initiated termination of Xlite-daemon process with PID {pid}.") - proc.wait(timeout=10) # Wait for the process to terminate with a timeout of 60 seconds - logging.info(f"Xlite-daemon process with PID {pid} has been terminated.") - except psutil.NoSuchProcess: - logging.warning(f"Xlite-daemon process with PID {pid} not found.") - except psutil.TimeoutExpired: - logging.warning(f"Force terminating Xlite-daemon process with PID {pid}.") - if proc: - proc.kill() - proc.wait() - logging.info(f"Xlite-daemon process with PID {pid} has been force terminated.") - except Exception as e: - logging.error(f"Error: {e}") - logging.info(f"Closed Xlite daemon") + self.helper.terminate_processes(self.xlite_daemon_pids, "Xlite-daemon") def download_xlite_bin(self): self.downloading_bin = True @@ -364,69 +282,15 @@ def download_xlite_bin(self): if url is None: raise ValueError(f"Unsupported OS or architecture {global_variables.system} {global_variables.machine}") - # Set timeout values in seconds - connection_timeout = 5 - read_timeout = 30 - response = requests.get(url, stream=True, timeout=(connection_timeout, read_timeout)) - response.raise_for_status() # Raise an exception for 4xx and 5xx status codes - if response.status_code == 200: - file_name = os.path.basename(url) - tmp_file_path = os.path.join(global_variables.aio_folder, "tmp_xl_bin") - try: - remote_file_size = int(response.headers.get('Content-Length', 0)) - # tmp_file_path = os.path.join(aio_folder, file_name + "_tmp") - logging.info(f"Downloading {url} to {tmp_file_path}, remote size: {int(remote_file_size / 1024)} kb") - bytes_downloaded = 0 - with open(tmp_file_path, "wb") as f: - for chunk in response.iter_content(chunk_size=8192): # Iterate over response content in chunks - if chunk: # Filter out keep-alive new chunks - f.write(chunk) - bytes_downloaded += len(chunk) - self.binary_percent_download = (bytes_downloaded / remote_file_size) * 100 - # print(self.binary_percent_download) - except requests.exceptions.RequestException as e: - logging.error(f"Error occurred during download: {str(e)}") - - self.binary_percent_download = None - - local_file_size = os.path.getsize(tmp_file_path) - if local_file_size != remote_file_size: - os.remove(tmp_file_path) - raise ValueError( - f"Downloaded {os.path.basename(url)} size doesn't match the expected size. Deleting it") - - logging.info(f"{os.path.basename(url)} downloaded successfully.") - - # Extract the archive - if url.endswith(".zip"): - with zipfile.ZipFile(tmp_file_path, "r") as zip_ref: - local_path = os.path.join(global_variables.aio_folder, - global_variables.conf_data.xlite_bin_path[global_variables.system]) - zip_ref.extractall(local_path) - logging.info("Zip file extracted successfully.") - os.remove(tmp_file_path) - elif url.endswith(".tar.gz"): - with tarfile.open(tmp_file_path, "r:gz") as tar: - tar.extractall(global_variables.aio_folder) - logging.info("Tar.gz file extracted successfully.") - os.remove(tmp_file_path) - elif url.endswith(".dmg"): - file_path = os.path.join(global_variables.aio_folder, file_name) - os.rename(tmp_file_path, file_path) - logging.info("DMG file saved successfully.") - else: - print("Failed to download the Xlite binary.") + tmp_path = os.path.join(global_variables.aio_folder, "tmp_xl_bin") + final_path = self.xlite_exe # For DMG + extract_to = global_variables.aio_folder # For zip/tar.gz + + self.helper.download_file( + url, tmp_path, final_path, extract_to, + global_variables.system, "binary_percent_download", self + ) self.downloading_bin = False def unmount_dmg(self): - if os.path.ismount(self.dmg_mount_path): - try: - subprocess.run(["hdiutil", "detach", self.dmg_mount_path], check=True) - logging.info("DMG unmounted successfully.") - except subprocess.CalledProcessError as e: - logging.error(f"Error: Failed to unmount DMG: {e}") - else: - logging.error("Error: DMG is not mounted.") - -# if __name__ == "__main__": -# install_vc_redist(vc_redist_win_url) + self.helper.handle_dmg(None, self.dmg_mount_path, "unmount")