test_dashboard/server/watcher.py

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):
        # New measurement/log files arriving after directory creation should update that single test.
        filename = os.path.basename(file_path).lower()
        is_duration_log = filename.endswith(".log") or filename.endswith(".txt") or "log" in filename
        if filename != "measurement.db" and not is_duration_log:
            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")