import os import threading from watchdog.events import FileSystemEventHandler from watchdog.observers.polling import PollingObserver from db_py import ( mark_tests_completed, reset_by_file_id_and_device, update_all_p2p_coe_pairs_sql, update_all_p3p_pairs_sql, ) from scanner import parse_deleted_result_dir_name, process_result_dir, resolve_runtime_path _OBSERVER = None _OBSERVER_LOCK = threading.RLock() class ResultsDirEventHandler(FileSystemEventHandler): def __init__(self, results_root): super().__init__() self.results_root = str(results_root) def _is_direct_child(self, path_value): parent = os.path.dirname(path_value.rstrip("\\/")) return os.path.normcase(parent) == os.path.normcase(self.results_root.rstrip("\\/")) def _mark_completed(self, dir_path): if not self._is_direct_child(dir_path): return dir_name = os.path.basename(dir_path.rstrip("\\/")) completion = process_result_dir(self.results_root, dir_name) if not completion: return mark_tests_completed([completion]) coe_pairs = update_all_p2p_coe_pairs_sql() p3p_pairs = update_all_p3p_pairs_sql() print( "[watcher] Marked completed from new result dir " f"{dir_name} (coe_pair={coe_pairs}, p3p_pair={p3p_pairs})" ) def _mark_completed_from_file(self, file_path): # measurement.db arriving after directory creation should update that single test. if os.path.basename(file_path).lower() != "measurement.db": return parent_dir = os.path.dirname(file_path.rstrip("\\/")) self._mark_completed(parent_dir) def _clear_completed(self, dir_path): if not self._is_direct_child(dir_path): return dir_name = os.path.basename(dir_path.rstrip("\\/")) test_id, device = parse_deleted_result_dir_name(dir_name) if not test_id: return reset_by_file_id_and_device(test_id, device) coe_pairs = update_all_p2p_coe_pairs_sql() p3p_pairs = update_all_p3p_pairs_sql() print( "[watcher] Cleared completed from removed result dir " f"{dir_name} (coe_pair={coe_pairs}, p3p_pair={p3p_pairs})" ) def on_created(self, event): if event.is_directory: self._mark_completed(event.src_path) return self._mark_completed_from_file(event.src_path) def on_deleted(self, event): if event.is_directory: self._clear_completed(event.src_path) def on_moved(self, event): if event.is_directory: self._clear_completed(event.src_path) self._mark_completed(event.dest_path) return self._mark_completed_from_file(event.dest_path) def stop_results_watchers(): global _OBSERVER with _OBSERVER_LOCK: if _OBSERVER is None: return try: _OBSERVER.stop() _OBSERVER.join(timeout=5) finally: _OBSERVER = None print("[watcher] Stopped results directory watcher") def start_results_watchers(results_dir, results_dir_ref): global _OBSERVER with _OBSERVER_LOCK: stop_results_watchers() resolved = [] seen = set() for raw_dir in (results_dir, results_dir_ref): if not raw_dir: continue path = resolve_runtime_path(raw_dir) if not path: continue normalized = os.path.normcase(str(path).rstrip("\\/")) if normalized and normalized not in seen: seen.add(normalized) resolved.append(str(path)) if not resolved: print("[watcher] No results directories configured, watcher not started") return observer = PollingObserver(timeout=1.0) scheduled_count = 0 for path in resolved: try: observer.schedule(ResultsDirEventHandler(path), path, recursive=False) scheduled_count += 1 print(f"[watcher] Watching results directory: {path}") except Exception as exc: print(f"[watcher] Skipping watch path {path}: {exc}") if scheduled_count == 0: print("[watcher] No valid watch paths available, watcher not started") return try: observer.start() except Exception as exc: print(f"[watcher] Failed to start results watcher: {exc}") try: observer.stop() observer.join(timeout=5) except Exception: pass return _OBSERVER = observer print("[watcher] Results directory watcher started")