molop.config¶

This page provides the API reference for the molop.config module.

The module exposes the process-wide molopconfig object. It controls progress display, the maximum parallel-job limit, logging, and structure-recovery options. Prefer explicit function arguments for per-call behavior; change the global configuration only when the policy should apply to subsequent calls.

Configuration overview¶

The MolOPConfig fields and defaults are listed below. from molop import molopconfig returns a process-wide shared object; changing it affects subsequent calls. Creating an independent MolOPConfig(...) applies its progress, file-logging, and native-logging settings during construction.

Option	Type	Default	Description
`show_progress_bar`	`bool`	`True`	Show batch progress bars; also controls MolOP’s console log handler.
`max_jobs`	`int \\| None`	`None`	Process-wide parallel-job ceiling; `None` uses the CPUs available to the process.
`graph_reconstruction_backend`	`"cpp" \\| "python"`	`"cpp"`	Backend used for molecular-graph reconstruction.
`reconstruction_failure_policy`	`"raise" \\| "return_suspicious"`	`"raise"`	Raise on reconstruction failure, or retain a fallback graph marked suspicious.
`prewarm_topologies`	`bool`	`False`	Prewarm graph-dependent topologies in the parent process; otherwise reconstruct lazily in the worker using the result.
`make_dative_bonds`	`bool`	`True`	Create dative bonds during graph reconstruction.
`make_stereochemistry`	`bool`	`True`	Assign stereochemistry during graph reconstruction.
`force_unit_transform`	`bool`	`False`	Force unit conversion.
`parallel_max_size`	`int`	`8 * 1024**2`	Size threshold used by parallel dispatch and joblib data transfer, in bytes (8 MiB).
`max_recursion_depth`	`int`	`3000`	Python recursion depth requested by `set_max_recursion_depth()`; import does not change the interpreter limit.
`log_to_file`	`bool`	`False`	Enable MolOP file logging.
`log_file_path`	`str`	`"molop.log"`	File-log path; effective when `log_to_file` is enabled or `enable_file_logging()` is called.
`suppress_rdkit_logs`	`bool`	`True`	Suppress RDKit native diagnostics.
`suppress_openbabel_logs`	`bool`	`True`	Suppress Open Babel native diagnostics.
`use_dof_effect_drawer`	`bool`	`True`	Prefer the `rdkit-dof` depth-of-field drawer; use the standard RDKit drawer when disabled.

effective_max_jobs and effective_molgr_max_jobs are read-only limits calculated from available CPU resources and the configured ceilings.

Import side effects and explicit initialization¶

Importing molop initializes the global molopconfig and suppresses RDKit/Open Babel native diagnostics by default. It does not create molop.log or change the host process’s Python recursion limit. rdkit-dof is loaded only when depth-aware drawing or an explicit related setting requires it. Native diagnostics belong to the native libraries’ own output and are not automatically copied to MolOP’s file log.

To retain native diagnostics, change the flags and apply the policy to the current process:

Python

from molop import molopconfig

molopconfig.suppress_rdkit_logs = False
molopconfig.suppress_openbabel_logs = False
molopconfig.configure_native_logging()

Keep the other flag set to True when only one native library should remain quiet. New MolOPConfig(...) instances apply both native-log flags during construction; call configure_native_logging() after changing flags on an existing object.

To enable file logging at the same time as quiet native diagnostics, create a configuration explicitly:

Python

from molop.config import MolOPConfig

config = MolOPConfig(
    log_to_file=True,
    log_file_path="run.log",
    suppress_rdkit_logs=True,
    suppress_openbabel_logs=True,
)

enable_file_logging() and disable_file_logging() own the file-handler lifecycle. Setting log_to_file = True on the global object does not create a handler automatically; call enable_file_logging() and use disable_file_logging() to turn it off. set_log_level() accepts DEBUG, INFO, WARNING, ERROR, or CRITICAL.

Use the methods below to switch the progress bar and MolOP’s console handler together:

Python

molopconfig.quiet()    # disable the progress bar and MolOP console logs
molopconfig.verbose()  # restore the progress bar and MolOP console logs

set_max_recursion_depth() changes the current process only when explicitly called.

Parent-process topology prewarming is controlled by prewarm_topologies and defaults to False. With the default, graph-dependent operations reconstruct lazily in the spawn-like loky worker that uses the result, avoiding a separate parent-process warmup step. Set it to True when a workflow needs the deterministic parent-side cache:

Python

from molop import molopconfig

molopconfig.prewarm_topologies = True

MolGR reconstruction failures are strict by default. Enable raw fallback retention when the workflow must preserve invalid structures for review:

Python

molopconfig.reconstruction_failure_policy = "return_suspicious"

Retained graphs are marked with topology_reconstruction_status == "suspicious_fallback"; they must not be treated as trusted chemistry.

Parallelism defaults¶

max_jobs defaults to None, which means automatic process-aware detection. MolOP takes the most restrictive observable value from joblib/loky’s scheduler and container limits, os.cpu_count(), and the current process affinity where the operating system exposes it. The direct psutil dependency enables joblib’s affinity fallback on Windows and other supported platforms. Thus n_jobs=-1 uses the CPUs available to the current process without exceeding a detected allocation.

Linux supports affinity and cgroup quota detection. Windows affinity is read through psutil and joblib also applies its safe worker limit for Windows process pools. macOS does not expose a portable per-process CPU-affinity API, so its automatic value uses the CPUs visible to the process and any joblib limit. On every platform, use max_jobs, MOLOP_MAX_JOBS, or LOKY_MAX_CPU_COUNT when the runtime allocation is not observable from the operating system.

Set a positive max_jobs to impose a process-wide ceiling:

Python

from molop import molopconfig

molopconfig.max_jobs = 32

For deployments that construct the default configuration from the environment, set MOLOP_MAX_JOBS=32. An explicit MolOPConfig(max_jobs=None) keeps automatic detection even when the environment variable is present. Positive n_jobs values still request a per-call limit, capped by effective_max_jobs; n_jobs=1 remains the serial debugging mode. CLI users can set the same process-wide ceiling for one invocation with molop --max-jobs 32 parse ....

Tasks that may enter MolGR use the separate effective_molgr_max_jobs budget. Automatic and explicit values are capped at floor(2 * available_cpu_count / 3) (with a minimum of one serial worker), then by max_jobs. This covers topology reconstruction, graph-dependent summaries, exports, transforms, and user callbacks such as filter_custom. File parsing and other operations that do not invoke MolGR keep the full effective_max_jobs limit.

Author: TMJ Date: 2025-01-15 23:01:22 LastEditors: TMJ LastEditTime: 2026-06-18 19:45:06 Description: 请填写简介

`MolOPConfig` ¶

Bases: BaseModel

Configuration class for MolOP operations. Used to manage settings related to molecule processing, fingerprint generation, and logging.

源代码位于： src/molop/config/__init__.py

Python

class MolOPConfig(BaseModel):
    """
    Configuration class for MolOP operations.
    Used to manage settings related to molecule processing, fingerprint generation, and logging.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True, validate_assignment=True)

    # --- General Settings ---
    show_progress_bar: bool = Field(default=True, description="Whether to display the progress bar")
    max_jobs: int | None = Field(
        default=None,
        ge=1,
        description="Maximum parallel jobs; None uses all CPUs available to the process",
    )

    # --- Advanced Settings ---
    graph_reconstruction_backend: Literal["cpp", "python"] = Field(
        default="cpp", description="Backend for graph reconstruction"
    )
    reconstruction_failure_policy: Literal["raise", "return_suspicious"] = Field(
        default="raise",
        description=(
            "Whether MolGR reconstruction failures raise or retain an untrusted "
            "suspicious fallback molecule"
        ),
    )
    prewarm_topologies: bool = Field(
        default=False,
        description=(
            "Whether graph-dependent operations prewarm coordinate-only topologies "
            "in the parent process"
        ),
    )
    make_dative_bonds: bool = Field(default=True, description="Whether to make dative bonds")
    make_stereochemistry: bool = Field(
        default=True, description="Whether to assign stereochemistry during graph reconstruction"
    )
    force_unit_transform: bool = Field(
        default=False, description="Whether to force unit conversion"
    )
    parallel_max_size: int = Field(
        default=8 * 1024**2,
        description="Maximum data size for parallel processing (bytes)",
    )
    max_recursion_depth: int = Field(default=3000, description="Maximum recursion depth for Python")

    # --- Log File Control ---
    log_to_file: bool = Field(default=False, description="Whether to write log messages to a file")
    log_file_path: str = Field(default="molop.log", description="Path used for optional file logs")

    # --- Native library logging control ---
    suppress_rdkit_logs: bool = Field(
        default=True,
        description="Whether to suppress RDKit native diagnostics",
    )
    suppress_openbabel_logs: bool = Field(
        default=True,
        description="Whether to suppress Open Babel native diagnostics",
    )

    # --- DOF Effect Drawer Control ---
    use_dof_effect_drawer: bool = Field(
        default=True, description="Whether to use DOF effect drawer"
    )

    def __init__(self, **data: Any):
        """
        Initializes the configuration object and configures the logger based on current settings.
        """
        if "max_jobs" not in data and (env_max_jobs := os.environ.get(MAX_JOBS_ENV_VAR)):
            data["max_jobs"] = env_max_jobs
        super().__init__(**data)
        # Set log state based on initial configuration values
        if self.show_progress_bar:
            self.verbose()
        else:
            self.quiet()

        if self.log_to_file:
            self.enable_file_logging()
        else:
            self.disable_file_logging()

        # ``rdkit-dof`` changes the interpreter recursion limit when imported.
        # Do not import it during a normal library import; explicit rendering
        # configuration can still enable it through this method.
        if not self.use_dof_effect_drawer:
            self.set_dof_effect_drawer(enable=False)

        self.configure_native_logging()

    def quiet(self):
        """
        Disables the progress bar and console log output.
        This allows the program to run silently in the background.
        """
        self.show_progress_bar = False
        if stream_handler in moloplogger.handlers:
            moloplogger.removeHandler(stream_handler)

    def verbose(self):
        """
        Enables the progress bar and console log output.
        """
        self.show_progress_bar = True
        if stream_handler not in moloplogger.handlers:
            moloplogger.addHandler(stream_handler)

    def enable_file_logging(self):
        """Enable logging to a file."""
        global file_handler

        self.log_to_file = True
        if file_handler is None or file_handler.baseFilename != os.path.abspath(self.log_file_path):
            if file_handler is not None:
                file_handler.close()
            file_handler = logging.FileHandler(self.log_file_path)
            file_handler.setLevel(logging.DEBUG)
            file_handler.setFormatter(formatter)
        if file_handler not in moloplogger.handlers:
            moloplogger.addHandler(file_handler)
        moloplogger.info(
            f"File logging enabled. Logs will be written to: {getattr(file_handler, 'baseFilename', 'N/A')}"
        )

    def disable_file_logging(self):
        """Disable logging to a file."""
        global file_handler

        self.log_to_file = False
        if file_handler is not None and file_handler in moloplogger.handlers:
            moloplogger.removeHandler(file_handler)
        if file_handler is not None:
            file_handler.close()
            file_handler = None

    def set_log_level(self, level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]):
        """
        Sets the level for the molop logger.

        Args:
            level (str): The logging level, must be one of 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'.
        """
        try:
            moloplogger.setLevel(level)
            logging.info(f"Log level set to {level}")
        except ValueError as e:
            logging.error(f"Error setting log level: Invalid level '{level}'")
            raise ValueError(f"Invalid log level: {level}") from e

    def set_max_recursion_depth(self, depth: int):
        """
        Set the maximum recursion depth for Python.

        Args:
            depth (int): The maximum recursion depth.
        """
        sys.setrecursionlimit(depth)
        self.max_recursion_depth = depth
        logging.info(f"Maximum recursion depth set to {depth}")

    def configure_native_logging(self) -> None:
        """Apply the configured native-library logging policy.

        Native diagnostics are suppressed by default.  Call this method after
        changing either suppression flag to apply the new policy to the current
        process.
        """

        if self.suppress_rdkit_logs:
            RDLogger.DisableLog("rdApp.*")  # type: ignore
        else:
            RDLogger.EnableLog("rdApp.*")  # type: ignore
        if self.suppress_openbabel_logs:
            pybel.ob.obErrorLog.StopLogging()
        else:
            pybel.ob.obErrorLog.StartLogging()

    @property
    def effective_max_jobs(self) -> int:
        """Current process-aware worker limit after applying ``max_jobs``."""

        available_jobs = available_cpu_count()
        return available_jobs if self.max_jobs is None else min(available_jobs, self.max_jobs)

    @property
    def effective_molgr_max_jobs(self) -> int:
        """Worker limit for tasks that may enter MolGR's native runtime.

        MolGR work intentionally leaves one third of the process-visible CPU
        budget available for the parent process, native helper threads, and
        the operating system.  The existing ``max_jobs`` setting remains an
        additional upper bound.
        """

        available_jobs = available_cpu_count()
        molgr_jobs = max(1, (available_jobs * 2) // 3)
        return min(self.effective_max_jobs, molgr_jobs)

    def set_n_jobs(self, n_jobs: int) -> int:
        """Resolve automatic or explicit parallelism within the effective worker limit."""

        return self.effective_max_jobs if n_jobs <= 0 else min(n_jobs, self.effective_max_jobs)

    def set_molgr_n_jobs(self, n_jobs: int) -> int:
        """Resolve parallelism for work that may invoke MolGR.

        Both automatic (non-positive) and explicit values are capped so a
        caller cannot accidentally bypass the native-runtime safety budget.
        """

        limit = self.effective_molgr_max_jobs
        return limit if n_jobs <= 0 else min(n_jobs, limit)

    def apply_molgr_reconstruction_policy(
        self,
        policy: Literal["raise", "return_suspicious"] | None = None,
    ) -> Literal["raise", "return_suspicious"]:
        """Apply MolOP's reconstruction policy to the shared MolGR config."""

        resolved_policy = self.reconstruction_failure_policy if policy is None else policy
        MOLGR_CONFIG.interface.reconstruction_failure_policy = resolved_policy
        return resolved_policy

    def set_dof_effect_drawer(self, enable: bool):
        """
        Set whether to use the DOF effect drawer.

        Args:
            enable (bool): Whether to use the DOF effect drawer.
        """
        dofconfig.enable_ipython_integration(enable)

`effective_max_jobs` `property` ¶

Current process-aware worker limit after applying max_jobs.

`effective_molgr_max_jobs` `property` ¶

Worker limit for tasks that may enter MolGR’s native runtime.

MolGR work intentionally leaves one third of the process-visible CPU budget available for the parent process, native helper threads, and the operating system. The existing max_jobs setting remains an additional upper bound.

`init(**data)` ¶

Initializes the configuration object and configures the logger based on current settings.

源代码位于： src/molop/config/__init__.py

Python

def __init__(self, **data: Any):
    """
    Initializes the configuration object and configures the logger based on current settings.
    """
    if "max_jobs" not in data and (env_max_jobs := os.environ.get(MAX_JOBS_ENV_VAR)):
        data["max_jobs"] = env_max_jobs
    super().__init__(**data)
    # Set log state based on initial configuration values
    if self.show_progress_bar:
        self.verbose()
    else:
        self.quiet()

    if self.log_to_file:
        self.enable_file_logging()
    else:
        self.disable_file_logging()

    # ``rdkit-dof`` changes the interpreter recursion limit when imported.
    # Do not import it during a normal library import; explicit rendering
    # configuration can still enable it through this method.
    if not self.use_dof_effect_drawer:
        self.set_dof_effect_drawer(enable=False)

    self.configure_native_logging()

`apply_molgr_reconstruction_policy(policy=None)` ¶

Apply MolOP’s reconstruction policy to the shared MolGR config.

源代码位于： src/molop/config/__init__.py

Python

def apply_molgr_reconstruction_policy(
    self,
    policy: Literal["raise", "return_suspicious"] | None = None,
) -> Literal["raise", "return_suspicious"]:
    """Apply MolOP's reconstruction policy to the shared MolGR config."""

    resolved_policy = self.reconstruction_failure_policy if policy is None else policy
    MOLGR_CONFIG.interface.reconstruction_failure_policy = resolved_policy
    return resolved_policy

`configure_native_logging()` ¶

Apply the configured native-library logging policy.

Native diagnostics are suppressed by default. Call this method after changing either suppression flag to apply the new policy to the current process.

源代码位于： src/molop/config/__init__.py

Python

def configure_native_logging(self) -> None:
    """Apply the configured native-library logging policy.

    Native diagnostics are suppressed by default.  Call this method after
    changing either suppression flag to apply the new policy to the current
    process.
    """

    if self.suppress_rdkit_logs:
        RDLogger.DisableLog("rdApp.*")  # type: ignore
    else:
        RDLogger.EnableLog("rdApp.*")  # type: ignore
    if self.suppress_openbabel_logs:
        pybel.ob.obErrorLog.StopLogging()
    else:
        pybel.ob.obErrorLog.StartLogging()

`disable_file_logging()` ¶

Disable logging to a file.

源代码位于： src/molop/config/__init__.py

Python

def disable_file_logging(self):
    """Disable logging to a file."""
    global file_handler

    self.log_to_file = False
    if file_handler is not None and file_handler in moloplogger.handlers:
        moloplogger.removeHandler(file_handler)
    if file_handler is not None:
        file_handler.close()
        file_handler = None

`enable_file_logging()` ¶

Enable logging to a file.

源代码位于： src/molop/config/__init__.py

Python

def enable_file_logging(self):
    """Enable logging to a file."""
    global file_handler

    self.log_to_file = True
    if file_handler is None or file_handler.baseFilename != os.path.abspath(self.log_file_path):
        if file_handler is not None:
            file_handler.close()
        file_handler = logging.FileHandler(self.log_file_path)
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(formatter)
    if file_handler not in moloplogger.handlers:
        moloplogger.addHandler(file_handler)
    moloplogger.info(
        f"File logging enabled. Logs will be written to: {getattr(file_handler, 'baseFilename', 'N/A')}"
    )

`quiet()` ¶

Disables the progress bar and console log output. This allows the program to run silently in the background.

源代码位于： src/molop/config/__init__.py

Python

def quiet(self):
    """
    Disables the progress bar and console log output.
    This allows the program to run silently in the background.
    """
    self.show_progress_bar = False
    if stream_handler in moloplogger.handlers:
        moloplogger.removeHandler(stream_handler)

`set_dof_effect_drawer(enable)` ¶

Set whether to use the DOF effect drawer.

参数：

名称	类型	描述	默认
`enable`	`bool`	Whether to use the DOF effect drawer.	必需

源代码位于： src/molop/config/__init__.py

Python

def set_dof_effect_drawer(self, enable: bool):
    """
    Set whether to use the DOF effect drawer.

    Args:
        enable (bool): Whether to use the DOF effect drawer.
    """
    dofconfig.enable_ipython_integration(enable)

`set_log_level(level)` ¶

Sets the level for the molop logger.

参数：

名称	类型	描述	默认
`level`	`str`	The logging level, must be one of ‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’, ‘CRITICAL’.	必需

源代码位于： src/molop/config/__init__.py

Python

def set_log_level(self, level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]):
    """
    Sets the level for the molop logger.

    Args:
        level (str): The logging level, must be one of 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'.
    """
    try:
        moloplogger.setLevel(level)
        logging.info(f"Log level set to {level}")
    except ValueError as e:
        logging.error(f"Error setting log level: Invalid level '{level}'")
        raise ValueError(f"Invalid log level: {level}") from e

`set_max_recursion_depth(depth)` ¶

Set the maximum recursion depth for Python.

参数：

名称	类型	描述	默认
`depth`	`int`	The maximum recursion depth.	必需

源代码位于： src/molop/config/__init__.py

Python

def set_max_recursion_depth(self, depth: int):
    """
    Set the maximum recursion depth for Python.

    Args:
        depth (int): The maximum recursion depth.
    """
    sys.setrecursionlimit(depth)
    self.max_recursion_depth = depth
    logging.info(f"Maximum recursion depth set to {depth}")

`set_molgr_n_jobs(n_jobs)` ¶

Resolve parallelism for work that may invoke MolGR.

Both automatic (non-positive) and explicit values are capped so a caller cannot accidentally bypass the native-runtime safety budget.

源代码位于： src/molop/config/__init__.py

Python

def set_molgr_n_jobs(self, n_jobs: int) -> int:
    """Resolve parallelism for work that may invoke MolGR.

    Both automatic (non-positive) and explicit values are capped so a
    caller cannot accidentally bypass the native-runtime safety budget.
    """

    limit = self.effective_molgr_max_jobs
    return limit if n_jobs <= 0 else min(n_jobs, limit)

`set_n_jobs(n_jobs)` ¶

Resolve automatic or explicit parallelism within the effective worker limit.

源代码位于： src/molop/config/__init__.py

Python

def set_n_jobs(self, n_jobs: int) -> int:
    """Resolve automatic or explicit parallelism within the effective worker limit."""

    return self.effective_max_jobs if n_jobs <= 0 else min(n_jobs, self.effective_max_jobs)

`verbose()` ¶

Enables the progress bar and console log output.

源代码位于： src/molop/config/__init__.py

Python

def verbose(self):
    """
    Enables the progress bar and console log output.
    """
    self.show_progress_bar = True
    if stream_handler not in moloplogger.handlers:
        moloplogger.addHandler(stream_handler)

`available_cpu_count()` ¶

Return joblib/loky’s process-aware logical CPU limit.

源代码位于： src/molop/config/__init__.py

Python

def available_cpu_count() -> int:
    """Return joblib/loky's process-aware logical CPU limit."""

    return max(int(joblib_cpu_count()), 1)

molop.config¶

Configuration overview¶

Import side effects and explicit initialization¶

Parallelism defaults¶

MolOPConfig ¶

effective_max_jobs property ¶

effective_molgr_max_jobs property ¶

__init__(**data) ¶

apply_molgr_reconstruction_policy(policy=None) ¶

configure_native_logging() ¶

disable_file_logging() ¶

enable_file_logging() ¶

quiet() ¶

set_dof_effect_drawer(enable) ¶

set_log_level(level) ¶

set_max_recursion_depth(depth) ¶

set_molgr_n_jobs(n_jobs) ¶

set_n_jobs(n_jobs) ¶

verbose() ¶

available_cpu_count() ¶

`MolOPConfig` ¶

`effective_max_jobs` `property` ¶

`effective_molgr_max_jobs` `property` ¶

`init(**data)` ¶

`apply_molgr_reconstruction_policy(policy=None)` ¶

`configure_native_logging()` ¶

`disable_file_logging()` ¶

`enable_file_logging()` ¶

`quiet()` ¶

`set_dof_effect_drawer(enable)` ¶

`set_log_level(level)` ¶

`set_max_recursion_depth(depth)` ¶

`set_molgr_n_jobs(n_jobs)` ¶

`set_n_jobs(n_jobs)` ¶

`verbose()` ¶

`available_cpu_count()` ¶