"""
Plugin SDK - Client and server helpers for the AIPC plugin system.
DEPRECATED: the platform does not ship the ``/run/aipc/plugins`` discovery
mechanism. This module is kept importable for compatibility, emits a
``DeprecationWarning`` on use, and will be removed in v0.8.0.
For plugin providers:
server = PluginServer("my-plugin")
server.start(my_grpc_servicer)
For plugin consumers:
discovery = PluginDiscovery()
endpoint = discovery.get("rtsp-server")
channel = endpoint.connect()
"""
from __future__ import annotations
import json
import os
import threading
import time
import warnings
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
import grpc
DISCOVERY_DIR = "/run/aipc/plugins"
DISCOVERY_FILE = "discovery.json"
_DEPRECATION_MESSAGE = (
"neoruntime_ipc_sdk.plugin is deprecated and will be removed in v0.8.0: "
"the platform does not ship the /run/aipc/plugins discovery mechanism"
)
def _warn_deprecated() -> None:
warnings.warn(_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=3)
[docs]
@dataclass
class PluginEndpoint:
"""Resolved endpoint for a discovered plugin capability."""
app_id: str
capability_id: str
version: str
transport: str
socket_path: str | None = None
grpc_service: str | None = None
event_publish: list[str] = field(default_factory=list)
event_subscribe: list[str] = field(default_factory=list)
state: str = "unknown"
[docs]
def connect(self, **kwargs) -> grpc.Channel:
"""Create a gRPC channel to this plugin."""
if not self.socket_path:
raise RuntimeError(
f"Plugin {self.app_id} capability {self.capability_id} has no gRPC endpoint"
)
return grpc.insecure_channel(f"unix://{self.socket_path}", **kwargs)
@property
def is_available(self) -> bool:
return self.state == "running"
[docs]
class PluginDiscovery:
"""Client-side plugin discovery using discovery.json + optional file watcher."""
[docs]
def __init__(self, discovery_dir: str = DISCOVERY_DIR):
_warn_deprecated()
self._dir = Path(discovery_dir)
self._file = self._dir / DISCOVERY_FILE
self._data: dict = {}
self._lock = threading.Lock()
self._watchers: list[Callable] = []
self._watch_thread: threading.Thread | None = None
self._running = False
self.reload()
[docs]
def reload(self):
"""Reload discovery data from file."""
with self._lock:
try:
if self._file.exists():
self._data = json.loads(self._file.read_text())
else:
self._data = {"plugins": {}}
except (json.JSONDecodeError, OSError):
self._data = {"plugins": {}}
[docs]
def get(self, capability_id: str) -> PluginEndpoint | None:
"""Find the first running plugin providing a capability."""
with self._lock:
plugins = self._data.get("plugins", {})
for _app_id, entry in plugins.items():
for cap in entry.get("capabilities", []):
if cap.get("id") == capability_id:
ep = PluginEndpoint(
app_id=entry["app_id"],
capability_id=cap["id"],
version=cap.get("version", ""),
transport=cap.get("transport", ""),
state=entry.get("state", "unknown"),
)
grpc_info = cap.get("grpc")
if grpc_info:
ep.socket_path = grpc_info.get("socket_path")
ep.grpc_service = grpc_info.get("service")
event_info = cap.get("event")
if event_info:
ep.event_publish = event_info.get("publish", [])
ep.event_subscribe = event_info.get("subscribe", [])
return ep
return None
[docs]
def require(self, capability_id: str, timeout: float = 30.0) -> PluginEndpoint:
"""Wait for a capability to become available. Raises TimeoutError."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
self.reload()
ep = self.get(capability_id)
if ep and ep.is_available:
return ep
time.sleep(1.0)
raise TimeoutError(f"Plugin capability {capability_id!r} not available within {timeout}s")
[docs]
def list_plugins(self) -> dict[str, dict]:
"""Return all known plugins."""
with self._lock:
return dict(self._data.get("plugins", {}))
[docs]
def list_capabilities(self) -> list[str]:
"""Return all known capability IDs."""
result = []
with self._lock:
for entry in self._data.get("plugins", {}).values():
for cap in entry.get("capabilities", []):
cap_id = cap.get("id")
if cap_id and cap_id not in result:
result.append(cap_id)
return result
[docs]
def watch(self, callback: Callable[[], None]):
"""Register a callback invoked when discovery.json changes."""
self._watchers.append(callback)
if not self._running:
self._start_watch()
def _start_watch(self):
self._running = True
self._watch_thread = threading.Thread(target=self._watch_loop, daemon=True)
self._watch_thread.start()
def _watch_loop(self):
last_mtime = 0.0
while self._running:
try:
if self._file.exists():
mtime = self._file.stat().st_mtime
if mtime != last_mtime:
last_mtime = mtime
self.reload()
for cb in self._watchers:
try:
cb()
except Exception:
pass
except OSError:
pass
time.sleep(2.0)
[docs]
def close(self):
self._running = False
[docs]
class PluginServer:
"""Helper for plugin containers to set up a gRPC server on the standard socket path."""
[docs]
def __init__(self, plugin_id: str, socket_dir: str = DISCOVERY_DIR):
_warn_deprecated()
self.plugin_id = plugin_id
self.socket_path = os.path.join(socket_dir, f"{plugin_id}.sock")
self._server: grpc.Server | None = None
[docs]
def create_server(self, max_workers: int = 4, **kwargs) -> grpc.Server:
"""Create a gRPC server bound to the plugin socket."""
from concurrent import futures
self._server = grpc.server(futures.ThreadPoolExecutor(max_workers=max_workers), **kwargs)
self._server.add_insecure_port(f"unix://{self.socket_path}")
return self._server
[docs]
def start(self):
"""Start the gRPC server."""
if self._server is None:
raise RuntimeError("Call create_server() first")
self._server.start()
[docs]
def wait(self):
"""Block until server terminates."""
if self._server:
self._server.wait_for_termination()
[docs]
def stop(self, grace: float = 5.0):
"""Gracefully stop the server."""
if self._server:
self._server.stop(grace)
# Clean up socket file
try:
os.unlink(self.socket_path)
except OSError:
pass