Aluode/PerceptionLabPortable

1from __future__ import annotations2 3import abc4import hashlib5 6from fsspec.implementations.local import make_path_posix7 8 9class AbstractCacheMapper(abc.ABC):10    """Abstract super-class for mappers from remote URLs to local cached11    basenames.12    """13 14    @abc.abstractmethod15    def __call__(self, path: str) -> str: ...16 17    def __eq__(self, other: object) -> bool:18        # Identity only depends on class. When derived classes have attributes19        # they will need to be included.20        return isinstance(other, type(self))21 22    def __hash__(self) -> int:23        # Identity only depends on class. When derived classes have attributes24        # they will need to be included.25        return hash(type(self))26 27 28class BasenameCacheMapper(AbstractCacheMapper):29    """Cache mapper that uses the basename of the remote URL and a fixed number30    of directory levels above this.31 32    The default is zero directory levels, meaning different paths with the same33    basename will have the same cached basename.34    """35 36    def __init__(self, directory_levels: int = 0):37        if directory_levels < 0:38            raise ValueError(39                "BasenameCacheMapper requires zero or positive directory_levels"40            )41        self.directory_levels = directory_levels42 43        # Separator for directories when encoded as strings.44        self._separator = "_@_"45 46    def __call__(self, path: str) -> str:47        path = make_path_posix(path)48        prefix, *bits = path.rsplit("/", self.directory_levels + 1)49        if bits:50            return self._separator.join(bits)51        else:52            return prefix  # No separator found, simple filename53 54    def __eq__(self, other: object) -> bool:55        return super().__eq__(other) and self.directory_levels == other.directory_levels56 57    def __hash__(self) -> int:58        return super().__hash__() ^ hash(self.directory_levels)59 60 61class HashCacheMapper(AbstractCacheMapper):62    """Cache mapper that uses a hash of the remote URL."""63 64    def __call__(self, path: str) -> str:65        return hashlib.sha256(path.encode()).hexdigest()66 67 68def create_cache_mapper(same_names: bool) -> AbstractCacheMapper:69    """Factory method to create cache mapper for backward compatibility with70    ``CachingFileSystem`` constructor using ``same_names`` kwarg.71    """72    if same_names:73        return BasenameCacheMapper()74    else:75        return HashCacheMapper()76