Module deeporigin.src.pocket_finder

Classes

class PocketFinderClient

Expand source code

class PocketFinderClient(Client):
    """
    A client class for finding and analyzing protein pockets using a pocket finding service.
    This class provides functionality to identify potential binding pockets in protein structures,
    process the results, and visualize them. It supports both synchronous and asynchronous
    operations for pocket finding requests.

    Attributes:
        None

    Methods:
        get_protein_data_object(protein: Protein) -> dict:
        pocket_finder_request(logger: Logger, protein_data) -> tuple:
        update_progress_bar(logger: Logger, progress: ProgressView, body: dict):
        async_find_pockets(protein: Union[str, Protein], results_dir: Optional[str]) -> str:
            Asynchronously finds pockets in a protein structure.
        async_get_find_pockets_response(request_id: str) -> tuple:
        wait_for_pocket_finder_response(logger: Logger, request_id: str, progress_view: ProgressView, interval: float) -> tuple:
        process_pocket_finder_request(logger: Logger, protein: Protein) -> tuple:
            Process a request to find pockets in a protein structure.
        preprocess(protein, results_dir: str) -> Protein:
        find_pockets(protein: Union[str, Protein], results_dir: Optional[str]) -> Tuple[Protein, List[Pocket]]:
            Find pockets in a protein structure using PocketFinder algorithm.
        show_pockets(protein: Protein, pockets: List[Pocket], protein_surface_alpha: float, pocket_surface_alpha: float) -> str:

        >>> client = PocketFinderClient()
        >>> protein = Protein("example.pdb")
        >>> protein, pockets = client.find_pockets(protein, "results")
        >>> client.show_pockets(protein, pockets)
    """
    def __init__(self):
        super().__init__()

    def get_protein_data_object(self, protein: Protein):
        """
        Creates a dictionary containing protein data information.

        Args:
            protein (Protein): A Protein object containing block type and content.

        Returns:
            dict: A dictionary with two keys:
                - 'extension': The block type of the protein
                - 'content': The block content of the protein
        """
        return {"extension": protein.block_type, "content": protein.block_content}

    def pocket_finder_request(self, logger: Logger, protein_data):
        """
        Sends a request to find pockets in a protein structure.

        This method communicates with a pocket finding service by sending protein data and processing
        the response to identify potential binding pockets in the protein structure.

        Args:
            logger (Logger): Logger object to track the execution flow and record events
            protein_data: Protein structure data to analyze for pockets

        Returns:
            tuple: A tuple containing:
                - data: The processed pocket finder results if successful, None otherwise
                - success (bool): True if the request was successful, False otherwise

        Raises:
            May raise exceptions related to HTTP requests or JSON processing

        Notes:
            - Handles 503 errors specifically for worker availability issues
            - Uses both console logging and provided logger for different logging purposes
            - Maintains logging depth for hierarchical logging structure
        """
        console_logger = Logger("INFO", None)

        logger.log_info("PocketFinder Request - Preparing pocket finder request.")
        logger.add_depth()

        body = {"protein_data": protein_data}

        data, success = None, False
        try:
            logger.log_info("PocketFinder Request - Sending pocket finder request.")
            response = self.post_request(
                endpoint="pocket_finder",
                logger=logger,
                data=body,
            )

            logger.log_info("PocketFinder Request - Received pocket finder response.")
            logger.sub_depth()

            data, success = response.json(), True
        except Exception as e:
            if response.status_code == 503:
                console_logger.log_error(WARNING_NUMBER_OF_AVAILABLE_WORKERS)
            else:
                console_logger.log_warning(WARNING_MESSAGE)

            logger.log_error(
                f"PocketFinder Request - Unexpected status code: {response.status_code}. \nMessage: {response.text}"
            )
            logger.sub_depth()

            data, success = None, False

        return data, success

    def update_progress_bar(self, logger: Logger, progress: ProgressView, body):
        """
        Updates the progress bar based on the received response body.

        Args:
            logger (Logger): Logger instance to log progress information
            progress (ProgressView): Progress bar view instance to be updated
            body (dict): Response body containing progress information with keys:
                - description (str): Description of current progress state
                - percentage (int/float): Percentage completion value

        Updates the progress bar's description and percentage based on the body content.
        When "Completed" status is received, updates with completion values.
        For ongoing progress, only updates description if different from current,
        and logs when progress increases.
        """
        if body.get("description", "") == "Completed":
            progress.update(new_desc=body["description"], new_percentage=body["percentage"])
            progress.display()
        else:
            new_percentage = int(body["percentage"])
            new_description = body["description"]

            if new_description == progress.description:
                new_description = None

            if new_percentage > progress.bar.n:
                logger.log_info("PocketFinder progress achieved.")

            progress.update(new_desc=new_description, new_percentage=new_percentage)
            progress.display()

    def async_find_pockets(self, protein: Union[str, Protein], results_dir: Optional[str] = ""):
        """
        Asynchronously finds pockets in a protein structure using the pocket finder service.

        Args:
            protein (Union[str, Protein]): Either a path to a protein structure file or a Protein object.
            results_dir (Optional[str], optional): Directory to store results. Defaults to "".

        Returns:
            str: Request ID for tracking the asynchronous pocket finding process.

        Raises:
            TypeError: If protein parameter is neither a string path nor a Protein object.
        """
        if not isinstance(protein, Protein):
            protein = Protein(file_path=protein)

        logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))

        protein = self.preprocess(protein, results_dir)
        protein_data = self.get_protein_data_object(protein)

        request_id, _ = self.pocket_finder_request(logger, protein_data)
        return request_id

    def async_get_find_pockets_response(self, request_id: str):
        """
        Asynchronously retrieves the response for a pocket finding request.

        This method queries the server for the status and results of a previously submitted
        pocket finding job identified by the request_id. It interprets various HTTP status
        codes to determine the current state of the request.

        Args:
            request_id (str): The unique identifier for the pocket finding request.

        Returns:
            tuple: A tuple containing two elements:
                - First element: JSON response data if successful (status 200 or 202), None if failed
                - Second element: String indicating status ("Completed", "Running", or "Error")

        Status Codes:
            - 200: Request completed successfully
            - 202: Request is still processing
            - 429: Rate limit exceeded
            - Other: Unexpected error

        Example:
            >>> response_data, status = client.async_get_find_pockets_response("request123")
            >>> if status == "Completed":
            ...     process_results(response_data)
        """
        response = None
        logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))

        response = self.get_request(endpoint=f"pocket_finder/{request_id}", logger=logger)

        match response.status_code:
            case 200:
                logger.log_info(f"PocketFinder request successfully completed: status_code = {response.status_code}")

                return response.json(), "Completed"
            case 202:
                return response.json(), "Running"
            case 429:
                logger.log_warning(WARNING_MESSAGE)
                logger.log_error(
                    f"PocketFinder request failed: status_code = {response.status_code}, message: {response.text}"
                )
                return None, "Error"
            case _:
                logger.log_warning(WARNING_MESSAGE)
                logger.log_error("PocketFinder request unexpected status code.")
                return None, "Error"

    def wait_for_pocket_finder_response(self, logger: Logger, request_id: str, progress_view: ProgressView, interval=0.5):
        """
        Waits for and handles responses from the pocket finder service with progress updates.

        This method continuously polls the pocket finder service until a final response is received,
        updating a progress bar during the process.

        Args:
            logger (Logger): Logger instance for tracking the process
            request_id (str): Unique identifier for the pocket finder request
            progress_view (ProgressView): Progress bar view instance to update status
            interval (float, optional): Polling interval in seconds. Defaults to 0.5

        Returns:
            tuple: A tuple containing:
                - data (dict|None): Response data if successful, None otherwise
                - success (bool): True if request completed successfully, False otherwise

        Status Codes:
            - 200: Request completed successfully
            - 202: Request is still processing
            - 429: Rate limit exceeded
            - Other: Unexpected error

        Note:
            The method will update the progress bar during processing and close it upon completion.
            It also handles logging at different levels based on the response status.
        """
        logger.add_depth()

        console_logger = Logger("INFO", None)
        logger.log_info("Waiting for pocket_finder service.")

        data, success = None, False
        while True:
            time.sleep(interval)
            response = self.get_request(endpoint=f"pocket_finder/{request_id}", logger=logger)

            match response.status_code:
                case 200:
                    logger.log_info(
                        f"PocketFinder request successfully completed: status_code = {response.status_code}"
                    )

                    data, success = response.json(), True
                    break
                case 202:
                    body = response.json()
                    self.update_progress_bar(logger, progress_view, body)
                case 429:
                    console_logger.log_warning(WARNING_MESSAGE)
                    logger.log_error(
                        f"PocketFinder request failed: status_code = {response.status_code}, message: {response.text}"
                    )

                    data, success = None, False
                    break
                case _:
                    console_logger.log_warning(WARNING_MESSAGE)
                    logger.log_error("PocketFinder request unexpected status code.")

                    data, success = None, False
                    break

        if success:
            self.update_progress_bar(logger, progress_view, {"description": "Completed", "percentage": 100})
            progress_view.close()

            logger.log_info("PocketFinder completed: Proceeding to the next step")
        logger.sub_depth()

        return data, success

    def process_pocket_finder_request(self, logger: Logger, protein: Protein):
        """Process a request to find pockets in a protein structure.

        This method handles the complete workflow of pocket finding, including making the initial request
        and waiting for the response from the pocket finder service.

        Args:
            logger (Logger): Logger instance to track the process and record any issues
            protein (Protein): Protein object containing the structure to analyze

        Returns:
            tuple: A tuple containing:
                - response: The server response containing pocket information if successful, None otherwise
                - completed (bool): Flag indicating if the process completed successfully

        Raises:
            Exception: If there are server-side errors (HTTP 503 for no available workers, or other server issues)

        Note:
            The method handles server errors internally, logging appropriate warnings and errors.
            It uses a ProgressView to track the request progress and will return (None, False)
            if any errors occur during processing.
        """
        console_logger = Logger("INFO", None)

        response, completed = None, False

        logger.log_info("PocketFinder: Starting pocket finder.")
        logger.add_depth()

        protein_data = self.get_protein_data_object(protein)

        progress_view = ProgressView()

        try:
            request_id, completed = self.pocket_finder_request(logger, protein_data)
            if completed:
                response, completed = self.wait_for_pocket_finder_response(logger, request_id, progress_view)
                if not completed:
                    console_logger.log_warning(WARNING_MESSAGE_SERVER)
                    logger.log_error(
                        "PocketFinder: Something went wrong. For more information please check backend service logs."
                    )
        except Exception as e:
            if response.status_code == 503:
                console_logger.log_error(WARNING_NUMBER_OF_AVAILABLE_WORKERS)
            else:
                console_logger.log_warning(WARNING_MESSAGE_SERVER)

            logger.log_error(f"Something went wrong on server side - {str(e)}")
            response, completed = None, False

        return response, completed

    def preprocess(self, protein, results_dir=""):
        """
        Preprocesses the input protein data for pocket finding.

        This method validates the results directory and creates it if necessary, and ensures
        the input protein is in the correct format (either a Protein object or a valid protein file path).

        Args:
            protein (Union[Protein, str]): Either a Protein object or a path to a protein file.
            results_dir (str, optional): Directory path for storing results. Defaults to empty string.

        Returns:
            Protein: A valid Protein object ready for pocket finding analysis.

        Raises:
            Exception: If results_dir is non-empty.
            Exception: If protein is neither a Protein object nor a valid file path.
            Exception: If protein file cannot be processed into a Protein object.
        """
        if len(results_dir) > 0:
            if os.path.exists(results_dir) and len(os.listdir(results_dir)) > 0:
                raise Exception("Non empty directory is given as results directory.")

            if not os.path.exists(results_dir):
                os.makedirs(results_dir)

        if not isinstance(protein, Protein):
            if not isinstance(protein, str):
                raise Exception("Find pockets can only be performed either on protein object or protein file.")

            try:
                protein = Protein(file_path=protein)
            except Exception as e:
                raise Exception(f"Failed to create protein object from the given file. \n Exception: {str(e)}")

        return protein

    def find_pockets(self, protein: Union[str, Protein], results_dir: Optional[str] = "pocket_results") -> Tuple[Protein, List[Pocket]]:
        """Find pockets in a protein structure using PocketFinder algorithm.
        This method processes a protein structure to identify and analyze potential binding pockets,
        creates visualization files, and generates a comprehensive report of the findings.
        Args:
            protein (Union[str, Protein]): Input protein structure, either as a file path string or Protein object
            results_dir (Optional[str]): Directory to store pocket finder results (default: "pocket_results")
        Returns:
            Tuple[Protein, List[Pocket]]: A tuple containing:
                - The processed protein object
                - A PocketFinderReport object containing the identified pockets and their properties
        Raises:
            Exception: If there is an error during pocket finding or result processing
        Notes:
            - Creates a unique directory for each protein analysis
            - Converts pocket atoms to Xenon for visualization purposes
            - Generates PDB files for each identified pocket
            - Creates a properties CSV file with pocket metrics
            - Uses custom color schemes for pocket visualization
        """
        console_logger = Logger("INFO", None)
        logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))
        logger.log_info("Executing pocket finder request.")

        base_path = Path(str(results_dir)) / protein.file_path.stem
        if not base_path.exists():
            base_path.mkdir(parents=True)
        else:
            number = len(list(Path(str(results_dir)).glob(f"{protein.file_path.stem}*")))
            base_path = Path(str(results_dir)) / f"{protein.file_path.stem}_#{number}"
            base_path.mkdir(parents=True)

        protein = self.preprocess(protein, str(base_path))
        props_file_csv = base_path / f"{protein.file_path.stem}_pocket_finder_props.csv"
        pocket_report = PocketFinderReport(protein, props_file_csv)
        pocket_dicts, completed = self.process_pocket_finder_request(logger, protein)
        if completed:
            try:
                logger.log_info("Parsing pocket finder results.")

                surface_colors = ProteinViewer("", format="pdb").get_pocket_visualization_config().surface_colors
                for index, pocket_dict in enumerate(pocket_dicts):
                    color = surface_colors[index % len(surface_colors)]
                    pocket = Pocket(
                        color=color,
                        block_type="pdb",
                        index=index,
                        name=f"{protein.file_path.stem}_pocket_{index}_color_{color}",
                        block_content=pocket_dict["content"],
                        props=pocket_dict.get("props", None),
                    )

                    pocket_path = base_path / f"{pocket.name}.pdb"

                    atoms = []
                    for atom in pocket.structure:
                        atom.atom_name = "Xe"
                        atom.element = "Xe"
                        atoms.append(atom)

                    pocket.structure = struc.array(atoms)
                    pocket.write_to_file(str(pocket_path))
                    pocket_report.add_pocket(pocket)

                pocket_report.save_props()
                logger.log_info("PocketFinder request successfully completed.")
            except Exception as e:
                error = str(e)
                logger.log_error("Something went wrong - " + error)
                console_logger.log_error(f"Something went wrong - Error: {error}")
        else:
            logger.log_error("Pocket finder terminated.")

        return protein, pocket_report


    @classmethod
    @jupyter_visualization
    def show_pockets(cls, protein: Protein, pockets: List[Pocket], protein_surface_alpha: float = 0, pocket_surface_alpha: float = 0.7):
        """
        Visualizes protein and its pockets in a 3D viewer.
        This class method creates an interactive 3D visualization of a protein structure
        along with its identified pockets using ProteinViewer.
        Args:
            protein (Protein): The protein object containing structural information.
            pockets (List[Pocket]): List of pocket objects to be visualized.
            protein_surface_alpha (float, optional): Transparency of the protein surface.
                0 is fully transparent, 1 is fully opaque. Defaults to 0.
            pocket_surface_alpha (float, optional): Transparency of the pocket surfaces.
                0 is fully transparent, 1 is fully opaque. Defaults to 0.7.
        Returns:
            str: HTML content containing the interactive 3D visualization that can be
                displayed in a web browser or notebook environment.
        Example:
            >>> protein = Protein("1abc.pdb")
            >>> pockets = [Pocket1, Pocket2]
            >>> html = PocketFinder.show_pockets(protein, pockets)
        """
        protein_viewer = ProteinViewer(data=str(protein.file_path), format=str(protein.block_type))
        pocket_paths = [str(pocket.file_path) for pocket in pockets]

        # Retrieve and customize pocket visualization configuration
        pocket_config = protein_viewer.get_pocket_visualization_config()
        pocket_config.surface_alpha = pocket_surface_alpha

        protein_config = protein_viewer.get_protein_visualization_config()
        protein_config.style_type = "cartoon"
        protein_config.surface_alpha = protein_surface_alpha
        pocket_config.surface_colors = [pocket.color for pocket in pockets]

        # Render the protein with pockets
        html_content = protein_viewer.render_protein_with_pockets(
            pocket_paths=pocket_paths, pocket_config=pocket_config, protein_config=protein_config
        )

        # Display the rendered HTML content
        return html_content

A client class for finding and analyzing protein pockets using a pocket finding service. This class provides functionality to identify potential binding pockets in protein structures, process the results, and visualize them. It supports both synchronous and asynchronous operations for pocket finding requests.

Attributes

None

Methods

get_protein_data_object(protein: Protein) -> dict: pocket_finder_request(logger: Logger, protein_data) -> tuple: update_progress_bar(logger: Logger, progress: ProgressView, body: dict): async_find_pockets(protein: Union[str, Protein], results_dir: Optional[str]) -> str: Asynchronously finds pockets in a protein structure. async_get_find_pockets_response(request_id: str) -> tuple: wait_for_pocket_finder_response(logger: Logger, request_id: str, progress_view: ProgressView, interval: float) -> tuple: process_pocket_finder_request(logger: Logger, protein: Protein) -> tuple: Process a request to find pockets in a protein structure. preprocess(protein, results_dir: str) -> Protein: find_pockets(protein: Union[str, Protein], results_dir: Optional[str]) -> Tuple[Protein, List[Pocket]]: Find pockets in a protein structure using PocketFinder algorithm. show_pockets(protein: Protein, pockets: List[Pocket], protein_surface_alpha: float, pocket_surface_alpha: float) -> str:

>>> client = PocketFinderClient()
>>> protein = Protein("example.pdb")
>>> protein, pockets = client.find_pockets(protein, "results")
>>> client.show_pockets(protein, pockets)

Ancestors

• Client

Static methods

def show_pockets(*args, **kwargs)

Methods

def async_find_pockets(self,
protein: str | Protein,
results_dir: str | None = '')

Expand source code

def async_find_pockets(self, protein: Union[str, Protein], results_dir: Optional[str] = ""):
    """
    Asynchronously finds pockets in a protein structure using the pocket finder service.

    Args:
        protein (Union[str, Protein]): Either a path to a protein structure file or a Protein object.
        results_dir (Optional[str], optional): Directory to store results. Defaults to "".

    Returns:
        str: Request ID for tracking the asynchronous pocket finding process.

    Raises:
        TypeError: If protein parameter is neither a string path nor a Protein object.
    """
    if not isinstance(protein, Protein):
        protein = Protein(file_path=protein)

    logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))

    protein = self.preprocess(protein, results_dir)
    protein_data = self.get_protein_data_object(protein)

    request_id, _ = self.pocket_finder_request(logger, protein_data)
    return request_id

Asynchronously finds pockets in a protein structure using the pocket finder service.

Args

protein : Union[str, Protein]

Either a path to a protein structure file or a Protein object.

results_dir : Optional[str], optional

Directory to store results. Defaults to "".

Returns

str

Request ID for tracking the asynchronous pocket finding process.

Raises

TypeError

If protein parameter is neither a string path nor a Protein object.

def async_get_find_pockets_response(self, request_id: str)

Expand source code

def async_get_find_pockets_response(self, request_id: str):
    """
    Asynchronously retrieves the response for a pocket finding request.

    This method queries the server for the status and results of a previously submitted
    pocket finding job identified by the request_id. It interprets various HTTP status
    codes to determine the current state of the request.

    Args:
        request_id (str): The unique identifier for the pocket finding request.

    Returns:
        tuple: A tuple containing two elements:
            - First element: JSON response data if successful (status 200 or 202), None if failed
            - Second element: String indicating status ("Completed", "Running", or "Error")

    Status Codes:
        - 200: Request completed successfully
        - 202: Request is still processing
        - 429: Rate limit exceeded
        - Other: Unexpected error

    Example:
        >>> response_data, status = client.async_get_find_pockets_response("request123")
        >>> if status == "Completed":
        ...     process_results(response_data)
    """
    response = None
    logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))

    response = self.get_request(endpoint=f"pocket_finder/{request_id}", logger=logger)

    match response.status_code:
        case 200:
            logger.log_info(f"PocketFinder request successfully completed: status_code = {response.status_code}")

            return response.json(), "Completed"
        case 202:
            return response.json(), "Running"
        case 429:
            logger.log_warning(WARNING_MESSAGE)
            logger.log_error(
                f"PocketFinder request failed: status_code = {response.status_code}, message: {response.text}"
            )
            return None, "Error"
        case _:
            logger.log_warning(WARNING_MESSAGE)
            logger.log_error("PocketFinder request unexpected status code.")
            return None, "Error"

Asynchronously retrieves the response for a pocket finding request.

This method queries the server for the status and results of a previously submitted pocket finding job identified by the request_id. It interprets various HTTP status codes to determine the current state of the request.

Args

request_id : str

The unique identifier for the pocket finding request.

Returns

tuple

A tuple containing two elements: - First element: JSON response data if successful (status 200 or 202), None if failed - Second element: String indicating status ("Completed", "Running", or "Error")

Status Codes: - 200: Request completed successfully - 202: Request is still processing - 429: Rate limit exceeded - Other: Unexpected error

Example

>>> response_data, status = client.async_get_find_pockets_response("request123")
>>> if status == "Completed":
...     process_results(response_data)

def find_pockets(self,
protein: str | Protein,
results_dir: str | None = 'pocket_results') ‑> Tuple[Protein, List[Pocket]]

Expand source code

def find_pockets(self, protein: Union[str, Protein], results_dir: Optional[str] = "pocket_results") -> Tuple[Protein, List[Pocket]]:
    """Find pockets in a protein structure using PocketFinder algorithm.
    This method processes a protein structure to identify and analyze potential binding pockets,
    creates visualization files, and generates a comprehensive report of the findings.
    Args:
        protein (Union[str, Protein]): Input protein structure, either as a file path string or Protein object
        results_dir (Optional[str]): Directory to store pocket finder results (default: "pocket_results")
    Returns:
        Tuple[Protein, List[Pocket]]: A tuple containing:
            - The processed protein object
            - A PocketFinderReport object containing the identified pockets and their properties
    Raises:
        Exception: If there is an error during pocket finding or result processing
    Notes:
        - Creates a unique directory for each protein analysis
        - Converts pocket atoms to Xenon for visualization purposes
        - Generates PDB files for each identified pocket
        - Creates a properties CSV file with pocket metrics
        - Uses custom color schemes for pocket visualization
    """
    console_logger = Logger("INFO", None)
    logger = Logger("INFO", os.getenv("LOG_BIOSIM_CLIENT"))
    logger.log_info("Executing pocket finder request.")

    base_path = Path(str(results_dir)) / protein.file_path.stem
    if not base_path.exists():
        base_path.mkdir(parents=True)
    else:
        number = len(list(Path(str(results_dir)).glob(f"{protein.file_path.stem}*")))
        base_path = Path(str(results_dir)) / f"{protein.file_path.stem}_#{number}"
        base_path.mkdir(parents=True)

    protein = self.preprocess(protein, str(base_path))
    props_file_csv = base_path / f"{protein.file_path.stem}_pocket_finder_props.csv"
    pocket_report = PocketFinderReport(protein, props_file_csv)
    pocket_dicts, completed = self.process_pocket_finder_request(logger, protein)
    if completed:
        try:
            logger.log_info("Parsing pocket finder results.")

            surface_colors = ProteinViewer("", format="pdb").get_pocket_visualization_config().surface_colors
            for index, pocket_dict in enumerate(pocket_dicts):
                color = surface_colors[index % len(surface_colors)]
                pocket = Pocket(
                    color=color,
                    block_type="pdb",
                    index=index,
                    name=f"{protein.file_path.stem}_pocket_{index}_color_{color}",
                    block_content=pocket_dict["content"],
                    props=pocket_dict.get("props", None),
                )

                pocket_path = base_path / f"{pocket.name}.pdb"

                atoms = []
                for atom in pocket.structure:
                    atom.atom_name = "Xe"
                    atom.element = "Xe"
                    atoms.append(atom)

                pocket.structure = struc.array(atoms)
                pocket.write_to_file(str(pocket_path))
                pocket_report.add_pocket(pocket)

            pocket_report.save_props()
            logger.log_info("PocketFinder request successfully completed.")
        except Exception as e:
            error = str(e)
            logger.log_error("Something went wrong - " + error)
            console_logger.log_error(f"Something went wrong - Error: {error}")
    else:
        logger.log_error("Pocket finder terminated.")

    return protein, pocket_report

Find pockets in a protein structure using PocketFinder algorithm. This method processes a protein structure to identify and analyze potential binding pockets, creates visualization files, and generates a comprehensive report of the findings.

Args

protein : Union[str, Protein]

Input protein structure, either as a file path string or Protein object

results_dir : Optional[str]

Directory to store pocket finder results (default: "pocket_results")

Returns

Tuple[Protein, List[Pocket]]

A tuple containing: - The processed protein object - A PocketFinderReport object containing the identified pockets and their properties

Raises

Exception

If there is an error during pocket finding or result processing

Notes

• Creates a unique directory for each protein analysis
• Converts pocket atoms to Xenon for visualization purposes
• Generates PDB files for each identified pocket
• Creates a properties CSV file with pocket metrics
• Uses custom color schemes for pocket visualization

def get_protein_data_object(self,
protein: Protein)

Expand source code

def get_protein_data_object(self, protein: Protein):
    """
    Creates a dictionary containing protein data information.

    Args:
        protein (Protein): A Protein object containing block type and content.

    Returns:
        dict: A dictionary with two keys:
            - 'extension': The block type of the protein
            - 'content': The block content of the protein
    """
    return {"extension": protein.block_type, "content": protein.block_content}

Creates a dictionary containing protein data information.

Args

protein : Protein

A Protein object containing block type and content.

Returns

dict

A dictionary with two keys: - 'extension': The block type of the protein - 'content': The block content of the protein

def pocket_finder_request(self,
logger: Logger,
protein_data)

Expand source code

def pocket_finder_request(self, logger: Logger, protein_data):
    """
    Sends a request to find pockets in a protein structure.

    This method communicates with a pocket finding service by sending protein data and processing
    the response to identify potential binding pockets in the protein structure.

    Args:
        logger (Logger): Logger object to track the execution flow and record events
        protein_data: Protein structure data to analyze for pockets

    Returns:
        tuple: A tuple containing:
            - data: The processed pocket finder results if successful, None otherwise
            - success (bool): True if the request was successful, False otherwise

    Raises:
        May raise exceptions related to HTTP requests or JSON processing

    Notes:
        - Handles 503 errors specifically for worker availability issues
        - Uses both console logging and provided logger for different logging purposes
        - Maintains logging depth for hierarchical logging structure
    """
    console_logger = Logger("INFO", None)

    logger.log_info("PocketFinder Request - Preparing pocket finder request.")
    logger.add_depth()

    body = {"protein_data": protein_data}

    data, success = None, False
    try:
        logger.log_info("PocketFinder Request - Sending pocket finder request.")
        response = self.post_request(
            endpoint="pocket_finder",
            logger=logger,
            data=body,
        )

        logger.log_info("PocketFinder Request - Received pocket finder response.")
        logger.sub_depth()

        data, success = response.json(), True
    except Exception as e:
        if response.status_code == 503:
            console_logger.log_error(WARNING_NUMBER_OF_AVAILABLE_WORKERS)
        else:
            console_logger.log_warning(WARNING_MESSAGE)

        logger.log_error(
            f"PocketFinder Request - Unexpected status code: {response.status_code}. \nMessage: {response.text}"
        )
        logger.sub_depth()

        data, success = None, False

    return data, success

Sends a request to find pockets in a protein structure.

This method communicates with a pocket finding service by sending protein data and processing the response to identify potential binding pockets in the protein structure.

Args

logger : Logger

Logger object to track the execution flow and record events

protein_data

Protein structure data to analyze for pockets

Returns

tuple

A tuple containing: - data: The processed pocket finder results if successful, None otherwise - success (bool): True if the request was successful, False otherwise

Raises

May raise exceptions related to HTTP requests or JSON processing

Notes

• Handles 503 errors specifically for worker availability issues
• Uses both console logging and provided logger for different logging purposes
• Maintains logging depth for hierarchical logging structure

def preprocess(self, protein, results_dir='')

Expand source code

def preprocess(self, protein, results_dir=""):
    """
    Preprocesses the input protein data for pocket finding.

    This method validates the results directory and creates it if necessary, and ensures
    the input protein is in the correct format (either a Protein object or a valid protein file path).

    Args:
        protein (Union[Protein, str]): Either a Protein object or a path to a protein file.
        results_dir (str, optional): Directory path for storing results. Defaults to empty string.

    Returns:
        Protein: A valid Protein object ready for pocket finding analysis.

    Raises:
        Exception: If results_dir is non-empty.
        Exception: If protein is neither a Protein object nor a valid file path.
        Exception: If protein file cannot be processed into a Protein object.
    """
    if len(results_dir) > 0:
        if os.path.exists(results_dir) and len(os.listdir(results_dir)) > 0:
            raise Exception("Non empty directory is given as results directory.")

        if not os.path.exists(results_dir):
            os.makedirs(results_dir)

    if not isinstance(protein, Protein):
        if not isinstance(protein, str):
            raise Exception("Find pockets can only be performed either on protein object or protein file.")

        try:
            protein = Protein(file_path=protein)
        except Exception as e:
            raise Exception(f"Failed to create protein object from the given file. \n Exception: {str(e)}")

    return protein

Preprocesses the input protein data for pocket finding.

This method validates the results directory and creates it if necessary, and ensures the input protein is in the correct format (either a Protein object or a valid protein file path).

Args

protein : Union[Protein, str]

Either a Protein object or a path to a protein file.

results_dir : str, optional

Directory path for storing results. Defaults to empty string.

Returns

Protein

A valid Protein object ready for pocket finding analysis.

Raises

Exception

If results_dir is non-empty.

Exception

If protein is neither a Protein object nor a valid file path.

Exception

If protein file cannot be processed into a Protein object.

def process_pocket_finder_request(self,
logger: Logger,
protein: Protein)

Expand source code

def process_pocket_finder_request(self, logger: Logger, protein: Protein):
    """Process a request to find pockets in a protein structure.

    This method handles the complete workflow of pocket finding, including making the initial request
    and waiting for the response from the pocket finder service.

    Args:
        logger (Logger): Logger instance to track the process and record any issues
        protein (Protein): Protein object containing the structure to analyze

    Returns:
        tuple: A tuple containing:
            - response: The server response containing pocket information if successful, None otherwise
            - completed (bool): Flag indicating if the process completed successfully

    Raises:
        Exception: If there are server-side errors (HTTP 503 for no available workers, or other server issues)

    Note:
        The method handles server errors internally, logging appropriate warnings and errors.
        It uses a ProgressView to track the request progress and will return (None, False)
        if any errors occur during processing.
    """
    console_logger = Logger("INFO", None)

    response, completed = None, False

    logger.log_info("PocketFinder: Starting pocket finder.")
    logger.add_depth()

    protein_data = self.get_protein_data_object(protein)

    progress_view = ProgressView()

    try:
        request_id, completed = self.pocket_finder_request(logger, protein_data)
        if completed:
            response, completed = self.wait_for_pocket_finder_response(logger, request_id, progress_view)
            if not completed:
                console_logger.log_warning(WARNING_MESSAGE_SERVER)
                logger.log_error(
                    "PocketFinder: Something went wrong. For more information please check backend service logs."
                )
    except Exception as e:
        if response.status_code == 503:
            console_logger.log_error(WARNING_NUMBER_OF_AVAILABLE_WORKERS)
        else:
            console_logger.log_warning(WARNING_MESSAGE_SERVER)

        logger.log_error(f"Something went wrong on server side - {str(e)}")
        response, completed = None, False

    return response, completed

Process a request to find pockets in a protein structure.

This method handles the complete workflow of pocket finding, including making the initial request and waiting for the response from the pocket finder service.

Args

logger : Logger

Logger instance to track the process and record any issues

protein : Protein

Protein object containing the structure to analyze

Returns

tuple

A tuple containing: - response: The server response containing pocket information if successful, None otherwise - completed (bool): Flag indicating if the process completed successfully

Raises

Exception

If there are server-side errors (HTTP 503 for no available workers, or other server issues)

Note

The method handles server errors internally, logging appropriate warnings and errors. It uses a ProgressView to track the request progress and will return (None, False) if any errors occur during processing.

def update_progress_bar(self,
logger: Logger,
progress: ProgressView,
body)

Expand source code

def update_progress_bar(self, logger: Logger, progress: ProgressView, body):
    """
    Updates the progress bar based on the received response body.

    Args:
        logger (Logger): Logger instance to log progress information
        progress (ProgressView): Progress bar view instance to be updated
        body (dict): Response body containing progress information with keys:
            - description (str): Description of current progress state
            - percentage (int/float): Percentage completion value

    Updates the progress bar's description and percentage based on the body content.
    When "Completed" status is received, updates with completion values.
    For ongoing progress, only updates description if different from current,
    and logs when progress increases.
    """
    if body.get("description", "") == "Completed":
        progress.update(new_desc=body["description"], new_percentage=body["percentage"])
        progress.display()
    else:
        new_percentage = int(body["percentage"])
        new_description = body["description"]

        if new_description == progress.description:
            new_description = None

        if new_percentage > progress.bar.n:
            logger.log_info("PocketFinder progress achieved.")

        progress.update(new_desc=new_description, new_percentage=new_percentage)
        progress.display()

Updates the progress bar based on the received response body.

Args

logger : Logger

Logger instance to log progress information

progress : ProgressView

Progress bar view instance to be updated

body : dict

Response body containing progress information with keys: - description (str): Description of current progress state - percentage (int/float): Percentage completion value

Updates the progress bar's description and percentage based on the body content. When "Completed" status is received, updates with completion values. For ongoing progress, only updates description if different from current, and logs when progress increases.

def wait_for_pocket_finder_response(self,
logger: Logger,
request_id: str,
progress_view: ProgressView,
interval=0.5)

Expand source code

def wait_for_pocket_finder_response(self, logger: Logger, request_id: str, progress_view: ProgressView, interval=0.5):
    """
    Waits for and handles responses from the pocket finder service with progress updates.

    This method continuously polls the pocket finder service until a final response is received,
    updating a progress bar during the process.

    Args:
        logger (Logger): Logger instance for tracking the process
        request_id (str): Unique identifier for the pocket finder request
        progress_view (ProgressView): Progress bar view instance to update status
        interval (float, optional): Polling interval in seconds. Defaults to 0.5

    Returns:
        tuple: A tuple containing:
            - data (dict|None): Response data if successful, None otherwise
            - success (bool): True if request completed successfully, False otherwise

    Status Codes:
        - 200: Request completed successfully
        - 202: Request is still processing
        - 429: Rate limit exceeded
        - Other: Unexpected error

    Note:
        The method will update the progress bar during processing and close it upon completion.
        It also handles logging at different levels based on the response status.
    """
    logger.add_depth()

    console_logger = Logger("INFO", None)
    logger.log_info("Waiting for pocket_finder service.")

    data, success = None, False
    while True:
        time.sleep(interval)
        response = self.get_request(endpoint=f"pocket_finder/{request_id}", logger=logger)

        match response.status_code:
            case 200:
                logger.log_info(
                    f"PocketFinder request successfully completed: status_code = {response.status_code}"
                )

                data, success = response.json(), True
                break
            case 202:
                body = response.json()
                self.update_progress_bar(logger, progress_view, body)
            case 429:
                console_logger.log_warning(WARNING_MESSAGE)
                logger.log_error(
                    f"PocketFinder request failed: status_code = {response.status_code}, message: {response.text}"
                )

                data, success = None, False
                break
            case _:
                console_logger.log_warning(WARNING_MESSAGE)
                logger.log_error("PocketFinder request unexpected status code.")

                data, success = None, False
                break

    if success:
        self.update_progress_bar(logger, progress_view, {"description": "Completed", "percentage": 100})
        progress_view.close()

        logger.log_info("PocketFinder completed: Proceeding to the next step")
    logger.sub_depth()

    return data, success

Waits for and handles responses from the pocket finder service with progress updates.

This method continuously polls the pocket finder service until a final response is received, updating a progress bar during the process.

Args

logger : Logger

Logger instance for tracking the process

request_id : str

Unique identifier for the pocket finder request

progress_view : ProgressView

Progress bar view instance to update status

interval : float, optional

Polling interval in seconds. Defaults to 0.5

Returns

tuple

A tuple containing: - data (dict|None): Response data if successful, None otherwise - success (bool): True if request completed successfully, False otherwise

Status Codes: - 200: Request completed successfully - 202: Request is still processing - 429: Rate limit exceeded - Other: Unexpected error

Note

The method will update the progress bar during processing and close it upon completion. It also handles logging at different levels based on the response status.

Inherited members

• Client:
  • get_request
  • get_url
  • post_request