93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
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)
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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),
|
|
}
|