AskaEth
f5800cc6c3
- headless: TUI skip now works correctly - example_plugin: keyword args compat (plugin_name=, config=, bridge=) - sysmon: psutil made optional (graceful degrade) - tui_service: SysMonWidget optional - All 22 modules import clean, 23 tests pass, integration verified
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
import logging
|
|
from typing import Dict
|
|
try:
|
|
from sdk.plugin_command_decorator import plugin_command, command
|
|
except ImportError:
|
|
def plugin_command(n=None,d=None,p=None):
|
|
def deco(f):
|
|
f._is_plugin_command=True;f._command_name=n or f.__name__
|
|
f._command_description=d or (f.__doc__ or "").strip();return f
|
|
return deco
|
|
command=plugin_command
|
|
|
|
logger=logging.getLogger(__name__)
|
|
|
|
class Plugin:
|
|
def __init__(self, plugin_name=None, config=None, bridge=None, n=None, c=None):
|
|
self.plugin_name=plugin_name or n;self.config=config or c;self.bridge=bridge
|
|
self.network_bridge=None;self.is_running=False
|
|
|
|
async def initialize(self):
|
|
logger.info(f"init: {self.plugin_name}")
|
|
try:
|
|
internet=self.bridge.service_manager.get_service("internet")
|
|
from bridges.plugin_network_bridge import PluginNetworkBridge
|
|
self.network_bridge=PluginNetworkBridge(self.plugin_name,internet,self.bridge)
|
|
await self.network_bridge.register_http_route(
|
|
"/api/example/info",self._api_info,methods=["GET"],require_auth=False)
|
|
except Exception as e:
|
|
logger.warning(f"network skip: {e}")
|
|
self.is_running=True
|
|
|
|
async def _api_info(self,req):
|
|
from aiohttp import web
|
|
return web.json_response({"plugin":self.plugin_name,"status":"running"})
|
|
|
|
@plugin_command(name="echo",description="echo input")
|
|
async def cmd_echo(self,*args):
|
|
return " ".join(args) if args else "echo: no input"
|
|
|
|
@plugin_command(name="plugin_status",description="show plugin status")
|
|
async def cmd_status(self,*args):
|
|
return f"{self.plugin_name} v{self.config.get('version','?')} - running"
|
|
|
|
async def shutdown(self):
|
|
self.is_running=False
|