Registry¶
Registry manages the registration and discovery of codecs for different chemical file formats.
源代码位于： 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,
priority=codec.priority,
)
def register_writer_factory(
self,
factory: WriterFactory,
*,
format_id: str,
required_level: StructureLevel,
priority: int = 0,
) -> None:
normalized_format_id = _normalize_format_id(format_id)
spec = _WriterSpec(
format_id=normalized_format_id,
required_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,
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,
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 = "prefer",
**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)
graph_writers = [
spec for spec in writer_specs if spec.required_level == StructureLevel.GRAPH
]
coords_writers = [
spec for spec in writer_specs if spec.required_level == StructureLevel.COORDS
]
if graph_policy == "coords":
if not coords_writers:
raise ConversionError(f"No coords-level writers registered for {format_id}.")
return _write_with_specs(coords_writers, raw_value, **kwargs)
if graph_policy == "strict":
if not graph_writers:
raise ConversionError(f"No graph-level writers registered for {format_id}.")
graph_value = _coerce_graph_value(raw_value, structure_level)
return _write_with_specs(graph_writers, graph_value, **kwargs)
if graph_writers:
try:
graph_value = _coerce_graph_value(raw_value, structure_level)
return _write_with_specs(graph_writers, graph_value, **kwargs)
except ConversionError:
pass
if coords_writers:
return _write_with_specs(coords_writers, raw_value, **kwargs)
if graph_writers:
raise ConversionError(f"Unable to upgrade value for graph writers of {format_id}.")
raise UnsupportedFormatError(f"No compatible writers registered for {format_id}.")
ensure_default_codecs_registered()
¶
Idempotently register default codecs.
The registry is intentionally empty until this is called.
源代码位于： 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.
源代码位于： 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.
源代码位于： 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.
源代码位于： 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='prefer', **kwargs)
¶
Write a value using registered writer codecs.
源代码位于： src/molop/io/codec_registry.py
Python
def write(
self,
format_id: str,
value: object,
*,
graph_policy: GraphPolicy = "prefer",
**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)
graph_writers = [
spec for spec in writer_specs if spec.required_level == StructureLevel.GRAPH
]
coords_writers = [
spec for spec in writer_specs if spec.required_level == StructureLevel.COORDS
]
if graph_policy == "coords":
if not coords_writers:
raise ConversionError(f"No coords-level writers registered for {format_id}.")
return _write_with_specs(coords_writers, raw_value, **kwargs)
if graph_policy == "strict":
if not graph_writers:
raise ConversionError(f"No graph-level writers registered for {format_id}.")
graph_value = _coerce_graph_value(raw_value, structure_level)
return _write_with_specs(graph_writers, graph_value, **kwargs)
if graph_writers:
try:
graph_value = _coerce_graph_value(raw_value, structure_level)
return _write_with_specs(graph_writers, graph_value, **kwargs)
except ConversionError:
pass
if coords_writers:
return _write_with_specs(coords_writers, raw_value, **kwargs)
if graph_writers:
raise ConversionError(f"Unable to upgrade value for graph writers of {format_id}.")
raise UnsupportedFormatError(f"No compatible writers registered for {format_id}.")
writer_factory(*, format_id, required_level, priority=0)
¶
Decorator that registers a writer factory into this registry.
源代码位于： src/molop/io/codec_registry.py
Python
def writer_factory(
self,
*,
format_id: str,
required_level: StructureLevel,
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,
priority=priority,
)
return factory
return decorator