Recipe hashes

pangeo_forge_recipes/patterns.py

@@ -248,6 +248,12 @@ def items(self):
         for key in self:
             yield key, self[key]
 
+    def sha256(self):
+        """Compute a sha256 hash for the instance."""
+        from .serialization import pattern_blockchain
+
+        return pattern_blockchain(self).pop(-1)
+
 
 def pattern_from_file_sequence(file_list, concat_dim, nitems_per_file=None, **kwargs):
     """Convenience function for creating a FilePattern from a list of files."""

pangeo_forge_recipes/recipes/base.py

@@ -12,6 +12,7 @@
 @dataclass
 class BaseRecipe(ABC):
     _compiler: ClassVar[RecipeCompiler]
+    _hash_exclude_ = ["file_pattern", "storage_config"]
 
     def to_function(self):
         from ..executors import FunctionPipelineExecutor
@@ -38,6 +39,11 @@ def to_beam(self):
 
         return BeamPipelineExecutor.compile(self._compiler())
 
+    def sha256(self):
+        from ..serialization import dataclass_sha256
+
+        return dataclass_sha256(self, ignore_keys=self._hash_exclude_)
+
 
 RecipeCompiler = Callable[[BaseRecipe], Pipeline]
 

pangeo_forge_recipes/serialization.py

@@ -0,0 +1,68 @@
+from dataclasses import asdict
+from enum import Enum
+from hashlib import sha256
+from json import dumps
+from typing import List
+
+
+def json_default(thing):
+    # custom serializer for FileType, CombineOp, etc.
+    if isinstance(thing, Enum):
+        return thing.value
+    raise TypeError(f"object of type {type(thing).__name__} not serializable")
+
+
+def dict_to_sha256(thing):
+    # https://death.andgravity.com/stable-hashing
+    b = dumps(
+        thing,
+        default=json_default,
+        ensure_ascii=False,
+        sort_keys=True,
+        indent=None,
+        separators=(",", ":"),
+    )
+    return sha256(b.encode("utf-8")).digest()
+
+
+def dataclass_sha256(dclass: type, ignore_keys: List[str]) -> bytes:
+    d = asdict(dclass)
+    for k in ignore_keys:
+        del d[k]
+    return dict_to_sha256(d)
+
+
+def pattern_blockchain(pattern):
+    # this contains general aspects of the file pattern
+    # we exclude the file pattern and concat dims because they are generated by iterating
+    # if they are generated in a different way, we ultimately don't care
+    root = {
+        "fsspec_open_kwargs": pattern.fsspec_open_kwargs,
+        "query_string_secrets": pattern.query_string_secrets,
+        "file_type": pattern.file_type,
+        "nitems_per_file": [
+            op.nitems_per_file for op in pattern.combine_dims if op.name in pattern.concat_dims
+        ],
+    }
+
+    root_sha256 = dict_to_sha256(root)
+
+    blockchain = [root_sha256]
+    for k, v in pattern.items():
+        key_hash = b"".join(
+            sorted([dataclass_sha256(dimindex, ignore_keys=["sequence_len"]) for dimindex in k])
+        )
+        value_hash = sha256(v.encode("utf-8")).digest()
+        new_hash = key_hash + value_hash
+        new_block = sha256(blockchain[-1] + new_hash).digest()
+        blockchain.append(new_block)
+
+    return blockchain
+
+
+def match_pattern_blockchain(old_pattern_last_hash, new_pattern):
+
+    new_chain = pattern_blockchain(new_pattern)
+    for k, h in zip(new_pattern, new_chain):
+        if h == old_pattern_last_hash:
+            return k

tests/test_serialization.py

@@ -0,0 +1,63 @@
+from datetime import datetime, timedelta
+
+import pandas as pd
+import pytest
+
+from pangeo_forge_recipes.patterns import ConcatDim, FilePattern
+from pangeo_forge_recipes.serialization import match_pattern_blockchain
+
+URL_FORMAT = (
+    "https://www.ncei.noaa.gov/data/sea-surface-temperature-optimum-interpolation/"
+    "v2.1/access/avhrr/{time:%Y%m}/oisst-avhrr-v02r01.{time:%Y%m%d}.nc"
+)
+
+
+def make_file_pattern(dates, nitems_per_file):
+    def make_url(time):
+        return URL_FORMAT.format(time=time)
+
+    time_concat_dim = ConcatDim("time", dates, nitems_per_file=nitems_per_file)
+    pattern = FilePattern(make_url, time_concat_dim)
+
+    return pattern
+
+
+@pytest.fixture
+def end_date():
+    return "2022-02-01"
+
+
+@pytest.fixture
+def old_pattern_sha256(end_date):
+    dates = pd.date_range("1981-09-01", end_date, freq="D")
+    pattern = make_file_pattern(dates, nitems_per_file=1)
+    return pattern.sha256()
+
+
+def get_new_pattern_with_next_url(end_date, nitems_per_file):
+
+    fmt = "%Y-%m-%d"
+
+    def increment_end_date(ndays):
+        return datetime.strptime(end_date, fmt) + timedelta(days=ndays)
+
+    next_day = increment_end_date(ndays=1)
+    new_end_date = increment_end_date(ndays=90).strftime(fmt)
+    new_dates = pd.date_range("1981-09-01", new_end_date, freq="D")
+    new_pattern = make_file_pattern(new_dates, nitems_per_file=nitems_per_file)
+    return new_pattern, URL_FORMAT.format(time=next_day)
+
+
+@pytest.mark.parametrize("new_pattern_nitems_per_file", [1, 2])
+def test_match_pattern_blockchain(
+    old_pattern_sha256,
+    end_date,
+    new_pattern_nitems_per_file,
+):
+    new_pattern, next_url = get_new_pattern_with_next_url(end_date, new_pattern_nitems_per_file)
+    matching_key = match_pattern_blockchain(old_pattern_sha256, new_pattern)
+
+    if new_pattern_nitems_per_file == 1:
+        assert new_pattern[matching_key] == next_url
+    elif new_pattern_nitems_per_file == 2:
+        assert matching_key is None