258 lines
12 KiB
Python
258 lines
12 KiB
Python
# 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...")
|
|
|
|
delay_ms = int(wait_time_seconds * 1000)
|
|
# Schedule the next step
|
|
self.step_delay_callback(
|
|
delay_ms,
|
|
self._execute_step,
|
|
steps,
|
|
step_index + 1
|
|
) |