WIP - Initial commit for parallel metrics.

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "scipy",
     "tables-io",
     "deprecated",
+    "pytdigest",
 ]
 
 # On a mac, install optional dependencies with `pip install '.[dev]'` (include the single quotes)

src/qp/metrics/base_metric_classes.py

@@ -2,6 +2,7 @@
 
 from abc import ABC
 
+from pytdigest import TDigest
 
 class MetricInputType(enum.Enum):
     """Defines the various combinations of input types that metric classes accept."""
@@ -160,3 +161,44 @@ class PointToDistMetric(BaseMetric):
 
     def evaluate(self, estimate, reference):
         raise NotImplementedError()
+
+    def eval_from_iterator(self, estimate, reference):
+        self.initialize()
+        for estimate, reference in zip(estimate, reference):
+            self.accumulate(estimate, reference)
+        return self.finalize()
+
+    def initialize(self):
+        pass
+
+    def accumulate(self, estimate, reference):
+        raise NotImplementedError()
+
+    def finalize(self):
+        raise NotImplementedError()
+
+
+class PointToPointMetricDigester(PointToPointMetric):
+
+    def accumulate(self, estimate, reference):
+        # centroids = estimate.ancil['z_mode'] - reference
+        # digest = TDigest.make_centroied(centroids)  # Or something like this
+        digest = TDigest.compute(estimate, compression=100)
+        centroids = digest.get_centroids()
+        return centroids
+
+    def finalize(self):
+        # ents = comm.gather()
+        # meta_digest = TDigest.from_centroid(cents)  # Or something like this
+        # return self.compute_from_digest(meta_digest)
+        centroids = comm.gather(centroids, root=0) # ???
+
+        #? Does this need to be the more complex version from the example? i.e.
+        # digests = (
+        #     TDigest.of_centroids(centroid, compression=COMPRESSION)
+        #     for centroid in chain.from_iterable(centroids)
+        # )
+        # digest = reduce(add, digests)
+        digest = TDigest.of_centroids(centroids, compression=100)
+
+        return self.compute_from_digest(digest)

src/qp/metrics/point_estimate_metric_classes.py

@@ -2,7 +2,9 @@
 from qp.metrics.base_metric_classes import (
     MetricOutputType,
     PointToPointMetric,
+    PointToPointMetricDigester
 )
+from pytdigest import TDigest
 
 
 class PointStatsEz(PointToPointMetric):
@@ -37,7 +39,7 @@ def evaluate(self, estimate, reference):
         return (estimate - reference) / (1.0 + reference)
 
 
-class PointSigmaIQR(PointToPointMetric):
+class PointSigmaIQR(PointToPointMetricDigester):
     """Calculate sigmaIQR"""
 
     metric_name = "point_stats_iqr"
@@ -68,6 +70,18 @@ def evaluate(self, estimate, reference):
         sigma_iqr = iqr / 1.349
         return sigma_iqr
 
+    def accumulate(self, estimate, reference):
+        ez = (estimate - reference) / (1.0 + reference)
+        digest = TDigest.compute(ez, compression=100)
+        centroids = digest.get_centroids()
+        return centroids
+
+    def compute_from_digest(self, digest):
+        x75, x25 = digest.inverse_cdf([0.75,0.25])
+        iqr = x75 - x25
+        sigma_iqr = iqr / 1.349
+        return sigma_iqr
+
 
 class PointBias(PointToPointMetric):
     """calculates the bias of the point stats ez samples.