Honestly don't know what I was last working on here

Levalicious · Levalicious · commit e58a439eb652 · 2025-06-04T10:25:38.000-04:00
diff --git a/notes/invsets.md b/notes/invsets.md
@@ -0,0 +1,87 @@
+# Invsets
+
+Given a set $U$, a permutation oracle $O \rightarrow \{0, 1\}$
+
+An order is defined as a series of sets $S_1$, $S_2$, ..., $S_n$ where $|S_n| = n$ and $S_n \subseteq U$, i.e. $\emptyset \subset S_1 \subset S_2 \subset ... \subset S_n \subseteq U$.
+
+The inversion set ($invset(\pi)$) of a permutation $\pi$ of $n$ elements is the set of pairs $(i, j)$ such that $i < j$ and $\pi(i) > \pi(j)$
+
+For each pair $(i, j) | i \neq j$, initialize counters:
+- $M^+_{(i, j)} \gets 0$ // positive (accepted) count
+- $M^-_{(i, j)} \gets 0$ // negative (rejected) count
+
+For each permutation $\pi$ queried:
+- If $O(\pi) = 1$ (accepted): for each $(i, j) \in invset(\pi)$, increment $M^+_{(i, j)} \gets M^+_{(i, j)} + 1$
+- If $O(\pi) = 0$ (rejected): for each $(i, j) \in invset(\pi)$, increment $M^-_{(i, j)} \gets M^-_{(i, j)} + 1$
+
+$P_{observed}^+(i, j) = M^+_{(i, j)} / N^+$ (frequency in positive samples)
+
+$P_{observed}^-(i, j) = M^-_{(i, j)} / N^-$ (frequency in negative samples)
+
+$P_{observed+}(x) = (1-\epsilon)P(1) + \epsilon(1 - P(1))$
+$P_{observed-}(x) = (1-\epsilon)P(0) + \epsilon(1 - P(0))$
+$P(1) = 1 - P(0)$
+
+
+Greatest Lower Bound
+
+ $meet(A, B) \rightarrow invset(A) \cap invset(B)$
+
+Greatest Upper Bound
+ $join(A, B) \rightarrow invset(A) \cup invset(B)$
+
+Assume oracle outputs are flipped with probability $\epsilon$. We can estimate the 'true' probability $P(x)$ from our observed samples $P_{obs}(x)$ as follows:
+
+$P_{observed}(x) = (1 - \epsilon) * P(x) + \epsilon * (1 - P(x))$
+
+$P_{observed}(x) = P(x) - \epsilon P(x)  + \epsilon - \epsilon P(x)$
+
+$P_{observed}(x) = P(x) - \epsilon P(x) - \epsilon P(x) + \epsilon$
+
+$P_{observed}(x) = P(x) (1 - \epsilon - \epsilon) + \epsilon$
+
+$P_{observed}(x) = P(x) (1 - 2\epsilon) + \epsilon$
+
+$P_{observed}(x) - \epsilon = P(x) (1 - 2\epsilon)$
+
+$\frac{P_{observed}(x) - \epsilon}{1 - 2\epsilon} = P(x)$
+
+
+$$ f = (1-p)q + p(1-q) $$
+$$ f = q - pq + p - pq $$
+$$ f = q - pq + p - pq $$
+$$ f = q(1-p-p) + p $$
+$$ f = q(1-2p) + p $$
+$$ f - p = q (1 - 2p) $$
+$$ \frac{f - p}{1 - 2p} = q $$
+x = (1 - p) * x + p * (x)
+
+
+$P(\text{observed} = 1 \mid \text{true} = 0) = p$
+$P(\text{observed} = 1 \mid \text{true} = 1) = 1-p$
+$P(\text{observed} = 0 \mid \text{true} = 0) = 1-p$
+$P(\text{observed} = 0 \mid \text{true} = 1) = p$
+
+
+if $O(A) = 1$ and $O(B) = 1$
+
+ ?
+
+if $O(A) = 0$ and $O(B) = 0$
+
+ ?
+
+if $O(A) = 1$ and $O(B) = 0$
+
+ ?
+
+if $O(A) = 0$ and $O(B) = 1$
+
+ ?
+
+# DB
+
+Record:
+- Key
+- Version
+- Payload
diff --git a/plugin_oracle/base/db.py b/plugin_oracle/base/db.py
@@ -2,6 +2,7 @@
 import pickle
 import random
 from typing import TypeVar, Generic, Protocol
+from copy import deepcopy
 
 from plugin_oracle.util.mod.mod import Mod
 
@@ -14,9 +15,9 @@ class HasHashAndInd(Protocol):
 class MSet(Generic[T]):
     def __init__(self) -> None:
         self._U: dict[bytes, T] = {}
-        self._fsets: dict[bytes, set[bytes]] = {}
-        self._isets: dict[bytes, set[bytes]] = {}
+        self._fsets: dict[bytes, dict[bytes, float]] = {}
         self._bound: tuple[list[bytes], list[bytes]] = ([], [])
+        self._e: float = 0.2
 
     @property
     def U(self) -> dict[bytes, T]:
@@ -30,42 +31,65 @@ def min(self) -> list[bytes]:
     def max(self) -> list[bytes]:
         return self._bound[1]
     
-    def __getitem__(self, hash: bytes) -> tuple[T | None, set[bytes] | None, set[bytes] | None]:
-        return (self._U.get(hash, None), self._fsets.get(hash, None), self._isets.get(hash, None))
+    def __getitem__(self, hash: bytes) -> tuple[T, dict[bytes, float]]:
+        return (self._U[hash], self._fsets[hash])
+    
+    def __contains__(self, hash: bytes) -> bool:
+        return hash in self._U
 
     def addU(self, v: T) -> None:
         if v.hash in self._U:
             return
         self._U[v.hash] = v
-        self._fsets[v.hash] = self._fsets.get(v.hash, set())
-        self._fsets[v.hash].update(self._U.keys())
-        self._fsets[v.hash].discard(v.hash)
-        for k in self._U.keys():
-            if k != v.hash:
-                self._fsets[k].add(v.hash)
-        self._isets[v.hash] = self._isets.get(v.hash, set())
-        self._isets[v.hash].update(self._U.keys())
-        self._isets[v.hash].discard(v.hash)
+        self._fsets[v.hash] = self._fsets.get(v.hash, {})
+        self._fsets[v.hash].update({k: 1.0 for k in self._U.keys()})
+        del self._fsets[v.hash][v.hash]
         for k in self._U.keys():
             if k != v.hash:
-                self._isets[k].add(v.hash)
+                self._fsets[k][v.hash] = 1.0
     
     def permutation(self, perm: list[bytes], state: bool) -> None:
         if any(hash not in self._U for hash in perm):
             raise ValueError('Unfiltered permutation encountered')
         if state:
-            if perm < self._bound[0]:
+            if perm < self._bound[0] or self._bound[0] == []:
                 self._bound = (perm, self._bound[1])
-            if perm > self._bound[1]:
+            if perm > self._bound[1] or self._bound[1] == []:
                 self._bound = (self._bound[0], perm)
-
-        for i in range(len(perm)):
-            fset = self._fsets[perm[i]] if state else self._isets[perm[i]]
-            for j in range(i):
-                fset.discard(perm[j])
+        if state:
+            for i in range(len(perm)):
+                fset = self._fsets[perm[i]]
+                for j in range(i):
+                    fset[perm[j]] *= self._e
+        else:
+            l = len(perm)
+            tU = set(self._U.keys())
+            tlen = len(tU)
+            for i in range(l):
+                tU.discard(perm[i])
+                fset = self._fsets[perm[i]]
+                tlen -= 1
+                for j in fset.keys():
+                    if j in tU:
+                        fset[j] += ((1 - self._e) / (tlen * (tlen - 1)))
+                        if fset[j] > 1.0:
+                            fset[j] = 1.0
+    
+    def edrop(self) -> 'MSet[T]':
+        out = MSet[T]()
+        out._U = self._U.copy() # pyright: ignore [reportConstantRedefinition]
+        out._fsets = deepcopy(self._fsets)
+        for k in out._U.keys():
+            fset = out._fsets[k]
+            follow = list(fset.keys())
+            for m in follow:
+                if fset[m] < (1 - self._e):
+                    del fset[m]
+        out._bound = self._bound
+        return out
     
-    def rtoposort(self, state: bool, seed: int = 0) -> list[bytes] | None:
-        adj = self._fsets if state else self._isets
+    def rtoposort(self, seed: int = 0) -> list[bytes] | None:
+        adj = self._fsets
         indegree = {k: 0 for k in adj}
         for vs in adj.values():
             for v in vs:
@@ -76,7 +100,7 @@ def rtoposort(self, state: bool, seed: int = 0) -> list[bytes] | None:
         while queue:
             n = queue.pop(random.randrange(len(queue)))
             L.append(n)
-            for m in adj.get(n, set()):
+            for m in adj.get(n, {}):
                 indegree[m] -= 1
                 if indegree[m] == 0:
                     queue.append(m)
@@ -85,8 +109,8 @@ def rtoposort(self, state: bool, seed: int = 0) -> list[bytes] | None:
             return None
         return L
     
-    def toposort(self, state: bool) -> list[bytes] | None:
-        adj = self._fsets if state else self._isets
+    def toposort(self) -> list[bytes] | None:
+        adj = self._fsets
         indegree = {k: 0 for k in adj}
         for vs in adj.values():
             for v in vs:
@@ -96,16 +120,71 @@ def toposort(self, state: bool) -> list[bytes] | None:
         while queue:
             n = queue.pop()
             L.append(n)
-            for m in adj.get(n, set()):
+            for m in adj.get(n, {}):
                 indegree[m] -= 1
                 if indegree[m] == 0:
                     queue.append(m)
         if len(L) != len(indegree):
             return None
         return L
+    
+    def assemble(self) -> None | list[tuple[bytes, bytes]]:
+        leximin: list[bytes] = self._bound[0]
+        leximax: list[bytes] = self._bound[1]
+        if leximin == [] or leximax == []:
+            return None
+        anc: dict[bytes, list[None | bytes]] = {b: [None, None] for b in leximin}
+        # Boolean markers for validation.
+        hv: dict[bytes, bool] = {b: False for b in leximin}
+        stack: list[bytes] = []
+
+        # First pass: based on leximin.
+        for b in leximin:
+            while stack and stack[-1] < b:
+                _ = stack.pop()
+            if stack:
+                anc[b][0] = stack[-1]
+            stack.append(b)
+
+        stack.clear()
+
+        # Second pass: based on leximax.
+        for b in leximax:
+            while stack and stack[-1] > b:
+                _ = stack.pop()
+            hv[b] = True
+            # If a parent was set in first pass but hasn't been marked yet, it's inconsistent.
+            r = anc[b][0]
+            if r is not None and not hv.get(r, False):
+                return None
+            if stack:
+                if anc[b][0] is not None:
+                    anc[b][1] = stack[-1]
+                else:
+                    anc[b][0] = stack[-1]
+            stack.append(b)
+
+        # Reset hv markers.
+        for b in leximin:
+            hv[b] = False
+
+        # Third pass: validate that all assigned parents appear earlier.
+        for b in leximin:
+            hv[b] = True
+            for parent in anc[b]:
+                if parent is not None and not hv[parent]:
+                    return None
+
+        # Build the edge list: for every byte b, each non-None parent becomes an edge (parent -> b).
+        elist: list[tuple[bytes, bytes]] = []
+        for b in leximin:
+            for parent in anc[b]:
+                if parent is not None:
+                    elist.append((parent, b))
+        return elist
 
 class MDB:
-    _fname: str = '/db_0.pkl'
+    _fname: str = '/db_1.pkl'
 
     def __init__(self) -> None:
         self.mod: MSet[Mod] = MSet[Mod]()
diff --git a/plugin_oracle/base/oracle/oracle.py b/plugin_oracle/base/oracle/oracle.py
@@ -93,7 +93,7 @@ def observe(self, result: bool, mlist: IModList, organizer: IOrganizer) -> None:
 
     def sample(self, mlist: IModList, plist: IPluginList, organizer: IOrganizer) -> None:
         t0 = time()
-        ld = self.db.mod.rtoposort(True)
+        ld = self.db.mod.edrop().rtoposort()
         if ld is None:
             self._log.warning('Failed to find a topological sort!')
             return
@@ -126,9 +126,9 @@ def predict(self, mlist: IModList, organizer: IOrganizer) -> str:
         loadorder = self.permutation(mlist, organizer)
         report: list[tuple[str, str]] = []
         for i, hash in enumerate(loadorder):
-            _, fset, _ = self.db.mod[hash]
+            _, fset = self.db.mod[hash]
             for j in range(i + 1, len(loadorder)):
-                if fset is None or loadorder[j] not in fset:
+                if loadorder[j] not in fset:
                     m0: Mod = self.db.mod.U.get(hash, Mod(hash))
                     m1: Mod = self.db.mod.U.get(loadorder[j], Mod(loadorder[j]))
                     self.db.mod.addU(m0)
diff --git a/plugin_oracle/base/window.py b/plugin_oracle/base/window.py
@@ -18,17 +18,28 @@ def __init__(self, oracle: Oracle, samplers: list[Callable[[bool], None]], repor
         layout = QVBoxLayout()
         self.setLayout(layout)
         tab_widget = QTabWidget()
-        tabs: list[QWidget] = [QWidget()]
-        layouts: list[QVBoxLayout] = [QVBoxLayout()]
-        tabnames: list[str] = ['Graph']
+        tabs: list[QWidget] = [QWidget(), QWidget()]
+        layouts: list[QVBoxLayout] = [QVBoxLayout(), QVBoxLayout()]
+        tabnames: list[str] = ['Graph', 'Bounds']
 
-        tsort = self.oracle.db.mod.rtoposort(True)
+        graph = self.oracle.db.mod.edrop()
+        tsort = graph.toposort()
         if tsort is None:
             label = QLabel("Cannot render: the mod graph contains a cycle.")
             layouts[0].addWidget(label)
         else:
-            graph_widget = OracleGraph(tsort, self.oracle.db.mod) # pyright: ignore [reportArgumentType]
+            graph_widget = OracleGraph(tsort, graph) # pyright: ignore [reportArgumentType]
             layouts[0].addWidget(graph_widget)
+        
+        elist = self.oracle.db.mod.assemble()
+        if not elist:
+            label = QLabel("Cannot render: the mod graph is empty.")
+            layouts[1].addWidget(label)
+        else:
+            labels = {k: v.name for k, v in self.oracle.db.mod.U.items()}
+            bound_graph_widget = OracleGraph.from_elist(elist, labels)
+            layouts[1].addWidget(bound_graph_widget)
+
         for i in range(len(tabs)):
             tabs[i].setLayout(layouts[i])
             _ = tab_widget.addTab(tabs[i], tabnames[i])
@@ -66,10 +77,22 @@ def on_predict(self):
             _ = QMessageBox.warning(self, "Predict Error", str(e))
 
 class OracleGraph(QWidget):
-    def __init__(self, order: list[bytes], mset: MSet[HasHashAndInd], parent: QWidget | None = None) -> None:
+    @staticmethod
+    def from_mset(order: list[bytes], mset: MSet[HasHashAndInd]) -> 'OracleGraph':
+        config: MetroConfig = MetroConfig()
+        renderer: MetroRender = MetroRender.from_mset(config, order, mset)
+        return OracleGraph(config, renderer)
+    
+    @staticmethod
+    def from_elist(elist: list[tuple[bytes, bytes]], labels: dict[bytes, str]) -> 'OracleGraph':
+        config: MetroConfig = MetroConfig()
+        renderer: MetroRender = MetroRender.from_edgelist_and_labels(config, elist, labels)
+        return OracleGraph(config, renderer)
+    
+    def __init__(self, config: MetroConfig, renderer: MetroRender, parent: QWidget | None = None) -> None:
         super().__init__(parent)
-        self.config: MetroConfig = MetroConfig()
-        self.renderer: MetroRender = MetroRender.from_mset(self.config, order, mset)
+        self.config: MetroConfig = config
+        self.renderer: MetroRender = renderer
         self._scale: float = 4.0
         self._offset_x: float = 100
         self._offset_y: float = 0
diff --git a/plugin_oracle/util/ml/order.py b/plugin_oracle/util/ml/order.py
@@ -0,0 +1,8 @@
+
+def inversion(x: list[bytes]) -> set[tuple[bytes, bytes]]:
+    oset: set[tuple[bytes, bytes]] = set()
+    for i in range(len(x)):
+        for j in range(i + 1, len(x)):
+            if x[i] > x[j]:
+                oset.add((x[i], x[j]))
+    return oset
