Source code for OCDocker.Console.session

#!/usr/bin/env python3

# Description
###############################################################################
'''
Console session helpers: namespace construction, REPL loop, and utilities.

Builds the preloaded interactive namespace (docking modules, Initialise
symbols, ``print_args``, ``clean_test_files``) and runs the command loop with
tab-completion and history.
'''

from __future__ import annotations

# Imports
###############################################################################
import code
import inspect
import os
import shutil
import sys
from glob import glob
from pprint import pprint
from typing import Any, Mapping

import OCDocker.Error as ocerror

from OCDocker.Console.commands import handle_command

# License
###############################################################################
'''Copyright (c) Federal University of Rio de Janeiro (UFRJ), Artur Duque Rossi, and Pedro Henrique Monteiro Torres.

SPDX-License-Identifier: BSD-3-Clause

See the LICENSE file for full terms.
'''

# Functions
###############################################################################
## Private ##


def _setup_readline(namespace: Mapping[str, Any]) -> None:
    '''Configure readline tab-completion and load command history.'''

    try:
        import readline
        import rlcompleter

        completer = rlcompleter.Completer(dict(namespace))
        readline.set_completer(completer.complete)
        readline.parse_and_bind("tab: complete")
        hist = os.path.expanduser("~/.ocdocker_console_history")
        try:
            readline.read_history_file(hist)
        except (OSError, FileNotFoundError):
            pass
    except (ImportError, AttributeError):
        pass


def _save_readline_history() -> None:
    '''Persist readline command history (best-effort).'''

    try:
        if "readline" in sys.modules:
            sys.modules["readline"].write_history_file(os.path.expanduser("~/.ocdocker_console_history"))
    except (OSError, AttributeError):
        pass


## Public ##




[docs]
def clean_test_files(baseProtPath: str, baseLigPath: str, baseDecPath: str, baseCanPath: str) -> None:
    '''Reset the test_files folder to its original state.

    Parameters
    ----------
    baseProtPath : str
        Path to the base protein folder.
    baseLigPath : str
        Path to the base ligand folder.
    baseDecPath : str
        Path to the base decoy folder.
    baseCanPath : str
        Path to the base candidates folder.
    '''

    for f in glob(f"{baseProtPath}/*"):
        if os.path.isfile(f) and not f.endswith(f"{baseProtPath}/receptor.pdb"):
            os.remove(f)

    for ligFolder in [baseLigPath, baseDecPath, baseCanPath]:
        for f in glob(f"{ligFolder}/*/*"):
            if os.path.isfile(f) and not f.endswith("ligand.smi"):
                os.remove(f)
            elif os.path.isdir(f) and not f.endswith("boxes"):
                shutil.rmtree(f)

    return None










[docs]
def build_namespace() -> dict[str, Any]:
    '''Build the interactive console namespace after environment bootstrap.

    Returns
    -------
    dict[str, Any]
        Mapping of names exposed in the REPL (Initialise symbols, docking
        modules, helpers).
    '''

    namespace: dict[str, Any] = {}

    import OCDocker.Initialise as ocinit

    for key, value in vars(ocinit).items():
        if not key.startswith("__"):
            namespace[key] = value

    import OCDocker.Docking.Gnina as ocgnina
    import OCDocker.Docking.PLANTS as ocplants
    import OCDocker.Docking.Smina as ocsmina
    import OCDocker.Docking.Vina as ocvina
    import OCDocker.Ligand as ocl
    import OCDocker.Processing.Preprocessing.RMSDClustering as ocrmsdclust
    import OCDocker.Receptor as ocr
    import OCDocker.Toolbox as octools
    import OCDocker.Toolbox.Conversion as occonversion
    import OCDocker.Toolbox.MoleculeProcessing as ocmolproc

    namespace.update(
        {
            "ocgnina": ocgnina,
            "ocplants": ocplants,
            "ocsmina": ocsmina,
            "ocvina": ocvina,
            "ocl": ocl,
            "ocrmsdclust": ocrmsdclust,
            "ocr": ocr,
            "octools": octools,
            "occonversion": occonversion,
            "ocmolproc": ocmolproc,
            "ocerror": ocerror,
            "inspect": inspect,
            "glob": glob,
            "pprint": pprint,
            "os": os,
            "sys": sys,
            "print_args": print_args,
            "clean_test_files": clean_test_files,
        }
    )

    try:
        import OCDocker.Rescoring.ODDT as ocoddt

        namespace["ocoddt"] = ocoddt
    except ModuleNotFoundError as exc:
        missing_mod = getattr(exc, "name", "")
        if missing_mod == "oddt" or missing_mod.startswith("oddt."):
            print("Warning: optional dependency 'oddt' is not installed; 'ocoddt' is unavailable in console mode.")
        else:
            raise

    return namespace






[docs]
def run_interactive(namespace: Mapping[str, Any], *, use_ipython: bool = False) -> int:
    '''Run the interactive console loop.

    Parameters
    ----------
    namespace : Mapping[str, Any]
        Preloaded namespace for Python execution.
    use_ipython : bool, optional
        When ``True`` and IPython is installed, use ``IPython.embed`` instead
        of the simple command loop.

    Returns
    -------
    int
        Exit code (0 for success).
    '''

    if use_ipython:
        try:
            from IPython import embed

            colors = "NoColor"
            if sys.stdout.isatty() and os.getenv("TERM") and "dumb" not in os.getenv("TERM", ""):
                colors = "Linux"
            embed(user_ns=dict(namespace), banner1="", colors=colors, display_banner=False)
            return 0
        except ImportError:
            pass

    print("Launching OCDocker Console. Type 'help' for commands, 'exit' to leave.")
    _setup_readline(namespace)
    interactive = code.InteractiveConsole(locals=dict(namespace))

    while True:
        try:
            line = input("ocdocker> ")
        except EOFError:
            print()
            break
        except KeyboardInterrupt:
            print()
            continue

        stripped = line.strip()
        if not stripped:
            continue

        if handle_command(stripped, namespace):
            if stripped.lower() in ("exit", "quit"):
                break
            continue

        try:
            interactive.push(stripped)
        except SystemExit:
            break

    _save_readline_history()
    return 0