test_dashboard/server/parser.py

import re

INTERFERENCE_TYPES = {"COE", "P2P", "P3P"}
DIRECTION_TYPES = {"UL", "DL"}


def parse_target_filename(filename, parent_dir):
    base_name = re.sub(r"\.ini$", "", filename, flags=re.IGNORECASE)
    segments = base_name.split("_")[2:]

    test_identifier = f"{parent_dir}/{base_name}"
    interference = segments[0] if segments and segments[0] in INTERFERENCE_TYPES else None
    test_id = next((s for s in segments if re.match(r"^R\d+[A-Z0-9]+$", s, re.IGNORECASE)), None)
    throttled = None
    if interference == "P3P" and test_id:
        # P3P test_id carries throttle marker: TH = throttled, otherwise UT.
        throttled = "TH" if "TH" in test_id.upper() else "UT"

    device = segments[1] if len(segments) > 1 else None
    test_point = next((s for s in segments if re.match(r"^TPT\w+$", s)), None)
    rssi = next((s for s in segments if re.match(r"^RSSI\d+$", s)), None)
    station = next((s for s in segments if re.match(r"^STA\d+$", s)), None)
    band = next((s for s in segments if re.match(r"^\dGHZ$", s)), None)
    channel = next((s for s in segments if re.match(r"^CH\d+$", s)), None)
    bandwidth = next((s for s in segments if re.match(r"^BW\d+$", s)), None)
    sp = next ((s for s in segments if re.match(r"^SP", s)), None)

    last_segment = segments[-1] if segments else None
    direction = last_segment if last_segment in DIRECTION_TYPES else None

    rotation = None
    if parent_dir:
        rotation = next((s for s in parent_dir.split("_") if re.match(r"^ROT\d+$", s)), None)

    return {
        "id": test_identifier,
        "test_id": test_id,
        "interference": interference,
        "device": device,
        "test_point": test_point,
        "rssi": rssi,
        "station": station,
        "band": band,
        "channel": channel,
        "bandwidth": bandwidth,
        "direction": direction,
        "rotation": rotation,
        "throttled": throttled,
        "sp": sp,
    }


def parse_result_filename(filename):
    segments = re.split(r"[_\-]", filename)
    test_id = next((s for s in segments if re.match(r"^R\d+[A-Z0-9]+$", s)), None)
    device = segments[1] if len(segments) > 1 else None
    return {"test_id": test_id, "device": device}


def parse_timestamp(filename):
    match = re.search(r"(\d{4}-\d{2}-\d{2}-\d{2}-\d{2}(?:-\d{2})?)(?:\.\w+)?$", filename)
    if not match:
        return None

    parts = match.group(1).split("-")
    year, month, day, hour, minute = parts[:5]
    second = parts[5] if len(parts) > 5 else "00"
    return f"{year}-{month}-{day}T{hour}:{minute}:{second}"


def parse_elapsed_time(time_str):
    match = re.match(r"^(\d+):(\d{2}):(\d{2})\.(\d+)$", time_str)
    if not match:
        return None

    hours, minutes, seconds, fraction = match.groups()
    return int(hours) * 3600 + int(minutes) * 60 + int(seconds) + float(f"0.{fraction}")


def parse_tput_rssi(line):
    match = re.search(
        r"STA(\d+) over angles >> AVG IxChariot TPUT: (\d+) Mbps, AVG DL RSSI: (-?\d+) dBm, AVG UL RSSI: (-?\d+) dBm",
        line,
    )
    if not match:
        return None

    station, tput, dl_rssi, ul_rssi = match.groups()
    return {
        "station": int(station),
        "tput": int(tput),
        "dlRssi": int(dl_rssi),
        "ulRssi": int(ul_rssi),
    }