Lift HDI calculation to its own module

New functionality from HDI allows the calculation of the smallest
interval that excludes a given value. Its added complexity adds code,
which makes putting it in its own module cleaner.

app.py

@@ -4,6 +4,7 @@ import itertools as it
 import functools as ft
 import collections as cl
 from pathlib import Path
+from dataclasses import fields, asdict
 
 import pandas as pd
 import gradio as gr
@@ -12,27 +13,9 @@ import matplotlib.pyplot as plt
 from datasets import load_dataset
 from scipy.special import expit
 
-HDI = cl.namedtuple('HDI', 'lower, upper')
-TabGroup = cl.namedtuple('TabGroup', 'name, docs, dataset')
-
-#
-# See https://cran.r-project.org/package=HDInterval
-#
-def hdi(values, ci=0.95):
-    values = sorted(filter(math.isfinite, values))
-    if not values:
-        raise ValueError('Empty data set')
-
-    n = len(values)
-    exclude = n - math.floor(n * ci)
+from hdinterval import HDI, HDInterval
 
-    left = it.islice(values, exclude)
-    right = it.islice(values, n - exclude, None)
-
-    diffs = ((x, y, y - x) for (x, y) in zip(left, right))
-    (*args, _) = min(diffs, key=op.itemgetter(-1))
-
-    return HDI(*args)
+TabGroup = cl.namedtuple('TabGroup', 'name, docs, dataset')
 
 #
 #
@@ -60,14 +43,15 @@ def load(repo):
 def summarize(df, ci=0.95):
     def _aggregate(i, g):
         values = g['value']
-        interval = hdi(values, ci)
+        hdi = HDInterval(values)
+        interval = hdi(ci)
 
         agg = {
             'model': i,
             'ability': values.median(),
-            'uncertainty': interval.upper - interval.lower,
+            'uncertainty': interval.width(),
         }
-        agg.update(interval._asdict())
+        agg.update(asdict(interval))
 
         return agg
 
@@ -150,17 +134,20 @@ class RankPlotter(DataPlotter):
 class ComparisonPlotter(DataPlotter):
     def __init__(self, df, model_1, model_2, ci=0.95):
         super().__init__(compare(df, model_1, model_2))
-        self.interval = hdi(self.df, ci)
+        self.hdi = HDInterval(self.df)
+        self.ci = ci
 
     def draw(self, ax):
+        interval = self.hdi(self.ci)
+
         sns.ecdfplot(self.df, ax=ax)
 
         (_, color, *_) = sns.color_palette()
         ax.axvline(x=self.df.median(),
                    color=color,
                    linestyle='dashed')
-        ax.axvspan(xmin=self.interval.lower,
-                   xmax=self.interval.upper,
+        ax.axvspan(xmin=interval.lower,
+                   xmax=interval.upper,
                    alpha=0.15,
                    color=color)
         ax.set_xlabel('Pr(M$_{1}$ \u003E M$_{2}$)')
@@ -205,7 +192,7 @@ def layout(tab):
 
     with gr.Row():
         view = rank(summarize(df), False)
-        columns = { x: f'HDI {x}' for x in HDI._fields }
+        columns = { x.name: f'HDI {x.name}' for x in fields(HDI) }
         for i in view.columns:
             columns.setdefault(i, i.title())
         view = (view

hdinterval.py

@@ -0,0 +1,87 @@
+import math
+import warnings
+import operator as op
+import itertools as it
+import functools as ft
+import statistics as st
+from dataclasses import dataclass
+
+@dataclass
+class HDI:
+    lower: float
+    upper: float
+
+    def __iter__(self):
+        yield from (self.lower, self.upper)
+
+    def __contains__(self, item):
+        return self.lower <= item <= self.upper
+
+    def width(self):
+        return self.upper - self.lower
+
+class HDInterval:
+    @ft.cached_property
+    def values(self):
+        view = sorted(filter(math.isfinite, self._values))
+        if not view:
+            raise AttributeError('Empty data set')
+
+        return view
+
+    def __init__(self, values):
+        self._values = values
+
+    #
+    # See https://cran.r-project.org/package=HDInterval
+    #
+    def __call__(self, ci=0.95):
+        if ci == 1:
+            args = (self.values[x] for x in (0, -1))
+        else:
+            n = len(self.values)
+            exclude = n - math.floor(n * ci)
+
+            left = it.islice(self.values, exclude)
+            right = it.islice(self.values, n - exclude, None)
+
+            diffs = ((x, y, y - x) for (x, y) in zip(left, right))
+            (*args, _) = min(diffs, key=op.itemgetter(-1))
+
+        return HDI(*args)
+
+    def _at(self, target, tolerance, ci, jump):
+        if ci > 1:
+            return 1
+
+        interval = self(ci)
+        if any(math.isclose(x, target, abs_tol=tolerance) for x in interval):
+            return ci
+
+        plus_minus = op.sub if target in interval else op.add
+        ci = plus_minus(ci, jump)
+        jump /= 2
+
+        return self._at(target, tolerance, ci, jump)
+
+    def at(self, target, tolerance=1e-3):
+        while tolerance < 1:
+            try:
+                return self._at(target, tolerance, 1, 1)
+            except RecursionError:
+                tolerance *= 10
+                warnings.warn(f'Tolerance reduced: {tolerance}')
+
+        raise OverflowError()
+
+if __name__ == '__main__':
+    import numpy as np
+
+    data = np.random.uniform(size=2000)
+    # data = list(filter(lambda x: x > 0.7, data))
+    # data = [0.5] * 10
+
+    interval = HDInterval(data)
+    point = interval.at(0.5)
+    hdi = interval(point)
+    print(point, hdi)