source:
from PIL import Image
import pillow_avif
import os
import sys
import threading
import time
MAX_THREADS = 25
def file_convert(img_dir, file_name, convert_from, convert_to):
tmp = Image.open(img_dir + "/" + file_name)
tmp.save(img_dir + "/" + file_name[:-len(convert_from)] + convert_to)
os.remove(file_name)
#run in interactive mode if called without arguments
if (len(sys.argv) == 1):
while(True):
#prompt for a directory path to search for images
print("Enter the directory to search for .webp files or enter 'cwd' to use the currect directory")
img_dir = input("Path to search: ")
if(img_dir == "cwd" or img_dir == "CWD"):
img_dir = os.getcwd()
break
#if path supplied is not valid, loop and prompt again
img_dir = os.path.expanduser(img_dir)
if(not os.path.isdir(img_dir)):
print("'" + img_dir + "' is not a valid directory.")
else:
break
img_from = ".webp"
img_to = ".png"
#arguments are supplied, don't prompt for input
else:
img_dir = sys.argv[1]
img_dir = os.path.expanduser(img_dir)
#check if first argument is a valid file path or an indicator to use the current directory
if(img_dir == "cwd" or img_dir == "CWD"):
img_dir = os.getcwd()
elif(not os.path.isdir(img_dir)):
print("'" + img_dir + "' is not a valid directory.")
sys.exit()
if(len(sys.argv) == 4):
img_from = sys.argv[2]
img_to = sys.argv[3]
else:
img_from = ".webp"
img_to = ".png"
#cd into supplied directory
os.chdir(img_dir)
dir_contents = os.listdir(img_dir)
dir_files = []
#compile list of files in folder
for x in dir_contents:
if (os.path.isfile(x)):
dir_files.append(x)
#display files in folder to user to confirm directory is correct
print("\nFiles in " + img_dir + ":")
for x in dir_files:
print(" > " + x)
#if in interactive mode, prompt for confirmation
if(len(sys.argv) == 1):
confirm = input("\nConvert all .webp to .png? (y/n):\n")
else:
confirm = "y"
#convert files from webp to png and list progress
if (confirm in ["y", "Y"]):
queue = []
threads = []
print("Queuing files for conversion")
for x in dir_files:
if (x[-len(img_from):] == img_from):
print("Queued: " + x)
queue.append(x)
conversions = len(queue)
iterations = 0
start = time.perf_counter()
while(len(queue) > 0):
threads = threading.enumerate()
#removing this print statement adds ~15 seconds to execution time
#why? who the fuck knows
print("Waiting for queued threads to finish", end="\r")
# MAX_THREADS + 1 because threading.enumerate returns all threads including the main thread (this one)
if(len(threads) < MAX_THREADS + 1):
tmp = threading.Thread(target=file_convert, args=(img_dir, queue.pop(), img_from, img_to))
tmp.start()
threads.append(tmp)
end = time.perf_counter()
print(f"Converted {conversions} images in {(end - start):.2f} seconds.")
#if user does not confirm, exit
else:
print("No operations performed. Exiting...")