molop¶
This page provides the API reference for the molop module.
The stable top-level exports are AutoFileParser, AutoMemoryParser, AutoTextParser,
AutoBytesParser, AutoParser, molopconfig, moloplogger, and __version__. Start with the Python API guide
for parsing workflows.
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.
源代码位于： 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.
源代码位于： 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.
源代码位于： 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.
源代码位于： src/molop/io/parse_outcomes.py
ParseOptions
dataclass
¶
Immutable options shared by batch dispatch and reader codecs.
源代码位于： 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.
源代码位于： 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.
源代码位于： 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.
引发：
| 类型 | 描述 |
|---|---|
TypeError
|
If |
ValueError
|
If |
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. |
源代码位于： 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.
源代码位于： 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.
源代码位于： 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.
源代码位于： 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,
)