p3p pairs

server/db_py.py

@@ -66,6 +66,7 @@ def _init_db():
             "tput_results TEXT",
             "throttled TEXT",
             "coe_pair TEXT",
+            "p3p_pair TEXT",
         ]:
             try:
                 _conn.execute(f"ALTER TABLE tests ADD COLUMN {col_def}")
@@ -152,6 +153,19 @@ def set_coe_pair(test_row_id, coe_pair):
         )
 
 
+def set_p3p_pair(test_row_id, p3p_pair):
+    json_value = json.dumps(p3p_pair or [])
+    with _tx():
+        _conn.execute(
+            """
+            UPDATE tests
+            SET p3p_pair = ?
+            WHERE id = ?
+            """,
+            (json_value, test_row_id),
+        )
+
+
 def get_station_for_test(test_id, device):
     with _lock:
         row = _conn.execute(

server/scanner.py

@@ -2,7 +2,14 @@ import os
 import re
 import smbclient
 
-from db_py import clear_tests, get_all_tests, mark_completed, set_coe_pair, upsert_test
+from db_py import (
+    clear_tests,
+    get_all_tests,
+    mark_completed,
+    set_coe_pair,
+    set_p3p_pair,
+    upsert_test,
+)
 from parser import (
     parse_elapsed_time,
     parse_result_filename,
@@ -111,6 +118,7 @@ def scan_results_only(results_dir, results_dir_ref=None):
     if results_dir_ref:
         scan_results(results_dir_ref)
     update_p2p_coe_pairs()
+    update_p3p_throttle_pairs()
 
 
 def full_scan(target_dir, results_dir, results_dir_ref=None):
@@ -132,6 +140,7 @@ def full_scan(target_dir, results_dir, results_dir_ref=None):
     if results_dir_ref:
         scan_results(results_dir_ref)
     update_p2p_coe_pairs()
+    update_p3p_throttle_pairs()
 
 
 def update_p2p_coe_pairs():
@@ -174,6 +183,58 @@ def update_p2p_coe_pairs():
 
     print(f"[scanner] coe_pair updated for {updated} P2P test(s)")
 
+def update_p3p_throttle_pairs():
+    tests = get_all_tests()
+
+    p3p_lookup = {}
+    for test in tests:
+        if test.get("interference") != "P3P":
+            continue
+
+        device = test.get("device")
+        test_id = test.get("test_id")
+        if not device or not test_id:
+            continue
+
+        p3p_lookup.setdefault(f"{str(device).upper()}_{str(test_id).upper()}", []).append(test)
+
+    def _pair_test_id(test_id, throttled):
+        if not test_id or not throttled:
+            return None
+
+        upper_id = str(test_id).upper()
+        upper_throttled = str(throttled).upper()
+
+        if upper_throttled == "TH":
+            return re.sub("TH", "UT", upper_id, count=1)
+        if upper_throttled == "UT":
+            return re.sub("UT", "TH", upper_id, count=1)
+        return None
+
+    updated = 0
+    for test in tests:
+        if test.get("interference") != "P3P":
+            continue
+
+        device = test.get("device")
+        test_id = test.get("test_id")
+        pair_test_id = _pair_test_id(test_id, test.get("throttled"))
+
+        pairs = []
+        if device and pair_test_id:
+            lookup_key = f"{str(device).upper()}_{str(pair_test_id).upper()}"
+            matches = p3p_lookup.get(lookup_key, [])
+            for match in matches:
+                match_device = match.get("device")
+                match_test_id = match.get("test_id")
+                if match_device and match_test_id:
+                    pairs.append(f"{match_device}_{match_test_id}")
+
+        set_p3p_pair(test.get("id"), sorted(set(pairs)))
+        updated += 1
+
+    print(f"[scanner] p3p_pair updated for {updated} P3P test(s)")
+
 
 def scan_targets(target_dir):
     target_dir = _normalize_input_path(target_dir)