diff --git a/.gitignore b/.gitignore index 8c143c6..af24acf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,10 +1,8 @@ /backend/.venv /backend/__pycache__ -/backend/_run_sched_test.py /backend/output.txt /backend/.env -IMPLEMENTATIONPLAN.md -scheduler.db -scheduler.db* -TARGET_TEST_DIR_STRUCTURE.md +/backend/tests +/backend/.pytest_cache +*.db .env \ No newline at end of file diff --git a/ONE_DAY_SCHEDULER.md b/ONE_DAY_SCHEDULER.md deleted file mode 100644 index c1cf6c1..0000000 --- a/ONE_DAY_SCHEDULER.md +++ /dev/null @@ -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) \ No newline at end of file diff --git a/backend/Dockerfile b/backend/Dockerfile index cf42ccc..dd62900 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -10,6 +10,8 @@ RUN pip install --no-cache-dir -r requirements.txt COPY . . +WORKDIR /app/src + EXPOSE 8000 CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/backend/_tmp_validate_pairing.py b/backend/_tmp_validate_pairing.py deleted file mode 100644 index fcdb1c0..0000000 --- a/backend/_tmp_validate_pairing.py +++ /dev/null @@ -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) diff --git a/backend/scanner.py b/backend/scanner.py deleted file mode 100644 index 17fcfdc..0000000 --- a/backend/scanner.py +++ /dev/null @@ -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 ///... 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 \\\... 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)") - diff --git a/backend/app.py b/backend/src/app.py similarity index 81% rename from backend/app.py rename to backend/src/app.py index dfffaac..f1dd1be 100644 --- a/backend/app.py +++ b/backend/src/app.py @@ -2,35 +2,38 @@ from contextlib import asynccontextmanager from pathlib import Path from typing import Any from datetime import date, datetime, timedelta +import io import os +import zipfile -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, Response from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field -import db -from parser import CsvValidationError, parse_target_csv +import db as db +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. -os.environ.setdefault("DUT", db.DEVICE_DUT) -os.environ.setdefault("REF", db.DEVICE_REF) +os.environ.setdefault("DUT", db.DUT) +os.environ.setdefault("REF", db.REF) 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_window import get_shift_capacity_for_date, get_shift_sequence, is_off_day from watcher import configure_result_watcher, stop_result_watcher - +from scanner import process_targets 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() REF = os.getenv("REF", "CGW452").strip() - - 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): @@ -66,7 +69,7 @@ def _runtime_overrides_from_settings(settings: dict[str, Any]) -> dict[str, dict } 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] = {} for test_type, suffix in (("P2P", "P2p"), ("COE", "Coe"), ("P3P", "P3p")): 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 -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): 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, "config": row.config, "status": row.status, - "priority": row.priority, "estimated_minutes": row.estimated_minutes, } @@ -349,29 +332,26 @@ def restart_and_clear_data() -> dict[str, Any]: @app.post("/api/tests/load") 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) smb_credentials = _smb_credentials_from_settings(settings) runtime_overrides = _runtime_overrides_from_settings(settings) - try: - 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 = process_targets(target_dir, resolved_csv_paths, smb_credentials=smb_credentials, runtime_overrides=runtime_overrides) - count = db.upsert_tests(parsed.tests, DB_PATH) - - print(f"Loaded {count} tests from {csv_path}, with {len(parsed.warnings)} warnings.") + print(f"Loaded {count} tests from {resolved_csv_paths}.") return { "loaded_tests": count, - "warnings": parsed.warnings, } @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), } +@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 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__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) diff --git a/backend/db.py b/backend/src/db.py similarity index 96% rename from backend/db.py rename to backend/src/db.py index 83a2d07..317bc02 100644 --- a/backend/db.py +++ b/backend/src/db.py @@ -10,15 +10,13 @@ from test_config import serialize_station_testpoint_map, resolve_test_config_key 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) DUT = os.getenv("DUT", "CGW453").strip() REF = os.getenv("REF", "CGW452").strip() # Device names for the test database (use hardware device names) -DEVICE_DUT = DUT -DEVICE_REF = REF MAX_SCHEDULE_VERSIONS = 50 @dataclass(frozen=True) @@ -31,7 +29,6 @@ class TestRecord: power_mode: str | None has_coe_pair: bool coe_pairing: list[str] - priority: int victim_band: str | None config: dict[str, dict[str, str | None]] throttled: bool @@ -40,6 +37,7 @@ class TestRecord: excluded: bool = False raw_payload: dict[str, Any] | None = None station_testpoint_map: str | None = None + file_path: str | None = None @dataclass(frozen=True) @@ -53,7 +51,6 @@ class ScheduleRow: rotation: str | None config: dict[str, dict[str, str | None]] status: str - priority: int estimated_minutes: int @contextmanager @@ -85,7 +82,6 @@ def init_db(db_path: str | Path = DB_PATH) -> None: power_mode TEXT, has_coe_pair INTEGER NOT NULL DEFAULT 0, coe_pairing_json TEXT, - priority INTEGER NOT NULL CHECK (priority BETWEEN 1 AND 5), victim_band TEXT, config_json TEXT, 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_priority ON tests(priority); 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", "throttled", "INTEGER NOT NULL DEFAULT 0") _ensure_column(conn, "tests", "station_testpoint_map", "TEXT") + _ensure_column(conn, "tests", "file_path", "TEXT") _ensure_column(conn, "schedules", "status_snapshot", "TEXT") 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: if not records: return 0 values: list[tuple[Any, ...]] = [] for r in records: - station_testpoint_map = _serialize_station_testpoint_map(r.config) + station_testpoint_map = serialize_station_testpoint_map(r.config) values.append( ( @@ -190,7 +181,6 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in int(r.has_coe_pair), json.dumps(r.coe_pairing or []), json.dumps(r.config), - r.priority, r.victim_band, int(r.throttled), r.estimated_minutes, @@ -198,6 +188,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in int(r.excluded), json.dumps(r.raw_payload or {}), 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( 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 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, coe_pairing_json = excluded.coe_pairing_json, config_json = excluded.config_json, - priority = excluded.priority, victim_band = excluded.victim_band, throttled = excluded.throttled, 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, raw_payload = excluded.raw_payload, station_testpoint_map = excluded.station_testpoint_map, + file_path = excluded.file_path, updated_at = CURRENT_TIMESTAMP """, values, @@ -407,7 +398,7 @@ def get_not_excluded_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis SELECT test_id, device, test_type, rotation, rx_tx, has_coe_pair, power_mode, config_json, victim_band, - coe_pairing_json, priority, throttled, estimated_minutes, + coe_pairing_json, throttled, estimated_minutes, excluded, status, raw_payload FROM tests 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"], has_coe_pair=bool(row["has_coe_pair"]), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), - priority=int(row["priority"]), victim_band=row["victim_band"], config=json.loads(row["config_json"] or "{}"), throttled=bool(row["throttled"]), @@ -450,7 +440,7 @@ def list_schedulable_tests(db_path: str | Path = DB_PATH, rule: str = "") -> lis SELECT test_id, device, test_type, rotation, rx_tx, has_coe_pair, power_mode, config_json, victim_band, - coe_pairing_json, priority, throttled, estimated_minutes, + coe_pairing_json, throttled, estimated_minutes, excluded, status, raw_payload FROM tests 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"], has_coe_pair=bool(row["has_coe_pair"]), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), - priority=int(row["priority"]), victim_band=row["victim_band"], config=json.loads(row["config_json"] or "{}"), throttled=bool(row["throttled"]), @@ -494,7 +483,7 @@ def list_tests_for_device(device: str, db_path: str | Path = DB_PATH) -> list[Te SELECT test_id, device, test_type, rotation, rx_tx, has_coe_pair, 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 FROM tests 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"], has_coe_pair=bool(row["has_coe_pair"]), coe_pairing=json.loads(row["coe_pairing_json"] or "[]"), - priority=int(row["priority"]), victim_band=row["victim_band"], config=json.loads(row["config_json"] or "{}"), throttled=bool(row["throttled"]), @@ -633,7 +621,6 @@ def _hydrate_schedule_rows(rows: list[sqlite3.Row]) -> list[ScheduleRow]: rotation=row["rotation"], config=json.loads(row["config_json"] or "{}"), status=row["status"], - priority=int(row["priority"]), estimated_minutes=int(row["estimated_minutes"]), ) ) @@ -716,7 +703,6 @@ def get_schedule_week( t.rotation, t.config_json, {status_sql} AS status, - t.priority, t.estimated_minutes FROM schedules s 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.config_json, {status_sql} AS status, - t.priority, t.estimated_minutes FROM schedules s JOIN tests t ON t.test_id = s.test_id AND t.device = s.device diff --git a/backend/src/file_manager.py b/backend/src/file_manager.py new file mode 100644 index 0000000..7fe8add --- /dev/null +++ b/backend/src/file_manager.py @@ -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 ///... 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 \\... 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 \ No newline at end of file diff --git a/backend/parser.py b/backend/src/parser.py similarity index 61% rename from backend/parser.py rename to backend/src/parser.py index b6d1877..4c0c689 100644 --- a/backend/parser.py +++ b/backend/src/parser.py @@ -5,13 +5,14 @@ from dataclasses import dataclass from pathlib import Path from typing import Any -try: - import smbclient # type: ignore[import-not-found] -except ModuleNotFoundError: - smbclient = None +from db import DUT, REF, TestRecord +from file_manager import ( + open_csv_handle +) -from db import DEVICE_DUT, DEVICE_REF, TestRecord -from scanner import resolve_runtime_path +TEST_TYPES = {"P2P", "COE", "P3P"} +POWER_MODES = {"LPI", "SP"} +RX_TX = {"RX", "TX"} P2P_COE_REQUIRED_COLUMNS = [ "Priority", @@ -92,131 +93,6 @@ class CsvValidationError(ValueError): _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 ///... 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]: return [ column @@ -240,35 +116,25 @@ def _detect_csv_format(normalized_fieldnames: set[str]) -> str: ) 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, runtime_overrides: dict[str, Any] | None = None, -) -> ParseResult: - paths = _resolve_csv_paths(csv_path, smb_credentials=smb_credentials) +) -> dict[str, Any]: runtime_defaults_by_device = _resolve_runtime_defaults(runtime_overrides) - all_tests: list[TestRecord] = [] - all_warnings: list[str] = [] - seen_test_keys: set[tuple[str, str]] = set() + results = {} + + if isinstance(paths, (str, Path)): + paths = [paths] for path in paths: - parsed = _parse_single_csv( + csv_result = _parse_single_csv( path, smb_credentials=smb_credentials, 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]: 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) if not runtime_overrides: return { - DEVICE_DUT: dict(legacy_defaults), - DEVICE_REF: dict(legacy_defaults), + DUT: dict(legacy_defaults), + REF: dict(legacy_defaults), } resolved: dict[str, dict[str, int]] = {} - for device in (DEVICE_DUT, DEVICE_REF): + for device in (DUT, REF): device_defaults = dict(legacy_defaults) raw_device_overrides = runtime_overrides.get(device) if isinstance(raw_device_overrides, dict): @@ -311,48 +177,13 @@ def _resolve_runtime_defaults(runtime_overrides: dict[str, Any] | None) -> dict[ 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( path: str, smb_credentials: dict[str, Any] | None, runtime_defaults_by_device: dict[str, dict[str, int]], -) -> ParseResult: - with _open_csv_handle(path, smb_credentials=smb_credentials) as handle: +) -> dict[str, Any]: + records: dict[str, Any] = {} + with open_csv_handle(path, smb_credentials=smb_credentials) as handle: reader = csv.DictReader(handle) if not reader.fieldnames: 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): test_id = (row.get("TC ID") or "").strip() 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 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 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 = _normalize_victim_band(_row_get(row, victim_band_source)) - # Default priority based on test type and COE pairing. - if test_type == "P2P": - priority = 2 if has_coe_pair else 3 - elif test_type == "COE": - priority = 3 - else: - priority = 4 - - record = TestRecord( - test_id=test_id, - device=DEVICE_DUT, - test_type=test_type, - rotation=rotation, - rx_tx=rx_tx, - power_mode=power_mode, - has_coe_pair=has_coe_pair, - coe_pairing=[], - 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)) + records[test_id] = { + "test_id": test_id, + "test_type": test_type, + "rotation": rotation, + "rx_tx": rx_tx, + "power_mode": power_mode, + "has_coe_pair": has_coe_pair, + "coe_pairing": [], + "victim_band": victim_band, + "config": config, + "throttled": throttled, + "estimated_minutes": {device: defaults[test_type] for device, defaults in runtime_defaults_by_device.items()}, + "status": "pending", + "excluded": False, + "raw_payload": row, + } + records_with_signature.append((records[test_id], signature)) # Build COE pairing based on full signature and RX/TX+band suffix. # Example key suffixes: RXAX, TXAX, RXBE, TXBE. coe_by_signature_and_suffix: dict[tuple[tuple[str, ...], str], list[str]] = {} for record, signature in records_with_signature: - if record.test_type == "COE": - suffix = _extract_pairing_suffix(record.test_id) + if record["test_type"] == "COE": + suffix = _extract_pairing_suffix(record["test_id"]) if not suffix: continue # 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) - 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: - pairs = record.coe_pairing - if record.test_type == "P2P" and signature is not None: - suffix = _extract_pairing_suffix(record.test_id) + pairs = record["coe_pairing"] + if record["test_type"] == "P2P" and signature is not None: + suffix = _extract_pairing_suffix(record["test_id"]) 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 [] - for device in (DEVICE_DUT, DEVICE_REF): - 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), - 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, - ) - ) + records[record["test_id"]].update( + has_coe_pair=bool(pairs), + coe_pairing=pairs, + ) - return ParseResult(tests=tests, warnings=warnings) + return records 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" -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 { "5G": { "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]]: if csv_format == "p3p": 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: @@ -670,3 +474,47 @@ def _row_get(row: dict[str, str], key: str) -> str | 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": {}, + } + diff --git a/backend/src/scanner.py b/backend/src/scanner.py new file mode 100644 index 0000000..990e567 --- /dev/null +++ b/backend/src/scanner.py @@ -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 to ROT 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 diff --git a/backend/scheduler.py b/backend/src/scheduler.py similarity index 100% rename from backend/scheduler.py rename to backend/src/scheduler.py diff --git a/backend/test_bundle.py b/backend/src/test_bundle.py similarity index 96% rename from backend/test_bundle.py rename to backend/src/test_bundle.py index 99b4654..e8da093 100644 --- a/backend/test_bundle.py +++ b/backend/src/test_bundle.py @@ -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]: """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_ref: set[str] = set() 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: if dut_test_id in processed_dut: 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( 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: if ref_test_id in processed_ref: 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( create_bundle([ref_test_id], REF, BUNDLE_PRIORITY_DUT_COMPLETED_MIRROR, bundle_index, active_dut, active_ref) ) diff --git a/backend/test_config.py b/backend/src/test_config.py similarity index 100% rename from backend/test_config.py rename to backend/src/test_config.py diff --git a/backend/test_window.py b/backend/src/test_window.py similarity index 100% rename from backend/test_window.py rename to backend/src/test_window.py diff --git a/backend/watcher.py b/backend/src/watcher.py similarity index 99% rename from backend/watcher.py rename to backend/src/watcher.py index 9fdab11..e849c0d 100644 --- a/backend/watcher.py +++ b/backend/src/watcher.py @@ -15,7 +15,8 @@ except ModuleNotFoundError: from watchdog.events import FileSystemEvent, FileSystemEventHandler, FileSystemMovedEvent 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") diff --git a/docker-compose.yml b/docker-compose.yml index cb50575..dcfd7cc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,7 +7,8 @@ services: environment: HOST_BROWSE_ROOT: ${HOST_BROWSE_ROOT} HOST_MOUNT_ROOT: /host - DB_PATH: /app/runtime/scheduler.db + DB_PATH: /app/data/scheduler.db + PYTHONPATH: /app/src volumes: - ${HOST_BROWSE_ROOT}:/host - scheduler-db:/app/runtime diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 70ae181..7bb94f8 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -201,6 +201,7 @@ function buildCalendarScheduleData(currentScheduleData = {}, previousScheduleDat } const DEFAULT_SETTINGS = { + targetDir: '', p2pCoeCsvPath: '', p3pCsvPath: '', refResultDir: '', @@ -394,11 +395,13 @@ export default function App() { .split(',').map(s => s.trim()).filter(Boolean) await api.saveHolidays(holidayDates) - if (sanitizedSettings.p2pCoeCsvPath?.trim()) { - await api.loadCsv(sanitizedSettings.p2pCoeCsvPath.trim()) - } - if (sanitizedSettings.p3pCsvPath?.trim()) { - await api.loadCsv(sanitizedSettings.p3pCsvPath.trim()) + const csvPaths = [ + sanitizedSettings.p2pCoeCsvPath?.trim(), + sanitizedSettings.p3pCsvPath?.trim(), + ].filter(Boolean) + + if (csvPaths.length > 0) { + await api.loadCsv(csvPaths, sanitizedSettings.targetDir?.trim() ?? '') } setSettings(sanitizedSettings) diff --git a/frontend/src/api.js b/frontend/src/api.js index b25bd54..e429732 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -27,7 +27,7 @@ export const api = { saveHolidays: (dates) => request('POST', '/holidays', { dates }), // Tests - loadCsv: (csv_path) => request('POST', '/tests/load', { csv_path }), + loadCsv: (csv_paths, target_dir) => request('POST', '/tests/load', { csv_paths, target_dir }), // Schedule compileSchedule: (opts) => request('POST', '/schedule/compile', opts), @@ -39,6 +39,20 @@ export const api = { return request('GET', `/schedule/week?${params.toString()}`) }, 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 diff --git a/frontend/src/components/SettingsModal.jsx b/frontend/src/components/SettingsModal.jsx index 9e283d8..d177bed 100644 --- a/frontend/src/components/SettingsModal.jsx +++ b/frontend/src/components/SettingsModal.jsx @@ -97,6 +97,15 @@ export default function SettingsModal({ isOpen, onClose, settings, onSave, onRes File Paths

+ + set('targetDir', e.target.value)} + /> + (Array.isArray(windowDetails?.tests) ? windowDetails.tests : []), [windowDetails], @@ -196,6 +199,10 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose const hasConfigMapImages = configMapImages.length > 0 const showMapPane = showConfigMap && hasConfigMapImages + useEffect(() => { + setExportError(null) + }, [windowDetails]) + useEffect(() => { if (!isOpen) return undefined @@ -209,6 +216,25 @@ export default function TestWindowDetailsPanel({ isOpen, windowDetails, onClose return () => window.removeEventListener('keydown', handleKeyDown) }, [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 (
- +
+ + +
+ {exportError && ( +
+ {exportError} +
+ )} + {windowDetails && (