服务插件开发

无界面后台任务的实现方式。

ServicePlugin 用于执行后台任务,不渲染任何界面。它适合文件监听、设备数据采集、定时同步、事件响应等场景。ServicePlugin 与 UiPlugin 一样随时间持续运行,但没有 viewstabseditor 等 UI 专属 API。

基本结构

from pyrite_sdk.core.plugin import ServicePlugin


class FileWatcher(ServicePlugin):
    def on_start(self):
        print("服务已启动")
        self.file.get_root_dir(lambda **result: self._scan(result.get("path")))

    def _scan(self, root):
        print("工作区根目录:", root)

    def on_dispose(self):
        print("服务已停止")


plugin = FileWatcher()
plugin.start()

对应的 plugin.toml

manifest_version = 2
id = "file-watcher"
name = "File Watcher"
version = "1.0.0"
type = "service"
protocol_version = 1
python_version = "3.14"
author = "Your name"
activation_events = ["onStartup"]
permissions = ["file.read", "persistence.read", "persistence.write"]
platforms = ["linux", "macos", "windows", "android"]

后台常驻插件用 activation_events = ["onStartup"],IDE 启动时即自动运行。

可用 API

ServicePlugin 继承 BasePlugin(公共 API),另含:fileboardpersistenceserialdocumentsruntimeenv

事件订阅

服务插件最常用 events 订阅 IDE 的事件主题:

self.events.subscribe(
    "editor.document.changed",
    lambda event: print("document changed:", event),
)

也可以用类型化订阅:

self.documents.on_opened(lambda event: print("opened:", event))
self.documents.on_changed(lambda event: print("changed:", event))
self.documents.on_saved(lambda event: print("saved:", event))
self.runtime.on_program_started(lambda event: print("program started"))
self.runtime.on_program_finished(lambda event: print("program finished"))
self.runtime.on_backend_restarted(lambda event: print("backend restarted"))

订阅支持释放:

sub = self.runtime.on_program_paused(handler)
# ...
sub.dispose()

定时任务

服务插件可在 on_start 中启动后台线程或 threading.Timer。回调中调用 SDK API 时保持线程安全:

import threading
import time

class PeriodicService(ServicePlugin):
    def on_start(self):
        self._stop = False
        self._t = threading.Thread(target=self._loop, daemon=True)
        self._t.start()

    def _loop(self):
        while not self._stop:
            self.board.get_root_dir(
                lambda **result: print("board root:", result.get("path"))
            )
            time.sleep(5)

    def on_dispose(self):
        self._stop = True

使用场景

文件操作

self.file.get_file_list("/", lambda **result: print(result))
self.file.write_file("/tmp/hello.txt", "hi", callback=...)

设备文件

self.board.get_dir_list("/", lambda **result: print(result))
self.board.download_file("main.py", "local.py", callback=...)

串口通信

self.serial.list_ports(lambda **result: print("ports:", result.get("ports", [])))
self.serial.connect("COM3", callback=...)
self.serial.send("AT\r\n", callback=...)
self.serial.send_command("import os; os.uname()", callback=...)
self.serial.read(timeout_ms=1000, callback=...)

运行时检查

self.runtime.sessions(lambda **cb: print("sessions:", cb.get("sessions")))

运行环境

env = self.env.get()        # 立即返回快照(或提供回调)
if self.env.is_desktop_platform:
    print("desktop platform")
print("layout:", self.env.layout_mode, self.env.os)
self.env.on_change(lambda snapshot: print("env changed:", snapshot.get("layoutMode")))

与 Data 插件的区别

  • ServicePlugin:持续运行,可监听事件、执行定时任务,含 file/board/persistence/serial/documents/runtime/env
  • DataPlugin:贡献一次数据后自动退出。

如果任务需要在贡献后保持运行,请用 ServicePlugin

On this page