/
Sturon
/
Diplom
Обзор
Документация
Войти
/
Sturon
/
Diplom
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
analyzer/app/desktop.py
796 строк
28 KB
Sturon
Initial
03 июн 2026, 15:01
03 июн 2026, 15:01
17b44a2
Код
Авторство
О чём код?
from __future__ import annotations import json import os import shutil import sys import threading import time from pathlib import Path from typing import Any from PySide6.QtCore import QObject, Qt, QThread, QTimer, Signal from PySide6.QtGui import QPixmap from PySide6.QtWidgets import ( QApplication, QFormLayout, QFrame, QGridLayout, QGroupBox, QHBoxLayout, QLabel, QLineEdit, QListWidget, QMainWindow, QMessageBox, QPushButton, QSizePolicy, QVBoxLayout, QWidget, QFileDialog, ) 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, Roi from app.runtime_models import AnalyzerConfig IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"} def data_dir() -> Path: return Path(os.environ.get("ANALYZER_DATA_DIR", "data")).resolve() def config_path() -> Path: return data_dir() / "config.json" def load_config() -> AnalyzerConfig: data_dir().mkdir(parents=True, exist_ok=True) path = config_path() if not path.exists(): config = AnalyzerConfig() save_config(config) return config return AnalyzerConfig(**json.loads(path.read_text(encoding="utf-8"))) def save_config(config: AnalyzerConfig) -> None: data_dir().mkdir(parents=True, exist_ok=True) config_path().write_text(config.model_dump_json(by_alias=True, indent=2), encoding="utf-8") class AppSignals(QObject): file_detected = Signal(str) status = Signal(str) class CreateCameraWorker(QThread): success = Signal(dict) failure = Signal(str) def __init__(self, config: AnalyzerConfig, camera_name: str) -> None: super().__init__() self.config = config.model_copy(deep=True) self.camera_name = camera_name def run(self) -> None: try: camera = BackendClient(self.config).create_camera(self.camera_name) self.success.emit(camera) except Exception as exc: self.failure.emit(str(exc)) class ProfilePoller(QThread): profile_loaded = Signal(object) failure = Signal(str) def __init__(self) -> None: super().__init__() self._config: AnalyzerConfig | None = None self._lock = threading.Lock() self._stop = threading.Event() def update_config(self, config: AnalyzerConfig) -> None: with self._lock: self._config = config.model_copy(deep=True) def stop(self) -> None: self._stop.set() def run(self) -> None: while not self._stop.is_set(): with self._lock: config = self._config.model_copy(deep=True) if self._config else None if config is None or config.camera_id is None: self.profile_loaded.emit(None) else: try: profile = BackendClient(config).find_active_profile_for_camera(config.camera_id) self.profile_loaded.emit(profile) except Exception as exc: self.failure.emit(str(exc)) self._stop.wait(3) class ImageProcessor(QThread): completed = Signal(object, str, object) skipped = Signal(str) failure = Signal(str) def __init__( self, config: AnalyzerConfig, profile: dict[str, Any] | None, image_path: Path, ) -> None: super().__init__() self.config = config.model_copy(deep=True) self.profile = profile self.image_path = image_path def run(self) -> None: client = BackendClient(self.config) try: if self.config.camera_id is not None: self.profile = client.find_active_profile_for_camera(self.config.camera_id) except Exception as exc: self.failure.emit(str(exc)) return if self.profile is None: self.skipped.emit("Кадр получен, но точка контроля для камеры еще не назначена") return if not self.profile.get("analysisEnabled", False): self.skipped.emit("Кадр получен, но анализатор для точки контроля выключен") return stable_path = wait_until_stable(self.image_path) if stable_path is None: self.failure.emit("Не удалось дождаться окончания записи файла") return try: image = read_image(stable_path.read_bytes()) reference_url = self.profile["referenceImage"]["imageUrl"] reference = read_image(client.download_media(reference_url)) parameters = build_parameters(self.config, self.profile, stable_path) result, annotated = compare_with_reference(image, reference, parameters) display_path = save_latest_image(annotated) backend_item = client.submit_result( result, stable_path, annotated_image_path=display_path, ) self.completed.emit(result, str(display_path), backend_item) except Exception as exc: self.failure.emit(str(exc)) class InboxHandler(FileSystemEventHandler): def __init__(self, signals: AppSignals) -> None: self.signals = signals self.last_seen: dict[str, float] = {} self.lock = threading.Lock() def on_created(self, event: FileSystemEvent) -> None: self._handle(event) def on_moved(self, event: FileSystemEvent) -> None: self._handle(event) def on_modified(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)) if path.suffix.lower() not in IMAGE_EXTENSIONS: return now = time.monotonic() key = str(path.resolve()) with self.lock: if now - self.last_seen.get(key, 0.0) < 1.0: return self.last_seen[key] = now self.signals.file_detected.emit(str(path)) class AnalyzerWindow(QMainWindow): def __init__(self) -> None: super().__init__() self.config = load_config() self.profile: dict[str, Any] | None = None self.last_image_path: str | None = None self.signals = AppSignals() self.observer: Observer | None = None self.active_processors: list[ImageProcessor] = [] self.inflight_file_keys: set[str] = set() self.processed_file_keys: set[str] = set() self.create_worker: CreateCameraWorker | None = None self.profile_poller = ProfilePoller() self.inbox_scan_timer = QTimer(self) self.setWindowTitle("Analyzer") self.resize(1280, 780) self._build_ui() self._apply_config_to_inputs() self._connect_signals() self._start_services() def closeEvent(self, event) -> None: # type: ignore[override] self._stop_services() super().closeEvent(event) def _build_ui(self) -> None: root = QWidget() self.setCentralWidget(root) layout = QVBoxLayout(root) layout.setContentsMargins(14, 14, 14, 14) layout.setSpacing(12) top = QFrame() top.setObjectName("topbar") top_layout = QGridLayout(top) top_layout.setColumnStretch(0, 1) top_layout.setColumnStretch(1, 1) top_layout.setColumnStretch(2, 1) backend_box = QGroupBox("Подключение") backend_form = QFormLayout(backend_box) self.backend_url_input = QLineEdit() self.watch_dir_input = QLineEdit() self.browse_button = QPushButton("...") folder_row = QHBoxLayout() folder_row.setContentsMargins(0, 0, 0, 0) folder_row.addWidget(self.watch_dir_input) folder_row.addWidget(self.browse_button) folder_widget = QWidget() folder_widget.setLayout(folder_row) self.save_settings_button = QPushButton("Сохранить") backend_form.addRow("Backend URL", self.backend_url_input) backend_form.addRow("Папка кадров", folder_widget) backend_form.addRow("", self.save_settings_button) camera_box = QGroupBox("Камера") camera_form = QFormLayout(camera_box) self.camera_name_input = QLineEdit() self.create_camera_button = QPushButton("Создать камеру") self.reset_button = QPushButton("Полный сброс") self.reset_button.setObjectName("resetButton") camera_form.addRow("Название", self.camera_name_input) camera_form.addRow("", self.create_camera_button) camera_form.addRow("", self.reset_button) status_box = QGroupBox("Статус") status_layout = QFormLayout(status_box) self.camera_status = QLabel("Не создана") self.profile_status = QLabel("Ожидание точки контроля") self.folder_status = QLabel("-") status_layout.addRow("Камера", self.camera_status) status_layout.addRow("Точка контроля", self.profile_status) status_layout.addRow("Папка", self.folder_status) top_layout.addWidget(backend_box, 0, 0) top_layout.addWidget(camera_box, 0, 1) top_layout.addWidget(status_box, 0, 2) layout.addWidget(top) content = QHBoxLayout() content.setSpacing(12) layout.addLayout(content, stretch=1) image_box = QGroupBox("Последний кадр") image_layout = QVBoxLayout(image_box) self.image_label = QLabel("Добавьте изображение в папку-триггер") self.image_label.setAlignment(Qt.AlignCenter) self.image_label.setMinimumSize(640, 420) self.image_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) self.image_label.setObjectName("imageStage") image_layout.addWidget(self.image_label) content.addWidget(image_box, stretch=1) result_box = QGroupBox("Результат анализа") result_layout = QVBoxLayout(result_box) self.verdict_label = QLabel("-") self.verdict_label.setObjectName("verdict") self.defect_count_label = QLabel("-") self.processing_time_label = QLabel("-") self.image_size_label = QLabel("-") self.summary_label = QLabel("Результатов пока нет") self.summary_label.setWordWrap(True) self.defects_list = QListWidget() result_form = QFormLayout() result_form.addRow("Вердикт", self.verdict_label) result_form.addRow("Дефекты", self.defect_count_label) result_form.addRow("Время", self.processing_time_label) result_form.addRow("Размер", self.image_size_label) result_layout.addLayout(result_form) result_layout.addWidget(self.summary_label) result_layout.addWidget(self.defects_list, stretch=1) content.addWidget(result_box, stretch=0) self.message_label = QLabel("Готово") layout.addWidget(self.message_label) self.setStyleSheet(DESKTOP_STYLE) def _apply_config_to_inputs(self) -> None: self.backend_url_input.setText(self.config.backend_url) self.watch_dir_input.setText(self.config.watch_dir) self.camera_name_input.setText(self.config.camera_name or "") self._update_camera_status() self.folder_status.setText(str(Path(self.config.watch_dir).expanduser().resolve())) def _connect_signals(self) -> None: self.browse_button.clicked.connect(self._choose_folder) self.save_settings_button.clicked.connect(self._save_settings) self.create_camera_button.clicked.connect(self._create_camera) self.reset_button.clicked.connect(self._reset_local_state) self.signals.file_detected.connect(self._on_file_detected) self.profile_poller.profile_loaded.connect(self._on_profile_loaded) self.profile_poller.failure.connect(self._on_profile_error) self.inbox_scan_timer.timeout.connect(self._scan_inbox_for_ready_frame) def _start_services(self) -> None: self._ensure_watch_dir() self._restart_observer() self.profile_poller.update_config(self.config) self.profile_poller.start() self.inbox_scan_timer.start(1000) def _stop_services(self) -> None: if self.observer is not None: self.observer.stop() self.observer.join(timeout=3) self.profile_poller.stop() self.profile_poller.wait(3000) for processor in self.active_processors: processor.wait(3000) self.inbox_scan_timer.stop() def _choose_folder(self) -> None: selected = QFileDialog.getExistingDirectory(self, "Выберите папку кадров", self.watch_dir_input.text()) if selected: self.watch_dir_input.setText(selected) def _save_settings(self) -> None: self.config.backend_url = self.backend_url_input.text().strip().rstrip("/") or "http://localhost:8080" self.config.watch_dir = self.watch_dir_input.text().strip() or "data/inbox" save_config(self.config) self._ensure_watch_dir() self._restart_observer() self.profile_poller.update_config(self.config) self._update_camera_status() self.folder_status.setText(str(Path(self.config.watch_dir).expanduser().resolve())) self._set_message("Настройки сохранены") def _create_camera(self) -> None: name = self.camera_name_input.text().strip() if not name: self._set_message("Введите название камеры") return self.create_camera_button.setEnabled(False) self._set_message("Создаю камеру в backend...") self.create_worker = CreateCameraWorker(self.config, name) self.create_worker.success.connect(self._on_camera_created) self.create_worker.failure.connect(self._on_camera_create_failed) self.create_worker.finished.connect(lambda: self.create_camera_button.setEnabled(True)) self.create_worker.start() def _on_camera_created(self, camera: dict[str, Any]) -> None: self.config.camera_id = camera["id"] self.config.camera_name = camera["name"] save_config(self.config) self.profile_poller.update_config(self.config) self._update_camera_status() self._set_message(f"Камера создана: {camera['name']}") def _on_camera_create_failed(self, message: str) -> None: self._set_message(f"Ошибка создания камеры: {message}") def _reset_local_state(self) -> None: answer = QMessageBox.question( self, "Полный сброс", "Сбросить локальные настройки analyzer, привязку камеры и последний результат? Данные backend не удаляются.", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: return self.config = AnalyzerConfig() self.profile = None self.last_image_path = None self.active_processors.clear() self.inflight_file_keys.clear() self.processed_file_keys.clear() save_config(self.config) shutil.rmtree(data_dir() / "latest", ignore_errors=True) self._ensure_watch_dir() self._restart_observer() self.profile_poller.update_config(self.config) self._apply_config_to_inputs() self.profile_status.setText("WAITING") self.profile_status.setProperty("state", "waiting") self.profile_status.style().unpolish(self.profile_status) self.profile_status.style().polish(self.profile_status) self.image_label.setPixmap(QPixmap()) self.image_label.setText("Добавьте изображение в папку-триггер") self._render_empty_result() self._set_message("Локальное состояние analyzer сброшено") def _on_profile_loaded(self, profile: object) -> None: self.profile = profile if isinstance(profile, dict) else None if self.profile is None: self.profile_status.setText("WAITING") self.profile_status.setProperty("state", "waiting") else: name = self.profile.get("name", "Точка контроля") mode = "ANALYZER ON" if self.profile.get("analysisEnabled", False) else "ANALYZER OFF" self.profile_status.setText(f"ACTIVE - {name} | {mode}") self.profile_status.setProperty("state", "active") self.profile_status.style().unpolish(self.profile_status) self.profile_status.style().polish(self.profile_status) if self._analysis_enabled(): QTimer.singleShot(0, self._scan_inbox_for_ready_frame) def _on_profile_error(self, message: str) -> None: self.profile_status.setText("BACKEND ERROR") self._set_message(f"Не удалось получить точку контроля: {message}") def _on_file_detected(self, raw_path: str) -> None: path = Path(raw_path) if not self._analysis_enabled(): self._set_message(f"Файл найден: {path.name}. Анализатор выключен в backend.") return self._set_message(f"Получен файл: {path.name}") self._start_processing_path(path) def _scan_inbox_for_ready_frame(self) -> None: if not self._analysis_enabled() or self.active_processors: return watch_dir = Path(self.config.watch_dir).expanduser().resolve() if not watch_dir.exists(): return candidates = sorted( ( path for path in watch_dir.iterdir() if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS ), key=lambda path: path.stat().st_mtime, ) for path in candidates: key = file_key(path) if key in self.processed_file_keys or key in self.inflight_file_keys: continue self._set_message(f"Найден готовый кадр в папке: {path.name}") self._start_processing_path(path) return def _start_processing_path(self, path: Path) -> None: key = file_key(path) if key in self.processed_file_keys or key in self.inflight_file_keys: return if self.active_processors: self._set_message("Кадр найден, но предыдущий кадр еще обрабатывается") return self.inflight_file_keys.add(key) processor = ImageProcessor(self.config, self.profile, path) processor.completed.connect( lambda result, image_path, backend_item, file_key_value=key: self._on_processing_completed( result, image_path, backend_item, file_key_value, ) ) processor.skipped.connect( lambda message, file_key_value=key: self._on_processing_skipped(message, file_key_value) ) processor.failure.connect( lambda message, file_key_value=key: self._on_processing_failed(message, file_key_value) ) processor.finished.connect(lambda worker=processor, file_key_value=key: self._remove_processor(worker, file_key_value)) self.active_processors.append(processor) processor.start() def _on_processing_completed( self, result: object, image_path: str, backend_item: object, file_key_value: str, ) -> None: self.processed_file_keys.add(file_key_value) self.last_image_path = image_path self._render_image() self._render_result(result) item_id = backend_item.get("id") if isinstance(backend_item, dict) else None suffix = f", item #{item_id}" if item_id is not None else "" self._set_message(f"Кадр обработан и отправлен в backend{suffix}") def _on_processing_skipped(self, message: str, file_key_value: str) -> None: self.inflight_file_keys.discard(file_key_value) self._set_message(message) def _on_processing_failed(self, message: str, file_key_value: str) -> None: self.processed_file_keys.add(file_key_value) self.inflight_file_keys.discard(file_key_value) self._set_message(f"Ошибка обработки: {message}") def _remove_processor(self, processor: ImageProcessor, file_key_value: str) -> None: self.inflight_file_keys.discard(file_key_value) if processor in self.active_processors: self.active_processors.remove(processor) QTimer.singleShot(0, self._scan_inbox_for_ready_frame) def _render_image(self) -> None: if self.last_image_path is None: return pixmap = QPixmap(self.last_image_path) if pixmap.isNull(): return scaled = pixmap.scaled(self.image_label.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation) self.image_label.setPixmap(scaled) def resizeEvent(self, event) -> None: # type: ignore[override] super().resizeEvent(event) QTimer.singleShot(0, self._render_image) def _render_result(self, result: object) -> None: if not hasattr(result, "result"): return ok = bool(result.result) self.verdict_label.setText("TRUE" if ok else "FALSE") self.verdict_label.setProperty("state", "ok" if ok else "bad") self.verdict_label.style().unpolish(self.verdict_label) self.verdict_label.style().polish(self.verdict_label) self.defect_count_label.setText(str(result.defect_count)) self.processing_time_label.setText(f"{result.processing_time_ms} ms") self.image_size_label.setText(f"{result.image_width} x {result.image_height}") self.summary_label.setText(result.summary) self.defects_list.clear() if not result.defects: self.defects_list.addItem("Дефекты не обнаружены") return for defect in result.defects: self.defects_list.addItem( f"{defect.defect_type} / {defect.roi_name} / {defect.confidence:.2f}" ) def _render_empty_result(self) -> None: self.verdict_label.setText("-") self.verdict_label.setProperty("state", "") self.verdict_label.style().unpolish(self.verdict_label) self.verdict_label.style().polish(self.verdict_label) self.defect_count_label.setText("-") self.processing_time_label.setText("-") self.image_size_label.setText("-") self.summary_label.setText("Результатов пока нет") self.defects_list.clear() def _ensure_watch_dir(self) -> None: Path(self.config.watch_dir).expanduser().resolve().mkdir(parents=True, exist_ok=True) def _restart_observer(self) -> None: if self.observer is not None: self.observer.stop() self.observer.join(timeout=3) watch_dir = Path(self.config.watch_dir).expanduser().resolve() self.observer = Observer() self.observer.schedule(InboxHandler(self.signals), str(watch_dir), recursive=False) self.observer.start() QTimer.singleShot(0, self._scan_inbox_for_ready_frame) def _update_camera_status(self) -> None: if self.config.camera_id is None: self.camera_status.setText("Не создана") else: self.camera_status.setText(f"#{self.config.camera_id} {self.config.camera_name or ''}".strip()) def _set_message(self, message: str) -> None: self.message_label.setText(message) def _analysis_enabled(self) -> bool: return bool(self.profile and self.profile.get("analysisEnabled", False)) def wait_until_stable(image_path: Path) -> Path | None: path = image_path.resolve() last_size = -1 for _ in range(30): if not path.exists() or not path.is_file(): time.sleep(0.04) continue size = path.stat().st_size if size > 0 and size == last_size: return path last_size = size time.sleep(0.04) return None def build_parameters( config: AnalyzerConfig, 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=config.max_defects, minConfidence=config.min_confidence, brightnessMin=config.brightness_min, brightnessMax=config.brightness_max, maxDarkRatio=config.max_dark_ratio, maxBrightRatio=config.max_bright_ratio, minTexture=config.min_texture, differenceThreshold=config.difference_threshold, minDefectArea=config.min_defect_area, alignmentSearchRadius=config.alignment_search_radius, alignmentCoarseStep=config.alignment_coarse_step, edgeDifferenceWeight=config.edge_difference_weight, structuredPixelWeight=config.structured_pixel_weight, ), ) def save_latest_image(image) -> Path: latest_dir = data_dir() / "latest" latest_dir.mkdir(parents=True, exist_ok=True) target = latest_dir / "latest.png" image.save(target, format="PNG") return target def file_key(path: Path) -> str: resolved = path.expanduser().resolve() try: stat = resolved.stat() return f"{str(resolved).casefold()}:{stat.st_size}:{stat.st_mtime_ns}" except FileNotFoundError: return str(resolved).casefold() DESKTOP_STYLE = """ QMainWindow { background: #eef2f6; } QWidget { color: #17202c; font-size: 14px; } QFrame#topbar { background: #eef2f6; } QLabel { color: #17202c; } QGroupBox { background: #ffffff; color: #243244; border: 1px solid #cbd5e1; border-radius: 8px; margin-top: 18px; padding: 14px 12px 12px 12px; font-weight: 700; } QGroupBox::title { subcontrol-origin: margin; left: 10px; padding: 0 6px; color: #243244; background: #eef2f6; } QGroupBox QLabel { color: #2f3d50; } QLineEdit { min-height: 30px; border: 1px solid #cbd5e1; border-radius: 6px; padding: 4px 8px; background: #ffffff; color: #17202c; selection-background-color: #206a5d; selection-color: #ffffff; } QPushButton { min-height: 32px; border: 0; border-radius: 6px; padding: 4px 12px; color: #ffffff; background: #206a5d; font-weight: 700; } QPushButton:disabled { background: #94a3b8; } QPushButton#resetButton { background: #b91c1c; } QPushButton#resetButton:hover { background: #991b1b; } QLabel#imageStage { background: #111827; color: #e5edf7; border-radius: 8px; font-size: 16px; font-weight: 700; } QLabel#verdict { min-width: 84px; padding: 8px 12px; border-radius: 8px; background: #edf2f7; color: #243244; font-size: 18px; font-weight: 900; } QLabel#verdict[state="ok"] { background: #dcfce7; color: #166534; } QLabel#verdict[state="bad"] { background: #fee2e2; color: #991b1b; } QLabel[state="active"] { color: #0f7b46; font-weight: 900; } QLabel[state="waiting"] { color: #8a5b00; font-weight: 900; } QListWidget { border: 1px solid #d9e0ea; border-radius: 8px; background: #ffffff; color: #17202c; } """ def main() -> None: app = QApplication(sys.argv) window = AnalyzerWindow() window.show() sys.exit(app.exec())