|
import os |
|
from pathlib import Path |
|
import logging |
|
import subprocess |
|
from tqdm import tqdm |
|
import sys |
|
|
|
|
|
logging.basicConfig( |
|
level=logging.INFO, |
|
format='%(asctime)s - %(levelname)s - %(message)s' |
|
) |
|
logger = logging.getLogger(__name__) |
|
|
|
def run_python_script(script_path, env_vars=None): |
|
"""Execute a single Python script with optional environment variables.""" |
|
try: |
|
logger.info(f"Running script: {script_path}") |
|
|
|
|
|
env = os.environ.copy() |
|
|
|
if env_vars: |
|
env.update(env_vars) |
|
|
|
|
|
|
|
|
|
|
|
completed = subprocess.run( |
|
['python', str(script_path)], |
|
env=env |
|
) |
|
|
|
if completed.returncode == 0: |
|
logger.info(f"Successfully executed: {script_path}") |
|
return True |
|
else: |
|
logger.error( |
|
f"Error executing {script_path} (return code: {completed.returncode})" |
|
) |
|
return False |
|
|
|
except Exception as e: |
|
logger.error(f"Error executing {script_path}: {str(e)}") |
|
return False |
|
|
|
def main(): |
|
|
|
preprocessing_dir = Path("preprocessing") |
|
if not preprocessing_dir.exists(): |
|
preprocessing_dir = Path(__file__).parent |
|
|
|
|
|
python_files = sorted([f for f in preprocessing_dir.glob("[0-9]*.py")]) |
|
|
|
logger.info(f"Found {len(python_files)} Python scripts to execute") |
|
|
|
|
|
results = [] |
|
for script_path in tqdm(python_files, desc="Executing scripts"): |
|
|
|
if "4_sciplex_SMILES" in script_path.name: |
|
|
|
success_true = run_python_script(script_path, {"LINCS_GENES": "true"}) |
|
results.append((f"{script_path.name} (LINCS_GENES=True)", success_true)) |
|
|
|
|
|
success_false = run_python_script(script_path, {"LINCS_GENES": "false"}) |
|
results.append((f"{script_path.name} (LINCS_GENES=False)", success_false)) |
|
else: |
|
|
|
success = run_python_script(script_path) |
|
results.append((script_path.name, success)) |
|
|
|
|
|
print("\nExecution Summary:") |
|
print("-----------------") |
|
for name, success in results: |
|
status = "β Success" if success else "β Failed" |
|
print(f"{name}: {status}") |
|
|
|
if __name__ == "__main__": |
|
main() |
|
|