import datetime import os import shutil import subprocess import tempfile import uuid import logging def set_env_variables(): if "DiffDock-Pocket-Dir" not in os.environ: work_dir = os.path.abspath(os.path.join("../DiffDock-Pocket")) if os.path.exists(work_dir): os.environ["DiffDock-Pocket-Dir"] = work_dir else: raise ValueError(f"DiffDock-Pocket-Dir {work_dir} not found") if "LOG_LEVEL" not in os.environ: os.environ["LOG_LEVEL"] = "INFO" def configure_logging(level=None): if level is None: level = getattr(logging, os.environ.get("LOG_LEVEL", "INFO")) # Note that this sets the universal logger, # which includes other libraries. logging.basicConfig( level=level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.StreamHandler(), # Outputs logs to stderr by default # If you also want to log to a file, uncomment the following line: # logging.FileHandler('my_app.log', mode='a', encoding='utf-8') ], ) def run_cli_command( protein_path: str, ligand: str, samples_per_complex: int, keep_local_structures: bool, save_visualisation: bool, work_dir=None, ): if work_dir is None: work_dir = os.environ.get( "DiffDock-Pocket-Dir", os.path.join(os.environ["HOME"], "DiffDock-Pocket") ) command = [ "python3", "inference.py", f"--protein_path={protein_path}", f"--ligand={ligand}", f"--samples_per_complex={samples_per_complex}", ] # Adding boolean arguments only if they are True if keep_local_structures: command.append("--keep_local_structures") if save_visualisation: command.append("--save_visualisation") with tempfile.TemporaryDirectory() as temp_dir: temp_dir_path = temp_dir command.append(f"--out_dir={temp_dir_path}") # Convert command list to string for printing command_str = " ".join(command) logging.info(f"Executing command: {command_str}") # Running the command try: result = subprocess.run( command, cwd=work_dir, check=False, text=True, capture_output=True, env=os.environ, ) logging.debug(f"Command output:\n{result.stdout}") if result.stderr: logging.error(f"Command error:\n{result.stderr}") except subprocess.CalledProcessError as e: logging.error(f"An error occurred while executing the command: {e}") # Zip the output directory # Generate a unique filename using a timestamp and a UUID timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") uuid_tag = str(uuid.uuid4())[0:8] unique_filename = f"output_{timestamp}_{uuid_tag}" zip_base_name = os.path.join("tmp", unique_filename) full_zip_path = shutil.make_archive(zip_base_name, "zip", temp_dir) logging.debug(f"Directory '{temp_dir}' zipped to '{full_zip_path}'") return full_zip_path if __name__ == "__main__": # Testing code set_env_variables() configure_logging() work_dir = os.path.abspath("../DiffDock-Pocket") os.environ["DiffDock-Pocket-Dir"] = work_dir protein_path = os.path.join(work_dir, "example_data", "3dpf_protein.pdb") ligand = os.path.join(work_dir, "example_data", "3dpf_ligand.sdf") run_cli_command( protein_path, ligand, samples_per_complex=1, keep_local_structures=True, save_visualisation=True, )