# LauncherTool/core/execution_handler.py """ Handles the execution of application sequences. """ import subprocess import os import shlex import logging import time # Per la pausa, ma verrà gestito da Tkinter per l'UI from typing import List, Dict, Callable, Any from .config_manager import ConfigManager from .exceptions import ( ApplicationNotFoundError, ExecutionError, CommandExecutionError, ApplicationPathNotFoundError ) logger = logging.getLogger(__name__) class ExecutionHandler: """ Manages the execution of a sequence of applications. """ def __init__(self, config_manager: ConfigManager, output_callback: Callable[[str], None] = None, step_delay_callback: Callable[[float, Callable, list, int], Any] = None): """ Initializes the ExecutionHandler. Args: config_manager (ConfigManager): The configuration manager instance. output_callback (Callable[[str], None], optional): A callback function to send output messages to (e.g., for GUI display). It takes a string message as an argument. step_delay_callback (Callable[[float, Callable, list, int], Any], optional): A callback function to handle delays between steps, typically `root.after` for Tkinter applications. It takes (delay_ms, next_function, *args_for_next_function). """ self.config_manager = config_manager self.output_callback = output_callback if output_callback else self._default_output # If no step_delay_callback is provided, we'll use a simple time.sleep # This makes the class usable even outside a Tkinter event loop, e.g., for tests # or a CLI version. self.step_delay_callback = step_delay_callback if step_delay_callback else self._default_delay def _default_output(self, message: str): """Default output method if none is provided (prints to console).""" print(message) logger.debug(f"ExecutionHandler (default output): {message}") def _default_delay(self, delay_seconds: float, next_callable: Callable, *args): """Default delay method if none is provided (uses time.sleep).""" time.sleep(delay_seconds) return next_callable(*args) def _log_and_output(self, message: str, level: int = logging.INFO): """Helper to log and send output via callback.""" if level == logging.ERROR: logger.error(message) elif level == logging.WARNING: logger.warning(message) else: logger.info(message) self.output_callback(message + "\n") def run_sequence_async(self, sequence_name: str): """ Starts the execution of a named sequence. This method is designed to be called, and then the execution proceeds step-by-step using the step_delay_callback. Args: sequence_name (str): The name of the sequence to run. Raises: SequenceNotFoundError: If the sequence is not found. """ try: sequence = self.config_manager.get_sequence_by_name(sequence_name) except NameNotFoundError as e: # Catching the specific error from config_manager self._log_and_output(f"Error: {e}", level=logging.ERROR) raise # Re-raise to be handled by the caller (e.g., GUI) self._log_and_output(f"Starting sequence: {sequence_name}") steps = sequence.get("steps", []) if not steps: self._log_and_output(f"Sequence '{sequence_name}' has no steps to execute.", level=logging.WARNING) self._log_and_output("Sequence completed (no steps).") return # Start the first step using the delay callback mechanism # The first step doesn't have a preceding delay from here, # but we use the callback structure for consistency. self.step_delay_callback(0, self._execute_step, steps, 0) def _execute_step(self, steps: List[Dict], step_index: int): """ Executes a single step of the sequence and schedules the next one. This method is intended to be called by the step_delay_callback. Args: steps (List[Dict]): The list of all steps in the sequence. step_index (int): The index of the current step to execute. """ if step_index >= len(steps): self._log_and_output("Sequence completed.") return # Sequence finished current_step = steps[step_index] app_name = current_step.get("application") wait_time_seconds = float(current_step.get("wait_time", 0.0)) if not app_name: self._log_and_output( f"Error in step {step_index + 1}: Application name is missing.", level=logging.ERROR ) # Optionally, decide whether to stop or skip to the next step # For now, we stop if a step is malformed. return try: application_config = self.config_manager.get_application_by_name(app_name) except ApplicationNotFoundError: self._log_and_output( f"Error: Application '{app_name}' in step {step_index + 1} not found in configuration.", level=logging.ERROR ) # Decide: stop sequence or try next step? For now, stop. return app_path = application_config.get("path") if not app_path or not os.path.exists(app_path): self._log_and_output( f"Error: Executable path for application '{app_name}' ('{app_path}') not found or invalid.", level=logging.ERROR ) # This is a critical error for this step. # Consider raising ApplicationPathNotFoundError(app_name, app_path) if the GUI should handle it. return # Build the command command = [app_path] # Parameters defined in the sequence step override defaults from application config step_parameters = current_step.get("parameters", {}) # Parameters defined in the application config (defaults) app_config_parameters = application_config.get("parameters", []) # Logic for assembling parameters: # 1. Use value from step_parameters if defined for a parameter name. # 2. Else, use default_value from app_config_parameters. # Parameter names in step_parameters should match names in app_config_parameters. # Create a dictionary of default values from app_config for quick lookup default_params_map = {p["name"]: p["default_value"] for p in app_config_parameters if "name" in p and "default_value" in p} # Add parameters defined in app_config_parameters, potentially overridden by step_parameters for app_param_config in app_config_parameters: param_name = app_param_config["name"] param_value_to_use = None if param_name in step_parameters: # Value is directly provided in the sequence step's parameters param_value_to_use = step_parameters[param_name] elif app_param_config.get("default_value"): # Use the default value from the application's configuration param_value_to_use = app_param_config["default_value"] if param_value_to_use is not None: # The original code appended the value directly. # If parameters need to be in the form "--name=value" or "/name:value", # this logic needs adjustment based on how your EXEs expect them. # Assuming the value itself is the full argument string (e.g., "/--NoRis", "/r:p=127.0.0.1") command.append(str(param_value_to_use)) self._log_and_output(f"Executing step {step_index + 1}: {' '.join(map(shlex.quote, command))}") try: app_dir = os.path.dirname(app_path) # Ensure current working directory is valid before changing original_cwd = os.getcwd() if os.path.isdir(app_dir): os.chdir(app_dir) logger.debug(f"Changed CWD to: {app_dir}") else: logger.warning(f"Application directory '{app_dir}' not found. Executing from CWD: {original_cwd}") # For Windows, DETACHED_PROCESS and CREATE_NEW_PROCESS_GROUP are useful # For Unix, subprocess.DEVNULL for stdout/stderr to detach cleanly if no output is needed # shell=True can be a security risk if command parts come from untrusted input. # Here, app_path comes from config, parameters too. shlex.quote helps. # If shell=False, command must be a list, and path resolution is stricter. # Given the original code used shell=True, we'll stick to it for now, but be mindful. # The original code joined command into a string then passed to Popen with shell=True. # It's generally safer to pass a list of args if shell=False. # If shell=True, command_string is better. command_string = ' '.join(map(shlex.quote, command)) if os.name == 'nt': # Windows process = subprocess.Popen( command_string, # or command list if shell=False creationflags=subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP, shell=True, # Consistent with original cwd=app_dir if os.path.isdir(app_dir) else None # Popen can handle cwd ) else: # Unix-like (Linux, macOS) process = subprocess.Popen( command_string, # or command list if shell=False stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, shell=True, # Consistent with original start_new_session=True, # Good for detaching on Unix cwd=app_dir if os.path.isdir(app_dir) else None ) self._log_and_output(f"Application '{app_name}' launched (PID: {process.pid if process else 'Unknown'}).") except FileNotFoundError: # Specifically for the executable not being found by Popen err_msg = f"Execution Error: Application executable '{app_path}' for '{app_name}' not found by the system." self._log_and_output(err_msg, level=logging.ERROR) # raise ApplicationPathNotFoundError(app_name, app_path) from fnfe # Let GUI handle # For now, stopping sequence return except Exception as e: err_msg = f"Execution Error for '{app_name}': {e}" self._log_and_output(err_msg, level=logging.ERROR) # raise CommandExecutionError(command, e) from e # Let GUI handle # For now, stopping sequence return finally: if 'original_cwd' in locals() and os.getcwd() != original_cwd : # Check if original_cwd was set os.chdir(original_cwd) # Change back to original CWD logger.debug(f"Restored CWD to: {original_cwd}") self._log_and_output(f"Waiting {wait_time_seconds} seconds before next step...") # Schedule the next step self.step_delay_callback( wait_time_seconds * 1000, # Convert to milliseconds for Tkinter's 'after' self._execute_step, steps, step_index + 1 )