implement coordinate hash

dascore/config.py

@@ -99,6 +99,10 @@ class DascoreConfig(BaseModel):
         default=1_048_576,
         description="Block size in bytes for general remote file downloads.",
     )
+    remote_download_timeout: float = Field(
+        default=60.0,
+        description="Timeout in seconds for blocking remote file downloads.",
+    )
     remote_hdf5_block_size: int = Field(
         default=5_242_880,
         description="Block size in bytes for remote HDF5 access on tuned protocols.",

dascore/core/coords.py

@@ -38,7 +38,11 @@
     get_quantity_str,
     percent,
 )
-from dascore.utils.array import _coerce_text_array, _is_text_coercible_array
+from dascore.utils.array import (
+    _coerce_text_array,
+    _is_text_coercible_array,
+    hash_array,
+)
 from dascore.utils.display import get_nice_text
 from dascore.utils.docs import compose_docstring
 from dascore.utils.misc import (
@@ -450,6 +454,57 @@ def __array__(self, dtype=None, copy=False):
         """Numpy method for getting array data with `np.array(coord)`."""
         return self.data
 
+    def _get_hashable_coord(self) -> Self:
+        """Return a coordinate normalized for stable hashing."""
+        if self.units is None or dtype_time_like(self.dtype):
+            return self
+        return self.simplify_units()
+
+    @staticmethod
+    def _hash_scalar(value) -> tuple[str, str | None]:
+        """Return a dtype-aware scalar hash token."""
+        if value is None:
+            return ("none", None)
+        return ("scalar", hash_array(np.asarray([value])))
+
+    @staticmethod
+    def _hash_array_token(values: np.ndarray) -> tuple[str, str | tuple[int, ...]]:
+        """Return a hash token consistent with coord array equality."""
+        dtype = np.dtype(values.dtype)
+        if np.issubdtype(dtype, np.inexact):
+            # Coord equality for inexact arrays uses np.isclose semantics, so
+            # hashing raw bytes would violate the hash contract. Fall back to
+            # metadata-only hashing for these dtypes.
+            return ("approx-array", str(dtype), values.shape)
+        return ("array", hash_array(values))
+
+    def __hash__(self):
+        coord = self._get_hashable_coord()
+        unit_str = coord.unit_str
+        if isinstance(coord, CoordRange):
+            payload = (
+                coord.__class__,
+                unit_str,
+                coord.shape,
+                self._hash_scalar(coord.start),
+                self._hash_scalar(coord.stop),
+                self._hash_scalar(coord.step),
+            )
+            return hash(payload)
+        if isinstance(coord, CoordPartial):
+            payload = (
+                coord.__class__,
+                unit_str,
+                coord.shape,
+                str(np.dtype(coord.dtype)),
+                self._hash_scalar(coord.start),
+                self._hash_scalar(coord.stop),
+                self._hash_scalar(coord.step),
+            )
+            return hash(payload)
+        payload = (coord.__class__, unit_str, self._hash_array_token(coord.values))
+        return hash(payload)
+
     @cached_method
     def min(self):
         """Return min value."""
@@ -1016,7 +1071,20 @@ def update(self, **kwargs):
     # Other operations that normally modify data do not in this case.
     update_limits = update
     set_units = update
-    convert_units = update
+
+    def convert_units(self, units) -> Self:
+        """Convert scalar metadata units, or set units if none exist."""
+        if self.units is None or dtype_time_like(self.dtype):
+            return self.set_units(units=units)
+        out = {"units": units}
+        for name in ("start", "stop", "step"):
+            value = getattr(self, name)
+            out[name] = (
+                value
+                if pd.isnull(value)
+                else convert_units(value, to_units=units, from_units=self.units)
+            )
+        return self.new(**out)
 
     def sort(self, reverse=False):
         """Sort dummy array. Does nothing."""

dascore/utils/array.py

@@ -622,7 +622,7 @@ def patch_array_function(self, func, types, args, kwargs):
     return apply_array_func(func, *args, **kwargs)
 
 
-def hash_numpy_array(arr: np.ndarray) -> str:
+def hash_array(arr: np.ndarray) -> str:
     """
     Return a stable hash for a NumPy array.
 
@@ -644,18 +644,18 @@ def hash_numpy_array(arr: np.ndarray) -> str:
     Examples
     --------
     >>> import numpy as np
-    >>> from dascore.utils.array import hash_numpy_array
+    >>> from dascore.utils.array import hash_array
     >>> a = np.array([1.0, 2.0, 3.0])
-    >>> h = hash_numpy_array(a)
+    >>> h = hash_array(a)
     >>> assert isinstance(h, str) and len(h) == 32
     >>> # Same data always produces the same hash
-    >>> assert hash_numpy_array(a) == hash_numpy_array(a.copy())
+    >>> assert hash_array(a) == hash_array(a.copy())
     >>> # Different dtype produces a different hash
-    >>> assert hash_numpy_array(a) != hash_numpy_array(a.astype(np.float32))
+    >>> assert hash_array(a) != hash_array(a.astype(np.float32))
     """
     arr = np.asarray(arr)
     if arr.dtype == object:
-        msg = "hash_numpy_array does not support object arrays."
+        msg = "hash_array does not support object arrays."
         raise ParameterError(msg)
 
     h = hashlib.blake2b(digest_size=16)
@@ -665,11 +665,17 @@ def hash_numpy_array(arr: np.ndarray) -> str:
     h.update(arr.dtype.str.encode("ascii"))
     h.update(np.asarray(arr.shape, dtype=np.int64).tobytes())
 
-    if arr.flags.c_contiguous:
+    if arr.flags.c_contiguous and arr.dtype.kind not in {"M", "m"}:
         # Zero-copy fast path
         h.update(memoryview(arr).cast("B"))
     else:
-        # Canonicalize layout; this copies once
+        # Canonicalize layout; this also handles datetime/timedelta dtypes,
+        # which do not expose a Python buffer directly.
         h.update(np.ascontiguousarray(arr).view(np.uint8))
 
     return h.hexdigest()
+
+
+def hash_numpy_array(arr: np.ndarray) -> str:
+    """Backward-compatible alias for `hash_array`."""
+    return hash_array(arr)

dascore/utils/hdf5.py

@@ -50,6 +50,7 @@
 from dascore.utils.remote_io import (
     FallbackFileObj,
     ensure_local_file,
+    get_cached_local_file,
     get_local_handle,
     is_no_range_http_error,
 )
@@ -63,6 +64,50 @@
 ns_to_timedelta = partial(pd.to_timedelta, unit="ns")
 
 
+class _ManagedH5pyFile:
+    """Proxy an h5py file while owning the underlying file object lifecycle."""
+
+    def __init__(self, handle: H5pyFile, owned_fileobj):
+        self._handle = handle
+        self._owned_fileobj = owned_fileobj
+        self._closed = False
+
+    def close(self):
+        """Close both the h5py file and the owned file object."""
+        if self._closed:
+            return
+        try:
+            self._handle.close()
+        finally:
+            with suppress(Exception):
+                self._owned_fileobj.close()
+            self._closed = True
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False
+
+    def __getitem__(self, item):
+        return self._handle[item]
+
+    def __contains__(self, item):
+        return item in self._handle
+
+    def __iter__(self):
+        return iter(self._handle)
+
+    @property
+    def closed(self):
+        """Return True when close has been called on the proxy."""
+        return self._closed
+
+    def __getattr__(self, item):
+        return getattr(self._handle, item)
+
+
 class _HDF5Store(pd.HDFStore):
     """
     A work-around for pandas HDF5 store not accepting
@@ -473,6 +518,12 @@ class H5Reader(PyTablesReader):
     mode = "r"
     constructor = H5pyFile
 
+    @classmethod
+    def _open_fileobj_handle(cls, fileobj):
+        """Open an h5py file and retain ownership of the wrapped file object."""
+        handle = cls.constructor(fileobj, mode=cls.mode, driver="fileobj")
+        return _ManagedH5pyFile(handle, fileobj)
+
     @staticmethod
     def _get_open_kwargs(resource: UPath) -> dict[str, object]:
         """Return backend-specific kwargs for remote HDF5 file objects."""
@@ -495,11 +546,13 @@ def get_handle(cls, resource):
         Unlike PyTablesReader, h5py can consume a binary file object via the
         ``fileobj`` driver, so remote UPath inputs stay streaming-based here.
         """
-        if isinstance(resource, cls | H5pyFile):
+        if isinstance(resource, cls | H5pyFile | _ManagedH5pyFile):
             return resource
         if isinstance(resource, io.IOBase):
-            return cls.constructor(resource, mode=cls.mode, driver="fileobj")
+            return cls._open_fileobj_handle(resource)
         if isinstance(resource, UPath):
+            if cached_path := get_cached_local_file(resource):
+                return super().get_handle(cached_path)
             mode = "rb" if cls.mode == "r" else "r+b"
             open_kwargs = cls._get_open_kwargs(resource)
             handle = FallbackFileObj(
@@ -508,7 +561,7 @@ def get_handle(cls, resource):
                 error_predicate=is_no_range_http_error,
             )
             try:
-                return cls.constructor(handle, mode=cls.mode, driver="fileobj")
+                return cls._open_fileobj_handle(handle)
             except Exception:
                 handle.close()
                 raise

dascore/utils/remote_io.py

@@ -12,6 +12,7 @@
 from functools import lru_cache
 from hashlib import sha256
 from pathlib import Path
+from urllib.request import Request, urlopen
 
 from dascore.compat import UPath
 from dascore.config import get_config
@@ -125,7 +126,6 @@ def _download_remote_file(path, local_path: Path):
     """Download a remote path into its cache location."""
     resource = coerce_to_upath(path)
     protocol = getattr(resource, "protocol", None)
-    open_kwargs = {"block_size": 0} if protocol in _HTTP_PROTOCOLS else {}
     local_path.parent.mkdir(parents=True, exist_ok=True)
     fd, temp_name = tempfile.mkstemp(
         dir=local_path.parent,
@@ -135,12 +135,20 @@ def _download_remote_file(path, local_path: Path):
     os.close(fd)
     tmp_path = Path(temp_name)
     try:
-        with (
-            resource.open("rb", **open_kwargs) as remote_fi,
-            tmp_path.open("wb") as local_fi,
-        ):
-            while chunk := remote_fi.read(get_config().remote_download_block_size):
-                local_fi.write(chunk)
+        if protocol in _HTTP_PROTOCOLS:
+            headers = dict(getattr(resource, "storage_options", {}) or {})
+            request = Request(str(resource), headers=headers)
+            timeout = get_config().remote_download_timeout
+            with (
+                urlopen(request, timeout=timeout) as remote_fi,
+                tmp_path.open("wb") as local_fi,
+            ):
+                while chunk := remote_fi.read(get_config().remote_download_block_size):
+                    local_fi.write(chunk)
+        else:
+            with resource.open("rb") as remote_fi, tmp_path.open("wb") as local_fi:
+                while chunk := remote_fi.read(get_config().remote_download_block_size):
+                    local_fi.write(chunk)
         tmp_path.replace(local_path)
     finally:
         tmp_path.unlink(missing_ok=True)
@@ -201,6 +209,19 @@ def ensure_local_file(resource) -> Path:
     raise TypeError(msg)
 
 
+def get_cached_local_file(resource) -> Path | None:
+    """Return the cached local path for one remote resource if it exists."""
+    if not is_pathlike(resource) or is_local_path(resource):
+        return None
+    remote = coerce_to_upath(resource)
+    cache_root = _normalize_cache_root(get_remote_cache_path())
+    remote_id = normalize_remote_id(remote)
+    local_path = (
+        cache_root / sha256(remote_id.encode()).hexdigest() / _safe_remote_name(remote)
+    )
+    return local_path if local_path.exists() else None
+
+
 def get_local_handle(resource, opener):
     """Materialize a resource locally, then pass it to an opener."""
     return opener(ensure_local_file(resource))