analyzer/app/runtime.py - master - Sturon/Diplom

analyzer/app/runtime.py
307 строк12 KB
Sturon
Initial
03 июн 2026, 15:01
03 июн 2026, 15:0117b44a2
from __future__ import annotations

import json
import os
import threading
import time
from pathlib import Path
from typing import Any

from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from app.analyzer import compare_with_reference, read_image
from app.backend_client import BackendClient
from app.models import DetectionRules, EvaluationParameters, EvaluationResponse, Roi
from app.runtime_models import AnalyzerConfig, AnalyzerStateResponse, ProcessingState, ProfileState

IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}


class AnalyzerRuntime:
    def __init__(self) -> None:
        self.data_dir = Path(os.environ.get("ANALYZER_DATA_DIR", "data")).resolve()
        self.config_path = self.data_dir / "config.json"
        self.latest_dir = self.data_dir / "latest"
        self.lock = threading.RLock()
        self.config = self._load_config()
        self.profile = ProfileState(active=False)
        self.profile_details: dict[str, Any] | None = None
        self.processing = ProcessingState()
        self.last_result: EvaluationResponse | None = None
        self.last_image_path: Path | None = None
        self.backend_item_url: str | None = None
        self.stop_event = threading.Event()
        self.profile_thread: threading.Thread | None = None
        self.observer: Observer | None = None

    def start(self) -> None:
        self.data_dir.mkdir(parents=True, exist_ok=True)
        self.latest_dir.mkdir(parents=True, exist_ok=True)
        self._ensure_watch_dir()
        self._start_profile_polling()
        self._restart_watcher()

    def stop(self) -> None:
        self.stop_event.set()
        if self.observer is not None:
            self.observer.stop()
            self.observer.join(timeout=5)
        if self.profile_thread is not None:
            self.profile_thread.join(timeout=5)

    def snapshot(self) -> AnalyzerStateResponse:
        with self.lock:
            return AnalyzerStateResponse(
                config=self.config,
                profile=self.profile,
                processing=self.processing,
                lastResult=self.last_result,
                lastImageUrl="/api/app/last-image" if self.last_image_path else None,
                backendItemUrl=self.backend_item_url,
            )

    def update_settings(self, backend_url: str | None, watch_dir: str | None) -> AnalyzerConfig:
        with self.lock:
            if backend_url:
                self.config.backend_url = backend_url.strip().rstrip("/")
            if watch_dir:
                self.config.watch_dir = watch_dir.strip()
            self._save_config()
            self._ensure_watch_dir()
            self._restart_watcher()
            return self.config

    def create_camera(self, name: str) -> dict[str, Any]:
        camera = BackendClient(self.config).create_camera(name)
        with self.lock:
            self.config.camera_id = camera["id"]
            self.config.camera_name = camera["name"]
            self._save_config()
        self.refresh_profile_once()
        return camera

    def refresh_profile_once(self) -> None:
        with self.lock:
            camera_id = self.config.camera_id
        if camera_id is None:
            with self.lock:
                self.profile = ProfileState(active=False, status="NO_CAMERA")
                self.profile_details = None
            return

        try:
            profile = BackendClient(self.config).find_active_profile_for_camera(camera_id)
        except Exception as exc:
            with self.lock:
                self.profile = ProfileState(active=False, status="BACKEND_ERROR")
                self.profile_details = None
                self.processing = ProcessingState(state="warning", message=f"Backend unavailable: {exc}")
            return

        with self.lock:
            if profile is None:
                self.profile = ProfileState(active=False, status="WAITING_PROFILE")
                self.profile_details = None
                return

            rois = profile.get("referenceImage", {}).get("rois") or []
            self.profile = ProfileState(
                active=True,
                id=profile.get("id"),
                name=profile.get("name"),
                status=profile.get("status"),
                roisCount=len(rois),
            )
            self.profile_details = profile

    def process_image(self, image_path: Path) -> None:
        if image_path.suffix.lower() not in IMAGE_EXTENSIONS:
            return

        stable_path = self._wait_until_stable(image_path)
        if stable_path is None:
            return

        with self.lock:
            camera_id = self.config.camera_id
            profile = self.profile_details
            self.processing = ProcessingState(state="processing", fileName=stable_path.name)

        if camera_id is not None:
            try:
                profile = BackendClient(self.config).find_active_profile_for_camera(camera_id)
                with self.lock:
                    self.profile_details = profile
            except Exception as exc:
                with self.lock:
                    self.processing = ProcessingState(
                        state="error",
                        message=f"Backend unavailable: {exc}",
                        fileName=stable_path.name,
                    )
                return

        if profile is None:
            with self.lock:
                self.processing = ProcessingState(
                    state="waiting-profile",
                    message="Кадр получен, но точка контроля для камеры еще не назначена",
                    fileName=stable_path.name,
                )
            return

        if not profile.get("analysisEnabled", False):
            with self.lock:
                self.processing = ProcessingState(
                    state="waiting-analysis",
                    message="Кадр получен, но анализатор для точки контроля выключен",
                    fileName=stable_path.name,
                )
            return

        try:
            image = read_image(stable_path.read_bytes())
            reference_url = profile["referenceImage"]["imageUrl"]
            reference = read_image(BackendClient(self.config).download_media(reference_url))
            parameters = self._build_parameters(profile, stable_path)
            result, annotated = compare_with_reference(image, reference, parameters)
            display_path = self._save_latest_image(annotated)
            backend_item = BackendClient(self.config).submit_result(
                result,
                stable_path,
                annotated_image_path=display_path,
            )

            with self.lock:
                self.last_result = result
                self.last_image_path = display_path
                self.backend_item_url = self._build_backend_item_url(backend_item)
                self.processing = ProcessingState(
                    state="done",
                    message="Кадр обработан и отправлен в backend",
                    fileName=stable_path.name,
                )
        except Exception as exc:
            with self.lock:
                self.processing = ProcessingState(
                    state="error",
                    message=str(exc),
                    fileName=stable_path.name,
                )

    def _load_config(self) -> AnalyzerConfig:
        self.data_dir.mkdir(parents=True, exist_ok=True)
        if not self.config_path.exists():
            config = AnalyzerConfig()
            self.config_path.write_text(config.model_dump_json(by_alias=True, indent=2), encoding="utf-8")
            return config
        return AnalyzerConfig(**json.loads(self.config_path.read_text(encoding="utf-8")))

    def _save_config(self) -> None:
        self.config_path.write_text(self.config.model_dump_json(by_alias=True, indent=2), encoding="utf-8")

    def _ensure_watch_dir(self) -> None:
        Path(self.config.watch_dir).expanduser().resolve().mkdir(parents=True, exist_ok=True)

    def _start_profile_polling(self) -> None:
        if self.profile_thread is not None:
            return
        self.profile_thread = threading.Thread(target=self._profile_poll_loop, daemon=True)
        self.profile_thread.start()

    def _profile_poll_loop(self) -> None:
        while not self.stop_event.is_set():
            self.refresh_profile_once()
            self.stop_event.wait(3)

    def _restart_watcher(self) -> None:
        if self.observer is not None:
            self.observer.stop()
            self.observer.join(timeout=5)

        watch_dir = Path(self.config.watch_dir).expanduser().resolve()
        handler = InboxEventHandler(self)
        observer = Observer()
        observer.schedule(handler, str(watch_dir), recursive=False)
        observer.start()
        self.observer = observer

    def _wait_until_stable(self, image_path: Path) -> Path | None:
        path = image_path.resolve()
        last_size = -1
        for _ in range(20):
            if not path.exists() or not path.is_file():
                time.sleep(0.05)
                continue
            size = path.stat().st_size
            if size > 0 and size == last_size:
                return path
            last_size = size
            time.sleep(0.05)
        return None

    def _build_parameters(self, profile: dict[str, Any], image_path: Path) -> EvaluationParameters:
        reference = profile.get("referenceImage") or {}
        rois = [
            Roi(
                name=roi.get("name") or f"ROI {index + 1}",
                order=roi.get("order", index),
                x=roi["x"],
                y=roi["y"],
                width=roi["width"],
                height=roi["height"],
            )
            for index, roi in enumerate(reference.get("rois") or [])
        ]
        return EvaluationParameters(
            requestId=f"{image_path.stem}-{int(time.time() * 1000)}",
            profileId=profile["id"],
            rois=rois,
            rules=DetectionRules(
                maxDefects=self.config.max_defects,
                minConfidence=self.config.min_confidence,
                brightnessMin=self.config.brightness_min,
                brightnessMax=self.config.brightness_max,
                maxDarkRatio=self.config.max_dark_ratio,
                maxBrightRatio=self.config.max_bright_ratio,
                minTexture=self.config.min_texture,
                differenceThreshold=self.config.difference_threshold,
                minDefectArea=self.config.min_defect_area,
                alignmentSearchRadius=self.config.alignment_search_radius,
                alignmentCoarseStep=self.config.alignment_coarse_step,
                edgeDifferenceWeight=self.config.edge_difference_weight,
                structuredPixelWeight=self.config.structured_pixel_weight,
            ),
        )

    def _save_latest_image(self, image) -> Path:
        target = self.latest_dir / "latest.png"
        image.save(target, format="PNG")
        return target

    def _build_backend_item_url(self, backend_item: dict[str, Any]) -> str | None:
        item_id = backend_item.get("id")
        if item_id is None:
            return None
        return f"{self.config.backend_url.rstrip('/')}/api/inspection-items/{item_id}"


class InboxEventHandler(FileSystemEventHandler):
    def __init__(self, runtime: AnalyzerRuntime) -> None:
        self.runtime = runtime

    def on_created(self, event: FileSystemEvent) -> None:
        self._handle(event)

    def on_moved(self, event: FileSystemEvent) -> None:
        self._handle(event)

    def _handle(self, event: FileSystemEvent) -> None:
        if event.is_directory:
            return
        path = Path(getattr(event, "dest_path", event.src_path))
        threading.Thread(target=self.runtime.process_image, args=(path,), daemon=True).start()


runtime = AnalyzerRuntime()