python migration

server/scanner.py

@@ -0,0 +1,173 @@
+import os
+import re
+
+from db_py import clear_tests, mark_completed, upsert_test
+from parser import (
+    parse_elapsed_time,
+    parse_result_filename,
+    parse_target_filename,
+    parse_timestamp,
+    parse_tput_rssi,
+)
+
+
+def full_scan(target_dir, results_dir):
+    clear_tests()
+    if not target_dir or not results_dir:
+        return
+
+    print(f"[scanner] target dir : {target_dir}")
+    print(f"[scanner] results dir: {results_dir}")
+
+    scan_targets(target_dir)
+    scan_results(results_dir)
+
+
+def scan_targets(target_dir):
+    try:
+        parent_entries = [
+            name
+            for name in os.listdir(target_dir)
+            if os.path.isdir(os.path.join(target_dir, name))
+        ]
+    except OSError as exc:
+        print(f"[scanner] Cannot read target dir: {exc}")
+        return
+
+    print(f"[scanner] subdirectories found: {len(parent_entries)}")
+
+    for parent_name in parent_entries:
+        parent_path = os.path.join(target_dir, parent_name)
+        try:
+            files = [
+                name
+                for name in os.listdir(parent_path)
+                if os.path.isfile(os.path.join(parent_path, name))
+                and not name.startswith("GLOBAL")
+                and name.endswith(".ini")
+            ]
+        except OSError as exc:
+            print(f"[scanner] Cannot read parent dir {parent_name}: {exc}")
+            continue
+
+        print(f"[scanner]   {parent_name} -> {len(files)} target test file(s)")
+
+        for filename in files:
+            parsed = parse_target_filename(filename, parent_name)
+            if not parsed["test_id"]:
+                print(f"[scanner] skip (no test_id): {parent_name}/{filename}")
+                continue
+
+            print(
+                f"[scanner] found: {parent_name}/{filename}  ->  test_id={parsed['test_id']}"
+            )
+            upsert_test(
+                {
+                    "id": parsed["id"],
+                    "test_id": parsed["test_id"],
+                    "parent_dir": parent_name,
+                    "filename": filename,
+                    "interference": parsed["interference"],
+                    "device": parsed["device"],
+                    "rotation": parsed["rotation"],
+                    "test_point": parsed["test_point"],
+                    "station": parsed["station"],
+                    "band": parsed["band"],
+                    "channel": parsed["channel"],
+                    "bandwidth": parsed["bandwidth"],
+                    "rssi": parsed["rssi"],
+                    "direction": parsed["direction"],
+                }
+            )
+
+
+def scan_results(results_dir):
+    try:
+        entries = [
+            name
+            for name in os.listdir(results_dir)
+            if os.path.isdir(os.path.join(results_dir, name))
+        ]
+    except OSError as exc:
+        print(f"[scanner] Cannot read results dir: {exc}")
+        return
+
+    print(f"[scanner] results: {len(entries)} result dir(s) found")
+    for dir_name in entries:
+        process_result_dir(results_dir, dir_name)
+
+
+def process_result_dir(results_dir, result_dir_name):
+    parsed = parse_result_filename(result_dir_name)
+    test_id = parsed["test_id"]
+    device = parsed["device"]
+    if not test_id or not device:
+        return
+
+    result_dir_path = os.path.join(results_dir, result_dir_name)
+
+    try:
+        log_files = [
+            name
+            for name in os.listdir(result_dir_path)
+            if os.path.isfile(os.path.join(result_dir_path, name))
+            and name.endswith(".txt")
+            and test_id in name
+        ]
+    except OSError as exc:
+        print(f"[scanner] Cannot read result dir {result_dir_name}: {exc}")
+        return
+
+    if not log_files:
+        print(f"[scanner] completed (no log yet): {test_id}")
+        mark_completed(test_id, device, None, None)
+        return
+
+    latest_log = sorted(log_files)[-1]
+    log_path = os.path.join(result_dir_path, latest_log)
+    completed_at = parse_timestamp(latest_log)
+    data = extract_log_data(log_path)
+
+    print(
+        f"[scanner] completed: {test_id}  device={device}  "
+        f"duration={data['duration_seconds']}s  stations={len(data['tputResults'])}  at={completed_at}"
+    )
+    mark_completed(
+        test_id,
+        device,
+        completed_at,
+        data["duration_seconds"],
+        data["tputResults"],
+    )
+
+
+def parse_deleted_result_dir_name(dir_name):
+    segments = re.split(r"[_\-]", dir_name)
+    test_id = next((s for s in segments if re.match(r"^R\d+[A-Z0-9]+$", s, re.IGNORECASE)), None)
+    device = next((s for s in segments if re.match(r"^CGW\d+$", s, re.IGNORECASE)), None)
+    return test_id, device
+
+
+def extract_log_data(log_file_path):
+    elapsed_regex = re.compile(r"\[.*?INFO\]\s+Elapsed time\s*:\s*([\d]+:[\d]{2}:[\d]{2}\.[\d]+)")
+    tput_regex = re.compile(
+        r"\[.*?INFO\]\s+STA(\d+) over angles >> AVG IxChariot TPUT: (\d+) Mbps, AVG DL RSSI: (-?\d+) dBm, AVG UL RSSI: (-?\d+) dBm"
+    )
+
+    result = {"duration_seconds": None, "tputResults": []}
+    try:
+        with open(log_file_path, "r", encoding="utf-8", errors="ignore") as fh:
+            for line in fh:
+                time_match = elapsed_regex.search(line)
+                if time_match:
+                    result["duration_seconds"] = parse_elapsed_time(time_match.group(1))
+                    continue
+
+                if tput_regex.search(line):
+                    parsed = parse_tput_rssi(line)
+                    if parsed:
+                        result["tputResults"].append(parsed)
+    except OSError as exc:
+        print(f"[scanner] Cannot read log file {log_file_path}: {exc}")
+
+    return result