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

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

import json
from pathlib import Path
from typing import Any

import requests

from app.models import EvaluationResponse
from app.runtime_models import AnalyzerConfig


class BackendClient:
    def __init__(self, config: AnalyzerConfig):
        self.config = config

    @property
    def base_url(self) -> str:
        return self.config.backend_url.rstrip("/")

    def create_camera(self, name: str) -> dict[str, Any]:
        response = requests.post(
            f"{self.base_url}/api/cameras",
            json={
                "name": name.strip(),
                "type": "MOCK_FOLDER",
                "location": self.config.watch_dir,
                "active": True,
                "notes": "Создано локальным приложением анализатора",
            },
            timeout=5,
        )
        response.raise_for_status()
        return response.json()

    def find_active_profile_for_camera(self, camera_id: int) -> dict[str, Any] | None:
        response = requests.get(f"{self.base_url}/api/profiles", timeout=5)
        response.raise_for_status()
        profiles = response.json()
        matches = [
            profile
            for profile in profiles
            if profile.get("cameraId") == camera_id
            and profile.get("status") == "ACTIVE"
            and profile.get("cameraActive") is True
        ]
        if not matches:
            return None

        profile_id = matches[0]["id"]
        details = requests.get(f"{self.base_url}/api/profiles/{profile_id}", timeout=5)
        details.raise_for_status()
        return details.json()

    def download_media(self, url: str) -> bytes:
        actual_url = url if url.startswith(("http://", "https://")) else f"{self.base_url}{url}"
        response = requests.get(actual_url, timeout=15)
        response.raise_for_status()
        return response.content

    def find_pending_inspection_for_camera(self, camera_id: int) -> dict[str, Any] | None:
        response = requests.get(
            f"{self.base_url}/api/external-inspections/pending",
            params={"cameraId": camera_id},
            timeout=5,
        )
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return response.json()

    def submit_result(
        self,
        result: EvaluationResponse,
        image_path: Path,
        inspection_item_id: int | None = None,
        annotated_image_path: Path | None = None,
    ) -> dict[str, Any]:
        defects = [
            {
                "defectType": defect.defect_type,
                "roiName": defect.roi_name,
                "roiOrder": defect.roi_order,
                "confidence": defect.confidence,
                "x": defect.x,
                "y": defect.y,
                "width": defect.width,
                "height": defect.height,
                "notes": json.dumps(defect.metrics, ensure_ascii=False),
            }
            for defect in result.defects
        ]
        payload = {
            "inspectionItemId": inspection_item_id,
            "profileId": result.profile_id,
            "result": result.result,
            "defectCount": result.defect_count,
            "processingTimeMs": result.processing_time_ms,
            "summary": result.summary,
            "defects": defects,
        }

        with image_path.open("rb") as image_file:
            files = {
                "result": ("result.json", json.dumps(payload, ensure_ascii=False), "application/json"),
                "image": (image_path.name, image_file, "application/octet-stream"),
            }
            annotated_file = None
            if annotated_image_path is not None:
                annotated_file = annotated_image_path.open("rb")
                files["annotatedImage"] = (
                    annotated_image_path.name,
                    annotated_file,
                    "image/png",
                )
            try:
                response = requests.post(
                    f"{self.base_url}/api/external-inspections",
                    files=files,
                    timeout=20,
                )
            finally:
                if annotated_file is not None:
                    annotated_file.close()
        response.raise_for_status()
        return response.json()