Mia.Wu/scheduler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #2 from Mia-Wu_wnc/tc_files_migration

Browse Source 
Tc files migration
This commit is contained in:
Mia Wu
2026-07-29 11:33:00 -04:00
committed by GitHub
parent 0e7f55ed8b 9f141ccc8b
commit b6c3cf9786
20 changed files with 841 additions and 749 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -5
View File
@@ -1,10 +1,8 @@
/backend/.venv /backend/.venv
/backend/__pycache__ /backend/__pycache__
/backend/_run_sched_test.py
/backend/output.txt /backend/output.txt
/backend/.env /backend/.env
IMPLEMENTATIONPLAN.md /backend/tests
scheduler.db /backend/.pytest_cache
scheduler.db* *.db
TARGET_TEST_DIR_STRUCTURE.md
.env .env
ONE_DAY_SCHEDULER.md
-40
View File
@@ -1,40 +0,0 @@
# Redesign Scheduler:
Instead of creating a whole schedule at once, schedule tests for today first, then fill in the rest with the original scheduler design. Add a button above remake schedule that opens up a modal and show the following. User has to confirm this new modal before able to click remake schedule.
## Backend:
- Device: keeps track of what device should tests be performed on today.
- DUT priority queue keeps track of:
- rerun required DUT tests
- REF priority queue keeps track of:
- DUT completed tests mirrored (priority 1)
- rerun required REF tests (prirority 2)
- rest of the test bundles in TC
- P2P COE test bundles priority 3
- P2P only test bundles priority 4
- P3P test bundles priority 5
Workflow:
1. fill pending queues:
- fill pending queues with all tests bundles in the current TC
- when there are DUT tests completed last test window, add to REF pending queue with priority 1
- when there are DUT tests rerun required for last test window, add to DUT pending queue with priority 2
- when there are REF tests rerun required for last test window, add to REF pending queue with priority 2
2. Create schedule for today
- Day time testing 9AM - 5PM
- take in user inputs for device and hours(capcity)
- fill in as many test from the chosen device pending queue as possible. If the chosen device pending queue is empty, use the original scheduler knapsack select from active list for that device
- Night time test window
- calculate capacity
- fill in as many tests from the today's device
- if still time available, fill with knapsack select
- if queue empty, fill with knapsack select
## Frontend
- Display DUT pending queue and REF pending queue
- DUT on the left, REF on the right
- list tests
- Display device: if last testing window was DUT then default device should be REF for tonight, vice versa. user has the choice to override this
- Display current Test Config (TC)
- Display day time available hours (default 8 hours)
- Diaplay night time test window available hours (16 hours for workday, 64 for weekends)
backend/Dockerfile
+2
View File
@@ -10,6 +10,8 @@ RUN pip install --no-cache-dir -r requirements.txt
COPY . . COPY . .
WORKDIR /app/src
EXPOSE 8000 EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"] CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
backend/_tmp_validate_pairing.py
-10
View File
@@ -1,10 +0,0 @@
from parser import parse_target_csv
from db import DEVICE_DUT, DEVICE_REF
result = parse_target_csv(['data/CGW453_P2P_COE_Tests.csv'])
by_id = {(t.test_id, t.device): t for t in result.tests}
dut = by_id.get(('P2PTXBE018', DEVICE_DUT))
ref = by_id.get(('COETXBE012', DEVICE_REF))
print('DUT', dut is not None, dut.coe_pairing if dut else None)
print('REF', ref is not None, ref.coe_pairing if ref else None)
backend/scanner.py
-349
View File
@@ -1,349 +0,0 @@
import os
import re
import threading
try:
import smbclient # type: ignore[import-not-found]
except ModuleNotFoundError:
smbclient = None
from db import DEVICE_DUT, DEVICE_REF, mark_tests_completed, mark_overdue_as_rerun, reset_completed_to_pending
_SMB_SESSIONS = set()
_SCAN_STATE_LOCK = threading.Lock()
_ACTIVE_SCAN_COUNT = 0
_RESULT_TEST_ID_PATTERN = re.compile(r"(?:COE|P2P|P3P)(?:RX|TX)?[A-Z]{2}\d{3}", re.IGNORECASE)
def _normalize_smb_credentials(smb_credentials=None):
username = ""
password = ""
domain = ""
if smb_credentials:
username = str(smb_credentials.get("username", "")).strip()
password = str(smb_credentials.get("password", ""))
domain = str(smb_credentials.get("domain", "")).strip()
if not username:
username = os.getenv("SMB_USERNAME", "").strip()
if not password:
password = os.getenv("SMB_PASSWORD", "")
if not domain:
domain = os.getenv("SMB_DOMAIN", "").strip()
if username and domain and "\\" not in username and "@" not in username:
username = f"{domain}\\{username}"
return username, password
def _extract_test_id_from_result_dir_name(dir_name):
if not dir_name:
return None
match = _RESULT_TEST_ID_PATTERN.search(str(dir_name).upper())
if match:
return match.group(0)
return None
def _build_match_tokens(parsed):
tokens = set()
def _add(value):
if value in (None, ""):
return
tokens.add(str(value).upper())
_add(parsed.get("interference"))
_add(parsed.get("device"))
_add(parsed.get("rotation"))
_add(parsed.get("test_point"))
_add(parsed.get("rssi"))
_add(parsed.get("station"))
_add(parsed.get("band"))
_add(parsed.get("channel"))
_add(parsed.get("bandwidth"))
_add(parsed.get("direction"))
_add(parsed.get("throttled"))
_add(parsed.get("test_id"))
for bw in parsed.get("extra_bandwidths") or []:
_add(bw)
# Add numeric alias for devices, e.g. CGW453 -> 453.
device = (parsed.get("device") or "").upper()
device_digits = re.sub(r"\D", "", device)
if device_digits:
tokens.add(device_digits)
# SP can appear as flag or specific token variant (e.g., SP40).
if parsed.get("sp"):
tokens.add("SP")
tokens.add(str(parsed.get("sp")).upper())
return tokens
def _rule_matches_target(rule, parsed_tokens):
# Rule parts are AND-ed: CGW453_P3P_ROT2 => device AND interference AND rotation.
# Support _, -, or spaces as condition separators.
parts = [p for p in re.split(r"[_\-\s]+", rule) if p]
if not parts:
return False
def _part_matches(part):
if part in parsed_tokens:
return True
# Support tag variants (e.g., BW80 should match BW80M/BW80+80 in source names).
# Keep this conservative for very short parts to avoid overmatching.
if len(part) >= 3:
for token in parsed_tokens:
if token.startswith(part) or part in token:
return True
return False
return all(_part_matches(part) for part in parts)
def _should_exclude_target(parsed, exclusions):
parsed_tokens = _build_match_tokens(parsed)
# Any matching rule excludes the testcase.
return any(_rule_matches_target(rule, parsed_tokens) for rule in exclusions)
def _scan_started():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT += 1
def _scan_finished():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT = max(0, _ACTIVE_SCAN_COUNT - 1)
def is_scan_in_progress():
with _SCAN_STATE_LOCK:
return _ACTIVE_SCAN_COUNT > 0
def resolve_runtime_path(path_value):
if not path_value:
return path_value
raw_path = str(path_value).strip()
if not raw_path:
return raw_path
# If the path is already valid in the current runtime, keep it.
if os.path.exists(raw_path):
return raw_path
# UNC/network paths are handled separately via smbclient.
if raw_path.startswith("\\\\") or raw_path.startswith("//"):
return raw_path
mount_root = (os.getenv("HOST_MOUNT_ROOT", "/host") or "/host").strip() or "/host"
host_root = (os.getenv("HOST_BROWSE_ROOT", "") or "").strip()
raw_norm = raw_path.replace("\\", "/")
# In container/Linux runtime, translate host-browse paths into mounted container paths.
if os.name != "nt" and host_root:
host_norm = host_root.replace("\\", "/").rstrip("/")
if raw_norm.lower() == host_norm.lower() or raw_norm.lower().startswith(host_norm.lower() + "/"):
relative = raw_norm[len(host_norm):].lstrip("/")
if relative:
return os.path.join(mount_root, *relative.split("/"))
return mount_root
# In native Windows runtime, accept /host/... paths coming from container-oriented settings
# and map them back to HOST_BROWSE_ROOT (for example C:/Users/... ).
if os.name == "nt" and host_root:
mount_norm = mount_root.replace("\\", "/").rstrip("/")
if mount_norm and (raw_norm.lower() == mount_norm.lower() or raw_norm.lower().startswith(mount_norm.lower() + "/")):
relative = raw_norm[len(mount_norm):].lstrip("/")
if relative:
return os.path.join(host_root, *relative.split("/"))
return host_root
return raw_path
def _normalize_input_path(path_value):
if not path_value:
return path_value
path = resolve_runtime_path(path_value)
path = str(path).strip()
# Accept //server/share style and normalize to UNC for smbclient.
if path.startswith("//"):
path = path.lstrip("/").replace("/", "\\")
return "\\\\" + path
# Accept \\server\share style UNC paths and ensure proper escaping.
if path.startswith("\\\\"):
# Clean up any doubled backslashes from replacement
while "\\\\\\" in path:
path = path.replace("\\\\\\", "\\\\")
return path
# Accept /<ipv4>/<share>/... and normalize to UNC for Linux-hosted inputs.
if re.match(r"^/\d{1,3}(?:\.\d{1,3}){3}/[^/]+", path):
return "\\\\" + path.lstrip("/").replace("/", "\\")
# Preserve existing slash-based local paths such as /host/... that are valid
# in the current runtime, so scanner matches watcher behavior.
if os.path.exists(path):
return path
# Preserve slash-based local mount paths even if they are temporarily missing.
if path.startswith("/"):
return path
# Normalize local paths after UNC checks.
path = path.replace("/", "\\")
# Accept \<ipv4>\<share>\... and normalize to UNC.
if re.match(r"^\d{1,3}(?:\.\d{1,3}){3}\\[^\\]+", path):
return "\\\\" + path
return path
def _is_unc_path(path):
return isinstance(path, str) and path.startswith("\\\\")
def _extract_unc_server(path):
if not _is_unc_path(path):
return None
rest = path[2:]
return rest.split("\\", 1)[0] if rest else None
def _register_smb_session_if_needed(path, smb_credentials=None):
if not _is_unc_path(path):
return
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
server = _extract_unc_server(path)
if not server or server in _SMB_SESSIONS:
return
username, password = _normalize_smb_credentials(smb_credentials)
if username:
smbclient.register_session(server, username=username, password=password)
else:
smbclient.register_session(server)
_SMB_SESSIONS.add(server)
def _iter_dir_entries(path, smb_credentials=None):
path = _normalize_input_path(path)
if _is_unc_path(path):
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
return list(smbclient.scandir(path))
return list(os.scandir(path))
def path_exists_with_smb(path, smb_credentials=None):
"""Check if a path exists, with SMB authentication for UNC paths."""
if not path:
return False
path = _normalize_input_path(path)
# For UNC paths, use SMB to check
if _is_unc_path(path):
try:
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
# Try to list entries; if successful, path exists
_iter_dir_entries(path, smb_credentials=smb_credentials)
return True
except Exception:
return False
# For local paths, use standard os.path.exists
return os.path.exists(path)
def scan_results(results_dir_dut, results_dir_ref, smb_credentials=None):
results_dir_dut = _normalize_input_path(results_dir_dut)
results_dir_ref = _normalize_input_path(results_dir_ref)
if not results_dir_dut or not results_dir_ref:
return
try:
dut_entries = [
entry.name
for entry in _iter_dir_entries(results_dir_dut, smb_credentials=smb_credentials)
if entry.is_dir()
]
except OSError as exc:
print(f"[scanner] Cannot read results dir: {exc}")
return
print(f"[scanner] results: {len(dut_entries)} result dir(s) found in DUT results dir")
try:
ref_entries = [
entry.name
for entry in _iter_dir_entries(results_dir_ref, smb_credentials=smb_credentials)
if entry.is_dir()
and not entry.name.startswith("obsolete")
]
except OSError as exc:
print(f"[scanner] Cannot read results dir: {exc}")
return
print(f"[scanner] results: {len(ref_entries)} result dir(s) found in reference results dir")
completed_batch = []
unmatched_entries = []
for entry_name in dut_entries:
test_id = _extract_test_id_from_result_dir_name(entry_name)
if test_id:
completed_batch.append((test_id, DEVICE_DUT))
else:
unmatched_entries.append(entry_name)
for entry_name in ref_entries:
test_id = _extract_test_id_from_result_dir_name(entry_name)
if test_id:
completed_batch.append((test_id, DEVICE_REF))
else:
unmatched_entries.append(entry_name)
if unmatched_entries:
print(f"[scanner] skipped {len(unmatched_entries)} result dir(s) with no recognizable test id")
if completed_batch:
print(f"[scanner] marking {len(completed_batch)} test(s) as completed:")
for test_id, device in completed_batch:
print(f" - {test_id} on {device}")
reset_count = reset_completed_to_pending()
if reset_count:
print(f"[scanner] reset {reset_count} previously-completed test(s) to pending before resync")
updated_count = mark_tests_completed(completed_batch)
print(f"[scanner] {updated_count} test(s) actually updated in database")
newly_rerun = mark_overdue_as_rerun()
if newly_rerun:
print(f"[scanner] {newly_rerun} test(s) marked as rerun-required (scheduled but not completed)")
backend/app.py → backend/src/app.py
+84 -47
View File
@@ -2,35 +2,38 @@ from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from datetime import date, datetime, timedelta from datetime import date, datetime, timedelta
import io
import os import os
import zipfile
from fastapi import FastAPI, HTTPException from fastapi import FastAPI, HTTPException, Response
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
import db import db as db
from parser import CsvValidationError, parse_target_csv from parser import CsvValidationError
from file_manager import resolve_requested_csv_path, read_file_bytes
# Ensure new_scheduler resolves the same DUT/REF labels as db records. # Ensure new_scheduler resolves the same DUT/REF labels as db records.
os.environ.setdefault("DUT", db.DEVICE_DUT) os.environ.setdefault("DUT", db.DUT)
os.environ.setdefault("REF", db.DEVICE_REF) os.environ.setdefault("REF", db.REF)
from scheduler import Scheduler, Test as SchedulerTest from scheduler import Scheduler, Test as SchedulerTest
from scanner import resolve_runtime_path from file_manager import resolve_runtime_path
from test_config import build_bundle_test_configs, build_config_rows from test_config import build_bundle_test_configs, build_config_rows
from test_window import get_shift_capacity_for_date, get_shift_sequence, is_off_day from test_window import get_shift_capacity_for_date, get_shift_sequence, is_off_day
from watcher import configure_result_watcher, stop_result_watcher from watcher import configure_result_watcher, stop_result_watcher
from scanner import process_targets
APP_ROOT = Path(__file__).resolve().parent APP_ROOT = Path(__file__).resolve().parent
DB_PATH = Path(os.getenv("DB_PATH", str(APP_ROOT / "scheduler.db"))) DB_PATH = Path(os.getenv("DB_PATH", str(APP_ROOT.parent / "data" / "scheduler.db")))
DUT = os.getenv("DUT", "CGW453").strip() DUT = os.getenv("DUT", "CGW453").strip()
REF = os.getenv("REF", "CGW452").strip() REF = os.getenv("REF", "CGW452").strip()
class LoadTestsRequest(BaseModel): class LoadTestsRequest(BaseModel):
csv_path: str = Field(..., description="Absolute or backend-relative path to target CSV") csv_path: str | None = Field(default=None, description="Absolute or backend-relative path to target CSV")
csv_paths: list[str] = Field(default_factory=list, description="One or more CSV paths to load together")
target_dir: str = Field(..., description="Absolute or backend-relative path to target directory")
class SaveSettingsRequest(BaseModel): class SaveSettingsRequest(BaseModel):
@@ -66,7 +69,7 @@ def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, dict
} }
overrides: dict[str, dict[str, int]] = {} overrides: dict[str, dict[str, int]] = {}
for device_key, field_prefix in ((db.DEVICE_DUT, "dut"), (db.DEVICE_REF, "ref")): for device_key, field_prefix in ((db.DUT, "dut"), (db.REF, "ref")):
device_overrides: dict[str, int] = {} device_overrides: dict[str, int] = {}
for test_type, suffix in (("P2P", "P2p"), ("COE", "Coe"), ("P3P", "P3p")): for test_type, suffix in (("P2P", "P2p"), ("COE", "Coe"), ("P3P", "P3p")):
minutes = _parse_positive_int(settings.get(f"{field_prefix}{suffix}RuntimeMinutes")) minutes = _parse_positive_int(settings.get(f"{field_prefix}{suffix}RuntimeMinutes"))
@@ -80,27 +83,8 @@ def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, dict
return overrides return overrides
def _is_explicit_path(path_value: str) -> bool:
return (
path_value.startswith(("/", "\\\\", "//"))
or "\\" in path_value
or ":" in path_value
)
def _resolve_requested_csv_path(requested_path: str) -> str | Path:
received_path = str(requested_path).strip()
resolved_path = str(resolve_runtime_path(received_path)).strip()
if _is_explicit_path(received_path) or _is_explicit_path(resolved_path):
return resolved_path
data_dir_path = APP_ROOT / "data" / received_path
if data_dir_path.exists():
return data_dir_path
return APP_ROOT / received_path
class CompileScheduleRequest(BaseModel): class CompileScheduleRequest(BaseModel):
start_date: str | None = Field(default=None, description="YYYY-MM-DD") start_date: str | None = Field(default=None, description="YYYY-MM-DD")
@@ -148,7 +132,6 @@ def _serialize_schedule_row(row: db.ScheduleRow) -> dict[str, Any]:
"rotation": row.rotation, "rotation": row.rotation,
"config": row.config, "config": row.config,
"status": row.status, "status": row.status,
"priority": row.priority,
"estimated_minutes": row.estimated_minutes, "estimated_minutes": row.estimated_minutes,
} }
@@ -349,29 +332,26 @@ def restart_and_clear_data() -> dict[str, Any]:
@app.post("/api/tests/load") @app.post("/api/tests/load")
def load_tests(request: LoadTestsRequest) -> dict[str, Any]: def load_tests(request: LoadTestsRequest) -> dict[str, Any]:
csv_path = _resolve_requested_csv_path(request.csv_path) requested_paths: list[str] = []
if request.csv_path and request.csv_path.strip():
requested_paths.append(request.csv_path.strip())
requested_paths.extend(path.strip() for path in request.csv_paths if path and path.strip())
if not requested_paths:
raise HTTPException(status_code=400, detail="At least one CSV path is required.")
resolved_csv_paths = [resolve_requested_csv_path(path) for path in requested_paths]
target_dir = resolve_runtime_path(request.target_dir)
settings = db.read_settings(DB_PATH) settings = db.read_settings(DB_PATH)
smb_credentials = _smb_credentials_from_settings(settings) smb_credentials = _smb_credentials_from_settings(settings)
runtime_overrides = _runtime_overrides_from_settings(settings) runtime_overrides = _runtime_overrides_from_settings(settings)
try: count = process_targets(target_dir, resolved_csv_paths, smb_credentials=smb_credentials, runtime_overrides=runtime_overrides)
parsed = parse_target_csv(
csv_path,
smb_credentials=smb_credentials,
runtime_overrides=runtime_overrides,
)
except FileNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except CsvValidationError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
count = db.upsert_tests(parsed.tests, DB_PATH) print(f"Loaded {count} tests from {resolved_csv_paths}.")
print(f"Loaded {count} tests from {csv_path}, with {len(parsed.warnings)} warnings.")
return { return {
"loaded_tests": count, "loaded_tests": count,
"warnings": parsed.warnings,
} }
@app.post("/api/schedule/active/remove") @app.post("/api/schedule/active/remove")
@@ -548,6 +528,63 @@ def get_schedule_week(start: str | None = None, version: int | None = None) -> d
"windows": _build_schedule_windows(week_start_date, all_rows, holiday_dates), "windows": _build_schedule_windows(week_start_date, all_rows, holiday_dates),
} }
@app.get("/api/schedule/export")
def export_window(window_id: str, version: int | None = None) -> Response:
resolved_version = db.resolve_schedule_version(version, DB_PATH)
if version is not None and resolved_version is None:
raise HTTPException(status_code=404, detail=f"Schedule version {version} was not found")
all_rows = db.get_schedule_rows(resolved_version, DB_PATH)
if not all_rows:
raise HTTPException(status_code=404, detail="No schedule found")
holiday_dates = db.list_holidays(DB_PATH)
week_start = min(datetime.strptime(r.scheduled_date, "%Y-%m-%d").date() for r in all_rows)
windows = _build_schedule_windows(week_start, all_rows, holiday_dates)
target_window = next((w for w in windows if w["window_id"] == window_id), None)
if target_window is None:
raise HTTPException(status_code=404, detail=f"Window '{window_id}' not found")
wanted = {(t["test_id"], t["device"]) for t in target_window["tests"]}
if not wanted:
raise HTTPException(status_code=400, detail="This window has no scheduled tests")
# Look up stored file paths from DB — avoids re-scanning and handles R<n> prefixes
with db.get_connection(DB_PATH) as conn:
rows = conn.execute(
"SELECT test_id, device, file_path FROM tests WHERE file_path IS NOT NULL"
).fetchall()
path_lookup: dict[tuple[str, str], str] = {
(row["test_id"], row["device"]): row["file_path"] for row in rows
}
file_paths = [path_lookup[k] for k in sorted(wanted) if k in path_lookup]
settings = db.read_settings(DB_PATH)
smb_credentials = _smb_credentials_from_settings(settings)
buf = io.BytesIO()
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
for fp in file_paths:
try:
data = read_file_bytes(fp, smb_credentials)
except OSError as exc:
print(f"[export] Could not read {fp}: {exc}")
continue
zf.writestr(os.path.basename(fp), data)
start_date_str = datetime.strptime(target_window["start_date"], "%Y-%m-%d").strftime("%m-%d-%Y")
devices = "_".join(sorted({t["device"] for t in target_window["tests"]}))
suffix = "daytime_tests" if target_window["window_type"] == "daytime" else "tests"
zip_name = f"{start_date_str}_{devices}_{suffix}.zip"
return Response(
content=buf.getvalue(),
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{zip_name}"'},
)
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn
uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)
backend/db.py → backend/src/db.py
+10 -25
View File
@@ -10,15 +10,13 @@ from test_config import serialize_station_testpoint_map, resolve_test_config_key
APP_ROOT = Path(__file__).resolve().parent APP_ROOT = Path(__file__).resolve().parent
DB_PATH = Path(os.getenv("DB_PATH", str(APP_ROOT / "scheduler.db"))) DB_PATH = Path(os.getenv("DB_PATH", str(APP_ROOT.parent / "data" / "scheduler.db")))
# Hardware device names (from environment or defaults) # Hardware device names (from environment or defaults)
DUT = os.getenv("DUT", "CGW453").strip() DUT = os.getenv("DUT", "CGW453").strip()
REF = os.getenv("REF", "CGW452").strip() REF = os.getenv("REF", "CGW452").strip()
# Device names for the test database (use hardware device names) # Device names for the test database (use hardware device names)
DEVICE_DUT = DUT
DEVICE_REF = REF
MAX_SCHEDULE_VERSIONS = 50 MAX_SCHEDULE_VERSIONS = 50
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -31,7 +29,6 @@ class TestRecord:
power_mode: str | None power_mode: str | None
has_coe_pair: bool has_coe_pair: bool
coe_pairing: list[str] coe_pairing: list[str]
priority: int
victim_band: str | None victim_band: str | None
config: dict[str, dict[str, str | None]] config: dict[str, dict[str, str | None]]
throttled: bool throttled: bool
@@ -40,6 +37,7 @@ class TestRecord:
excluded: bool = False excluded: bool = False
raw_payload: dict[str, Any] | None = None raw_payload: dict[str, Any] | None = None
station_testpoint_map: str | None = None station_testpoint_map: str | None = None
file_path: str | None = None
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -53,7 +51,6 @@ class ScheduleRow:
rotation: str | None rotation: str | None
config: dict[str, dict[str, str | None]] config: dict[str, dict[str, str | None]]
status: str status: str
priority: int
estimated_minutes: int estimated_minutes: int
@contextmanager @contextmanager
@@ -85,7 +82,6 @@ def init_db(db_path: str | Path = DB_PATH) -> None:
power_mode TEXT, power_mode TEXT,
has_coe_pair INTEGER NOT NULL DEFAULT 0, has_coe_pair INTEGER NOT NULL DEFAULT 0,
coe_pairing_json TEXT, coe_pairing_json TEXT,
priority INTEGER NOT NULL CHECK (priority BETWEEN 1 AND 5),
victim_band TEXT, victim_band TEXT,
config_json TEXT, config_json TEXT,
throttled INTEGER NOT NULL DEFAULT 0, throttled INTEGER NOT NULL DEFAULT 0,
@@ -127,7 +123,6 @@ def init_db(db_path: str | Path = DB_PATH) -> None:
); );
CREATE INDEX IF NOT EXISTS idx_tests_status ON tests(status); CREATE INDEX IF NOT EXISTS idx_tests_status ON tests(status);
CREATE INDEX IF NOT EXISTS idx_tests_priority ON tests(priority);
CREATE INDEX IF NOT EXISTS idx_schedules_date_shift ON schedules(scheduled_date, shift_index); CREATE INDEX IF NOT EXISTS idx_schedules_date_shift ON schedules(scheduled_date, shift_index);
""" """
) )
@@ -140,6 +135,7 @@ def init_db(db_path: str | Path = DB_PATH) -> None:
_ensure_column(conn, "tests", "excluded", "INTEGER NOT NULL DEFAULT 0") _ensure_column(conn, "tests", "excluded", "INTEGER NOT NULL DEFAULT 0")
_ensure_column(conn, "tests", "throttled", "INTEGER NOT NULL DEFAULT 0") _ensure_column(conn, "tests", "throttled", "INTEGER NOT NULL DEFAULT 0")
_ensure_column(conn, "tests", "station_testpoint_map", "TEXT") _ensure_column(conn, "tests", "station_testpoint_map", "TEXT")
_ensure_column(conn, "tests", "file_path", "TEXT")
_ensure_column(conn, "schedules", "status_snapshot", "TEXT") _ensure_column(conn, "schedules", "status_snapshot", "TEXT")
conn.execute( conn.execute(
""" """
@@ -166,18 +162,13 @@ def init_db(db_path: str | Path = DB_PATH) -> None:
""" """
) )
def _serialize_station_testpoint_map(config: dict[str, dict[str, str | None]]) -> str:
"""Compute and serialize the station-to-testpoint map from a test config."""
return serialize_station_testpoint_map(config)
def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> int: def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> int:
if not records: if not records:
return 0 return 0
values: list[tuple[Any, ...]] = [] values: list[tuple[Any, ...]] = []
for r in records: for r in records:
station_testpoint_map = _serialize_station_testpoint_map(r.config) station_testpoint_map = serialize_station_testpoint_map(r.config)
values.append( values.append(
( (
@@ -190,7 +181,6 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in
int(r.has_coe_pair), int(r.has_coe_pair),
json.dumps(r.coe_pairing or []), json.dumps(r.coe_pairing or []),
json.dumps(r.config), json.dumps(r.config),
r.priority,
r.victim_band, r.victim_band,
int(r.throttled), int(r.throttled),
r.estimated_minutes, r.estimated_minutes,
@@ -198,6 +188,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in
int(r.excluded), int(r.excluded),
json.dumps(r.raw_payload or {}), json.dumps(r.raw_payload or {}),
station_testpoint_map, station_testpoint_map,
r.file_path,
) )
) )
@@ -206,7 +197,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in
""" """
INSERT INTO tests( INSERT INTO tests(
test_id, device, test_type, rotation, rx_tx, power_mode, has_coe_pair, test_id, device, test_type, rotation, rx_tx, power_mode, has_coe_pair,
coe_pairing_json, config_json, priority, victim_band, throttled, estimated_minutes, status, excluded, raw_payload, station_testpoint_map coe_pairing_json, config_json, victim_band, throttled, estimated_minutes, status, excluded, raw_payload, station_testpoint_map, file_path
) )
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(test_id, device) DO UPDATE SET ON CONFLICT(test_id, device) DO UPDATE SET
@@ -218,7 +209,6 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in
has_coe_pair = excluded.has_coe_pair, has_coe_pair = excluded.has_coe_pair,
coe_pairing_json = excluded.coe_pairing_json, coe_pairing_json = excluded.coe_pairing_json,
config_json = excluded.config_json, config_json = excluded.config_json,
priority = excluded.priority,
victim_band = excluded.victim_band, victim_band = excluded.victim_band,
throttled = excluded.throttled, throttled = excluded.throttled,
estimated_minutes = excluded.estimated_minutes, estimated_minutes = excluded.estimated_minutes,
@@ -229,6 +219,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in
excluded = excluded.excluded, excluded = excluded.excluded,
raw_payload = excluded.raw_payload, raw_payload = excluded.raw_payload,
station_testpoint_map = excluded.station_testpoint_map, station_testpoint_map = excluded.station_testpoint_map,
file_path = excluded.file_path,
updated_at = CURRENT_TIMESTAMP updated_at = CURRENT_TIMESTAMP
""", """,
values, values,
@@ -407,7 +398,7 @@ def get_not_excluded_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis
SELECT SELECT
test_id, device, test_type, rotation, rx_tx, has_coe_pair, test_id, device, test_type, rotation, rx_tx, has_coe_pair,
power_mode, config_json, victim_band, power_mode, config_json, victim_band,
coe_pairing_json, priority, throttled, estimated_minutes, coe_pairing_json, throttled, estimated_minutes,
excluded, status, raw_payload excluded, status, raw_payload
FROM tests FROM tests
WHERE excluded = 0 WHERE excluded = 0
@@ -426,7 +417,6 @@ def get_not_excluded_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis
power_mode=row["power_mode"], power_mode=row["power_mode"],
has_coe_pair=bool(row["has_coe_pair"]), has_coe_pair=bool(row["has_coe_pair"]),
coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"),
priority=int(row["priority"]),
victim_band=row["victim_band"], victim_band=row["victim_band"],
config=json.loads(row["config_json"] or "{}"), config=json.loads(row["config_json"] or "{}"),
throttled=bool(row["throttled"]), throttled=bool(row["throttled"]),
@@ -450,7 +440,7 @@ def list_schedulable_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis
SELECT SELECT
test_id, device, test_type, rotation, rx_tx, has_coe_pair, test_id, device, test_type, rotation, rx_tx, has_coe_pair,
power_mode, config_json, victim_band, power_mode, config_json, victim_band,
coe_pairing_json, priority, throttled, estimated_minutes, coe_pairing_json, throttled, estimated_minutes,
excluded, status, raw_payload excluded, status, raw_payload
FROM tests FROM tests
WHERE excluded = 0 WHERE excluded = 0
@@ -470,7 +460,6 @@ def list_schedulable_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis
power_mode=row["power_mode"], power_mode=row["power_mode"],
has_coe_pair=bool(row["has_coe_pair"]), has_coe_pair=bool(row["has_coe_pair"]),
coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"),
priority=int(row["priority"]),
victim_band=row["victim_band"], victim_band=row["victim_band"],
config=json.loads(row["config_json"] or "{}"), config=json.loads(row["config_json"] or "{}"),
throttled=bool(row["throttled"]), throttled=bool(row["throttled"]),
@@ -494,7 +483,7 @@ def list_tests_for_device(device: str, db_path: str | Path = DB_PATH) -> list[Te
SELECT SELECT
test_id, device, test_type, rotation, rx_tx, has_coe_pair, test_id, device, test_type, rotation, rx_tx, has_coe_pair,
power_mode, config_json, victim_band, power_mode, config_json, victim_band,
coe_pairing_json, priority, throttled, estimated_minutes, coe_pairing_json, throttled, estimated_minutes,
excluded, status, raw_payload, station_testpoint_map excluded, status, raw_payload, station_testpoint_map
FROM tests FROM tests
WHERE device = ? WHERE device = ?
@@ -514,7 +503,6 @@ def list_tests_for_device(device: str, db_path: str | Path = DB_PATH) -> list[Te
power_mode=row["power_mode"], power_mode=row["power_mode"],
has_coe_pair=bool(row["has_coe_pair"]), has_coe_pair=bool(row["has_coe_pair"]),
coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"),
priority=int(row["priority"]),
victim_band=row["victim_band"], victim_band=row["victim_band"],
config=json.loads(row["config_json"] or "{}"), config=json.loads(row["config_json"] or "{}"),
throttled=bool(row["throttled"]), throttled=bool(row["throttled"]),
@@ -633,7 +621,6 @@ def _hydrate_schedule_rows(rows: list[sqlite3.Row]) -> list[ScheduleRow]:
rotation=row["rotation"], rotation=row["rotation"],
config=json.loads(row["config_json"] or "{}"), config=json.loads(row["config_json"] or "{}"),
status=row["status"], status=row["status"],
priority=int(row["priority"]),
estimated_minutes=int(row["estimated_minutes"]), estimated_minutes=int(row["estimated_minutes"]),
) )
) )
@@ -716,7 +703,6 @@ def get_schedule_week(
t.rotation, t.rotation,
t.config_json, t.config_json,
{status_sql} AS status, {status_sql} AS status,
t.priority,
t.estimated_minutes t.estimated_minutes
FROM schedules s FROM schedules s
JOIN tests t ON t.test_id = s.test_id AND t.device = s.device JOIN tests t ON t.test_id = s.test_id AND t.device = s.device
@@ -791,7 +777,6 @@ def get_schedule_rows(
t.rotation, t.rotation,
t.config_json, t.config_json,
{status_sql} AS status, {status_sql} AS status,
t.priority,
t.estimated_minutes t.estimated_minutes
FROM schedules s FROM schedules s
JOIN tests t ON t.test_id = s.test_id AND t.device = s.device JOIN tests t ON t.test_id = s.test_id AND t.device = s.device
backend/src/file_manager.py
+243
View File
@@ -0,0 +1,243 @@
import os
import re
from pathlib import Path
from typing import Any
import smbclient # type: ignore[import-not-found]
_SMB_SESSIONS: set[str] = set()
APP_ROOT = Path(__file__).resolve().parent
def normalize_smb_credentials(smb_credentials: dict[str, Any] | None) -> tuple[str, str, str]:
username = ""
password = ""
domain = ""
if smb_credentials:
username = str(smb_credentials.get("username", "")).strip()
password = str(smb_credentials.get("password", ""))
domain = str(smb_credentials.get("domain", "")).strip()
if not username:
username = os.getenv("SMB_USERNAME", "").strip()
if not password:
password = os.getenv("SMB_PASSWORD", "")
if not domain:
domain = os.getenv("SMB_DOMAIN", "").strip()
if username and domain and "\\" not in username and "@" not in username:
username = f"{domain}\\{username}"
return username, password, domain
def resolve_runtime_path(path_value):
if not path_value:
return path_value
raw_path = str(path_value).strip()
if not raw_path:
return raw_path
if os.path.exists(raw_path):
return raw_path
if raw_path.startswith("\\\\") or raw_path.startswith("//"):
return raw_path
mount_root = (os.getenv("HOST_MOUNT_ROOT", "/host") or "/host").strip() or "/host"
host_root = (os.getenv("HOST_BROWSE_ROOT", "") or "").strip()
raw_norm = raw_path.replace("\\", "/")
if os.name != "nt" and host_root:
host_norm = host_root.replace("\\", "/").rstrip("/")
if raw_norm.lower() == host_norm.lower() or raw_norm.lower().startswith(host_norm.lower() + "/"):
relative = raw_norm[len(host_norm):].lstrip("/")
if relative:
return os.path.join(mount_root, *relative.split("/"))
return mount_root
if os.name == "nt" and host_root:
mount_norm = mount_root.replace("\\", "/").rstrip("/")
if mount_norm and (raw_norm.lower() == mount_norm.lower() or raw_norm.lower().startswith(mount_norm.lower() + "/")):
relative = raw_norm[len(mount_norm):].lstrip("/")
if relative:
return os.path.join(host_root, *relative.split("/"))
return host_root
return raw_path
def normalize_input_path(path_value: str | Path) -> str:
path = str(path_value).strip()
if not path:
return path
path = str(resolve_runtime_path(path)).strip()
# Accept //server/share style and normalize to UNC for smbclient.
if path.startswith("//"):
path = path.lstrip("/").replace("/", "\\")
return "\\\\" + path
# Accept \\server\share style UNC paths and ensure proper escaping.
if path.startswith("\\\\"):
while "\\\\\\" in path:
path = path.replace("\\\\\\", "\\\\")
return path
# Accept /<ipv4>/<share>/... and normalize to UNC for Linux-hosted inputs.
if re.match(r"^/\d{1,3}(?:\.\d{1,3}){3}/[^/]+", path):
return "\\\\" + path.lstrip("/").replace("/", "\\")
if os.path.exists(path):
return path
if path.startswith("/"):
return path
path = path.replace("/", "\\")
# Accept <ipv4>\<share>\... and normalize to UNC.
if re.match(r"^\d{1,3}(?:\.\d{1,3}){3}\\[^\\]+", path):
return "\\\\" + path
return path
def is_unc_path(path: str) -> bool:
return path.startswith("\\\\")
def extract_unc_server(path: str) -> str | None:
if not is_unc_path(path):
return None
rest = path[2:]
return rest.split("\\", 1)[0] if rest else None
def _register_smb_session_if_needed(path: str, smb_credentials: dict[str, Any] | None = None) -> None:
if not is_unc_path(path):
return
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
server = extract_unc_server(path)
if not server or server in _SMB_SESSIONS:
return
username, password, _domain = normalize_smb_credentials(smb_credentials)
if username:
smbclient.register_session(server, username=username, password=password)
else:
smbclient.register_session(server)
_SMB_SESSIONS.add(server)
def is_dir(path: str, smb_credentials: dict[str, Any] | None) -> bool:
if is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
try:
entry_iter = smbclient.scandir(path)
for _ in entry_iter:
break
return True
except OSError:
return False
return Path(path).is_dir()
def path_name(path: str) -> str:
trimmed = path.rstrip("\\/")
if not trimmed:
return path
parts = re.split(r"[\\/]", trimmed)
return parts[-1] if parts else trimmed
def csv_paths_from_dir(path: str, smb_credentials: dict[str, Any] | None) -> list[str]:
if is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
entries = []
for entry in smbclient.scandir(path):
name = getattr(entry, "name", "")
if name and name.lower().endswith(".csv") and entry.is_file():
entries.append(path.rstrip("\\/") + "\\" + name)
return sorted(entries)
return [str(p) for p in sorted(Path(path).glob("*.csv"))]
def path_exists_with_smb(path, smb_credentials=None):
"""Check if a path exists, with SMB authentication for UNC paths."""
if not path:
return False
path = normalize_input_path(path)
# For UNC paths, use SMB to check
if is_unc_path(path):
try:
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
# Try to list entries; if successful, path exists
iter_dir_entries(path, smb_credentials=smb_credentials)
return True
except Exception:
return False
# For local paths, use standard os.path.exists
return os.path.exists(path)
def iter_dir_entries(path, smb_credentials=None):
path = normalize_input_path(path)
if is_unc_path(path):
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to scan UNC result paths")
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
return list(smbclient.scandir(path))
return list(os.scandir(path))
def open_csv_handle(path: str, smb_credentials: dict[str, Any] | None):
if is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
return smbclient.open_file(path, mode="r", encoding="utf-8-sig", newline="")
return Path(path).open("r", encoding="utf-8-sig", newline="")
def read_file_bytes(path: str, smb_credentials: dict[str, Any] | None = None) -> bytes:
if is_unc_path(path):
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to read UNC paths")
_register_smb_session_if_needed(path, smb_credentials=smb_credentials)
with smbclient.open_file(path, mode="rb") as fh:
return fh.read()
with open(path, "rb") as fh:
return fh.read()
def join_path(path, name):
if is_unc_path(path):
base = path.rstrip("\\")
return f"{base}\\{name}"
return str(Path(path) / name)
def _is_explicit_path(path_value: str) -> bool:
return (
path_value.startswith(("/", "\\\\", "//"))
or "\\" in path_value
or ":" in path_value
)
def resolve_requested_csv_path(requested_path: str) -> str | Path:
received_path = str(requested_path).strip()
resolved_path = str(resolve_runtime_path(received_path)).strip()
if _is_explicit_path(received_path) or _is_explicit_path(resolved_path):
return resolved_path
data_dir_path = APP_ROOT / "data" / received_path
if data_dir_path.exists():
return data_dir_path
return APP_ROOT / received_path
backend/parser.py → backend/src/parser.py
+97 -249
View File
@@ -5,13 +5,14 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
try: from db import DUT, REF, TestRecord
import smbclient # type: ignore[import-not-found] from file_manager import (
except ModuleNotFoundError: open_csv_handle
smbclient = None )
from db import DEVICE_DUT, DEVICE_REF, TestRecord TEST_TYPES = {"P2P", "COE", "P3P"}
from scanner import resolve_runtime_path POWER_MODES = {"LPI", "SP"}
RX_TX = {"RX", "TX"}
P2P_COE_REQUIRED_COLUMNS = [ P2P_COE_REQUIRED_COLUMNS = [
"Priority", "Priority",
@@ -92,131 +93,6 @@ class CsvValidationError(ValueError):
_SMB_SESSIONS: set[str] = set() _SMB_SESSIONS: set[str] = set()
def _normalize_input_path(path_value: str | Path) -> str:
path = str(path_value).strip()
if not path:
return path
path = str(resolve_runtime_path(path)).strip()
# Accept //server/share style and normalize to UNC for smbclient.
if path.startswith("//"):
return "\\\\" + path.lstrip("/").replace("/", "\\")
# Accept /<ipv4>/<share>/... and normalize to UNC for Linux-hosted inputs.
if re.match(r"^/\d{1,3}(?:\.\d{1,3}){3}/[^/]+", path):
return "\\\\" + path.lstrip("/").replace("/", "\\")
if os.path.exists(path):
return path
if path.startswith("/"):
return path
return path
def _is_unc_path(path: str) -> bool:
return path.startswith("\\\\")
def _extract_unc_server(path: str) -> str | None:
if not _is_unc_path(path):
return None
rest = path[2:]
return rest.split("\\", 1)[0] if rest else None
def _normalize_smb_credentials(smb_credentials: dict[str, Any] | None) -> tuple[str, str, str]:
username = ""
password = ""
domain = ""
if smb_credentials:
username = str(smb_credentials.get("username", "")).strip()
password = str(smb_credentials.get("password", ""))
domain = str(smb_credentials.get("domain", "")).strip()
if not username:
username = os.getenv("SMB_USERNAME", "").strip()
if not password:
password = os.getenv("SMB_PASSWORD", "")
if not domain:
domain = os.getenv("SMB_DOMAIN", "").strip()
if username and domain and "\\" not in username and "@" not in username:
username = f"{domain}\\{username}"
return username, password, domain
def _register_smb_session_if_needed(path: str, smb_credentials: dict[str, Any] | None) -> None:
if not _is_unc_path(path):
return
if smbclient is None:
raise ModuleNotFoundError("smbclient is required to read CSV files from UNC paths")
server = _extract_unc_server(path)
if not server or server in _SMB_SESSIONS:
return
username, password, _ = _normalize_smb_credentials(smb_credentials)
if username:
smbclient.register_session(server, username=username, password=password)
else:
smbclient.register_session(server)
_SMB_SESSIONS.add(server)
def _path_exists(path: str, smb_credentials: dict[str, Any] | None) -> bool:
if _is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
try:
smbclient.stat(path)
return True
except OSError:
return False
return Path(path).exists()
def _is_dir(path: str, smb_credentials: dict[str, Any] | None) -> bool:
if _is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
try:
entry_iter = smbclient.scandir(path)
for _ in entry_iter:
break
return True
except OSError:
return False
return Path(path).is_dir()
def _path_name(path: str) -> str:
trimmed = path.rstrip("\\/")
if not trimmed:
return path
parts = re.split(r"[\\/]", trimmed)
return parts[-1] if parts else trimmed
def _csv_paths_from_dir(path: str, smb_credentials: dict[str, Any] | None) -> list[str]:
if _is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
entries = []
for entry in smbclient.scandir(path):
name = getattr(entry, "name", "")
if name and name.lower().endswith(".csv") and entry.is_file():
entries.append(path.rstrip("\\/") + "\\" + name)
return sorted(entries)
return [str(p) for p in sorted(Path(path).glob("*.csv"))]
def _columns_missing(normalized_fieldnames: set[str], required_columns: list[str]) -> list[str]: def _columns_missing(normalized_fieldnames: set[str], required_columns: list[str]) -> list[str]:
return [ return [
column column
@@ -240,35 +116,25 @@ def _detect_csv_format(normalized_fieldnames: set[str]) -> str:
) )
def parse_target_csv( def parse_target_csv(
csv_path: str | Path | list[str | Path] | tuple[str | Path, ...], paths: str | Path | list[str | Path] | tuple[str | Path, ...],
smb_credentials: dict[str, Any] | None = None, smb_credentials: dict[str, Any] | None = None,
runtime_overrides: dict[str, Any] | None = None, runtime_overrides: dict[str, Any] | None = None,
) -> ParseResult: ) -> dict[str, Any]:
paths = _resolve_csv_paths(csv_path, smb_credentials=smb_credentials)
runtime_defaults_by_device = _resolve_runtime_defaults(runtime_overrides) runtime_defaults_by_device = _resolve_runtime_defaults(runtime_overrides)
all_tests: list[TestRecord] = [] results = {}
all_warnings: list[str] = []
seen_test_keys: set[tuple[str, str]] = set() if isinstance(paths, (str, Path)):
paths = [paths]
for path in paths: for path in paths:
parsed = _parse_single_csv( csv_result = _parse_single_csv(
path, path,
smb_credentials=smb_credentials, smb_credentials=smb_credentials,
runtime_defaults_by_device=runtime_defaults_by_device, runtime_defaults_by_device=runtime_defaults_by_device,
) )
all_warnings.extend(parsed.warnings)
for record in parsed.tests:
key = (record.test_id, record.device)
if key in seen_test_keys:
all_warnings.append(
f"File {_path_name(path)}: duplicate test/device '{record.test_id}/{record.device}', record skipped."
)
continue
seen_test_keys.add(key)
all_tests.append(record)
return ParseResult(tests=all_tests, warnings=all_warnings)
results.update(csv_result)
return results
def _coerce_runtime_defaults(raw_overrides: dict[str, Any] | None) -> dict[str, int]: def _coerce_runtime_defaults(raw_overrides: dict[str, Any] | None) -> dict[str, int]:
defaults = dict(RUNTIME_DEFAULTS) defaults = dict(RUNTIME_DEFAULTS)
@@ -297,12 +163,12 @@ def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[
legacy_defaults = _coerce_runtime_defaults(runtime_overrides) legacy_defaults = _coerce_runtime_defaults(runtime_overrides)
if not runtime_overrides: if not runtime_overrides:
return { return {
DEVICE_DUT: dict(legacy_defaults), DUT: dict(legacy_defaults),
DEVICE_REF: dict(legacy_defaults), REF: dict(legacy_defaults),
} }
resolved: dict[str, dict[str, int]] = {} resolved: dict[str, dict[str, int]] = {}
for device in (DEVICE_DUT, DEVICE_REF): for device in (DUT, REF):
device_defaults = dict(legacy_defaults) device_defaults = dict(legacy_defaults)
raw_device_overrides = runtime_overrides.get(device) raw_device_overrides = runtime_overrides.get(device)
if isinstance(raw_device_overrides, dict): if isinstance(raw_device_overrides, dict):
@@ -311,48 +177,13 @@ def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[
return resolved return resolved
def _resolve_csv_paths(
csv_path: str | Path | list[str | Path] | tuple[str | Path, ...],
smb_credentials: dict[str, Any] | None,
) -> list[str]:
if isinstance(csv_path, (str, Path)):
paths = [_normalize_input_path(csv_path)]
else:
paths = [_normalize_input_path(item) for item in csv_path]
resolved_paths: list[str] = []
for path in paths:
if _is_dir(path, smb_credentials=smb_credentials):
resolved_paths.extend(_csv_paths_from_dir(path, smb_credentials=smb_credentials))
else:
resolved_paths.append(path)
if not resolved_paths:
raise FileNotFoundError("No CSV files found to parse.")
for path in resolved_paths:
if not _path_exists(path, smb_credentials=smb_credentials):
raise FileNotFoundError(f"CSV file not found: {path}")
return resolved_paths
def _open_csv_handle(path: str, smb_credentials: dict[str, Any] | None):
if _is_unc_path(path):
_register_smb_session_if_needed(path, smb_credentials)
return smbclient.open_file(path, mode="r", encoding="utf-8-sig", newline="")
return Path(path).open("r", encoding="utf-8-sig", newline="")
def _parse_single_csv( def _parse_single_csv(
path: str, path: str,
smb_credentials: dict[str, Any] | None, smb_credentials: dict[str, Any] | None,
runtime_defaults_by_device: dict[str, dict[str, int]], runtime_defaults_by_device: dict[str, dict[str, int]],
) -> ParseResult: ) -> dict[str, Any]:
with _open_csv_handle(path, smb_credentials=smb_credentials) as handle: records: dict[str, Any] = {}
with open_csv_handle(path, smb_credentials=smb_credentials) as handle:
reader = csv.DictReader(handle) reader = csv.DictReader(handle)
if not reader.fieldnames: if not reader.fieldnames:
raise CsvValidationError("CSV is missing a header row.") raise CsvValidationError("CSV is missing a header row.")
@@ -369,11 +200,11 @@ def _parse_single_csv(
for row_num, row in enumerate(reader, start=2): for row_num, row in enumerate(reader, start=2):
test_id = (row.get("TC ID") or "").strip() test_id = (row.get("TC ID") or "").strip()
if not test_id: if not test_id:
warnings.append(f"File {_path_name(path)}, row {row_num}: missing TC ID, row skipped.") warnings.append(f"File {path}, row {row_num}: missing TC ID, row skipped.")
continue continue
if test_id in seen_test_ids: if test_id in seen_test_ids:
warnings.append(f"File {_path_name(path)}, row {row_num}: duplicate TC ID '{test_id}', row skipped.") warnings.append(f"File {path}, row {row_num}: duplicate TC ID '{test_id}', row skipped.")
continue continue
seen_test_ids.add(test_id) seen_test_ids.add(test_id)
@@ -388,79 +219,52 @@ def _parse_single_csv(
victim_band_source = "Victim Band" if csv_format == "p2p_coe" else "Band" victim_band_source = "Victim Band" if csv_format == "p2p_coe" else "Band"
victim_band = _normalize_victim_band(_row_get(row, victim_band_source)) victim_band = _normalize_victim_band(_row_get(row, victim_band_source))
# Default priority based on test type and COE pairing. records[test_id] = {
if test_type == "P2P": "test_id": test_id,
priority = 2 if has_coe_pair else 3 "test_type": test_type,
elif test_type == "COE": "rotation": rotation,
priority = 3 "rx_tx": rx_tx,
else: "power_mode": power_mode,
priority = 4 "has_coe_pair": has_coe_pair,
"coe_pairing": [],
record = TestRecord( "victim_band": victim_band,
test_id=test_id, "config": config,
device=DEVICE_DUT, "throttled": throttled,
test_type=test_type, "estimated_minutes": {device: defaults[test_type] for device, defaults in runtime_defaults_by_device.items()},
rotation=rotation, "status": "pending",
rx_tx=rx_tx, "excluded": False,
power_mode=power_mode, "raw_payload": row,
has_coe_pair=has_coe_pair, }
coe_pairing=[], records_with_signature.append((records[test_id], signature))
priority=priority,
victim_band=victim_band,
config=config,
throttled=throttled,
estimated_minutes=RUNTIME_DEFAULTS[test_type],
status="pending",
excluded=False,
raw_payload=row,
)
records_with_signature.append((record, signature))
# Build COE pairing based on full signature and RX/TX+band suffix. # Build COE pairing based on full signature and RX/TX+band suffix.
# Example key suffixes: RXAX, TXAX, RXBE, TXBE. # Example key suffixes: RXAX, TXAX, RXBE, TXBE.
coe_by_signature_and_suffix: dict[tuple[tuple[str, ...], str], list[str]] = {} coe_by_signature_and_suffix: dict[tuple[tuple[str, ...], str], list[str]] = {}
for record, signature in records_with_signature: for record, signature in records_with_signature:
if record.test_type == "COE": if record["test_type"] == "COE":
suffix = _extract_pairing_suffix(record.test_id) suffix = _extract_pairing_suffix(record["test_id"])
if not suffix: if not suffix:
continue continue
# Index all populated COE band signatures so any one can match the selected P2P signature. # Index all populated COE band signatures so any one can match the selected P2P signature.
for coe_signature in _all_band_signatures(record.raw_payload or {}): for coe_signature in _all_band_signatures(record["raw_payload"] or {}):
key = (coe_signature, suffix) key = (coe_signature, suffix)
coe_by_signature_and_suffix.setdefault(key, []).append(record.test_id) coe_by_signature_and_suffix.setdefault(key, []).append(record["test_id"])
for record, signature in records_with_signature: for record, signature in records_with_signature:
pairs = record.coe_pairing pairs = record["coe_pairing"]
if record.test_type == "P2P" and signature is not None: if record["test_type"] == "P2P" and signature is not None:
suffix = _extract_pairing_suffix(record.test_id) suffix = _extract_pairing_suffix(record["test_id"])
key = (signature, suffix) if suffix else None key = (signature, suffix) if suffix else None
print(f"[parser] P2P test {record['test_id']} with signature {signature} and suffix {suffix} has COE pairs: {coe_by_signature_and_suffix.get(key, [])}")
pairs = sorted(set(coe_by_signature_and_suffix.get(key, []))) if key else [] pairs = sorted(set(coe_by_signature_and_suffix.get(key, []))) if key else []
for device in (DEVICE_DUT, DEVICE_REF): records[record["test_id"]].update(
device_runtime_defaults = runtime_defaults_by_device.get(device, RUNTIME_DEFAULTS)
tests.append(
TestRecord(
test_id=record.test_id,
device=device,
test_type=record.test_type,
rotation=record.rotation,
rx_tx=record.rx_tx,
power_mode=record.power_mode,
has_coe_pair=bool(pairs), has_coe_pair=bool(pairs),
coe_pairing=pairs, coe_pairing=pairs,
priority=record.priority,
victim_band=record.victim_band,
config=record.config,
throttled=record.throttled,
estimated_minutes=device_runtime_defaults.get(record.test_type, RUNTIME_DEFAULTS[record.test_type]),
status=record.status,
excluded=record.excluded,
raw_payload=record.raw_payload,
)
) )
return ParseResult(tests=tests, warnings=warnings) return records
def _infer_test_type(test_id: str) -> str: def _infer_test_type(test_id: str) -> str:
@@ -586,7 +390,7 @@ def _normalize_yes_no(value: str | None) -> bool:
return _normalize_value(value) == "YES" return _normalize_value(value) == "YES"
def _build_legacy_config(row: dict[str, str]) -> dict[str, dict[str, str | None]]: def _build_p2p_coe_config(row: dict[str, str]) -> dict[str, dict[str, str | None]]:
return { return {
"5G": { "5G": {
"test_point": _empty_to_none(_row_get(row, "5G_Test Point")), "test_point": _empty_to_none(_row_get(row, "5G_Test Point")),
@@ -650,7 +454,7 @@ def _build_p3p_config(row: dict[str, str]) -> dict[str, dict[str, str | None]]:
def _build_config(row: dict[str, str], csv_format: str) -> dict[str, dict[str, str | None]]: def _build_config(row: dict[str, str], csv_format: str) -> dict[str, dict[str, str | None]]:
if csv_format == "p3p": if csv_format == "p3p":
return _build_p3p_config(row) return _build_p3p_config(row)
return _build_legacy_config(row) return _build_p2p_coe_config(row)
def _row_get(row: dict[str, str], key: str) -> str | None: def _row_get(row: dict[str, str], key: str) -> str | None:
@@ -670,3 +474,47 @@ def _row_get(row: dict[str, str], key: str) -> str | None:
return None return None
def parse_target_filename(filename, parent_dir):
base_name = re.sub(r"\.ini$", "", filename, flags=re.IGNORECASE).upper()
segments = base_name.split("_")
test_type = next((s for s in segments if s in TEST_TYPES), None)
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 s in (DUT, REF)), None)
if test_id is None or test_type is None or device is None:
print(f"Warning: Could not parse test_id, test_type, or device from filename '{filename}'")
return None
test_id = test_id[2:] if test_id else None
throttled = None
if test_type == "P3P" and test_id:
# P3P test_id carries throttle marker: TH = throttled, otherwise UT.
throttled = True if "TH" in test_id else False
rx_tx = "RX" if "RX" in test_id else "TX"
band = next((s for s in segments if re.match(r"^\dGHZ$", s)), None)
bandwidth = next((s for s in segments if re.match(r"^BW\d+$", s)), None)
pm = next ((s for s in segments if s in POWER_MODES), None)
rotation = None
if parent_dir:
rotation = next((s for s in parent_dir.split("_") if re.match(r"^ROT\d+$", s)), None)
return {
"test_id": test_id,
"device": device,
"test_type": test_type,
"rotation": rotation,
"rx_tx": rx_tx,
"power_mode": pm,
"has_coe_pair": None,
"coe_pairing": [],
"victim_band": band,
"config_json": {},
"throttled": throttled,
"estimated_minutes": None,
"status": "pending",
"excluded": False,
"raw_payload": {},
"station_testpoint_map": {},
}
backend/src/scanner.py
+297
View File
@@ -0,0 +1,297 @@
import os
import re
import threading
import json
try:
import smbclient # type: ignore[import-not-found]
except ModuleNotFoundError:
smbclient = None
from parser import parse_target_csv, parse_target_filename, RUNTIME_DEFAULTS
from db import DUT, REF, mark_tests_completed, mark_overdue_as_rerun, reset_completed_to_pending, upsert_tests, TestRecord, get_connection
from file_manager import (
iter_dir_entries,
normalize_input_path,
join_path
)
_SCAN_STATE_LOCK = threading.Lock()
_ACTIVE_SCAN_COUNT = 0
_RESULT_TEST_ID_PATTERN = re.compile(r"(?:COE|P2P|P3P)(?:RX|TX)?[A-Z]{2}\d{3}", re.IGNORECASE)
def _extract_test_id_from_result_dir_name(dir_name):
if not dir_name:
return None
match = _RESULT_TEST_ID_PATTERN.search(str(dir_name).upper())
if match:
return match.group(0)
return None
def _build_match_tokens(parsed):
tokens = set()
def _add(value):
if value in (None, ""):
return
tokens.add(str(value).upper())
_add(parsed.get("interference"))
_add(parsed.get("device"))
_add(parsed.get("rotation"))
_add(parsed.get("test_point"))
_add(parsed.get("rssi"))
_add(parsed.get("station"))
_add(parsed.get("band"))
_add(parsed.get("channel"))
_add(parsed.get("bandwidth"))
_add(parsed.get("direction"))
_add(parsed.get("throttled"))
_add(parsed.get("test_id"))
for bw in parsed.get("extra_bandwidths") or []:
_add(bw)
# Add numeric alias for devices, e.g. CGW453 -> 453.
device = (parsed.get("device") or "").upper()
device_digits = re.sub(r"\D", "", device)
if device_digits:
tokens.add(device_digits)
# SP can appear as flag or specific token variant (e.g., SP40).
if parsed.get("sp"):
tokens.add("SP")
tokens.add(str(parsed.get("sp")).upper())
return tokens
def _rule_matches_target(rule, parsed_tokens):
# Rule parts are AND-ed: CGW453_P3P_ROT2 => device AND interference AND rotation.
# Support _, -, or spaces as condition separators.
parts = [p for p in re.split(r"[_\-\s]+", rule) if p]
if not parts:
return False
def _part_matches(part):
if part in parsed_tokens:
return True
# Support tag variants (e.g., BW80 should match BW80M/BW80+80 in source names).
# Keep this conservative for very short parts to avoid overmatching.
if len(part) >= 3:
for token in parsed_tokens:
if token.startswith(part) or part in token:
return True
return False
return all(_part_matches(part) for part in parts)
def _should_exclude_target(parsed, exclusions):
parsed_tokens = _build_match_tokens(parsed)
# Any matching rule excludes the testcase.
return any(_rule_matches_target(rule, parsed_tokens) for rule in exclusions)
def _scan_started():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT += 1
def _scan_finished():
global _ACTIVE_SCAN_COUNT
with _SCAN_STATE_LOCK:
_ACTIVE_SCAN_COUNT = max(0, _ACTIVE_SCAN_COUNT - 1)
def is_scan_in_progress():
with _SCAN_STATE_LOCK:
return _ACTIVE_SCAN_COUNT > 0
def scan_results(results_dir_dut, results_dir_ref, smb_credentials=None):
results_dir_dut = normalize_input_path(results_dir_dut)
results_dir_ref = normalize_input_path(results_dir_ref)
if not results_dir_dut or not results_dir_ref:
return
try:
dut_entries = [
entry.name
for entry in iter_dir_entries(results_dir_dut, smb_credentials=smb_credentials)
if entry.is_dir()
]
except OSError as exc:
print(f"[scanner] Cannot read results dir: {exc}")
return
print(f"[scanner] results: {len(dut_entries)} result dir(s) found in DUT results dir")
try:
ref_entries = [
entry.name
for entry in iter_dir_entries(results_dir_ref, smb_credentials=smb_credentials)
if entry.is_dir()
and not entry.name.startswith("obsolete")
]
except OSError as exc:
print(f"[scanner] Cannot read results dir: {exc}")
return
print(f"[scanner] results: {len(ref_entries)} result dir(s) found in reference results dir")
completed_batch = []
unmatched_entries = []
for entry_name in dut_entries:
test_id = _extract_test_id_from_result_dir_name(entry_name)
if test_id:
completed_batch.append((test_id, DUT))
else:
unmatched_entries.append(entry_name)
for entry_name in ref_entries:
test_id = _extract_test_id_from_result_dir_name(entry_name)
if test_id:
completed_batch.append((test_id, REF))
else:
unmatched_entries.append(entry_name)
if unmatched_entries:
print(f"[scanner] skipped {len(unmatched_entries)} result dir(s) with no recognizable test id")
if completed_batch:
print(f"[scanner] marking {len(completed_batch)} test(s) as completed:")
for test_id, device in completed_batch:
print(f" - {test_id} on {device}")
reset_count = reset_completed_to_pending()
if reset_count:
print(f"[scanner] reset {reset_count} previously-completed test(s) to pending before resync")
updated_count = mark_tests_completed(completed_batch)
print(f"[scanner] {updated_count} test(s) actually updated in database")
newly_rerun = mark_overdue_as_rerun()
if newly_rerun:
print(f"[scanner] {newly_rerun} test(s) marked as rerun-required (scheduled but not completed)")
def _normalize_rotation(rotation: str) -> str:
"""Normalize rotation string from R<n> to ROT<n> format."""
if not rotation:
return ""
rotation = str(rotation).strip().upper()
if rotation.startswith("R") and len(rotation) > 1 and rotation[1:].isdigit():
return f"ROT{rotation[1:]}"
return rotation
def process_targets(target_dir, csv_paths, smb_credentials=None, runtime_overrides=None, db_path=None):
target_dir = normalize_input_path(target_dir)
csv_results = parse_target_csv(csv_paths, smb_credentials=smb_credentials, runtime_overrides=runtime_overrides)
batch_tests = []
skipped_missing_csv = 0
existing_pairings: dict[tuple[str, str], tuple[bool, list[str]]] = {}
try:
with get_connection(db_path) if db_path is not None else get_connection() as conn:
rows = conn.execute(
"""
SELECT test_id, device, has_coe_pair, coe_pairing_json
FROM tests
"""
).fetchall()
for row in rows:
existing_pairings[(row["test_id"], row["device"])] = (
bool(row["has_coe_pair"]),
json.loads(row["coe_pairing_json"] or "[]"),
)
except Exception as exc:
print(f"[scanner] Could not read existing pairings for merge: {exc}")
try:
parent_entries = [
entry.name
for entry in iter_dir_entries(target_dir, smb_credentials=smb_credentials)
if entry.is_dir()
]
except (OSError, ValueError) as exc:
print(f"[scanner] Cannot read target dir: {exc}")
return 0
print(f"[scanner] subdirectories found: {len(parent_entries)}")
for parent_name in parent_entries:
parent_path = join_path(target_dir, parent_name)
try:
files = [
entry.name
for entry in iter_dir_entries(parent_path, smb_credentials=smb_credentials)
if entry.is_file()
and not entry.name.startswith("GLOBAL")
and entry.name.endswith(".ini")
]
except (OSError, ValueError) as exc:
print(f"[scanner] Cannot read parent dir {parent_name}: {exc}")
continue
for filename in files:
parsed = parse_target_filename(filename, parent_name)
if not parsed or not parsed["test_id"]:
continue
csv_entry = csv_results.get(parsed["test_id"], {})
if not csv_entry:
skipped_missing_csv += 1
continue
em_map = csv_entry.get("estimated_minutes", {})
if isinstance(em_map, dict):
estimated_minutes = em_map.get(parsed["device"])
else:
estimated_minutes = em_map
if estimated_minutes is None:
estimated_minutes = RUNTIME_DEFAULTS.get(parsed["test_type"], 80)
incoming_pairs = csv_entry.get("coe_pairing", []) or []
incoming_has_pair = bool(csv_entry.get("has_coe_pair", False))
existing_has_pair, existing_pairs = existing_pairings.get((parsed["test_id"], parsed["device"]), (False, []))
# Keep existing non-empty pairings when a later CSV load omits them for the same test.
if not incoming_pairs and not incoming_has_pair and existing_pairs:
incoming_pairs = existing_pairs
incoming_has_pair = existing_has_pair
batch_tests.append(TestRecord(
test_id=parsed["test_id"],
device=parsed["device"],
test_type=parsed["test_type"],
rotation=parsed["rotation"] if parsed["rotation"] else _normalize_rotation(csv_entry.get("rotation")),
rx_tx=parsed["rx_tx"],
power_mode=parsed["power_mode"],
has_coe_pair=incoming_has_pair,
coe_pairing=incoming_pairs,
victim_band=parsed["victim_band"],
config=csv_entry.get("config", {}),
throttled=parsed["throttled"] or False,
estimated_minutes=estimated_minutes,
status=parsed["status"],
excluded=False,
raw_payload=csv_entry.get("raw_payload", None),
file_path=join_path(parent_path, filename),
))
count = upsert_tests(batch_tests) if db_path is None else upsert_tests(batch_tests, db_path)
if skipped_missing_csv:
print(f"[scanner] skipped {skipped_missing_csv} target file(s) not present in provided CSV input")
return count
backend/scheduler.py → backend/src/scheduler.py
View File
backend/test_bundle.py → backend/src/test_bundle.py
+4 -3
View File
@@ -44,7 +44,8 @@ class TestBundle:
def build_test_bundles(active_dut: dict[str, Test], active_ref: dict[str, Test], top_priority_tests: set[tuple[str, str]]) -> list[TestBundle]: def build_test_bundles(active_dut: dict[str, Test], active_ref: dict[str, Test], top_priority_tests: set[tuple[str, str]]) -> list[TestBundle]:
"""Build deterministic bundles for DP scheduling.""" """Build deterministic bundles for DP scheduling."""
print(f"Top priority tests: {top_priority_tests}") print(f"[test_bundle] Top priority tests: {top_priority_tests}")
print(f"[debug] Tests with coe_pairing: {[(tid, t.coe_pairing) for tid, t in active_dut.items() if t.coe_pairing]}")
processed_dut: set[str] = set() processed_dut: set[str] = set()
processed_ref: set[str] = set() processed_ref: set[str] = set()
test_bundles: list[TestBundle] = [] test_bundles: list[TestBundle] = []
@@ -175,7 +176,7 @@ def build_test_bundles(active_dut: dict[str, Test], active_ref: dict[str, Test],
for dut_test_id in active_dut: for dut_test_id in active_dut:
if dut_test_id in processed_dut: if dut_test_id in processed_dut:
continue continue
print(f"Orphan COE-only DUT test found: {dut_test_id}") print(f"[test_bundle] Orphan COE-only DUT test found: {dut_test_id}")
test_bundles.append( test_bundles.append(
create_bundle([dut_test_id], DUT, BUNDLE_PRIORITY_COE_ONLY, bundle_index, active_dut, active_ref) create_bundle([dut_test_id], DUT, BUNDLE_PRIORITY_COE_ONLY, bundle_index, active_dut, active_ref)
) )
@@ -225,7 +226,7 @@ def build_test_bundles(active_dut: dict[str, Test], active_ref: dict[str, Test],
for ref_test_id in active_ref: for ref_test_id in active_ref:
if ref_test_id in processed_ref: if ref_test_id in processed_ref:
continue continue
print(f"Orphan COE-only REF test found: {ref_test_id}") print(f"[test-bundle]Orphan COE-only REF test found: {ref_test_id}")
test_bundles.append( test_bundles.append(
create_bundle([ref_test_id], REF, BUNDLE_PRIORITY_DUT_COMPLETED_MIRROR, bundle_index, active_dut, active_ref) create_bundle([ref_test_id], REF, BUNDLE_PRIORITY_DUT_COMPLETED_MIRROR, bundle_index, active_dut, active_ref)
) )
backend/test_config.py → backend/src/test_config.py
View File
backend/test_window.py → backend/src/test_window.py
View File
backend/watcher.py → backend/src/watcher.py
+2 -1
View File
@@ -15,7 +15,8 @@ except ModuleNotFoundError:
from watchdog.events import FileSystemEvent, FileSystemEventHandler, FileSystemMovedEvent from watchdog.events import FileSystemEvent, FileSystemEventHandler, FileSystemMovedEvent
from watchdog.observers.polling import PollingObserver from watchdog.observers.polling import PollingObserver
from scanner import resolve_runtime_path, scan_results from file_manager import resolve_runtime_path
from scanner import scan_results
LOGGER = logging.getLogger("scheduler.watcher") LOGGER = logging.getLogger("scheduler.watcher")
docker-compose.yml
+2 -1
View File
@@ -7,7 +7,8 @@ services:
environment: environment:
HOST_BROWSE_ROOT: ${HOST_BROWSE_ROOT} HOST_BROWSE_ROOT: ${HOST_BROWSE_ROOT}
HOST_MOUNT_ROOT: /host HOST_MOUNT_ROOT: /host
DB_PATH: /app/runtime/scheduler.db DB_PATH: /app/data/scheduler.db
PYTHONPATH: /app/src
volumes: volumes:
- ${HOST_BROWSE_ROOT}:/host - ${HOST_BROWSE_ROOT}:/host
- scheduler-db:/app/runtime - scheduler-db:/app/runtime
frontend/src/App.jsx
+8 -5
View File
@@ -201,6 +201,7 @@ function buildCalendarScheduleData(currentScheduleData = {}, previousScheduleDat
} }
const DEFAULT_SETTINGS = { const DEFAULT_SETTINGS = {
targetDir: '',
p2pCoeCsvPath: '', p2pCoeCsvPath: '',
p3pCsvPath: '', p3pCsvPath: '',
refResultDir: '', refResultDir: '',
@@ -394,11 +395,13 @@ export default function App() {
.split(',').map(s => s.trim()).filter(Boolean) .split(',').map(s => s.trim()).filter(Boolean)
await api.saveHolidays(holidayDates) await api.saveHolidays(holidayDates)
if (sanitizedSettings.p2pCoeCsvPath?.trim()) { const csvPaths = [
await api.loadCsv(sanitizedSettings.p2pCoeCsvPath.trim()) sanitizedSettings.p2pCoeCsvPath?.trim(),
} sanitizedSettings.p3pCsvPath?.trim(),
if (sanitizedSettings.p3pCsvPath?.trim()) { ].filter(Boolean)
await api.loadCsv(sanitizedSettings.p3pCsvPath.trim())
if (csvPaths.length > 0) {
await api.loadCsv(csvPaths, sanitizedSettings.targetDir?.trim() ?? '')
} }
setSettings(sanitizedSettings) setSettings(sanitizedSettings)
frontend/src/api.js
+15 -1
View File
@@ -27,7 +27,7 @@ export const api = {
saveHolidays: (dates) => request('POST', '/holidays', { dates }), saveHolidays: (dates) => request('POST', '/holidays', { dates }),
// Tests // Tests
loadCsv: (csv_path) => request('POST', '/tests/load', { csv_path }), loadCsv: (csv_paths, target_dir) => request('POST', '/tests/load', { csv_paths, target_dir }),
// Schedule // Schedule
compileSchedule: (opts) => request('POST', '/schedule/compile', opts), compileSchedule: (opts) => request('POST', '/schedule/compile', opts),
@@ -39,6 +39,20 @@ export const api = {
return request('GET', `/schedule/week?${params.toString()}`) return request('GET', `/schedule/week?${params.toString()}`)
}, },
getRerunTests: () => request('GET', '/tests/rerun'), getRerunTests: () => request('GET', '/tests/rerun'),
exportWindow: async (windowId) => {
const params = new URLSearchParams({ window_id: windowId })
const res = await fetch(`${BASE}/schedule/export?${params}`)
if (!res.ok) {
const err = await res.json().catch(() => ({ detail: res.statusText }))
throw new Error(err.detail || `HTTP ${res.status}`)
}
const blob = await res.blob()
const disposition = res.headers.get('Content-Disposition') ?? ''
const match = disposition.match(/filename="([^"]+)"/)
const filename = match ? match[1] : 'tests.zip'
return { blob, filename }
},
} }
// Transform the flat items array from GET /api/schedule/week into the // Transform the flat items array from GET /api/schedule/week into the
frontend/src/components/SettingsModal.jsx
+9
View File
@@ -97,6 +97,15 @@ export default function SettingsModal({ isOpen, onClose, settings, onSave, onRes
File Paths File Paths
</p> </p>
<div className="flex flex-col gap-3"> <div className="flex flex-col gap-3">
<Field label="Target Directory" required>
<input
type="text"
className={INPUT_CLS}
placeholder="/path/to/target/directory"
value={form.targetDir ?? ''}
onChange={(e) => set('targetDir', e.target.value)}
/>
</Field>
<Field label="P2P / COE CSV" required> <Field label="P2P / COE CSV" required>
<input <input
type="text" type="text"
frontend/src/components/TestWindowDetailsPanel.jsx
+52
View File
@@ -1,5 +1,6 @@
import { useEffect, useMemo, useState } from 'react' import { useEffect, useMemo, useState } from 'react'
import { getDeviceAccentClass } from './TestCard' import { getDeviceAccentClass } from './TestCard'
import { api } from '../api'
const CONFIG_MAP_IMAGE_MODULES = import.meta.glob('../assets/TC*_map/*.png', { const CONFIG_MAP_IMAGE_MODULES = import.meta.glob('../assets/TC*_map/*.png', {
eager: true, eager: true,
@@ -183,6 +184,8 @@ function TestRow({ test }) {
export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose }) { export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose }) {
const [showConfigMap, setShowConfigMap] = useState(false) const [showConfigMap, setShowConfigMap] = useState(false)
const [isExporting, setIsExporting] = useState(false)
const [exportError, setExportError] = useState(null)
const windowTests = useMemo( const windowTests = useMemo(
() => (Array.isArray(windowDetails?.tests) ? windowDetails.tests : []), () => (Array.isArray(windowDetails?.tests) ? windowDetails.tests : []),
[windowDetails], [windowDetails],
@@ -196,6 +199,10 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose
const hasConfigMapImages = configMapImages.length > 0 const hasConfigMapImages = configMapImages.length > 0
const showMapPane = showConfigMap && hasConfigMapImages const showMapPane = showConfigMap && hasConfigMapImages
useEffect(() => {
setExportError(null)
}, [windowDetails])
useEffect(() => { useEffect(() => {
if (!isOpen) return undefined if (!isOpen) return undefined
@@ -209,6 +216,25 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose
return () => window.removeEventListener('keydown', handleKeyDown) return () => window.removeEventListener('keydown', handleKeyDown)
}, [isOpen, onClose]) }, [isOpen, onClose])
async function handleExport() {
if (!windowDetails?.window_id) return
setIsExporting(true)
setExportError(null)
try {
const { blob, filename } = await api.exportWindow(windowDetails.window_id)
const url = URL.createObjectURL(blob)
const a = document.createElement('a')
a.href = url
a.download = filename
a.click()
URL.revokeObjectURL(url)
} catch (err) {
setExportError(err.message || 'Export failed')
} finally {
setIsExporting(false)
}
}
return ( return (
<div <div
className={`absolute inset-0 z-40 transition-opacity duration-300 ${ className={`absolute inset-0 z-40 transition-opacity duration-300 ${
@@ -238,6 +264,25 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose
{windowDetails ? formatWindowHeader(windowDetails) : 'Window details'} {windowDetails ? formatWindowHeader(windowDetails) : 'Window details'}
</h2> </h2>
</div> </div>
<div className="flex items-center gap-2">
<button
type="button"
onClick={handleExport}
disabled={isExporting || !windowDetails?.tests?.length}
className="inline-flex items-center gap-1.5 rounded-full border border-gray-700 px-3 py-2 text-xs font-semibold text-gray-300 hover:border-gray-500 hover:text-white transition-colors disabled:cursor-not-allowed disabled:opacity-50"
aria-label="Export .ini files for this window"
>
{isExporting ? (
<svg xmlns="http://www.w3.org/2000/svg" className="h-3.5 w-3.5 animate-spin" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="2">
<path strokeLinecap="round" strokeLinejoin="round" d="M12 3v3m0 12v3M4.22 4.22l2.12 2.12m11.32 11.32 2.12 2.12M3 12h3m12 0h3M4.22 19.78l2.12-2.12M17.66 6.34l2.12-2.12" />
</svg>
) : (
<svg xmlns="http://www.w3.org/2000/svg" className="h-3.5 w-3.5" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8">
<path strokeLinecap="round" strokeLinejoin="round" d="M4 16v2a2 2 0 0 0 2 2h12a2 2 0 0 0 2-2v-2M7 10l5 5 5-5M12 15V3" />
</svg>
)}
{isExporting ? 'Exporting…' : 'Export Tests'}
</button>
<button <button
type="button" type="button"
onClick={onClose} onClick={onClose}
@@ -249,6 +294,13 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose
</svg> </svg>
</button> </button>
</div> </div>
</div>
{exportError && (
<div className="border-b border-red-900/50 bg-red-950/40 px-5 py-2 text-xs text-red-300">
{exportError}
</div>
)}
{windowDetails && ( {windowDetails && (
<div className="flex-1 min-h-0 overflow-y-auto"> <div className="flex-1 min-h-0 overflow-y-auto">