import json import time from dataclasses import dataclass import numpy as np SEED = 115 @dataclass class Node: lo: float hi: float slopes: list left: object = None right: object = None depth: int = 0 @property def splitting(self): return self.left is not None and self.right is not None def build_tree(slopes, lo=0.0, hi=1.0, depth=0, max_depth=32): """Build a binary M-adic interval tree, retaining unary chains as shared nodes.""" vals = sorted(float(x) for x in slopes) node = Node(lo, hi, vals, depth=depth) if len(vals) <= 1 or depth >= max_depth: return node mid = (lo + hi) / 2.0 left = [x for x in vals if x < mid] right = [x for x in vals if x >= mid] if not left or not right: # A unary chain has no genuine branch decision. node.left = build_tree(vals, lo, hi, depth + 1, max_depth) else: node.left = build_tree(left, lo, mid, depth + 1, max_depth) node.right = build_tree(right, mid, hi, depth + 1, max_depth) return node def split_counts(node, count=0): count += int(node.splitting) if node.left is None: return [count] if node.right is None: return split_counts(node.left, count) return split_counts(node.left, count) + split_counts(node.right, count) def leaves(node): if node.left is None: return [node] if node.right is None: return leaves(node.left) return leaves(node.left) + leaves(node.right) def tree_summary(root): counts = split_counts(root) return {"leaves": len(leaves(root)), "path_split_counts": counts, "max_splits": max(counts), "min_splits": min(counts)} def lacunarity_ratios(slopes, alpha=0.0): vals = np.asarray(slopes, dtype=float) d = np.abs(vals - alpha) return d[1:] / np.maximum(d[:-1], 1e-12) def sample_line(field, y, x, slope, radius=2): """Nearest-neighbour finite line average for offsets -radius..radius.""" h, w = field.shape vals = [] for t in range(-radius, radius + 1): yy = int(np.clip(round(y + t * slope), 0, h - 1)) xx = int(np.clip(round(x + t), 0, w - 1)) vals.append(field[yy, xx]) return float(np.mean(np.abs(vals))) def directional_responses(field, slopes, radius=2): h, w = field.shape out = np.empty((h, w, len(slopes)), dtype=np.float32) for y in range(h): for x in range(w): for k, slope in enumerate(slopes): out[y, x, k] = sample_line(field, y, x, slope, radius) return out def routed_response(responses, slopes, root, beam=2): """Tree route: select the best beam child/subtree by cheap representative scores. The returned value uses only beam leaves, while dense responses serve as oracle data. """ slope_to_k = {float(s): k for k, s in enumerate(slopes)} selected = [] def visit(node, candidates): if node.left is None: selected.append(node.slopes[0]); return if node.right is None: visit(node.left, candidates); return children = [node.left, node.right] scores = [] for child in children: ks = [slope_to_k[float(s)] for s in child.slopes] scores.append(float(np.mean(responses[..., ks]))) order = np.argsort(scores)[::-1][:min(beam, 2)] for j in order: visit(children[int(j)], candidates) # Routing is applied globally in this compact MVP; leaf outputs are spatial maps. visit(root, None) selected = sorted(set(selected)) ks = [slope_to_k[s] for s in selected] return np.max(responses[..., ks], axis=2), len(ks), selected def sparse_from_proxy(target, proxy, beam): """Per-query routing using a cheap proxy, then evaluate only selected target lines.""" h, w, k = target.shape chosen = np.argpartition(proxy, -beam, axis=2)[:, :, -beam:] out = np.take_along_axis(target, chosen, axis=2).max(axis=2) return out, int(h*w*beam) def run(): rng = np.random.default_rng(SEED) k = 32 # Unique M-adic directions. For a binary tree, 32 distinct leaves necessarily # require at least ceil(log2(32))=5 splitting vertices on some route. slopes = np.linspace(0.0, 1.0, k, endpoint=False).tolist() root = build_tree(slopes) summary = tree_summary(root) lac_seq = [0.5**j for j in range(1, 10)] lac = lacunarity_ratios(lac_seq) math_check = { "geometric_max_ratio": float(lac.max()), "geometric_expected_ratio": 0.5, "geometric_ratio_allclose": bool(np.allclose(lac, 0.5)), "tree_split_bound": summary["max_splits"], "tree_paths": summary["path_split_counts"], "binary_lower_bound_for_32_leaves": 5, "bound_is_valid": summary["max_splits"] >= 5, } assert np.allclose(lac, 0.5) assert summary["max_splits"] >= 5 h = w = 24 yy, xx = np.mgrid[:h, :w] field = (np.sin((xx + 0.35 * yy) / 2.2) + 0.7 * np.cos((xx - 0.8 * yy) / 3.1) + 0.15 * rng.standard_normal((h, w))).astype(np.float32) # Oracle target and a cheaper-radius proxy. The proxy is a routing surrogate, # while target responses represent the expensive line-attention computation. t0 = time.perf_counter(); target = directional_responses(field, slopes, radius=2); target_time = time.perf_counter()-t0 t0 = time.perf_counter(); proxy = directional_responses(field, slopes, radius=1); proxy_time = time.perf_counter()-t0 dense = np.max(target, axis=2) idea, idea_samples = sparse_from_proxy(target, proxy, beam=2) rng2 = np.random.default_rng(SEED) random_idx = rng2.integers(0, k, size=(h, w, 2)) random_sparse = np.take_along_axis(target, random_idx, axis=2).max(axis=2) def rel_error(a): return float(np.mean(np.abs(a-dense)) / (np.mean(np.abs(dense))+1e-8)) metrics = { "K": k, "tree": summary, "dense_samples": int(h*w*k), "idea_target_samples": idea_samples, "random_target_samples": idea_samples, "sample_reduction": float(k/2), "dense_target_ms": 1000*target_time, "proxy_routing_ms": 1000*proxy_time, "idea_relative_error": rel_error(idea), "random_relative_error": rel_error(random_sparse), "idea_better_than_random": rel_error(idea) < rel_error(random_sparse), } result = {"seed": SEED, "math_check": math_check, "metrics": metrics} with open("results.json", "w") as f: json.dump(result, f, indent=2) print(json.dumps(result, indent=2)) if __name__ == "__main__": run()