381 lines
11 KiB
Python
381 lines
11 KiB
Python
# =============================================================================
|
|
# SCRIPT OF EXECUTION - ADVANCED CLUSTERING
|
|
# =============================================================================
|
|
"""
|
|
Helper script to run advanced clustering analysis
|
|
with different configurations and datasets.
|
|
"""
|
|
|
|
import subprocess
|
|
import sys
|
|
import os
|
|
|
|
def print_phase_menu():
|
|
"""Show options menu"""
|
|
print("\nAvailable options:")
|
|
print(" 1. Run phase V analysis")
|
|
print(" 2. Run phase R analysis")
|
|
print(" 3. Back to main menu")
|
|
|
|
def print_model_menu():
|
|
"""Show model selection menu"""
|
|
print("\n" + "="*80)
|
|
print(" SUPERVISED LEARNING - Hazelnut")
|
|
print("="*80)
|
|
print("\nSelect model:")
|
|
print(" 1. MobileNet")
|
|
print(" 2. EfficientNet")
|
|
print(" 3. ResNet")
|
|
print(" 4. Vision Transformer (ViT)")
|
|
print(" 5. Exit")
|
|
|
|
def print_data_menu():
|
|
"""Show data selection menu"""
|
|
print("\nSelect dataset:")
|
|
print(" 1. San Damiano")
|
|
print(" 2. GBIF")
|
|
print(" 3. AV")
|
|
print(" 4. Combined")
|
|
print(" 5. Combined/AV for testing")
|
|
print(" 6. Back to main menu")
|
|
|
|
def install_dependencies():
|
|
"""Install necessary dependencies"""
|
|
print("\nInstalling dependencies...")
|
|
|
|
dependencies = [
|
|
'tensorflow',
|
|
'scikit-learn',
|
|
'pandas',
|
|
'numpy',
|
|
'matplotlib',
|
|
'seaborn',
|
|
'tqdm',
|
|
'umap-learn',
|
|
'pillow'
|
|
]
|
|
|
|
for dep in dependencies:
|
|
print(f"Installing {dep}...")
|
|
subprocess.run([sys.executable, '-m', 'pip', 'install', dep], check=False)
|
|
|
|
print("Dependencies installed")
|
|
|
|
def verify_configuration():
|
|
"""Verify that the configuration is correct"""
|
|
print("\nVerifying configuration...")
|
|
|
|
# Verify TensorFlow
|
|
try:
|
|
import tensorflow as tf
|
|
print(f"TensorFlow installed: {tf.__version__}")
|
|
except ImportError:
|
|
print("TensorFlow NOT installed")
|
|
return False
|
|
|
|
# Verify GPU
|
|
gpus = tf.config.list_physical_devices('GPU')
|
|
if gpus:
|
|
print(f"GPU available: {len(gpus)} GPU(s)")
|
|
else:
|
|
print("GPU not available, CPU will be used (slower)")
|
|
|
|
return True
|
|
|
|
def run_model_V(image_dir: str, test: bool):
|
|
"""Run model"""
|
|
print("\nRunning MobileNet model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'MobileNetV1_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(image_dir, f"results_mobilenet_fase_V_{os.path.basename(image_dir)}")
|
|
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", "fase V"]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def run_model_R(image_dir: str, test: bool):
|
|
"""Run model"""
|
|
print("\nRunning MobileNet model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'MobileNetV1_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(os.getcwd(), f"results_mobilenet_fase_R_{os.path.basename(image_dir)}")
|
|
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", "fase R"]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def verify_paths(path: str) -> bool:
|
|
#Verify images directory
|
|
if os.path.exists(path):
|
|
print(f"Images directory found: {path}")
|
|
else:
|
|
print(f"Images directory NOT found: {path}")
|
|
return False
|
|
|
|
if os.path.exists(os.path.join(path, 'tags.csv')):
|
|
print(f"CSV found: {os.path.join(path, 'tags.csv')}")
|
|
return True
|
|
else:
|
|
print(f"CSV NOT found: {os.path.join(path, 'tags.csv')}")
|
|
return False
|
|
|
|
def run_EffiNet(image_dir: str, test: bool, phase: str):
|
|
"""Run EfficientNet model"""
|
|
print("\nRunning EfficientNet model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'EfficientNetV2_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(image_dir, f"results_efficientnet_{phase}_{os.path.basename(image_dir)}")
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
phase = phase.replace("_", " ")
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", phase]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def run_MobiNet(image_dir: str, test: bool, phase: str):
|
|
"""Run MobileNet model"""
|
|
print("\nRunning MobileNet model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'MobileNetV1_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(image_dir, f"results_mobilenet_{phase}_{os.path.basename(image_dir)}")
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
phase = phase.replace("_", " ")
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", phase]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def run_ResNet(image_dir: str, test: bool, phase: str):
|
|
"""Run ResNet model"""
|
|
print("\nRunning ResNet model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'ResNet50_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(image_dir, f"results_resnet_{phase}_{os.path.basename(image_dir)}")
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
phase = phase.replace("_", " ")
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", phase]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def run_ViT(image_dir: str, test: bool, phase: str):
|
|
"""Run Vision Transformer model"""
|
|
print("\nRunning Vision Transformer model...")
|
|
script_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'ViT_Nocciola.py'
|
|
)
|
|
|
|
csv_path = os.path.join(image_dir, 'tags.csv')
|
|
out_dir = os.path.join(image_dir, f"results_vit_{phase}_{os.path.basename(image_dir)}")
|
|
if test:
|
|
out_dir += "_AV"
|
|
|
|
for count in range(1, 100):
|
|
if os.path.exists(out_dir):
|
|
out_dir += f"_{str(count)}"
|
|
else:
|
|
break
|
|
|
|
phase = phase.replace("_", " ")
|
|
args = [sys.executable, script_path, "--images_dir", image_dir, "--csv_path", csv_path, "--output_dir", out_dir,
|
|
"--phase", phase]
|
|
|
|
if test:
|
|
args.append("--test")
|
|
|
|
try:
|
|
subprocess.run(args, check=True)
|
|
print("\nExecution completed successfully.")
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"\nError during execution: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""Main function"""
|
|
# Change this to your datasets directory
|
|
data_dir = os.path.join("")
|
|
|
|
while True:
|
|
print_model_menu()
|
|
try:
|
|
model_sel = int(input("\nSelect an option (1-5): ").strip())
|
|
except ValueError:
|
|
model_sel = 0
|
|
|
|
if model_sel == 5:
|
|
print("\nGoodbye!")
|
|
break
|
|
elif model_sel not in [1, 2, 3, 4]:
|
|
print("\nInvalid option. Please try again.")
|
|
continue
|
|
|
|
print_phase_menu()
|
|
phase_sel = input("\nSelect an option (1-3): ").strip()
|
|
|
|
match phase_sel:
|
|
case '1':
|
|
phase = "fase_V"
|
|
case '2':
|
|
phase = "fase_R"
|
|
case '3':
|
|
continue
|
|
case _:
|
|
print("\nInvalid option. Please try again.")
|
|
continue
|
|
|
|
print_data_menu()
|
|
data_sel = input("\nSelect dataset (1-6): ").strip()
|
|
|
|
test_AV = False
|
|
|
|
match data_sel:
|
|
case '1':
|
|
image_dir = os.path.join(data_dir, "SanDam")
|
|
case '2':
|
|
image_dir = os.path.join(data_dir, "GBIF")
|
|
case '3':
|
|
image_dir = os.path.join(data_dir, "AV")
|
|
case '4':
|
|
image_dir = os.path.join(data_dir, "combi")
|
|
case '5':
|
|
image_dir = os.path.join(data_dir, "combi")
|
|
test_AV = True
|
|
case '6':
|
|
continue
|
|
case _:
|
|
print("\nInvalid dataset option. Returning to main menu.")
|
|
continue
|
|
|
|
if verify_configuration():
|
|
if verify_paths(image_dir):
|
|
if model_sel == 1:
|
|
run_MobiNet(image_dir, test_AV, phase)
|
|
elif model_sel == 2:
|
|
run_EffiNet(image_dir, test_AV, phase)
|
|
elif model_sel == 3:
|
|
run_ResNet(image_dir, test_AV, phase)
|
|
elif model_sel == 4:
|
|
run_ViT(image_dir, test_AV, phase)
|
|
else:
|
|
print("\nError while launching model. Returning to main menu.")
|
|
continue
|
|
else:
|
|
print("\nDataset paths verification failed. Please check the dataset directory and CSV file.")
|
|
else:
|
|
print("\nThere are configuration issues. Please fix them before continuing.")
|
|
|
|
input("\nPress Enter to continue...")
|
|
|
|
if __name__ == '__main__':
|
|
main() |