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

export working on network files

Browse Source
This commit is contained in:
Mia Wu
2026-07-29 16:12:15 -04:00
parent b6c3cf9786
commit 0d2a1253cd
6 changed files with 72 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/app.py
+36 -6
View File
@@ -12,7 +12,7 @@ from pydantic import BaseModel, Field
import db as db
from parser import CsvValidationError
from file_manager import resolve_requested_csv_path, read_file_bytes
from file_manager import resolve_requested_csv_path, read_file_bytes, path_name
# Ensure new_scheduler resolves the same DUT/REF labels as db records.
os.environ.setdefault("DUT", db.DUT)
@@ -22,8 +22,8 @@ from scheduler import Scheduler, Test as SchedulerTest
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
from watcher import configure_result_watcher, stop_result_watcher, trigger_scan
from scanner import process_targets, scan_results
APP_ROOT = Path(__file__).resolve().parent
DB_PATH = Path(os.getenv("DB_PATH", str(APP_ROOT.parent / "data" / "scheduler.db")))
@@ -349,6 +349,13 @@ def load_tests(request: LoadTestsRequest) -> dict[str, Any]:
count = process_targets(target_dir, resolved_csv_paths, smb_credentials=smb_credentials, runtime_overrides=runtime_overrides)
# Synchronously scan result directories before returning so completed tests are
# written to the DB before the caller can compile a new schedule.
dut_result_dir = resolve_runtime_path(settings.get("dutResultDir", "") or "")
ref_result_dir = resolve_runtime_path(settings.get("refResultDir", "") or "")
if dut_result_dir and ref_result_dir:
scan_results(dut_result_dir, ref_result_dir, smb_credentials=smb_credentials)
print(f"Loaded {count} tests from {resolved_csv_paths}.")
return {
"loaded_tests": count,
@@ -560,18 +567,41 @@ def export_window(window_id: str, version: int | None = None) -> Response:
}
file_paths = [path_lookup[k] for k in sorted(wanted) if k in path_lookup]
# Include one GLOBAL.ini from the first parent directory that has test files.
# GLOBAL.ini is shared config — only one copy belongs in the export.
global_paths: list[str] = []
for fp in file_paths:
normalized = str(fp).replace("/", "\\")
if "\\" not in normalized:
continue
parent_dir = normalized.rsplit("\\", 1)[0]
global_paths.append(f"{parent_dir}\\GLOBAL.ini")
break
all_export_paths = file_paths + global_paths
settings = db.read_settings(DB_PATH)
smb_credentials = _smb_credentials_from_settings(settings)
buf = io.BytesIO()
added_count = 0
read_errors: list[str] = []
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
for fp in file_paths:
for fp in all_export_paths:
try:
data = read_file_bytes(fp, smb_credentials)
except OSError as exc:
except Exception as exc:
print(f"[export] Could not read {fp}: {exc}")
read_errors.append(f"{fp}: {exc}")
continue
zf.writestr(os.path.basename(fp), data)
zf.writestr(path_name(fp), data)
added_count += 1
if all_export_paths and added_count == 0:
raise HTTPException(
status_code=502,
detail=f"Failed to read scheduled network files. First error: {read_errors[0] if read_errors else 'unknown error'}",
)
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"]}))
backend/src/file_manager.py
+9 -8
View File
@@ -207,14 +207,15 @@ def open_csv_handle(path: str, smb_credentials: dict[str, Any] | None):
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()
path = normalize_input_path(path)
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):
backend/src/scheduler.py
+3 -3
View File
@@ -64,7 +64,7 @@ class Scheduler:
def compile_schedule(self) -> str | None:
bundles = build_test_bundles(self.active_dut, self.active_ref, self.top_priority_tests)
bundles = build_test_bundles(self.active_dut, self.active_ref, self.top_priority_tests, log=True)
if not bundles:
return f"Error: No test bundles could be created from the provided tests."
@@ -87,7 +87,7 @@ class Scheduler:
tc_order = self.get_tc_order(self.all_tests)
for tc in tc_order:
tc_bundles = all_bundles_by_tc[tc]
tc_bundles = all_bundles_by_tc.get(tc)
if tc_bundles is None or len(tc_bundles) == 0:
print(f"[scheduler] No bundles found for TC: {tc}. Skipping to next TC.")
continue
@@ -366,7 +366,7 @@ class Scheduler:
# Implement the logic for getting the test case order
all_dut = {test.test_id: test for test in all_tests if test.device == DUT}
all_ref = {test.test_id: test for test in all_tests if test.device == REF}
bundles = build_test_bundles(all_dut, all_ref, [])
bundles = build_test_bundles(all_dut, all_ref, [], log=False)
all_tcs = set()
for bundle in bundles:
backend/src/test_bundle.py
+4 -3
View File
@@ -41,11 +41,12 @@ class TestBundle:
priority: int
completed: int
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]], log: bool = False) -> list[TestBundle]:
"""Build deterministic bundles for DP scheduling."""
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]}")
if log:
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] = []
backend/src/test_config.py
+15 -4
View File
@@ -4,9 +4,9 @@ import json
from typing import Any
TEST_CONFIG_DEFINITION = {
"TC1": {"T1D": "STA5", "T1F": "STA56", "T1I": "STA58", "T1L": "STA64", "T2A":"STA63", "T2O":"STA65", "T2J": "STA59", "T2E": "STA6", "T3E": "STA4"},
"TC2": {"T1D": "STA64", "T1F": "STA4", "T1I": "STA5", "T1L": "STA58", "T2A":"STA56", "T2O":"STA59", "T2J": "STA6", "T2E": "STA65", "T3E": "STA63"},
"TC3": {"T1D": "STA58", "T1F": "STA63", "T1I": "STA64", "T2J": "STA65", "T2E": "STA59", "T3E": "STA56"},
"TC1": {"T1P": "STA5", "T1F": "STA56", "T1I": "STA58", "T1L": "STA64", "T2A":"STA63", "T2O":"STA65", "T2J": "STA59", "T2E": "STA6", "T3E": "STA4"},
"TC2": {"T1P": "STA64", "T1F": "STA4", "T1I": "STA5", "T1L": "STA58", "T2A":"STA56", "T2O":"STA59", "T2J": "STA6", "T2E": "STA65", "T3E": "STA63"},
"TC3": {"T1P": "STA58", "T1F": "STA63", "T1I": "STA64", "T2J": "STA65", "T2E": "STA59", "T3E": "STA56"},
"TC4": {"T1B": "STA56", "T1C": "STA4", "T1D": "STA63"},
"TC5": {"T1B": "STA4", "T1C": "STA63", "T1D": "STA56"},
"TC6": {"T1B": "STA63", "T1C": "STA56", "T1D": "STA4"},
@@ -26,6 +26,17 @@ def _norm(value: Any) -> str:
return " ".join(str(value).strip().upper().split())
# Canonical aliases: any key is normalised to its value before lookup.
_TESTPOINT_ALIASES: dict[str, str] = {
"T2Q": "T2E",
}
def _norm_testpoint(tp: str) -> str:
"""Return the canonical testpoint name, resolving known aliases."""
return _TESTPOINT_ALIASES.get(tp, tp)
def _get_test_config(test: Any) -> dict[str, dict[str, str | None]]:
if isinstance(test, dict):
config = test.get("config") or {}
@@ -38,7 +49,7 @@ def _testpoint_to_station_sets(config: dict[str, dict[str, str | None]]) -> dict
result: dict[str, set[str]] = {}
def _add_entry(entry: dict[str, str | None]) -> None:
testpoint = _norm(entry.get("test_point") or entry.get("Test Point"))
testpoint = _norm_testpoint(_norm(entry.get("test_point") or entry.get("Test Point")))
sta_raw = _norm(entry.get("sta") or entry.get("STA"))
if not testpoint or not sta_raw:
return
backend/src/watcher.py
+5
View File
@@ -271,6 +271,11 @@ def configure_result_watcher(settings: dict[str, Any]) -> None:
_WATCHER.configure_from_settings(settings)
def trigger_scan(reason: str = "manual") -> None:
"""Immediately schedule a result scan on the background thread."""
_WATCHER.scan_now(reason=reason)
def stop_result_watcher() -> None:
_WATCHER.stop()