Streaming Polyphase Factor Decomposition / SKAT

Download original file

This is the original source text. Reading it does not run or verify the program.

Read the typeset paper

1 import unittest
2 import numpy as np
3 
4 from budget_scheduler import (
5     additive_scalar_bound,
6     brute_force_unit_optimum,
7     depth_edge_risk_vector,
8     double_buffer_pipeline_estimate,
9     edge_risks_depth,
10     exact_depth_transfer_envelope,
11     exact_envelope_refinement,
12     prune_pairs,
13     symmetric_offdiag_pairs,
14     unit_cost_optimal_additive_schedule,
15 )
16 
17 
18 class BudgetSchedulerTests(unittest.TestCase):
19     def setUp(self):
20         rng = np.random.default_rng(123)
21         q = 6
22         B = np.zeros((q, q))
23         for i in range(q):
24             B[i, i] = 0.70 + 0.10 * rng.random()
25         for i in range(q):
26             for j in range(i + 1, q):
27                 d = j - i
28                 scale = 0.12 if d == 1 else 0.02 / d
29                 B[i, j] = B[j, i] = scale * (0.7 + 0.6 * rng.random())
30         self.B = B
31         self.rho = np.array([1.2, 0, 0, 0, 0, 0], dtype=float)
32         self.depth = 4
33         self.pairs = symmetric_offdiag_pairs(q)
34 
35     def test_edge_additive_vector_bound(self):
36         risks = edge_risks_depth(self.B, self.rho, self.depth)
37         rng = np.random.default_rng(4)
38         for _ in range(30):
39             omitted = [p for p in self.pairs if rng.random() < 0.35]
40             B0 = prune_pairs(self.B, omitted)
41             exact_vec = (
42                 np.linalg.matrix_power(self.B, self.depth)
43                 - np.linalg.matrix_power(B0, self.depth)
44             ) @ self.rho
45             add_vec = np.sum(
46                 [r.vector for r in risks if r.pair in set(omitted)],
47                 axis=0,
48             ) if omitted else np.zeros_like(self.rho)
49             self.assertTrue(np.all(exact_vec <= add_vec + 1e-12))
50 
51     def test_unit_cost_optimum_matches_bruteforce_additive(self):
52         budget = 0.012
53         sched = unit_cost_optimal_additive_schedule(
54             self.B, self.rho, self.depth, budget
55         )
56         brute = brute_force_unit_optimum(
57             self.B, self.rho, self.depth, budget,
58             use_exact_envelope=False,
59         )
60         self.assertEqual(len(sched.omitted_pairs), len(brute.omitted_pairs))
61         self.assertLessEqual(sched.additive_risk_bound, budget + 1e-15)
62 
63     def test_exact_refinement_remains_certified(self):
64         budget = 0.012
65         sched = unit_cost_optimal_additive_schedule(
66             self.B, self.rho, self.depth, budget
67         )
68         refined = exact_envelope_refinement(
69             self.B, self.rho, self.depth, budget, sched.omitted_pairs
70         )
71         self.assertTrue(refined.certificate_valid)
72         self.assertGreaterEqual(len(refined.omitted_pairs), len(sched.omitted_pairs))
73 
74     def test_double_buffer_formula(self):
75         io = [2.0, 2.0, 2.0, 2.0]
76         comp = [3.0, 3.0, 3.0, 3.0]
77         r = double_buffer_pipeline_estimate(
78             io, comp, resident_block_bytes=100, fixed_working_bytes=50
79         )
80         self.assertAlmostEqual(r.single_buffer_time, 20.0)
81         self.assertAlmostEqual(r.double_buffer_time, 14.0)
82         self.assertEqual(r.resident_bytes_single, 150)
83         self.assertEqual(r.resident_bytes_double, 250)
84 
85 
86 if __name__ == "__main__":
87     unittest.main(verbosity=2)

Back to source controls

Back to Streaming Polyphase Factor Decomposition / SKAT