File size: 6,314 Bytes
c5414cc add7d16 c5414cc 6413bf6 c5414cc 6413bf6 c5414cc e1d0aa6 6413bf6 e1d0aa6 c5414cc e1d0aa6 c5414cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | import unittest
from services.scoring.compression_score_cache import (
find_duplicate_compression_scores,
)
class CompressionScoreCacheTests(unittest.TestCase):
def find_duplicates(
self,
cache,
uid,
input_id="video-1",
vmaf_score=85.89,
base_vmaf_score=88.22,
vmaf_threshold=85.0,
compression_rate=0.0907,
final_score=0.5399,
):
return find_duplicate_compression_scores(
[uid],
[input_id],
[vmaf_score],
[base_vmaf_score],
[vmaf_threshold],
[compression_rate],
[final_score],
cache,
)
def test_subsequent_uid_is_duplicate_across_scorer_batches(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83), {})
self.assertEqual(self.find_duplicates(cache, uid=41), {0: 83})
def test_hidden_float_differences_and_derived_score_drift_still_match(self):
cache = {}
self.assertEqual(
self.find_duplicates(
cache,
uid=41,
vmaf_score=89.110001,
base_vmaf_score=91.110001,
vmaf_threshold=89.0,
compression_rate=0.08240001,
final_score=0.5714,
),
{},
)
self.assertEqual(
self.find_duplicates(
cache,
uid=83,
vmaf_score=89.109999,
base_vmaf_score=91.109999,
vmaf_threshold=89.0,
compression_rate=0.08239999,
final_score=0.5715,
),
{0: 41},
)
def test_subsequent_uid_loses_each_matching_synthetic_input_slot(self):
cache = {}
input_ids = [f"video-{idx}" for idx in range(5)]
vmaf_scores = [85.89, 85.16, 85.88, 85.13, 85.04]
base_vmaf_scores = [88.22, 87.50, 88.20, 87.46, 87.45]
compression_rates = [0.0907, 0.0708, 0.0905, 0.0713, 0.0787]
final_scores = [0.5399, 0.6341, 0.5409, 0.6307, 0.5887]
first_duplicates = find_duplicate_compression_scores(
[83] * 5,
input_ids,
vmaf_scores,
base_vmaf_scores,
[85.0] * 5,
compression_rates,
final_scores,
cache,
)
subsequent_duplicates = find_duplicate_compression_scores(
[41] * 5,
input_ids,
vmaf_scores,
base_vmaf_scores,
[85.0] * 5,
compression_rates,
final_scores,
cache,
)
self.assertEqual(first_duplicates, {})
self.assertEqual(subsequent_duplicates, {idx: 83 for idx in range(5)})
def test_same_metrics_for_different_inputs_get_separate_slots(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83, input_id="video-1"), {})
self.assertEqual(self.find_duplicates(cache, uid=41, input_id="video-2"), {})
def test_compression_rate_outside_tolerance_gets_a_separate_slot(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83), {})
self.assertEqual(
self.find_duplicates(cache, uid=41, compression_rate=0.1010),
{},
)
def test_vmaf_outside_tolerance_gets_a_separate_slot(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83), {})
self.assertEqual(
self.find_duplicates(cache, uid=41, vmaf_score=86.10),
{},
)
def test_small_compression_rate_drift_across_batches_is_duplicate(self):
cache = {}
input_ids = [f"video-{idx}" for idx in range(5)]
vmaf_scores = [85.27, 85.40, 85.48, 85.22, 85.06]
base_vmaf_scores = [87.49, 87.39, 87.61, 87.32, 86.92]
first_duplicates = find_duplicate_compression_scores(
[23] * 5,
input_ids,
vmaf_scores,
base_vmaf_scores,
[85.0] * 5,
[0.0909, 0.1403, 0.1170, 0.1331, 0.1229],
[0.5359, 0.4097, 0.4574, 0.4217, 0.4415],
cache,
)
subsequent_duplicates = find_duplicate_compression_scores(
[164] * 5,
input_ids,
vmaf_scores,
base_vmaf_scores,
[85.0] * 5,
[0.0912, 0.1406, 0.1173, 0.1334, 0.1232],
[0.5347, 0.4092, 0.4566, 0.4211, 0.4409],
cache,
)
self.assertEqual(first_duplicates, {})
self.assertEqual(subsequent_duplicates, {idx: 23 for idx in range(5)})
def test_small_vmaf_and_compression_drift_across_batches_is_duplicate(self):
cache = {}
input_ids = [f"video-{idx}" for idx in range(5)]
first_duplicates = find_duplicate_compression_scores(
[23] * 5,
input_ids,
[89.01, 89.31, 89.26, 89.23, 89.24],
[91.15, 91.53, 91.51, 91.38, 91.51],
[89.0] * 5,
[0.1259, 0.1073, 0.1116, 0.0969, 0.1160],
[0.4348, 0.4824, 0.4703, 0.5142, 0.4589],
cache,
)
subsequent_duplicates = find_duplicate_compression_scores(
[164] * 5,
input_ids,
[89.02, 89.21, 89.24, 89.27, 89.14],
[91.17, 91.43, 91.50, 91.43, 91.43],
[89.0] * 5,
[0.1267, 0.1080, 0.1131, 0.0977, 0.1158],
[0.4333, 0.4796, 0.4664, 0.5117, 0.4588],
cache,
)
self.assertEqual(first_duplicates, {})
self.assertEqual(subsequent_duplicates, {idx: 23 for idx in range(5)})
def test_non_positive_result_does_not_claim_a_slot(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83, final_score=0.0), {})
self.assertEqual(self.find_duplicates(cache, uid=41), {})
self.assertEqual(
cache["video-1"].popitem()[1],
[(85.89, 88.22, 0.0907, 41)],
)
def test_same_uid_can_reuse_its_own_signature(self):
cache = {}
self.assertEqual(self.find_duplicates(cache, uid=83), {})
self.assertEqual(self.find_duplicates(cache, uid=83), {})
if __name__ == "__main__":
unittest.main()
|