엣지 디바이스 선택

import subprocess
import json
from dataclasses import dataclass
from enum import Enum

class JetsonModel(Enum):
    NANO = "nano"
    XAVIER_NX = "xavier_nx"
    ORIN_NANO = "orin_nano"
    AGX_ORIN = "agx_orin"

class JetsonDeviceManager:
    """Jetson 디바이스 관리 및 모니터링"""

    SPECS = {
        JetsonModel.NANO: {"tops": 0.5, "power": [5, 10], "ram": 4},
        JetsonModel.XAVIER_NX: {"tops": 21, "power": [10, 20], "ram": 8},
        JetsonModel.ORIN_NANO: {"tops": 40, "power": [7, 15], "ram": 8},
        JetsonModel.AGX_ORIN: {"tops": 275, "power": [15, 60], "ram": 64}
    }

    def __init__(self, model: JetsonModel):
        self.model = model
        self.specs = self.SPECS[model]

    def get_power_mode(self) -> dict:
        """현재 전력 모드 조회"""
        result = subprocess.run(
            ["nvpmodel", "-q"],
            capture_output=True, text=True
        )
        return self._parse_nvpmodel(result.stdout)

    def set_power_mode(self, mode: int):
        """전력 모드 설정 (0=MAX, 1=보통, 2=저전력)"""
        subprocess.run(["sudo", "nvpmodel", "-m", str(mode)])

    def get_gpu_utilization(self) -> dict:
        """GPU 사용률 조회 (tegrastats 파싱)"""
        result = subprocess.run(
            ["tegrastats", "--interval", "100"],
            capture_output=True, text=True, timeout=1
        )
        return self._parse_tegrastats(result.stdout)

    def estimate_inference_fps(self, model_tops: float) -> float:
        """모델 연산량 기반 추론 FPS 추정"""
        # 실제 효율은 이론치의 60-80%
        efficiency = 0.7
        return (self.specs["tops"] * efficiency) / model_tops

from openvino.runtime import Core, CompiledModel
import numpy as np
from typing import List, Optional

class OpenVINODeviceSelector:
    """OpenVINO 디바이스 선택 및 벤치마크"""

    def __init__(self):
        self.core = Core()
        self.available_devices = self.core.available_devices

    def get_device_info(self) -> dict:
        """사용 가능한 디바이스 정보"""
        info = {}
        for device in self.available_devices:
            info[device] = {
                "full_name": self.core.get_property(device, "FULL_DEVICE_NAME"),
                "supported_properties": self.core.get_property(
                    device, "SUPPORTED_PROPERTIES"
                )
            }
        return info

    def benchmark_model(
        self,
        model_path: str,
        device: str,
        iterations: int = 100
    ) -> dict:
        """특정 디바이스에서 모델 벤치마크"""
        import time

        # 모델 컴파일
        model = self.core.read_model(model_path)
        compiled = self.core.compile_model(model, device)
        infer_request = compiled.create_infer_request()

        # 입력 준비
        input_tensor = np.random.randn(*model.inputs[0].shape).astype(np.float32)

        # 웜업
        for _ in range(10):
            infer_request.infer({0: input_tensor})

        # 벤치마크
        latencies = []
        for _ in range(iterations):
            start = time.perf_counter()
            infer_request.infer({0: input_tensor})
            latencies.append((time.perf_counter() - start) * 1000)

        return {
            "device": device,
            "mean_latency_ms": np.mean(latencies),
            "p99_latency_ms": np.percentile(latencies, 99),
            "throughput_fps": 1000 / np.mean(latencies)
        }

    def auto_select_device(
        self,
        model_path: str,
        target_latency_ms: float
    ) -> str:
        """지연시간 요구사항 기반 자동 디바이스 선택"""
        results = []
        for device in self.available_devices:
            try:
                benchmark = self.benchmark_model(model_path, device)
                if benchmark["p99_latency_ms"] <= target_latency_ms:
                    results.append((device, benchmark))
            except Exception:
                continue

        # 지연시간 충족하면서 가장 저전력 선택
        priority = ["MYRIAD", "GPU", "CPU"]
        for p in priority:
            for device, _ in results:
                if p in device:
                    return device
        return "CPU"

from pycoral.adapters import common, classify
from pycoral.utils.edgetpu import make_interpreter, list_edge_tpus
from PIL import Image
import numpy as np
import time

class CoralTPUInference:
    """Google Coral Edge TPU 추론 엔진"""

    def __init__(self, model_path: str, device_index: int = 0):
        # Edge TPU 디바이스 목록
        self.edge_tpus = list_edge_tpus()
        if not self.edge_tpus:
            raise RuntimeError("Edge TPU가 감지되지 않음")

        # 인터프리터 초기화
        # 모델명에 _edgetpu.tflite 포함 필수
        self.interpreter = make_interpreter(
            model_path,
            device=f":0" if device_index == 0 else f":{device_index}"
        )
        self.interpreter.allocate_tensors()

        # 입출력 정보
        self.input_details = self.interpreter.get_input_details()[0]
        self.output_details = self.interpreter.get_output_details()[0]

    def preprocess(self, image: Image.Image) -> np.ndarray:
        """이미지 전처리"""
        size = common.input_size(self.interpreter)
        image = image.resize(size, Image.LANCZOS)
        return np.asarray(image)

    def infer(self, image: Image.Image) -> tuple:
        """추론 실행"""
        # 전처리
        input_data = self.preprocess(image)

        # 추론
        start = time.perf_counter()
        common.set_input(self.interpreter, input_data)
        self.interpreter.invoke()
        latency = (time.perf_counter() - start) * 1000

        # 결과
        output = common.output_tensor(self.interpreter, 0)
        return output, latency

    def get_device_info(self) -> dict:
        """연결된 Edge TPU 정보"""
        return {
            "count": len(self.edge_tpus),
            "devices": [
                {"type": tpu["type"], "path": tpu["path"]}
                for tpu in self.edge_tpus
            ]
        }

from dataclasses import dataclass
from enum import Enum
from typing import List, Optional

class ApplicationType(Enum):
    VISION_INSPECTION = "vision_inspection"
    MULTI_CAMERA = "multi_camera"
    PREDICTIVE_MAINTENANCE = "predictive_maintenance"
    ROBOT_CONTROL = "robot_control"
    ANOMALY_DETECTION = "anomaly_detection"

@dataclass
class Requirements:
    application: ApplicationType
    latency_ms: float
    power_budget_w: float
    camera_count: int = 1
    model_size_mb: float = 50
    need_industrial_cert: bool = False
    budget_usd: float = 1000

@dataclass
class DeviceRecommendation:
    device: str
    score: float
    reasons: List[str]
    estimated_fps: float
    estimated_power: float

class EdgeDeviceSelector:
    """요구사항 기반 엣지 디바이스 추천"""

    DEVICES = {
        "jetson_orin_nano": {
            "tops": 40, "power": 15, "price": 500,
            "cameras": 4, "industrial": False
        },
        "jetson_agx_orin": {
            "tops": 275, "power": 60, "price": 2000,
            "cameras": 16, "industrial": True
        },
        "coral_tpu": {
            "tops": 4, "power": 2, "price": 60,
            "cameras": 1, "industrial": False
        },
        "intel_ncs2": {
            "tops": 4, "power": 2, "price": 70,
            "cameras": 1, "industrial": False
        },
        "industrial_ipc_gpu": {
            "tops": 50, "power": 100, "price": 1500,
            "cameras": 8, "industrial": True
        }
    }

    def recommend(self, req: Requirements) -> List[DeviceRecommendation]:
        """요구사항 기반 디바이스 추천"""
        recommendations = []

        for name, spec in self.DEVICES.items():
            score, reasons = self._evaluate(name, spec, req)
            if score > 0:
                est_fps = self._estimate_fps(spec, req)
                recommendations.append(DeviceRecommendation(
                    device=name,
                    score=score,
                    reasons=reasons,
                    estimated_fps=est_fps,
                    estimated_power=spec["power"]
                ))

        return sorted(recommendations, key=lambda x: x.score, reverse=True)

    def _evaluate(self, name: str, spec: dict, req: Requirements) -> tuple:
        """디바이스 적합성 평가"""
        score = 100
        reasons = []

        # 필수 조건 확인
        if req.need_industrial_cert and not spec["industrial"]:
            return 0, ["산업 인증 미충족"]

        if spec["cameras"] < req.camera_count:
            return 0, ["카메라 수 미충족"]

        if spec["price"] > req.budget_usd:
            score -= 30
            reasons.append("예산 초과")

        if spec["power"] > req.power_budget_w:
            score -= 20
            reasons.append("전력 예산 초과")
        else:
            reasons.append("전력 요구사항 충족")

        # 성능 여유도
        required_tops = req.model_size_mb / 10  # 대략적 추정
        if spec["tops"] >= required_tops * 2:
            score += 10
            reasons.append("충분한 성능 여유")

        return max(0, score), reasons

    def _estimate_fps(self, spec: dict, req: Requirements) -> float:
        """예상 FPS 계산"""
        # 간단한 추정: TOPS / 모델 크기 기반
        ops_per_frame = req.model_size_mb * 1e6  # 대략적 연산량
        return (spec["tops"] * 1e12 * 0.7) / ops_per_frame