// build.gradle
// implementation 'com.google.mlkit:image-labeling:17.0.8'
 
import com.google.mlkit.vision.common.InputImage
import com.google.mlkit.vision.label.ImageLabeling
import com.google.mlkit.vision.label.defaults.ImageLabelerOptions
 
fun labelImage(bitmap: Bitmap) {
    val image = InputImage.fromBitmap(bitmap, 0)
    val options = ImageLabelerOptions.Builder()
        .setConfidenceThreshold(0.7f) // 70%以上の信頼度のみ
        .build()
    val labeler = ImageLabeling.getClient(options)
 
    labeler.process(image)
        .addOnSuccessListener { labels ->
            for (label in labels) {
                println("${label.text}: ${label.confidence}")
                // 例: "Cat: 0.95", "Animal: 0.92"
            }
        }
        .addOnFailureListener { e ->
            println("エラー: ${e.message}")
        }
}

import CoreML
import Vision
 
func classifyImage(_ image: CGImage) {
    // MobileNetV3 モデルをロード
    guard let model = try? VNCoreMLModel(
        for: MobileNetV3(configuration: .init()).model
    ) else { return }
 
    let request = VNCoreMLRequest(model: model) { request, error in
        guard let results = request.results as? [VNClassificationObservation] else { return }
 
        // 上位3件を表示
        for result in results.prefix(3) {
            print("\(result.identifier): \(result.confidence * 100)%")
        }
    }
 
    let handler = VNImageRequestHandler(cgImage: image)
    try? handler.perform([request])
}

import tensorflow as tf
 
# モデルの量子化（FP32 → INT8）でNPU最適化
converter = tf.lite.TFLiteConverter.from_saved_model("model_dir")
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.int8]  # NPU向け量子化
 
# 代表的なデータセットで量子化キャリブレーション
def representative_dataset():
    for data in calibration_data:
        yield [data.astype("float32")]
 
converter.representative_dataset = representative_dataset
tflite_model = converter.convert()
 
# 保存
with open("model_quantized.tflite", "wb") as f:
    f.write(tflite_model)
 
print(f"モデルサイズ: {len(tflite_model) / 1024 / 1024:.1f} MB")

import org.tensorflow.lite.Interpreter
import org.tensorflow.lite.gpu.CompatibilityList
import org.tensorflow.lite.nnapi.NnApiDelegate
 
fun benchmarkNPU(modelPath: String) {
    // NNAPI デリゲート（NPU優先）
    val nnApiDelegate = NnApiDelegate(
        NnApiDelegate.Options().apply {
            setAllowFp16(true)             // FP16演算を許可
            setExecutionPreference(         // NPU優先実行
                NnApiDelegate.Options.EXECUTION_PREFERENCE_SUSTAINED_SPEED
            )
        }
    )
 
    val options = Interpreter.Options().apply {
        addDelegate(nnApiDelegate)
        setNumThreads(4)
    }
 
    val interpreter = Interpreter(loadModelFile(modelPath), options)
 
    // ウォームアップ + 計測
    val input = FloatArray(224 * 224 * 3)
    val output = Array(1) { FloatArray(1000) }
 
    repeat(10) { interpreter.run(input, output) } // ウォームアップ
 
    val start = System.nanoTime()
    repeat(100) { interpreter.run(input, output) }
    val elapsed = (System.nanoTime() - start) / 1_000_000.0 / 100
 
    println("平均推論時間: ${elapsed}ms")  // NPU: ~3ms, CPU: ~25ms
}

// Android AICore API（Pixel 8 Pro 以降）
import com.google.android.gms.aicore.GenerativeModel
import com.google.android.gms.aicore.GenerateContentRequest
 
suspend fun summarizeOnDevice(text: String): String {
    val model = GenerativeModel("gemini-nano")
 
    val request = GenerateContentRequest.newBuilder()
        .addText("以下のテキストを3行で要約してください:\n$text")
        .build()
 
    val response = model.generateContent(request)
    return response.text ?: "要約を生成できませんでした"
}
 
// 使用例
val article = "長い記事のテキスト..."
val summary = summarizeOnDevice(article)
println(summary)
// → "1. AIスマートフォンはNPUを搭載..."
// → "2. オンデバイス処理でプライバシーを保護..."
// → "3. クラウド不要で低遅延応答を実現..."

# モバイル向けWhisperモデルの最適化パイプライン
import torch
import whisper
import coremltools as ct
 
def convert_whisper_for_mobile():
    """Whisper tiny/base をモバイル向けに変換"""
    # Whisper tiny モデル（39M パラメータ、~150MB）
    model = whisper.load_model("tiny")
    model.eval()
 
    # エンコーダ部分をCore ML用にエクスポート
    # 入力: メルスペクトログラム (1, 80, 3000)
    dummy_input = torch.randn(1, 80, 3000)
 
    # TorchScript に変換
    traced_encoder = torch.jit.trace(model.encoder, dummy_input)
 
    # Core ML に変換（Neural Engine で高速実行）
    mlmodel = ct.convert(
        traced_encoder,
        inputs=[ct.TensorType(shape=(1, 80, 3000), name="mel_input")],
        compute_precision=ct.precision.FLOAT16,  # FP16で高速化
        compute_units=ct.ComputeUnit.ALL,  # CPU + GPU + Neural Engine
    )
    mlmodel.save("WhisperEncoder.mlpackage")
    print("Core ML モデル保存完了")
 
    # モデルサイズ比較
    # FP32 (元): ~150MB
    # FP16 (Neural Engine最適化): ~75MB
    # INT8 (最大圧縮): ~40MB
 
convert_whisper_for_mobile()

import org.tensorflow.lite.Interpreter
import org.tensorflow.lite.nnapi.NnApiDelegate
import org.tensorflow.lite.gpu.GpuDelegate
import java.io.FileInputStream
import java.nio.MappedByteBuffer
import java.nio.channels.FileChannel
 
class AIModelManager(private val context: Context) {
    private var interpreter: Interpreter? = null
    private var currentDelegate: String = "cpu"
 
    /**
     * 利用可能なアクセラレータを検出して最適なデリゲートを選択
     */
    fun loadModel(modelPath: String): Boolean {
        val modelBuffer = loadModelFile(modelPath)
        val options = Interpreter.Options()
 
        // NPU (NNAPI) を最優先で試行
        try {
            val nnApiOptions = NnApiDelegate.Options().apply {
                setAllowFp16(true)
                setExecutionPreference(
                    NnApiDelegate.Options.EXECUTION_PREFERENCE_SUSTAINED_SPEED
                )
                setUseNnapiCpu(false) // CPU フォールバックを無効化
            }
            val nnApiDelegate = NnApiDelegate(nnApiOptions)
            options.addDelegate(nnApiDelegate)
            interpreter = Interpreter(modelBuffer, options)
            currentDelegate = "npu"
            println("NPU (NNAPI) デリゲートで実行")
            return true
        } catch (e: Exception) {
            println("NPU利用不可: ${e.message}")
        }
 
        // GPU を次に試行
        try {
            val gpuDelegate = GpuDelegate()
            options.addDelegate(gpuDelegate)
            interpreter = Interpreter(modelBuffer, options)
            currentDelegate = "gpu"
            println("GPU デリゲートで実行")
            return true
        } catch (e: Exception) {
            println("GPU利用不可: ${e.message}")
        }
 
        // CPU フォールバック
        options.setNumThreads(4)
        interpreter = Interpreter(modelBuffer, options)
        currentDelegate = "cpu"
        println("CPU フォールバックで実行")
        return true
    }
 
    /**
     * 推論実行と性能計測
     */
    fun runInference(input: FloatArray): Pair<FloatArray, Long> {
        val output = Array(1) { FloatArray(1000) }
        val startTime = System.nanoTime()
        interpreter?.run(arrayOf(input), output)
        val elapsed = (System.nanoTime() - startTime) / 1_000_000
 
        println("推論完了: ${elapsed}ms (デリゲート: $currentDelegate)")
        return Pair(output[0], elapsed)
    }
 
    private fun loadModelFile(path: String): MappedByteBuffer {
        val assetFd = context.assets.openFd(path)
        val inputStream = FileInputStream(assetFd.fileDescriptor)
        val fileChannel = inputStream.channel
        return fileChannel.map(
            FileChannel.MapMode.READ_ONLY,
            assetFd.startOffset,
            assetFd.declaredLength
        )
    }
}

import MLX
import MLXLLM
import Foundation
 
/// Apple Silicon (Neural Engine + GPU) でローカルLLM推論
class OnDeviceLLMEngine {
    private var model: LLMModel?
    private var tokenizer: Tokenizer?
 
    func loadModel() async throws {
        // Phi-3 Mini (3.8B, 4bit量子化 ≈ 2.3GB)
        let configuration = ModelConfiguration(
            id: "mlx-community/Phi-3-mini-4k-instruct-4bit"
        )
 
        // モデルをダウンロード & ロード
        let (model, tokenizer) = try await LLM.load(configuration: configuration)
        self.model = model
        self.tokenizer = tokenizer
 
        print("モデルロード完了: Phi-3 Mini 4bit")
        print("メモリ使用量: \(MLX.GPU.activeMemory / 1_000_000)MB")
    }
 
    func generate(prompt: String, maxTokens: Int = 256) async -> String {
        guard let model = model, let tokenizer = tokenizer else {
            return "モデル未ロード"
        }
 
        let input = tokenizer.encode(prompt)
        var output: [Int] = []
        var generationTime: Double = 0
 
        let startTime = CFAbsoluteTimeGetCurrent()
 
        // トークン単位で逐次生成
        for token in try! model.generate(input: MLXArray(input), parameters: .init(
            temperature: 0.7,
            topP: 0.9,
            repetitionPenalty: 1.1
        )) {
            output.append(token)
            if output.count >= maxTokens { break }
        }
 
        generationTime = CFAbsoluteTimeGetCurrent() - startTime
        let tokensPerSecond = Double(output.count) / generationTime
 
        let result = tokenizer.decode(output)
        print("生成速度: \(String(format: "%.1f", tokensPerSecond)) tokens/sec")
        print("生成トークン数: \(output.count)")
 
        return result
    }
}
 
// 使用例
let engine = OnDeviceLLMEngine()
try await engine.loadModel()
let response = await engine.generate(
    prompt: "AIスマートフォンのNPUについて簡潔に説明してください。"
)
print(response)
// iPhone 15 Pro: ~15 tokens/sec (Neural Engine + GPU)

# Android: AI処理のバッテリー最適化パターン
 
class BatteryAwareAIManager:
    """
    バッテリー残量に応じてAI処理レベルを調整
    """
    def __init__(self):
        self.ai_level = "full"  # full / balanced / minimal
 
    def adjust_ai_level(self, battery_percentage: int, is_charging: bool):
        """バッテリー残量に基づくAI処理レベルの動的調整"""
        if is_charging:
            self.ai_level = "full"
        elif battery_percentage > 50:
            self.ai_level = "full"
            # 全AIモデルをNPUで実行
            # リアルタイムカメラAI有効
        elif battery_percentage > 20:
            self.ai_level = "balanced"
            # バックグラウンドAIの頻度を1/2に
            # カメラAIはユーザー操作時のみ
            # 推論バッチサイズを削減
        else:
            self.ai_level = "minimal"
            # バックグラウンドAIを停止
            # 必須AI（転倒検出等）のみ維持
            # モデルをより軽量なものに切り替え
 
        return self.ai_level
 
    def get_model_config(self, task: str) -> dict:
        """現在のAIレベルに応じたモデル設定を返す"""
        configs = {
            "image_classification": {
                "full":     {"model": "efficientnet_b4", "resolution": 380},
                "balanced": {"model": "mobilenet_v3_large", "resolution": 224},
                "minimal":  {"model": "mobilenet_v3_small", "resolution": 224},
            },
            "text_generation": {
                "full":     {"model": "gemini-nano-3.25b", "max_tokens": 512},
                "balanced": {"model": "gemini-nano-1.8b", "max_tokens": 256},
                "minimal":  {"model": "none", "max_tokens": 0},  # クラウドフォールバック
            }
        }
        return configs.get(task, {}).get(self.ai_level, {})

class ProgressiveModelLoader:
    """
    段階的モデルロードパターン
    ユーザー体験を損なわずに高品質AIを提供
    """
 
    def __init__(self):
        self.models = {}
        self.loading_priority = [
            ("tiny_classifier", "model_tiny.tflite", 0),      # 即座にロード
            ("standard_classifier", "model_std.tflite", 3),    # 3秒後にロード
            ("high_quality_model", "model_hq.tflite", 10),     # 10秒後にロード
        ]
 
    async def progressive_load(self):
        """アプリ起動後、段階的にモデルをロード"""
        for name, path, delay in self.loading_priority:
            await asyncio.sleep(delay)
            self.models[name] = load_tflite_model(path)
            print(f"ロード完了: {name}")
 
    def get_best_available_model(self, task: str):
        """現在ロード済みの最高品質モデルを返す"""
        for name in reversed(self.loading_priority):
            if name[0] in self.models:
                return self.models[name[0]]
        return None  # まだ何もロードされていない

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]