跳转至

molop

本页面提供了 molop 模块的 API 参考。

顶层稳定导出包括 AutoFileParser、AutoMemoryParser、AutoTextParser、AutoBytesParser、AutoParser、 molopconfig、moloplogger 和 __version__。解析工作流从 Python API 初体验开始。

Author: TMJ Date: 2024-10-19 09:57:26 LastEditors: TMJ LastEditTime: 2026-02-05 16:37:11 Description: MolOP is a toolbox for molecule operations and QM information extraction.

BatchParseError

Bases: RuntimeError

Raised by a fail-fast outcome iterator at its first unsuccessful input.

Source code in src/molop/io/parse_outcomes.py
Python
class BatchParseError(RuntimeError):
    """Raised by a fail-fast outcome iterator at its first unsuccessful input."""

    def __init__(self, outcome: FileParseOutcome[Any]) -> None:
        self.outcome = outcome
        detail = outcome.failure.message if outcome.failure is not None else outcome.status
        super().__init__(f"Parsing failed for {outcome.file_path}: {detail}")

BatchParseResult dataclass

Bases: Generic[B, T]

A successful batch plus one outcome for every supplied input path.

Source code in src/molop/io/parse_outcomes.py
Python
@dataclass(frozen=True, slots=True)
class BatchParseResult(Generic[B, T]):
    """A successful batch plus one outcome for every supplied input path."""

    batch: B
    outcomes: tuple[FileParseOutcome[T], ...]

    @property
    def failures(self) -> tuple[FileParseOutcome[T], ...]:
        return tuple(outcome for outcome in self.outcomes if not outcome.succeeded)

    @property
    def succeeded(self) -> tuple[FileParseOutcome[T], ...]:
        return tuple(outcome for outcome in self.outcomes if outcome.succeeded)

FileParseOutcome dataclass

Bases: Generic[T]

Result of attempting to parse one input path.

Source code in src/molop/io/parse_outcomes.py
Python
@dataclass(frozen=True, slots=True)
class FileParseOutcome(Generic[T]):
    """Result of attempting to parse one input path."""

    file_path: str
    status: ParseStatus
    value: T | None = None
    warnings: tuple[ParseWarning, ...] = ()
    detected_format: str | None = None
    failure: ParseFailure | None = None
    input_index: int | None = None

    @property
    def succeeded(self) -> bool:
        return self.status == "ok"

ParseFailure dataclass

Serializable description of one failed parse attempt.

Source code in src/molop/io/parse_outcomes.py
Python
@dataclass(frozen=True, slots=True)
class ParseFailure:
    """Serializable description of one failed parse attempt."""

    kind: str
    message: str
    reader_format: str | None = None
    exception_type: str | None = None

ParseOptions dataclass

Immutable options shared by batch dispatch and reader codecs.

Source code in src/molop/io/codec_types.py
Python
@dataclass(frozen=True, slots=True)
class ParseOptions:
    """Immutable options shared by batch dispatch and reader codecs."""

    total_charge: int | None = None
    total_multiplicity: int | None = None
    only_extract_structure: bool = False
    only_last_frame: bool = False
    capture_source_evidence: bool = False
    source_encoding: str = "utf-8"
    release_file_content: bool = True
    force_unit_transform: bool | None = None
    graph_reconstruction_backend: Literal["cpp", "python"] | None = None
    reconstruction_failure_policy: Literal["raise", "return_suspicious"] | None = None
    make_dative_bonds: bool | None = None
    make_stereochemistry: bool | None = None
    source_decode_errors: Literal["strict", "surrogateescape"] = "strict"

    def __post_init__(self) -> None:
        if self.source_decode_errors not in {"strict", "surrogateescape"}:
            raise ValueError("source_decode_errors must be 'strict' or 'surrogateescape'")

    def resolved(self) -> ParseOptions:
        """Snapshot process-global defaults that affect parse-time model behavior."""

        from molop.config import molopconfig

        return replace(
            self,
            force_unit_transform=(
                molopconfig.force_unit_transform
                if self.force_unit_transform is None
                else self.force_unit_transform
            ),
            graph_reconstruction_backend=(
                molopconfig.graph_reconstruction_backend
                if self.graph_reconstruction_backend is None
                else self.graph_reconstruction_backend
            ),
            reconstruction_failure_policy=(
                molopconfig.reconstruction_failure_policy
                if self.reconstruction_failure_policy is None
                else self.reconstruction_failure_policy
            ),
            make_dative_bonds=(
                molopconfig.make_dative_bonds
                if self.make_dative_bonds is None
                else self.make_dative_bonds
            ),
            make_stereochemistry=(
                molopconfig.make_stereochemistry
                if self.make_stereochemistry is None
                else self.make_stereochemistry
            ),
        )

resolved()

Snapshot process-global defaults that affect parse-time model behavior.

Source code in src/molop/io/codec_types.py
Python
def resolved(self) -> ParseOptions:
    """Snapshot process-global defaults that affect parse-time model behavior."""

    from molop.config import molopconfig

    return replace(
        self,
        force_unit_transform=(
            molopconfig.force_unit_transform
            if self.force_unit_transform is None
            else self.force_unit_transform
        ),
        graph_reconstruction_backend=(
            molopconfig.graph_reconstruction_backend
            if self.graph_reconstruction_backend is None
            else self.graph_reconstruction_backend
        ),
        reconstruction_failure_policy=(
            molopconfig.reconstruction_failure_policy
            if self.reconstruction_failure_policy is None
            else self.reconstruction_failure_policy
        ),
        make_dative_bonds=(
            molopconfig.make_dative_bonds
            if self.make_dative_bonds is None
            else self.make_dative_bonds
        ),
        make_stereochemistry=(
            molopconfig.make_stereochemistry
            if self.make_stereochemistry is None
            else self.make_stereochemistry
        ),
    )

AutoBytesParser(raw_bytes, *, total_charge=None, total_multiplicity=None, only_extract_structure=False, only_last_frame=False, capture_source_evidence=False, source_encoding='utf-8', release_file_content=True, parser_detection='auto', parse_options=None)

Parse bytes already held by the caller without a filesystem round trip.

Source code in src/molop/io/memory_parser.py
Python
def AutoBytesParser(
    raw_bytes: BytesSource,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> BaseChemFile[Any]:
    """Parse bytes already held by the caller without a filesystem round trip."""

    return AutoMemoryParser(
        _coerce_bytes(raw_bytes),
        total_charge=total_charge,
        total_multiplicity=total_multiplicity,
        only_extract_structure=only_extract_structure,
        only_last_frame=only_last_frame,
        capture_source_evidence=capture_source_evidence,
        source_encoding=source_encoding,
        release_file_content=release_file_content,
        parser_detection=parser_detection,
        parse_options=parse_options,
    )

AutoFileParser(file_path, *, total_charge=None, total_multiplicity=None, only_extract_structure=False, only_last_frame=False, capture_source_evidence=False, source_encoding='utf-8', release_file_content=True, parser_detection='auto', parse_options=None)

Parse one file with automatic format detection.

Unlike :func:AutoParser, this entrypoint accepts exactly one file path and returns the parsed file model directly. It performs no glob expansion, batch preparation, or process scheduling, so callers can place the parse inside their own worker or process model.

parser_detection="auto" first uses the path extension and reader probes to narrow the candidates, then lets the parser’s format checks validate the complete source. Pass a format id to select a reader explicitly while retaining the same file-level return type.

Raises:

Type Description
TypeError

If file_path is not a text path-like object.
ValueError

If file_path is not an existing regular file.
UnsupportedFormatError

If no reader is registered for the input.
FormatMismatchError

If all candidate readers reject the input format.
ParseError

If a selected reader fails during parsing.
Source code in src/molop/io/__init__.py
Python
def AutoFileParser(
    file_path: PathSpec,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> "FileDiskObj":
    """Parse one file with automatic format detection.

    Unlike :func:`AutoParser`, this entrypoint accepts exactly one file path and
    returns the parsed file model directly. It performs no glob expansion,
    batch preparation, or process scheduling, so callers can place the parse
    inside their own worker or process model.

    ``parser_detection="auto"`` first uses the path extension and reader
    probes to narrow the candidates, then lets the parser's format checks
    validate the complete source. Pass a format id to select a reader
    explicitly while retaining the same file-level return type.

    Raises:
        TypeError: If ``file_path`` is not a text path-like object.
        ValueError: If ``file_path`` is not an existing regular file.
        UnsupportedFormatError: If no reader is registered for the input.
        FormatMismatchError: If all candidate readers reject the input format.
        ParseError: If a selected reader fails during parsing.
    """
    try:
        path_value = os.fspath(file_path)
    except TypeError as exc:
        raise TypeError("file_path must be a path-like object") from exc
    if not isinstance(path_value, str):
        raise TypeError("file_path must resolve to a text path, not bytes")

    path = Path(os.path.abspath(path_value))
    if not path.is_file():
        raise ValueError(f"File {path} does not exist or is not a regular file.")

    hint_format = None if parser_detection == "auto" else parser_detection
    possible_readers = cast(
        tuple[_FileReaderCodec, ...],
        codec_registry.select_reader(path, hint_format=hint_format),
    )
    if parser_detection == "auto":
        possible_readers = _filter_readers_by_probe(str(path), possible_readers)

    outcome = _single_file_parser(
        file_path=str(path),
        possible_readers=possible_readers,
        total_charge=total_charge,
        total_multiplicity=total_multiplicity,
        only_extract_structure=only_extract_structure,
        only_last_frame=only_last_frame,
        capture_source_evidence=capture_source_evidence,
        source_encoding=source_encoding,
        release_file_content=release_file_content,
        parse_options=parse_options,
        return_outcome=True,
    )
    if outcome.succeeded and outcome.value is not None:
        return outcome.value

    failure = outcome.failure
    message = failure.message if failure is not None else f"Failed to parse file {path}."
    if outcome.status == "unsupported":
        raise UnsupportedFormatError(message)
    if outcome.status == "mismatch":
        raise FormatMismatchError(message)
    raise ParseError(message)

AutoMemoryParser(source, *, total_charge=None, total_multiplicity=None, only_extract_structure=False, only_last_frame=False, capture_source_evidence=False, source_encoding='utf-8', release_file_content=True, parser_detection='auto', parse_options=None)

Python
AutoMemoryParser(
    source: TextSource,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> BaseChemFile[Any]
Python
AutoMemoryParser(
    source: BytesSource,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> BaseChemFile[Any]

Parse already-loaded text or bytes with automatic format detection.

A string is treated as source text, never as a path. Use :func:AutoFileParser for path input. Binary and text streams are consumed from their current position.

Source code in src/molop/io/memory_parser.py
Python
def AutoMemoryParser(
    source: MemorySource,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> BaseChemFile[Any]:
    """Parse already-loaded text or bytes with automatic format detection.

    A string is treated as source text, never as a path. Use :func:`AutoFileParser`
    for path input. Binary and text streams are consumed from their current position.
    """

    options = _resolved_options(
        total_charge=total_charge,
        total_multiplicity=total_multiplicity,
        only_extract_structure=only_extract_structure,
        only_last_frame=only_last_frame,
        capture_source_evidence=capture_source_evidence,
        source_encoding=source_encoding,
        release_file_content=release_file_content,
        parse_options=parse_options,
    )
    if isinstance(source, str):
        return _parse_loaded_source(
            source,
            source_kind="text",
            parser_detection=parser_detection,
            options=options,
        )
    if isinstance(source, (bytes, bytearray, memoryview)):
        return _parse_loaded_source(
            bytes(source),
            source_kind="bytes",
            parser_detection=parser_detection,
            options=options,
        )
    read = getattr(source, "read", None)
    if callable(read):
        value = read()
        if isinstance(value, str):
            return _parse_loaded_source(
                value,
                source_kind="text",
                parser_detection=parser_detection,
                options=options,
            )
        if isinstance(value, (bytes, bytearray, memoryview)):
            return _parse_loaded_source(
                bytes(value),
                source_kind="bytes",
                parser_detection=parser_detection,
                options=options,
            )
        raise TypeError("source stream must return str or bytes-like data")
    return _parse_loaded_source(
        _coerce_bytes(cast(BytesSource, source)),
        source_kind="bytes",
        parser_detection=parser_detection,
        options=options,
    )

AutoTextParser(text, *, total_charge=None, total_multiplicity=None, only_extract_structure=False, only_last_frame=False, capture_source_evidence=False, source_encoding='utf-8', release_file_content=True, parser_detection='auto', parse_options=None)

Parse text already held by the caller without creating a temporary file.

Source code in src/molop/io/memory_parser.py
Python
def AutoTextParser(
    text: TextSource,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
) -> BaseChemFile[Any]:
    """Parse text already held by the caller without creating a temporary file."""

    return AutoMemoryParser(
        _coerce_text(text),
        total_charge=total_charge,
        total_multiplicity=total_multiplicity,
        only_extract_structure=only_extract_structure,
        only_last_frame=only_last_frame,
        capture_source_evidence=capture_source_evidence,
        source_encoding=source_encoding,
        release_file_content=release_file_content,
        parser_detection=parser_detection,
        parse_options=parse_options,
    )

iter_parse_outcomes(file_path, *, total_charge=None, total_multiplicity=None, n_jobs=-1, only_extract_structure=False, only_last_frame=False, capture_source_evidence=False, source_encoding='utf-8', release_file_content=True, parser_detection='auto', parse_options=None, fail_fast=False)

Stream one structured parse outcome for each normalized input path.

The returned iterator yields parallel results as they complete and closes MolOP’s worker/result resources when the consumer stops early. Input paths are expanded in caller order and repeated paths produce repeated outcomes. Unlike :func:AutoParser, this function does not construct or retain a successful batch. With fail_fast=True, the first unsuccessful input raises :class:BatchParseError after cancelling active work.

Source code in src/molop/io/__init__.py
Python
def iter_parse_outcomes(
    file_path: PathInput,
    *,
    total_charge: int | None = None,
    total_multiplicity: int | None = None,
    n_jobs: int = -1,
    only_extract_structure: bool = False,
    only_last_frame: bool = False,
    capture_source_evidence: bool = False,
    source_encoding: str = "utf-8",
    release_file_content: bool = True,
    parser_detection: str = "auto",
    parse_options: ParseOptions | None = None,
    fail_fast: bool = False,
) -> Iterator[FileParseOutcome["FileDiskObj"]]:
    """Stream one structured parse outcome for each normalized input path.

    The returned iterator yields parallel results as they complete and closes
    MolOP's worker/result resources when the consumer stops early.  Input
    paths are expanded in caller order and repeated paths produce repeated
    outcomes. Unlike :func:`AutoParser`, this function does not construct or
    retain a successful batch. With ``fail_fast=True``, the first unsuccessful
    input raises :class:`BatchParseError` after cancelling active work.
    """

    files = _expand_file_paths_preserving_duplicates(file_path)
    return FileBatchParserDisk(n_jobs=n_jobs).iter_outcomes(
        files,
        total_charge=total_charge,
        total_multiplicity=total_multiplicity,
        only_extract_structure=only_extract_structure,
        only_last_frame=only_last_frame,
        capture_source_evidence=capture_source_evidence,
        source_encoding=source_encoding,
        release_file_content=release_file_content,
        parser_detection=parser_detection,
        parse_options=parse_options,
        fail_fast=fail_fast,
    )