Source code for daf.storage.chains

"""
A chain of storage objects (first one wins).

This serves two purposes. First, it is only a `.StorageReader`, so it protects the wrapped storage objects against
accidental modification, even if they implement `.StorageWriter` as well. Second, it allows delta-encoding, where data
in early storage objects enhances or even overrides the data in later objects. For example, this is used in `.DafReader`
to present a unified view of the data in the derived cache and the base storage.
"""

# pylint: disable=duplicate-code,cyclic-import

from __future__ import annotations

from typing import Any
from typing import Collection
from typing import Dict
from typing import List
from typing import Sequence
from typing import Set
from typing import Tuple

import numpy as np

from ..typing import Known1D
from ..typing import Known2D
from ..typing import Vector
from ..typing import as_vector
from . import interface as _interface
from . import none as _none

# pylint: enable=duplicate-code,cyclic-import

__all__ = ["StorageChain"]


# pylint: disable=protected-access


[docs]class StorageChain(_interface.StorageReader): """ Low-level read-only access to a ``chain`` of storage objects (first one wins). If the ``name`` ends with ``#``, we append the object id to it to make it unique. This is different from other storage implementations in that modifications to the wrapped storage objects are guaranteed to be immediately visible in the chain (in contrast to, for example, `.StorageView` where this is **not** the case). The implementation of `.DafWriter` relies on this fact. """ def __init__(self, chain: Sequence[_interface.StorageReader], *, name: str = "chain#") -> None: if name.endswith("#"): name += str(id(self)) super().__init__(name=name) unique_readers: List[_interface.StorageReader] = [] unique_ids: Set[int] = set() for reader in chain: StorageChain._add_reader(unique_readers, unique_ids, reader) #: The unique chained `.StorageReader` objects (first one wins). self.chain: Tuple[_interface.StorageReader, ...] = tuple(unique_readers) self._verify() @staticmethod def _add_reader( unique_readers: List[_interface.StorageReader], unique_ids: Set[int], reader: _interface.StorageReader ) -> None: if isinstance(reader, _none.NoStorage): return if isinstance(reader, StorageChain): for chained_reader in reader.chain: StorageChain._add_reader(unique_readers, unique_ids, chained_reader) return reader_id = id(reader) if reader_id in unique_ids: return unique_readers.append(reader) unique_ids.add(reader_id) def _verify(self) -> None: axes_entries: Dict[str, Tuple[str, Vector]] = {} for storage in self.chain: new_name = storage.name for axis in storage._axis_names(): new_entries = as_vector(storage._axis_entries(axis)) old_data = axes_entries.get(axis) if old_data is None: axes_entries[axis] = (new_name, new_entries) else: old_name, old_entries = old_data assert np.all(old_entries == new_entries), ( f"inconsistent entries for the axis: {axis} " f"between the storage: {old_name} " f"and the storage: {new_name} " f"in the storage chain: {self.name}" ) def _self_description(self, self_description: Dict, *, detail: bool) -> None: self_description["chain"] = [storage.name for storage in self.chain] def _deep_description(self, description: Dict, self_description: Dict, *, detail: bool) -> None: for storage in self.chain: storage.description(description=description, detail=detail) def _item_names(self) -> Collection[str]: names: Set[str] = set() for storage in self.chain: names.update(storage._item_names()) return names def _has_item(self, name: str) -> bool: for storage in self.chain: if storage._has_item(name): return True return False def _get_item(self, name: str) -> Any: for storage in self.chain: if storage._has_item(name): return storage._get_item(name) assert False, "never happens" def _axis_names(self) -> Collection[str]: names: Set[str] = set() for storage in self.chain: names.update(storage._axis_names()) return names def _has_axis(self, axis: str) -> bool: for storage in self.chain: if storage._has_axis(axis): return True return False def _axis_size(self, axis: str) -> int: for storage in self.chain: if storage._has_axis(axis): return storage._axis_size(axis) assert False, "never happens" def _axis_entries(self, axis: str) -> Known1D: for storage in self.chain: if storage._has_axis(axis): return storage._axis_entries(axis) assert False, "never happens" def _data1d_names(self, axis: str) -> Collection[str]: names: Set[str] = set() for storage in self.chain: if storage._has_axis(axis): names.update(storage._data1d_names(axis)) return names def _has_data1d(self, axis: str, name: str) -> bool: for storage in self.chain: if storage._has_axis(axis) and storage._has_data1d(axis, name): return True return False def _get_data1d(self, axis: str, name: str) -> Known1D: for storage in self.chain: if storage._has_axis(axis) and storage.has_data1d(name): return storage._get_data1d(axis, name) assert False, "never happens" def _data2d_names(self, axes: Tuple[str, str]) -> Collection[str]: names: Set[str] = set() for storage in self.chain: if storage._has_axis(axes[0]) and storage.has_axis(axes[1]): names.update(storage._data2d_names(axes)) return names def _has_data2d(self, axes: Tuple[str, str], name: str) -> bool: for storage in self.chain: if storage._has_axis(axes[0]) and storage._has_axis(axes[1]) and storage._has_data2d(axes, name): return True return False def _get_data2d(self, axes: Tuple[str, str], name: str) -> Known2D: for storage in self.chain: if storage._has_axis(axes[0]) and storage._has_axis(axes[1]) and storage._has_data2d(axes, name): return storage._get_data2d(axes, name) assert False, "never happens"