A Python script providing some helpful utils for a SeedQueue instance

helper.py

# Define the settings
# instance_path = r"C:\PrismLauncher\instances\seed q"
print_benchmark = True
# delete_all_saves = True
# print_mods = True
run_ahk_script = True

# Actual script

import os, shutil, re, hashlib, time, subprocess

# Start timing the script
start_time = time.time()

# Allow commenting out lines
if not "instance_path" in globals(): instance_path = os.path.abspath(os.path.join(os.getcwd(), os.pardir))
if not "print_benchmark" in globals(): print_benchmark = False
if not "delete_all_saves" in globals(): delete_all_saves = False
if not "print_mods" in globals(): print_mods = False
if not "run_ahk_script" in globals(): run_ahk_script = False

# Paths based on the instance path
saves_path = os.path.join(instance_path, r".minecraft", r"saves")
log_file = os.path.join(instance_path, r".minecraft", r"logs", r"latest.log")
mods_path = os.path.join(instance_path, r".minecraft", r"mods")
script_dir = os.path.dirname(os.path.abspath(__file__))

def log_message(prefix, message):
    current_time = time.strftime('%H:%M:%S', time.localtime(time.time()))
    print(f"[{current_time}] [{prefix}] {message}", flush=True)

def sha256_hash(file_path):
    sha256 = hashlib.sha256()
    with open(file_path, 'rb') as f:
        for block in iter(lambda: f.read(4096), b""):
            sha256.update(block)
    return sha256.hexdigest()

def get_folder_size(folder_path):
    total_size = 0
    for dirpath, dirnames, filenames in os.walk(folder_path):
        for f in filenames:
            fp = os.path.join(dirpath, f)
            total_size += os.path.getsize(fp)
    return total_size

# Log start of script
log_message("script", "Starting the script...")

# Check for the "BENCHMARK | {}\n" pattern in latest.log
log_message("log checker", "Checking if latest.log exists...")
if os.path.exists(log_file):
    try:
        with open(log_file, 'r') as f:
            content = f.read()
        
        # Count the number of lines in latest.log
        line_count = content.count('\n')
        log_message("log checker", f"latest.log contains {line_count} lines.")
        
        # Check for the "BENCHMARK | {}\n" pattern in latest.log, if print_benchmark is True
        if print_benchmark:
            log_message("log checker", "Checking for benchmark patterns in latest.log...")
            try:
                # Find all occurrences of the benchmark pattern (case insensitive)
                benchmark_pattern = re.compile(r"Reset (\d+) seeds in ([\d.]+) seconds\.", re.IGNORECASE)
                benchmarks = benchmark_pattern.findall(content)

                # Convert benchmarks to a list of pairs of integers and floats
                parsed_benchmarks = []
                for seeds, seconds in benchmarks:
                    try:
                        seeds_int = int(seeds)
                        seconds_float = float(seconds)
                        parsed_benchmarks.append((seeds_int, seconds_float))
                    except ValueError:
                        log_message("log checker", f"Failed to convert benchmark: 'Reset {seeds} seeds in {seconds} seconds.'")

                # Check if parsed_benchmarks is not empty before processing
                if parsed_benchmarks:
                    # Filter and sort the list to keep only increasing seeds counts
                    sorted_benchmarks = []
                    for i in range(len(parsed_benchmarks) - 1):
                        if parsed_benchmarks[i][0] <= parsed_benchmarks[i + 1][0]:
                            sorted_benchmarks.append(parsed_benchmarks[i])

                    # Include the last one if it has the highest seed count
                    if not sorted_benchmarks or parsed_benchmarks[-1][0] >= sorted_benchmarks[-1][0]:
                        sorted_benchmarks.append(parsed_benchmarks[-1])

                    # Print the sorted benchmarks with formatting and comparisons
                    if sorted_benchmarks:
                        log_message("log checker", "Formatted benchmark results:")
                        previous_seeds, previous_seconds = 0, 0
                        for seeds, seconds in sorted_benchmarks:
                            # Calculate and print the total and last benchmark comparison
                            resets_per_second = seeds / seconds
                            diff_seeds = seeds - previous_seeds
                            diff_seconds = seconds - previous_seconds
                            if diff_seconds != 0:
                                diff_resets_per_second = diff_seeds / diff_seconds
                            else:
                                diff_resets_per_second = 0

                            print(f"  total: {seeds} seeds in {seconds:.2f}s, {resets_per_second:.2f} seeds/s; "
                                  f"last: {diff_seeds} seeds in {diff_seconds:.2f}s, {diff_resets_per_second:.2f} seeds/s")

                            # Update the previous benchmark values for the next iteration
                            previous_seeds, previous_seconds = seeds, seconds
                    else:
                        log_message("log checker", "No valid benchmark patterns found or seeds count did not increase sequentially.")
                else:
                    log_message("log checker", "No valid benchmark patterns found in latest.log.")
            
            except Exception as e:
                log_message("log checker", f"Error processing benchmarks: {str(e)}")
        
        
        # Check for "OpenGL debug message" (case insensitive)
        if re.search(r"OpenGL debug message", content, re.IGNORECASE):
            log_message("log checker", "OpenGL debug message found in latest.log.")
            exit(1)
        else:
            log_message("log checker", "No OpenGL debug message found in latest.log.")
    
    except Exception as e:
        log_message("log checker", f"Error reading latest.log: {str(e)}")
        log_message("log checker", "Please manually check for \"OpenGL debug message\"")
        exit(1)
else:
    log_message("log checker", "latest.log does not exist.")


# Delete the saves directory if it exists and count the worlds
if not os.path.exists(saves_path):
    log_message("saves deleter", f"Directory {saves_path} does not exist.")
elif delete_all_saves:
    log_message("saves deleter", "Deleting the entire saves directory...")
    try:
        # Count only the immediate subdirectories (folders) in /saves/
        folder_count = len([name for name in os.listdir(saves_path) if os.path.isdir(os.path.join(saves_path, name))])
        log_message("saves deleter", f"Deleting {folder_count} saves...")
        
        if folder_count > 500:
            speed = 360 # folders per second
            eta = folder_count / speed
            log_message("saves deleter", f"Estimated time to delete {folder_count} saves: {eta:.2f} seconds")

        # Measure the time taken to delete the directory
        start_delete_time = time.time()
        shutil.rmtree(saves_path)
        end_delete_time = time.time()

        # Calculate and log the time taken
        delete_time = end_delete_time - start_delete_time
        if delete_time > 1e-6:
            log_message("saves deleter", f"Deleted {folder_count} saves successfully in {delete_time:.2f} seconds with {folder_count / delete_time:.2f} saves per second.")
        else:
            log_message("saves deleter", f"Deleted {folder_count} saves successfully in {delete_time:.2f} seconds.")

        os.makedirs(saves_path, exist_ok=True)
    except Exception as e:
        log_message("saves deleter", f"Failed to delete saves: {str(e)}")

else:
    size_threshold = 11 * 1024 * 1024  # 11 MB in bytes
    deleted_count = 0
    skipped_count = 0

    # Iterate over all folders in saves_path
    for folder_name in os.listdir(saves_path):
        folder_path = os.path.join(saves_path, folder_name)
        if os.path.isdir(folder_path):
            folder_size = get_folder_size(folder_path)
            
            if folder_size < size_threshold:
                shutil.rmtree(folder_path)
                deleted_count += 1
            else:
                skipped_count += 1
    
    # Log the summary
    log_message("saves deleter", f"Deleted {deleted_count} folder(s). Skipped {skipped_count} folder(s).")
    


if print_mods:
    # Compute SHA-256 hashes for .jar files in the mods directory
    log_message("mod checker", "Checking for .jar files in the mods directory...")
    if os.path.exists(mods_path):
        jar_files = [f for f in os.listdir(mods_path) if f.endswith('.jar')]
        if jar_files:
            for jar_file in jar_files:
                jar_path = os.path.join(mods_path, jar_file)
                hash_value = sha256_hash(jar_path)
                log_message("mod checker", f"SHA-256 for {jar_file}: {hash_value}")
        else:
            log_message("mod checker", "No .jar files found in the mods directory.")
    else:
        log_message("mod checker", "Mods directory does not exist.")
else:
    log_message("mod checker", "Skipping mod check...")


# Execute the AutoHotkey script
if run_ahk_script:
    ahk_script = os.path.join(script_dir, "lc.ahk")
    if os.path.exists(ahk_script):
        log_message("ahk", "Executing AutoHotkey script...")
        try:
            subprocess.Popen([r"C:\Program Files\AutoHotkey\AutoHotkey.exe", ahk_script], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            log_message("helper", "lc.ahk script started successfully.")
        except subprocess.CalledProcessError as e:
            log_message("helper", f"Failed to execute lc.ahk: {str(e)}")
    else:
        log_message("ahk", "AutoHotkey script not found.")


# End of script
end_time = time.time()
elapsed_time = end_time - start_time
log_message("script", f"Script completed in {elapsed_time:.2f} seconds.")