Spaces:
Running
Running
| """Open an ordered list of local or remote tablebase sources. | |
| Every material and recursive dependency uses the same search order, | |
| filtered by castling rights for paired root and root/castling sources. | |
| The upstream-compatible adapter retains per-material opening locks and | |
| one shared decoded-block cache without piece-count or castling tiers. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| from typing import Any, List, Optional, Sequence, Union | |
| import chess.chesstb as chesstb | |
| from tablebase_opening import PerMaterialOpening, block_cache_usage | |
| log = logging.getLogger(__name__) | |
| #: A source exposing _find()/close() and optionally clear_caches()/ | |
| #: cache_stats(). | |
| TablebaseLike = Any | |
| #: A TABLEBASE_PATH config | |
| #: value: one path (local directory or http(s):// URL), or a list of | |
| #: them for a hybrid source. | |
| PathSpec = Union[str, Sequence[str]] | |
| def _as_path_list(spec: PathSpec) -> List[str]: | |
| """Expand a tablebase root, or preserve an explicit source list.""" | |
| if not isinstance(spec, str): | |
| return [p for p in spec if p] | |
| if not spec: | |
| return [] | |
| remote = spec.startswith(("http://", "https://")) | |
| root = spec.rstrip("/\\") or spec | |
| join = (lambda *parts: "/".join(parts)) if remote else os.path.join | |
| paths = [join(root, "full"), join(root, "full", "castling"), | |
| root, join(root, "castling")] | |
| return [p for p in paths if remote or p == root or os.path.isdir(p)] | |
| def open_one(path: str, *, label: str, cfg: Any, remote_source: Any, | |
| remote_direct: Any, remote_download: Any) -> TablebaseLike: | |
| """Open one local directory or HTTP(S) source with attribution in logs.""" | |
| if remote_source.looks_like_remote(path): | |
| remote_backend_name = cfg.remote_mode | |
| if remote_backend_name == "download": | |
| tb = remote_download.open_tablebase( | |
| path, | |
| block_cache_bytes=0, | |
| remote_page_cache_bytes=cfg.remote_page_cache_bytes, | |
| remote_page_size=cfg.remote_page_size_bytes, | |
| remote_timeout=cfg.remote_timeout_secs, | |
| remote_max_retries=cfg.remote_max_retries, | |
| remote_pool_maxsize=cfg.remote_pool_maxsize, | |
| ) | |
| else: | |
| tb = remote_direct.open_tablebase( | |
| path, | |
| block_cache_bytes=0, | |
| remote_page_cache_bytes=cfg.remote_page_cache_bytes, | |
| remote_page_size=cfg.remote_page_size_bytes, | |
| remote_timeout=cfg.remote_timeout_secs, | |
| remote_max_retries=cfg.remote_max_retries, | |
| remote_pool_maxsize=cfg.remote_pool_maxsize, | |
| ) | |
| log.info( | |
| "%s opened remotely at: %s (mode=%s, %s, block_cache_bytes=%d, " | |
| "remote_page_cache_bytes=%d, remote_page_size=%d)", | |
| label, path, remote_backend_name, | |
| "byte-range, nothing written to disk" if remote_backend_name == "direct" | |
| else "whole-file download, disk-cached", | |
| 0, cfg.remote_page_cache_bytes, cfg.remote_page_size_bytes, | |
| ) | |
| else: | |
| tb = chesstb.open_tablebase(path, block_cache_bytes=0) | |
| log.info("%s opened at: %s (resolver block cache disabled)", label, path) | |
| return tb | |
| def open_many(spec: PathSpec, *, label: str, cfg: Any, remote_source: Any, | |
| remote_direct: Any, remote_download: Any) -> Optional[TablebaseLike]: | |
| """Open TABLEBASE_PATH, combining multiple entries in search order.""" | |
| paths = _as_path_list(spec) | |
| if not paths: | |
| return None | |
| sources: List[TablebaseLike] = [] | |
| try: | |
| for i, p in enumerate(paths): | |
| sources.append(open_one( | |
| p, label=f"{label}[{i}]" if len(paths) > 1 else label, | |
| cfg=cfg, remote_source=remote_source, | |
| remote_direct=remote_direct, remote_download=remote_download, | |
| )) | |
| except Exception: | |
| # A later entry in a hybrid list failing to open must not leak | |
| # the sources earlier entries already opened. | |
| for source in sources: | |
| try: | |
| source.close() | |
| except Exception: | |
| log.exception("Failed closing partially opened source for %s", label) | |
| raise | |
| return sources[0] if len(sources) == 1 else CompositeTablebase( | |
| sources, remote_source=remote_source, paths=paths, | |
| ) | |
| class _RemoteTableSource: | |
| __slots__ = ("remote_source", "file") | |
| def __init__(self, remote_source: Any, file: Any) -> None: | |
| self.remote_source = remote_source | |
| self.file = file | |
| class _HybridSourced: | |
| def _open_source(self, path: Any) -> Any: | |
| if isinstance(path, _RemoteTableSource): | |
| # Upstream's _open_source contract: _TableFile.close() releases | |
| # _data. RemoteFile.close() is intentionally a no-op. | |
| self._data = path.file | |
| return path.remote_source.RemoteFileView(path.file) | |
| return super()._open_source(path) | |
| class _HybridWDLFile(_HybridSourced, chesstb.WDLFile): | |
| pass | |
| class _HybridDTZFile(_HybridSourced, chesstb.DTZFile): | |
| pass | |
| class _HybridDTCFile(_HybridSourced, chesstb.DTCFile): | |
| pass | |
| class _HybridDTMFile(_HybridSourced, chesstb.DTMFile): | |
| pass | |
| class _HybridDTM50File(_HybridSourced, chesstb.DTM50File): | |
| pass | |
| class CompositeTablebase: | |
| """A hybrid TABLEBASE_PATH entry: an ordered list of already-opened | |
| tablebase sources - any mix of a local chesstb.Tablebase and a | |
| remote_direct/remote_fallback backend - searched left-to-right until | |
| one resolves the material. Presents the same _find()/close()/ | |
| clear_caches()/cache_stats() surface as a single chesstb.Tablebase, | |
| so SourcedTablebase and app.py's admin routes don't need to know | |
| whether the resolver uses one source or several. | |
| A RemoteSourceError from one source (a network failure, not a "this | |
| material doesn't exist" answer) doesn't abort the walk - the next | |
| source is tried too, since resilience against exactly that is the | |
| point of listing more than one source. Only once every source has | |
| either answered "not found" or errored does _find() give up: it | |
| re-raises the last transport error if there was one - silently | |
| reporting "not found" instead would misrepresent a live network | |
| problem as a coverage gap - otherwise it returns None, the normal | |
| "position not covered" answer. | |
| """ | |
| def __init__(self, sources: List[TablebaseLike], *, remote_source: Any, | |
| paths: Optional[Sequence[str]] = None) -> None: | |
| self._sources = sources | |
| self._remote_errors = (remote_source.RemoteSourceError,) | |
| # A paired root/castling layout separates ordinary and castling | |
| # material. Unpaired custom roots may contain either, so retain | |
| # their original behavior. Keep _sources intact for lifecycle/stats. | |
| self._material_sources = {False: sources, True: sources} | |
| if paths is not None: | |
| normalized = [p.rstrip("/\\") for p in paths] | |
| paired_roots = { | |
| p for p in normalized | |
| if p + "/castling" in normalized or p + "\\castling" in normalized | |
| } | |
| castling_paths = { | |
| p + suffix for p in paired_roots | |
| for suffix in ("/castling", "\\castling") | |
| } | |
| self._material_sources = { | |
| False: [s for s, p in zip(sources, normalized) if p not in castling_paths], | |
| True: [s for s, p in zip(sources, normalized) if p not in paired_roots], | |
| } | |
| def _find(self, kind: str, name: str, ext: str) -> Optional[Any]: | |
| last_error: Optional[BaseException] = None | |
| for source in self._material_sources["r" in name]: | |
| try: | |
| found = source._find(kind, name, ext) | |
| except self._remote_errors as exc: | |
| last_error = exc | |
| continue | |
| if found is not None: | |
| return found | |
| if last_error is not None: | |
| raise last_error | |
| return None | |
| def close(self) -> None: | |
| for source in self._sources: | |
| source.close() | |
| def clear_caches(self) -> None: | |
| for source in self._sources: | |
| clear_fn = getattr(source, "clear_caches", None) | |
| if clear_fn is not None: | |
| clear_fn() | |
| def cache_stats(self) -> dict: | |
| """Per-source cache stats, keyed "source_0", "source_1", ... in | |
| list order - a source with no cache of its own (a plain local | |
| chesstb.Tablebase) contributes no key rather than an empty one.""" | |
| stats = {} | |
| for i, source in enumerate(self._sources): | |
| stats_fn = getattr(source, "cache_stats", None) | |
| if stats_fn is not None: | |
| stats[f"source_{i}"] = stats_fn() | |
| return stats | |
| class SourcedTablebase(PerMaterialOpening, chesstb.Tablebase): | |
| """An upstream-compatible tablebase backed by one ordered resolver.""" | |
| WDL_FILE = _HybridWDLFile | |
| DTZ_FILE = _HybridDTZFile | |
| DTC_FILE = _HybridDTCFile | |
| DTM_FILE = _HybridDTMFile | |
| DTM50_FILE = _HybridDTM50File | |
| def __init__(self, source: TablebaseLike, *, remote_source: Any, | |
| block_cache_bytes: int) -> None: | |
| self.source = source | |
| self._remote_source = remote_source | |
| # A disk-cache resolver returns a pathname before the file is mapped. | |
| # Do not increase its open concurrency without a separate file-lease | |
| # design protecting that gap against eviction. Direct HTTP handles | |
| # and ordinary local files do not have that temporary-path lifetime. | |
| def uses_download(source: Any) -> bool: | |
| if isinstance(source, CompositeTablebase): | |
| return any(uses_download(s) for s in source._sources) | |
| return "_disk_cache" in getattr(source, "__dict__", {}) | |
| self._serialize_downloads = uses_download(source) | |
| super().__init__("", block_cache_bytes=block_cache_bytes) | |
| def add_directory(self, directory: str) -> None: | |
| """No-op: material lookup goes through the ordered source resolver.""" | |
| def _table_open_lock(self, kind: str, cache_key: Any) -> Any: | |
| # _open_locks is upstream's per-kind first-open lock table. | |
| if self._serialize_downloads: | |
| return self._open_locks[kind] | |
| return super()._table_open_lock(kind, cache_key) | |
| def _find(self, kind: str, name: str, ext: str) -> Optional[Any]: | |
| found = self.source._find(kind, name, ext) | |
| if found is None: | |
| return None | |
| if isinstance(found, str): | |
| return found | |
| if isinstance(found, _RemoteTableSource): | |
| return found | |
| return _RemoteTableSource(self._remote_source, found) | |
| def close(self) -> None: | |
| try: | |
| super().close() | |
| finally: | |
| self.source.close() | |
| def clear_caches(self) -> None: | |
| self._block_cache.clear() | |
| clear_fn = getattr(self.source, "clear_caches", None) | |
| if clear_fn is not None: | |
| clear_fn() | |
| def cache_stats(self) -> dict: | |
| blocks, size = block_cache_usage(self) | |
| stats = {"block_cache": {"blocks": blocks, "bytes": size}} | |
| stats_fn = getattr(self.source, "cache_stats", None) | |
| if stats_fn is not None: | |
| stats["source"] = stats_fn() | |
| return stats | |