add a metadata comparison method

extra_requirements/requirements-tests.txt

@@ -1,3 +1,4 @@
 # File for the requirements of strax with the automated tests
 git+https://github.com/XENONnT/ax_env
+deepdiff==6.3.0
 ipython==8.12.1

requirements.txt

@@ -1,4 +1,6 @@
 blosc
+click
+deepdiff
 dill
 fsspec
 immutabledict

strax/context.py

@@ -5,8 +5,11 @@
 from functools import partial
 import typing as ty
 import time
+import json
 import numpy as np
 import pandas as pd
+import click
+import deepdiff
 import strax
 import inspect
 import types
@@ -1664,6 +1667,66 @@ def get_meta(self, run_id, target) -> dict:
 
     get_metadata = get_meta
 
+    def compare_metadata(self, run_id, target, old_metadata):
+        """
+        Compare the metadata between two strax data
+
+        :param run_id: run id to get
+        :param target: data type to get
+        :param old_metadata: path to metadata to compare, or a dictionary, or a tuple with
+            another run_id, target to compare against the metadata of the first id-target pair
+        """
+        color_values = lambda oldval, newval: (
+            click.style(oldval, fg='red', bold=True), click.style(newval, fg='green', bold=True))
+        underline = lambda text, bold=True: click.style(text, bold=bold, underline=True)
+
+        # new metadata for the given runid + target; fetch from context
+        new_metadata = self.get_metadata(run_id, target)
+        # old metadata to compare
+        if isinstance(old_metadata, str):
+            with open(old_metadata) as json_file:
+                old_metadata = json.load(json_file)
+        elif isinstance(old_metadata, dict):
+            old_metadata = old_metadata
+        elif isinstance(old_metadata, (tuple, list)):
+            old_metadata = self.get_metadata(old_metadata[0], old_metadata[1])
+        else:
+            raise ValueError(f"Expected old_metadata as `str` or `dict` got {type(old_metadata)}")
+
+        differences = deepdiff.DeepDiff(old_metadata, new_metadata)
+        for key, value in differences.items():
+            if key in ['values_changed', 'iterable_item_added', 'iterable_item_removed']:
+                print(underline(f"\n> {key}"))
+                for kk, vv in value.items():
+                    if key == "values_changed":
+                        old_values = vv['old_value']
+                        new_values = vv['new_value']
+                    elif key == "iterable_item_added":
+                        old_values = "-"
+                        new_values = vv
+                    else:  # if key == "iterable_item_removed":
+                        old_values = vv
+                        new_values = "-"
+                    old, new = color_values(old_values, new_values)
+                    click.secho(f"\t in {kk[4:]}", bold=False)
+                    print(f"\t\t{old} -> {new}")
+            elif key in ['dictionary_item_added', 'dictionary_item_removed']:
+                color = "red" if "removed" in key else "green"
+                print(underline(f"\n> {key:25s}"), end="->")
+                click.secho(f"\t{', '.join(value)}", fg=color)
+            elif key in ['type_changes']:
+                print(underline(f"\n> {key}"))
+                for kk, vv in value.items():
+                    click.secho(f"\t{kk}")
+                    oldtype = vv['old_type']
+                    newtype = vv['new_type']
+                    keyold, keynew = color_values('old_type', 'new_type')
+                    valueold, valuenew = color_values(vv['old_value'], vv['new_value'])
+                    print(f"\t\t{keyold:10s} : {oldtype} ({valueold})")
+                    print(f"\t\t{keynew:10s} : {newtype} ({valuenew})")
+            else:
+                raise KeyError(f"Unkown key in comparison {key}")
+
     def run_metadata(self, run_id, projection=None) -> dict:
         """
         Return run-level metadata for run_id, or raise DataNotAvailable

tests/test_context.py

@@ -439,10 +439,17 @@ def compute(self, **kwargs):
         st3 = st.new_context(register=DevelopRecords)
         assert key.lineage != st3.key_for(run_id, 'records').lineage
 
-
     @staticmethod
     def get_dummy_peaks_dependency():
         class DummyDependsOnPeaks(strax.CutPlugin):
             depends_on = 'peaks'
             provides = 'cut_peaks'
         return DummyDependsOnPeaks
+
+    def test_compare_metadata(self):
+        st = self.get_context(True)
+        st.register(Records)
+        st.make(run_id, 'records')
+        old_metadata = st.get_metadata(run_id, 'records')
+        old_metadata.pop('strax_version')
+        st.compare_metadata(run_id, 'records', old_metadata)