EvoSpikeNet-Core プラグイン化実装ロードマップ

作成日: 2026年5月15日
対象: Loihi, IBM Quantum, 量子コンピュータをサポートするためのプラグイン統合
推奨アプローチ: Scenario A (最小限統合) → Scenario B (完全統合)

実装進捗 (2026年8月21日)

Phase 1 DevicePlugin 統合: 完了（DevicePlugin 基底 + CPU/GPU/Loihi/Jetson/EdgeTPU/G-QuAT/IBM NorthPole + UniversalIntegrator ブリッジ）
Phase 2 依存関係管理: 完了（optional dependencies + requirements 分割 + DependencyChecker + CI matrix）
Phase 3 ニューロン層完成: 完了（Entangled 互換修正 + HH/Conductance 層 + plugin 追加）
Phase 4 量子層統合: 完了（QuantumNeuronLayer + QAOANeuronLayer + VQENeuronLayer + QuantumAnnealingPlasticity + 統合テスト）
Phase 5 IBM Quantum: 完了（IBMQuantumPlugin + QAOA/VQE ノード + Runtime/Sampler/Estimator + simulator fallback、ただし実機認証・ジョブ実行は環境依存）
Phase 6 最適化パイプライン: 完了（OptimizationPlugin + Quantization/Pruning/Fusion + YAML パイプライン）

📋 実装アーキテクチャの全体図

現在の構造 (非統合):
┌─────────────────────────────────────────┐
│ PluginFactory / PluginSystem            │
│ ├─ NeuronLayerPlugin                    │
│ │  ├─ LIFNeuron                         │
│ │  ├─ IzhikevichNeuron                  │
│ │  └─ EntangledSynchrony [不完全]       │
│ ├─ EncoderPlugin                        │
│ └─ PlasticityPlugin                     │
└─────────────────────────────────────────┘
        ↓
        × (分離)
        ↓
┌─────────────────────────────────────────┐
│ UniversalIntegrator                     │
│ ├─ CPUAdapter                           │
│ ├─ GPUAdapter                           │
│ ├─ LoihiAdapter [硬コード化]             │
│ ├─ JetsonAdapter [硬コード化]           │
│ └─ EdgeTPUAdapter [硬コード化]          │
└─────────────────────────────────────────┘

        ↓ [必須: 統合化]

提案の構造 (統合):
┌─────────────────────────────────────────┐
│ PluginFactory / PluginSystem            │
│ ├─ NeuronLayerPlugin                    │
│ │  ├─ LIFNeuron                         │
│ │  ├─ IzhikevichNeuron                  │
│ │  ├─ EntangledSynchrony [完成版]       │
│ │  ├─ QuantumNeuronLayer                │
│ │  ├─ QAOANeuronLayer (IBM)             │
│ │  └─ VQENeuronLayer (IBM)             │
│ ├─ EncoderPlugin                        │
│ ├─ PlasticityPlugin                     │
│ │  └─ QuantumAnnealingPlasticity        │
│ └─ DevicePlugin [新規]                  │
│    ├─ CPUPlugin                         │
│    ├─ GPUPlugin                         │
│    ├─ LoihiPlugin (with LAVA)           │
│    ├─ JetsonPlugin (with TensorRT)      │
│    ├─ EdgeTPUPlugin                     │
│    ├─ GQuATPlugin                       │
│    └─ IBMQuantumPlugin (with Qiskit)    │
└─────────────────────────────────────────┘

🔴 Phase 1: DevicePlugin 統合 (3-4週間, 8-10人月)

目標: プラットフォームアダプタをプラグインシステムに統合

1.1 PluginType 拡張

ファイル: evospikenet/plugins/init.py

変更内容:

class PluginType(Enum):
    """Types of plugins supported by the system."""
    NEURON = "neuron"              # 既存
    ENCODER = "encoder"            # 既存
    PLASTICITY = "plasticity"      # 既存
    FUNCTIONAL = "functional"      # 既存
    LEARNING = "learning"          # 既存
    MONITORING = "monitoring"      # 既存
    COMMUNICATION = "communication" # 既存

    # ↓ 新規追加 (Phase 1)
    DEVICE = "device"                      # CPU, GPU, Loihi, Jetson等
    OPTIMIZATION = "optimization"         # Quantization, Pruning等
    QUANTUM_LAYER = "quantum_layer"       # 量子ニューロン層
    SYNAPSE = "synapse"                   # シナプス型
    CHANNEL = "channel"                   # 通信チャネル

工数: 1日, テスト: enum値の重複なし確認

1.2 DevicePlugin 基底クラス作成

ファイル: evospikenet/plugins/device_plugin.py (新規)

実装:

from abc import abstractmethod
from typing import Dict, Any, Optional
import torch
from .device_plugin import BasePlugin, PluginMetadata, PluginStatus, PluginType

class DevicePlugin(BasePlugin):
    """Abstract base class for device/platform plugins."""

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        super().__init__(config)
        self.platform_info: Optional[Dict[str, Any]] = None

    @abstractmethod
    def get_capabilities(self) -> Dict[str, Any]:
        """
        Return platform capabilities.

        Returns:
            {
                "platform": str,              # "cpu" | "gpu" | "loihi" | ...
                "max_neurons": int,
                "supported_precisions": List[str],  # ["FP32", "INT8", ...]
                "optimization_support": List[str],  # ["quantization", ...]
                "compute_type": str,          # "cpu" | "gpu" | "neuromorphic" | ...
                "available": bool,            # hardware/dependency available?
            }
        """
        pass

    @abstractmethod
    def optimize_model(self, 
                      model: torch.nn.Module, 
                      config: Optional[Dict[str, Any]] = None
                      ) -> torch.nn.Module:
        """
        Optimize model for target platform.

        Args:
            model: PyTorch model
            config: optimization config (precision, quantization bits, etc.)

        Returns:
            Optimized model
        """
        pass

    @abstractmethod
    def convert_format(self, 
                      model: torch.nn.Module, 
                      output_path: str) -> str:
        """Convert model to platform-specific format."""
        pass

    @abstractmethod
    def deploy_model(self, model_path: str, **kwargs) -> bool:
        """Deploy model to platform hardware."""
        pass

    def validate_deployment(self, model_path: str) -> bool:
        """Optional: Validate deployment success."""
        return True

工数: 2-3日, テスト: 抽象メソッド呼び出し時に NotImplementedError

1.3 CPUPlugin 実装

ファイル: evospikenet/plugins/builtin/device_plugins.py (新規)

実装:

import torch
from ..device_plugin import DevicePlugin
from ... import PluginMetadata, PluginType

class CPUPlugin(DevicePlugin):
    """CPU platform plugin."""

    def get_metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="cpu",
            version="1.0.0",
            plugin_type=PluginType.DEVICE,
            description="CPU backend (multi-threading supported)",
            author="Moonlight Technologies Inc.",
            config_schema={
                "num_threads": int,
                "enable_mkl": bool,
            },
        )

    def initialize(self) -> bool:
        """Initialize CPU plugin."""
        try:
            self.num_threads = self.config.get("num_threads", 4)
            torch.set_num_threads(self.num_threads)
            return True
        except Exception as e:
            logger.error(f"Failed to initialize CPUPlugin: {e}")
            return False

    def activate(self) -> bool:
        """Activate CPU device."""
        self.status = PluginStatus.ACTIVE
        return True

    def deactivate(self) -> bool:
        """Deactivate CPU device."""
        self.status = PluginStatus.UNLOADED
        return True

    def get_capabilities(self) -> Dict[str, Any]:
        return {
            "platform": "cpu",
            "max_neurons": 1000000,  # theoretical limit
            "supported_precisions": ["FP32", "FP16", "INT8"],
            "optimization_support": ["quantization", "pruning"],
            "compute_type": "cpu",
            "available": True,
            "threads": self.num_threads,
        }

    def optimize_model(self, model, config=None):
        """CPU optimization (minimal, mainly for reference)."""
        if config is None:
            return model

        # FP16 conversion if requested
        precision = config.get("precision", "FP32")
        if precision == "FP16":
            return model.half()

        return model

    def convert_format(self, model, output_path):
        """Save model as PyTorch .pt format."""
        torch.save(model.state_dict(), output_path)
        return output_path

    def deploy_model(self, model_path, **kwargs):
        """CPU deployment is always available."""
        logger.info(f"CPU deployment ready for {model_path}")
        return True

工数: 1週間, テスト: get_capabilities, optimize_model, deploy_model の各メソッド

1.4 LoihiPlugin 実装 (with LAVA検出)

ファイル: evospikenet/plugins/builtin/device_plugins.py (続編)

実装:

class LoihiPlugin(DevicePlugin):
    """Intel Loihi neuromorphic chip plugin."""

    # Class-level LAVA availability check
    _lava_available: bool
    _lava_version: Optional[str] = None
    try:
        import lava.lib.dl.slayer as _slayer
        from lava.lib.dl import snn
        from lava import simulator
        _lava_available = True
        try:
            import lava
            _lava_version = lava.__version__
        except:
            pass
    except ImportError:
        _lava_available = False

    def get_metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="loihi",
            version="1.0.0",
            plugin_type=PluginType.DEVICE,
            description="Intel Loihi neuromorphic chip (LAVA-NC framework)",
            author="Moonlight Technologies Inc.",
            dependencies=["lava-nc>=0.5.0"] if self._lava_available else [],
            config_schema={
                "use_hardware": bool,
                "spike_precision": str,  # "int8" | "binary"
                "num_chips": int,
            },
        )

    def initialize(self) -> bool:
        """Initialize Loihi plugin."""
        try:
            self.use_hardware = self.config.get("use_hardware", False)
            self.spike_precision = self.config.get("spike_precision", "int8")
            self.num_chips = self.config.get("num_chips", 1)

            if self.use_hardware and not self._lava_available:
                logger.warning("LAVA not available; will use CPU fallback")
                self.use_hardware = False

            return True
        except Exception as e:
            logger.error(f"Failed to initialize LoihiPlugin: {e}")
            return False

    def activate(self) -> bool:
        if self._lava_available:
            logger.info(f"Activated Loihi plugin (LAVA {self._lava_version})")
        else:
            logger.warning("Loihi plugin activated but LAVA not available (INT8 fallback mode)")
        self.status = PluginStatus.ACTIVE
        return True

    def deactivate(self) -> bool:
        self.status = PluginStatus.UNLOADED
        return True

    def get_capabilities(self) -> Dict[str, Any]:
        return {
            "platform": "loihi",
            "max_neurons": 131072 * self.num_chips,  # 131K per chip
            "supported_precisions": ["INT8", "binary"],
            "optimization_support": ["spike_quantization", "on_chip_learning", "stdp"],
            "compute_type": "neuromorphic",
            "available": self._lava_available,
            "lava_version": self._lava_version,
            "hardware_available": self.use_hardware,
            "note": "Requires lava-nc >= 0.5.0 for full support",
        }

    def optimize_model(self, model, config=None):
        """Optimize model for Loihi execution."""
        if not self._lava_available:
            logger.warning("LAVA not available; using INT8 dynamic quantization")
            return torch.quantization.quantize_dynamic(
                model, {torch.nn.Linear}, dtype=torch.qint8
            )

        # LAVA available: mark model for spike quantization
        # (actual conversion in deploy_model)
        model._loihi_optimized = True
        model._spike_precision = self.spike_precision
        logger.info("Model marked for Loihi spike quantization")

        return model

    def convert_format(self, model, output_path):
        """Convert to NxSDK-compatible JSON format."""
        import json

        net_desc = {
            "platform": "loihi",
            "lava_version": self._lava_version,
            "layers": [
                {
                    "name": name,
                    "type": type(m).__name__,
                    "shape": tuple(p.shape) for p in m.parameters(),
                }
                for name, m in model.named_modules()
                if not list(m.children())
            ],
        }

        with open(output_path, "w") as f:
            json.dump(net_desc, f, indent=2, default=str)

        logger.info(f"Model exported to {output_path} (NxSDK format)")
        return output_path

    def deploy_model(self, model_path, **kwargs):
        """Deploy model to Loihi via LAVA runtime."""
        if not self._lava_available:
            logger.error("LAVA not available; cannot deploy to hardware")
            return False

        try:
            logger.info(f"Deploying {model_path} to Loihi via LAVA")
            # Actual LAVA deployment logic here
            # (requires loading NxSDK JSON and executing on hardware)
            return True
        except Exception as e:
            logger.error(f"Loihi deployment failed: {e}")
            return False

工数: 1週間, テスト: LAVA availability check, INT8 fallback, 形式変換

1.5 JetsonPlugin と EdgeTPUPlugin (同様の実装)

ファイル: evospikenet/plugins/builtin/device_plugins.py (続編)

JetsonPlugin:

class JetsonPlugin(DevicePlugin):
    """NVIDIA Jetson edge GPU plugin."""

    _tensorrt_available: bool
    try:
        import tensorrt
        _tensorrt_available = True
    except ImportError:
        _tensorrt_available = False

    # ... 同様の実装 (TensorRT使用)

EdgeTPUPlugin:

class EdgeTPUPlugin(DevicePlugin):
    """Google Coral Edge TPU plugin."""

    _edgetpu_available: bool
    try:
        from edgetpu.basic.basic_edge_tpu_interpreter import BasicEdgeTpuInterpreter
        _edgetpu_available = True
    except ImportError:
        _edgetpu_available = False

    # ... 同様の実装 (edgetpu_compiler使用)

工数: 各1週間 (合計 2週間)

1.6 DeviceFactory 統合

ファイル: evospikenet/plugin_factory.py (修正)

追加内容:

class DeviceFactory:
    """Factory for creating device plugins."""

    _device_plugins: Dict[str, Type[DevicePlugin]] = {}

    @classmethod
    def register_device_plugin(cls, plugin_type: str, plugin_class: Type[DevicePlugin]):
        """Register a device plugin."""
        cls._device_plugins[plugin_type] = plugin_class

    @classmethod
    def create_device_plugin(cls, 
                            plugin_type: str,
                            config: Optional[Dict[str, Any]] = None) -> Optional[DevicePlugin]:
        """Create and initialize a device plugin."""
        if plugin_type not in cls._device_plugins:
            logger.warning(f"Device plugin '{plugin_type}' not found")
            return None

        plugin_class = cls._device_plugins[plugin_type]
        plugin = plugin_class(config)

        if not plugin.initialize():
            logger.error(f"Failed to initialize {plugin_type} plugin")
            return None

        return plugin

    @classmethod
    def get_available_devices(cls) -> List[str]:
        """Get list of available device plugins."""
        return list(cls._device_plugins.keys())

# Automatic registration of built-in device plugins
def _register_builtin_device_plugins():
    """Register built-in device plugins."""
    from evospikenet.plugins.builtin.device_plugins import (
        CPUPlugin, GPUPlugin, LoihiPlugin, JetsonPlugin, EdgeTPUPlugin
    )

    DeviceFactory.register_device_plugin("cpu", CPUPlugin)
    DeviceFactory.register_device_plugin("gpu", GPUPlugin)
    DeviceFactory.register_device_plugin("loihi", LoihiPlugin)
    DeviceFactory.register_device_plugin("jetson", JetsonPlugin)
    DeviceFactory.register_device_plugin("edge_tpu", EdgeTPUPlugin)

# Call on module import
_register_builtin_device_plugins()

工数: 3-4日, テスト: device plugin registration/creation

1.7 後方互換性レイヤー

ファイル: evospikenet/universal_integration.py (修正)

修正内容:

# 従来コードとの互換性を保つため、UniversalIntegrator を legacy mode で動作させる

class UniversalIntegrator:
    """Legacy compatibility layer for existing code."""

    def __init__(self):
        self.device_factory = DeviceFactory()
        self.adapters = self._create_legacy_adapters()

    def _create_legacy_adapters(self):
        """Create adapter objects from device plugins."""
        adapters = {}
        for device_type in ["cpu", "gpu", "loihi", "jetson", "edge_tpu"]:
            plugin = self.device_factory.create_device_plugin(device_type)
            if plugin:
                adapters[PlatformType(device_type)] = plugin
        return adapters

    def detect_platform(self) -> PlatformInfo:
        """Detect available platform (unchanged)."""
        # 従来通りの実装
        ...

    @property
    def adapters(self) -> Dict:
        """Access adapters (deprecated, use DeviceFactory instead)."""
        logger.warning("UniversalIntegrator.adapters is deprecated; use DeviceFactory")
        return self._adapters

工数: 3-4日, テスト: 既存コードの動作確認 (回帰テスト)

1.8 設定ファイル拡張

ファイル: config/device_plugins.yaml (新規)

内容:

# Device Plugin Configuration

plugins:
  cpu:
    enabled: true
    config:
      num_threads: 4
      enable_mkl: false

  gpu:
    enabled: true
    config:
      device_id: 0
      memory_fraction: 0.8

  loihi:
    enabled: false  # Requires LAVA-NC
    config:
      use_hardware: false
      spike_precision: "int8"
      num_chips: 1
    required_packages:
      - "lava-nc>=0.5.0"

  jetson:
    enabled: false  # Requires TensorRT
    config:
      precision: "FP16"
    required_packages:
      - "tensorrt>=8.0"

  edge_tpu:
    enabled: false
    config:
      precision: "INT8"

# Device selection priority (first available is used)
device_priority:
  - "gpu"
  - "loihi"
  - "jetson"
  - "cpu"

工数: 1日, テスト: YAML パース、バリデーション

1.9 統合テスト

ファイル: tests/unit/test_device_plugins.py (新規)

テストケース: - [ ] CPUPlugin initialization - [ ] CPUPlugin optimize_model (FP16 conversion) - [ ] CPUPlugin deploy_model (always succeeds) - [ ] LoihiPlugin with LAVA available - [ ] LoihiPlugin without LAVA (INT8 fallback) - [ ] JetsonPlugin with TensorRT available - [ ] JetsonPlugin without TensorRT (error handling) - [ ] EdgeTPUPlugin edge cases - [ ] DeviceFactory registration/creation - [ ] Backwards compatibility (UniversalIntegrator) - [ ] YAML config loading + validation - [ ] Device priority selection

工数: 1週間, 目標カバレッジ: ≥ 85%

🟡 Phase 2: 依存関係管理 (1-2週間, 2-3人月)

2.1 pyproject.toml オプション依存追加

ファイル: EvoSpikeNet-Core/pyproject.toml (修正)

追加内容:

[project.optional-dependencies]
# 既存
zenoh = ["eclipse-zenoh>=1.0.0,<1.8"]
test = [...]
docs = [...]
jupyter = [...]
dev = [...]

# 新規: ハードウェアサポート
loihi = [
    "lava-nc>=0.5.0,<0.6",
]
jetson = [
    "tensorrt>=8.0,<9.0",
    "torch2trt>=0.5.0",
]
edge_tpu = [
    "edgetpu>=15.0,<16.0",
]
quantum = [
    "qiskit>=1.0,<2.0",
    "qiskit-machine-learning>=0.7.0,<0.8",
    "qiskit-aer>=0.13.0",
]

# 新規: Meta groups
hardware = ["loihi", "jetson", "edge_tpu"]
all_hardware = ["loihi", "jetson", "edge_tpu", "quantum"]

工数: 2-3日

2.2 DependencyChecker クラス

ファイル: evospikenet/plugins/setup_awareness.py (新規)

実装:

from typing import Tuple, Optional, Dict, List
import logging

logger = logging.getLogger(__name__)

class DependencyChecker:
    """Check availability of optional dependencies."""

    _cache: Dict[str, Tuple[bool, Optional[str]]] = {}

    @staticmethod
    def check_package(package_name: str, min_version: Optional[str] = None) -> Tuple[bool, Optional[str]]:
        """
        Check if a package is installed and optionally verify minimum version.

        Args:
            package_name: Package name (e.g., "lava-nc", "qiskit")
            min_version: Minimum required version (e.g., "0.5.0")

        Returns:
            (is_available, version_string_or_error)
        """
        if package_name in DependencyChecker._cache:
            return DependencyChecker._cache[package_name]

        try:
            module = __import__(package_name.replace("-", "_"))
            version = getattr(module, "__version__", "unknown")

            if min_version and version != "unknown":
                from packaging import version as pkg_version
                if pkg_version.parse(version) < pkg_version.parse(min_version):
                    result = (False, f"Version {version} < {min_version}")
                    DependencyChecker._cache[package_name] = result
                    return result

            result = (True, version)
            DependencyChecker._cache[package_name] = result
            return result

        except ImportError as e:
            result = (False, str(e))
            DependencyChecker._cache[package_name] = result
            return result

    @staticmethod
    def check_lava() -> Tuple[bool, Optional[str]]:
        return DependencyChecker.check_package("lava", "0.5.0")

    @staticmethod
    def check_qiskit() -> Tuple[bool, Optional[str]]:
        return DependencyChecker.check_package("qiskit", "1.0.0")

    @staticmethod
    def check_tensorrt() -> Tuple[bool, Optional[str]]:
        return DependencyChecker.check_package("tensorrt", "8.0.0")

    @staticmethod
    def check_edgetpu() -> Tuple[bool, Optional[str]]:
        return DependencyChecker.check_package("edgetpu")

    @staticmethod
    def get_missing_optional_deps(hardware_targets: List[str]) -> List[str]:
        """
        Get list of missing dependencies for target hardware.

        Args:
            hardware_targets: e.g., ["loihi", "jetson", "quantum"]

        Returns:
            List of missing package names
        """
        checkers = {
            "loihi": ("lava", "0.5.0"),
            "jetson": ("tensorrt", "8.0.0"),
            "edge_tpu": ("edgetpu", "15.0"),
            "quantum": ("qiskit", "1.0.0"),
        }

        missing = []
        for target in hardware_targets:
            if target in checkers:
                pkg_name, min_version = checkers[target]
                available, _ = DependencyChecker.check_package(pkg_name, min_version)
                if not available:
                    missing.append(f"{pkg_name} >= {min_version}")

        return missing

工数: 2-3日

2.3 CI/CD での依存関係検証

ファイル: .github/workflows/device-plugin-ci.yml (新規)

内容:

name: Device Plugin CI

on:
  push:
    paths:
      - 'evospikenet/plugins/**'
      - 'pyproject.toml'
  pull_request:
    paths:
      - 'evospikenet/plugins/**'

jobs:
  test-device-plugins:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]
        hardware: ["cpu", "gpu", "loihi", "jetson", "quantum"]

    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install base dependencies
        run: |
          pip install -e .

      - name: Install optional hardware dependencies
        run: |
          case "${{ matrix.hardware }}" in
            loihi)
              pip install -e ".[loihi]"
              ;;
            jetson)
              pip install -e ".[jetson]"
              ;;
            quantum)
              pip install -e ".[quantum]"
              ;;
          esac

      - name: Run device plugin tests
        run: |
          pytest tests/unit/test_device_plugins.py -v --hardware=${{ matrix.hardware }}

工数: 1-2日

🟠 Phase 3: ニューロン層完成 (2-3週間, 4-5人月)

(Previous content continues with Phase 3, 4, 5, 6...)

📊 全体工数サマリー

Phase	内容	工数 (人月)	期間	依存
1	DevicePlugin統合	8-10	3-4週	-
2	依存関係管理	2-3	1-2週	Phase 1
3	ニューロン層完成	4-5	2-3週	Phase 2
4	量子層統合	6-8	3-4週	Phase 3 (並列可)
5	IBM Quantum	8-10	4-6週	Phase 3, 2
6	最適化パイプライン	4-5	2-3週	Phase 4
合計	全統合	32-41	16-22週	並列実行で3ヶ月短縮可能

🎯 推奨実装シナリオ

Scenario A: MVP (5-6週間, 10-12人月)

Phase 1: DevicePlugin フレームワーク完成
Phase 2: 依存関係管理
成果: 基本的なハードウェア abstraction が使用可能に

Scenario B: フルサポート (16-22週間, 32-41人月)

Phase 1-6 すべて実装
並列開発により工期短縮

推奨: Scenario A → Scenario B への段階的実装

EvoSpikeNet-Core プラグイン化 実装ロードマップ