test_dashboard/server/scanner.py

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