Registry¶
Registry 管理不同化学文件格式的编解码器的注册和发现。
Source code in src/molop/io/codec_registry.py
Python
class Registry:
def __init__(self, *, autoload_defaults: bool = True) -> None:
self._registration_counter = count()
self._readers_by_format: dict[str, list[_ReaderSpec]] = {}
self._readers_by_extension: dict[str, list[_ReaderSpec]] = {}
self._fallback_readers: list[_ReaderSpec] = []
self._writers_by_format: dict[str, list[_WriterSpec]] = {}
self._openbabel_fallback_factory: ReaderFactory | None = None
self._default_codecs_registered = False
self._autoload_defaults = autoload_defaults
def ensure_default_codecs_registered(self) -> None:
"""Idempotently register default codecs.
The registry is intentionally empty until this is called.
"""
if self._default_codecs_registered:
return
# Do NOT flip the flag before successful activation.
from molop.io.codecs import catalog
catalog.load_builtin_codecs(self)
catalog.load_plugin_codecs(self)
self._default_codecs_registered = True
def register_reader(self, codec: ReaderCodec, *, fallback: bool = False) -> None:
self.register_reader_factory(
lambda: codec,
format_id=codec.format_id,
extensions=codec.extensions,
priority=codec.priority,
fallback=fallback,
)
def register_reader_factory(
self,
factory: ReaderFactory,
*,
format_id: str,
extensions: Iterable[str] = (),
priority: int = 0,
fallback: bool = False,
) -> None:
"""Register a reader factory keyed by format id and extension."""
normalized_format_id = _normalize_format_id(format_id)
normalized_extensions = _normalize_extensions(extensions)
spec = _ReaderSpec(
format_id=normalized_format_id,
extensions=normalized_extensions,
priority=priority,
factory=factory,
order=next(self._registration_counter),
)
self._readers_by_format.setdefault(normalized_format_id, []).append(spec)
self._readers_by_format[normalized_format_id].sort(key=_reader_sort_key)
for extension in normalized_extensions:
self._readers_by_extension.setdefault(extension, []).append(spec)
self._readers_by_extension[extension].sort(key=_reader_sort_key)
if fallback:
self._fallback_readers.append(spec)
self._fallback_readers.sort(key=_reader_sort_key)
def register_writer(self, codec: WriterCodec) -> None:
self.register_writer_factory(
lambda: codec,
format_id=codec.format_id,
required_level=codec.required_level,
default_graph_policy=getattr(codec, "default_graph_policy", None),
priority=codec.priority,
)
def register_writer_factory(
self,
factory: WriterFactory,
*,
format_id: str,
required_level: StructureLevel,
domain: WriterDomain = "file",
default_graph_policy: GraphPolicy | None = None,
priority: int = 0,
) -> None:
normalized_format_id = _normalize_format_id(format_id)
spec = _WriterSpec(
format_id=normalized_format_id,
required_level=required_level,
domain=domain,
default_graph_policy=default_graph_policy
or _default_graph_policy_for_level(required_level),
priority=priority,
factory=factory,
order=next(self._registration_counter),
)
self._writers_by_format.setdefault(normalized_format_id, []).append(spec)
self._writers_by_format[normalized_format_id].sort(key=_writer_sort_key)
def register_openbabel_fallback(self, factory: ReaderFactory) -> None:
self._openbabel_fallback_factory = factory
def reader_factory(
self,
*,
format_id: str,
extensions: Iterable[str] = (),
priority: int = 0,
fallback: bool = False,
):
"""Decorator that registers a reader factory into this registry."""
def decorator(factory: ReaderFactory) -> ReaderFactory:
self.register_reader_factory(
factory,
format_id=format_id,
extensions=extensions,
priority=priority,
fallback=fallback,
)
return factory
return decorator
def writer_factory(
self,
*,
format_id: str,
required_level: StructureLevel,
domain: WriterDomain = "file",
default_graph_policy: GraphPolicy | None = None,
priority: int = 0,
):
"""Decorator that registers a writer factory into this registry."""
def decorator(factory: WriterFactory) -> WriterFactory:
self.register_writer_factory(
factory,
format_id=format_id,
required_level=required_level,
domain=domain,
default_graph_policy=default_graph_policy,
priority=priority,
)
return factory
return decorator
def get_supported_writer_formats(self) -> list[str]:
"""Return a sorted list of all registered writer format IDs."""
if self._autoload_defaults:
self.ensure_default_codecs_registered()
return sorted(self._writers_by_format.keys())
def select_reader(
self, path: str | Path, hint_format: str | None = None
) -> tuple[ReaderCodec, ...]:
"""Return reader codecs ordered by deterministic precedence."""
if self._autoload_defaults:
self.ensure_default_codecs_registered()
candidates: list[_ReaderSpec] = []
seen: set[int] = set()
if hint_format:
normalized_hint = _normalize_format_id(hint_format)
_extend_unique(candidates, self._readers_by_format.get(normalized_hint, ()), seen)
extension = _normalize_extension(Path(path).suffix)
if extension:
_extend_unique(candidates, self._readers_by_extension.get(extension, ()), seen)
if self._fallback_readers:
_extend_unique(candidates, self._fallback_readers[:MAX_FALLBACK_READERS], seen)
lazy_codecs = [
_LazyReaderCodec(
format_id=spec.format_id,
extensions=frozenset(spec.extensions),
priority=spec.priority,
_factory=spec.factory,
)
for spec in candidates
]
if not lazy_codecs and self._openbabel_fallback_factory is not None:
lazy_codecs.append(
_LazyReaderCodec(
format_id=_OPENBABEL_FALLBACK_FORMAT_ID,
extensions=frozenset(),
priority=-10_000,
_factory=self._openbabel_fallback_factory,
)
)
if not lazy_codecs:
raise UnsupportedFormatError(f"No reader codecs registered for {path}.")
return tuple(lazy_codecs)
def write(
self,
format_id: str,
value: object,
*,
graph_policy: GraphPolicy | None = None,
**kwargs: Any,
) -> object:
"""Write a value using registered writer codecs."""
if self._autoload_defaults:
self.ensure_default_codecs_registered()
normalized_format_id = _normalize_format_id(format_id)
writer_specs = self._writers_by_format.get(normalized_format_id, [])
if not writer_specs:
raise UnsupportedFormatError(f"No writer codecs registered for {format_id}.")
raw_value, structure_level = _unwrap_parse_result(value)
file_writer_specs = [spec for spec in writer_specs if spec.domain == "file"]
if not file_writer_specs:
raise UnsupportedFormatError(f"No file writer codecs registered for {format_id}.")
return _write_with_domain_specs(
file_writer_specs,
raw_value,
structure_level=structure_level,
graph_policy=_resolve_graph_policy(file_writer_specs, graph_policy),
**kwargs,
)
def write_frame(
self,
format_id: str,
value: object,
*,
graph_policy: GraphPolicy | None = None,
**kwargs: Any,
) -> object:
if self._autoload_defaults:
self.ensure_default_codecs_registered()
normalized_format_id = _normalize_format_id(format_id)
writer_specs = self._writers_by_format.get(normalized_format_id, [])
if not writer_specs:
raise UnsupportedFormatError(f"No writer codecs registered for {format_id}.")
raw_value, structure_level = _unwrap_parse_result(value)
frame_writer_specs = [spec for spec in writer_specs if spec.domain == "frame"]
if not frame_writer_specs:
raise UnsupportedFormatError(f"No frame writer codecs registered for {format_id}.")
return _write_with_domain_specs(
frame_writer_specs,
raw_value,
structure_level=structure_level,
graph_policy=_resolve_graph_policy(frame_writer_specs, graph_policy),
**kwargs,
)
ensure_default_codecs_registered()
¶
Idempotently register default codecs.
The registry is intentionally empty until this is called.
Source code in src/molop/io/codec_registry.py
Python
def ensure_default_codecs_registered(self) -> None:
"""Idempotently register default codecs.
The registry is intentionally empty until this is called.
"""
if self._default_codecs_registered:
return
# Do NOT flip the flag before successful activation.
from molop.io.codecs import catalog
catalog.load_builtin_codecs(self)
catalog.load_plugin_codecs(self)
self._default_codecs_registered = True
get_supported_writer_formats()
¶
Return a sorted list of all registered writer format IDs.
reader_factory(*, format_id, extensions=(), priority=0, fallback=False)
¶
Decorator that registers a reader factory into this registry.
Source code in src/molop/io/codec_registry.py
Python
def reader_factory(
self,
*,
format_id: str,
extensions: Iterable[str] = (),
priority: int = 0,
fallback: bool = False,
):
"""Decorator that registers a reader factory into this registry."""
def decorator(factory: ReaderFactory) -> ReaderFactory:
self.register_reader_factory(
factory,
format_id=format_id,
extensions=extensions,
priority=priority,
fallback=fallback,
)
return factory
return decorator
register_reader_factory(factory, *, format_id, extensions=(), priority=0, fallback=False)
¶
Register a reader factory keyed by format id and extension.
Source code in src/molop/io/codec_registry.py
Python
def register_reader_factory(
self,
factory: ReaderFactory,
*,
format_id: str,
extensions: Iterable[str] = (),
priority: int = 0,
fallback: bool = False,
) -> None:
"""Register a reader factory keyed by format id and extension."""
normalized_format_id = _normalize_format_id(format_id)
normalized_extensions = _normalize_extensions(extensions)
spec = _ReaderSpec(
format_id=normalized_format_id,
extensions=normalized_extensions,
priority=priority,
factory=factory,
order=next(self._registration_counter),
)
self._readers_by_format.setdefault(normalized_format_id, []).append(spec)
self._readers_by_format[normalized_format_id].sort(key=_reader_sort_key)
for extension in normalized_extensions:
self._readers_by_extension.setdefault(extension, []).append(spec)
self._readers_by_extension[extension].sort(key=_reader_sort_key)
if fallback:
self._fallback_readers.append(spec)
self._fallback_readers.sort(key=_reader_sort_key)
select_reader(path, hint_format=None)
¶
Return reader codecs ordered by deterministic precedence.
Source code in src/molop/io/codec_registry.py
Python
def select_reader(
self, path: str | Path, hint_format: str | None = None
) -> tuple[ReaderCodec, ...]:
"""Return reader codecs ordered by deterministic precedence."""
if self._autoload_defaults:
self.ensure_default_codecs_registered()
candidates: list[_ReaderSpec] = []
seen: set[int] = set()
if hint_format:
normalized_hint = _normalize_format_id(hint_format)
_extend_unique(candidates, self._readers_by_format.get(normalized_hint, ()), seen)
extension = _normalize_extension(Path(path).suffix)
if extension:
_extend_unique(candidates, self._readers_by_extension.get(extension, ()), seen)
if self._fallback_readers:
_extend_unique(candidates, self._fallback_readers[:MAX_FALLBACK_READERS], seen)
lazy_codecs = [
_LazyReaderCodec(
format_id=spec.format_id,
extensions=frozenset(spec.extensions),
priority=spec.priority,
_factory=spec.factory,
)
for spec in candidates
]
if not lazy_codecs and self._openbabel_fallback_factory is not None:
lazy_codecs.append(
_LazyReaderCodec(
format_id=_OPENBABEL_FALLBACK_FORMAT_ID,
extensions=frozenset(),
priority=-10_000,
_factory=self._openbabel_fallback_factory,
)
)
if not lazy_codecs:
raise UnsupportedFormatError(f"No reader codecs registered for {path}.")
return tuple(lazy_codecs)
write(format_id, value, *, graph_policy=None, **kwargs)
¶
Write a value using registered writer codecs.
Source code in src/molop/io/codec_registry.py
Python
def write(
self,
format_id: str,
value: object,
*,
graph_policy: GraphPolicy | None = None,
**kwargs: Any,
) -> object:
"""Write a value using registered writer codecs."""
if self._autoload_defaults:
self.ensure_default_codecs_registered()
normalized_format_id = _normalize_format_id(format_id)
writer_specs = self._writers_by_format.get(normalized_format_id, [])
if not writer_specs:
raise UnsupportedFormatError(f"No writer codecs registered for {format_id}.")
raw_value, structure_level = _unwrap_parse_result(value)
file_writer_specs = [spec for spec in writer_specs if spec.domain == "file"]
if not file_writer_specs:
raise UnsupportedFormatError(f"No file writer codecs registered for {format_id}.")
return _write_with_domain_specs(
file_writer_specs,
raw_value,
structure_level=structure_level,
graph_policy=_resolve_graph_policy(file_writer_specs, graph_policy),
**kwargs,
)
writer_factory(*, format_id, required_level, domain='file', default_graph_policy=None, priority=0)
¶
Decorator that registers a writer factory into this registry.
Source code in src/molop/io/codec_registry.py
Python
def writer_factory(
self,
*,
format_id: str,
required_level: StructureLevel,
domain: WriterDomain = "file",
default_graph_policy: GraphPolicy | None = None,
priority: int = 0,
):
"""Decorator that registers a writer factory into this registry."""
def decorator(factory: WriterFactory) -> WriterFactory:
self.register_writer_factory(
factory,
format_id=format_id,
required_level=required_level,
domain=domain,
default_graph_policy=default_graph_policy,
priority=priority,
)
return factory
return decorator