5 分钟打造你的第一个 Tune 插件

Tune Server 从设计之初就是开放的。自 0.7.90 版本起，一套 Python 插件系统让你无需修改服务器源码即可扩展它：新增音频输出、集成第三方服务、创建自定义 API 端点，或者对播放事件做出响应。在本文中，我们将用几分钟时间创建一个可运行的插件。

前置条件

机器上已安装 Python 3.11+
pip（随 Python 一同提供）
一台可正常运行的 Tune 服务器（0.7.90 或更高版本）
一个文本编辑器（VS Code、PyCharm、vim……）

检查你的 Python 版本：

python3 --version
# Python 3.11.x 或更高

概念

Tune 的插件系统基于三个要素：

1. `TunePlugin` Protocol

每个插件都是一个实现了 TunePlugin protocol 的 Python 类。该 protocol 定义了两个属性（name 和 version）以及两个生命周期方法（setup 和 teardown）。

from typing import Protocol, runtime_checkable

@runtime_checkable
class TunePlugin(Protocol):
    name: str       # identifiant unique, ex: "hello"
    version: str    # version semver, ex: "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """Appelé au démarrage du serveur."""
        ...

    async def teardown(self) -> None:
        """Appelé à l'arrêt du serveur."""
        ...

2. `PluginContext`

在 setup 期间，你的插件会收到一个 PluginContext 对象，它以安全的方式提供对 Tune 内部服务的访问：

ctx.event_bus — 订阅播放事件（曲目开始、区域创建等）
ctx.zone_manager — 注册新的音频输出类型
ctx.api_app — 挂载自定义的 FastAPI 路由
ctx.settings — 读取服务器配置（只读）
ctx.db — 访问数据库

3. 生命周期

启动时，Tune 通过 Python 的 entry points 自动发现已安装的插件，然后按顺序初始化它们：

发现 — Tune 扫描 entry_points(group="tune_server.plugins")
setup(ctx) — 插件注册其路由、钩子和输出类型
正常运行 — 服务器运行中，插件处于活动状态
teardown() — 关闭时，插件释放其资源

不支持热重载：任何修改之后都需要重启服务器。这是为保证稳定性而做出的刻意选择（避免播放过程中出现状态损坏）。

快速上手，一步步来

我们来创建一个 "Hello" 插件，它会添加一个 API 端点 /api/v1/hello，并在每次切换曲目时记录一条日志。

第 1 步 — 创建项目结构

mkdir tune-hello-plugin
cd tune-hello-plugin
mkdir -p src/tune_hello

最终结构如下：

tune-hello-plugin/
├── pyproject.toml
└── src/
    └── tune_hello/
        ├── __init__.py
        └── plugin.py

第 2 步 — 编写 `pyproject.toml`

pyproject.toml 文件向 Python 和 Tune 声明你的插件。[project.entry-points] 一节是关键：它让 Tune 能在启动时发现你的插件。

[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"

[project]
name = "tune-hello-plugin"
version = "0.1.0"
description = "Mon premier plugin Tune"
requires-python = ">=3.11"
dependencies = []

[project.entry-points."tune_server.plugins"]
hello = "tune_hello:HelloPlugin"

[tool.setuptools.packages.find]
where = ["src"]

hello = "tune_hello:HelloPlugin" 这一行的含义是：“注册名为 'hello' 的插件，其类为 tune_hello 包中的 HelloPlugin”。

第 3 步 — 编写插件

创建 src/tune_hello/__init__.py：

from tune_hello.plugin import HelloPlugin

__all__ = ["HelloPlugin"]

然后是 src/tune_hello/plugin.py：

import logging
from fastapi import APIRouter

logger = logging.getLogger("tune.plugin.hello")


class HelloPlugin:
    """Un plugin minimal qui ajoute un endpoint et logge les pistes."""

    name = "hello"
    version = "0.1.0"

    def __init__(self):
        self.router = APIRouter(tags=["hello"])
        self._setup_routes()

    def _setup_routes(self):
        @self.router.get("/hello")
        async def hello():
            return {"message": "Bonjour depuis le plugin Hello !"}

        @self.router.get("/hello/status")
        async def status():
            return {
                "plugin": self.name,
                "version": self.version,
                "status": "running",
            }

    async def setup(self, ctx) -> None:
        """Appelé au démarrage de Tune."""
        # Enregistrer nos routes API
        ctx.register_router(self.router, prefix="/api/v1")

        # S'abonner aux événements de lecture
        ctx.event_bus.on("track.started", self._on_track_started)

        logger.info("Plugin Hello démarré (v%s)", self.version)

    async def teardown(self) -> None:
        """Appelé à l'arrêt de Tune."""
        logger.info("Plugin Hello arrêté")

    async def _on_track_started(self, event) -> None:
        """Appelé à chaque nouvelle piste."""
        track = event.track
        logger.info(
            "Nouvelle piste : %s - %s",
            track.artist,
            track.title,
        )

第 4 步 — 安装插件

在插件目录中，以开发模式将其安装到与 Tune 相同的 Python 环境里：

# 如有需要，激活 Tune 的 virtualenv
source /chemin/vers/tune-server/venv/bin/activate

# 以 editable 模式安装插件
pip install -e .

-e（editable）模式让你修改代码后无需每次重新安装。

第 5 步 — 重启 Tune

重启 Tune 服务器。在日志中，你应该会看到：

INFO  plugin_loaded  name=hello  version=0.1.0
INFO  Plugin Hello démarré (v0.1.0)

如果你使用 systemd：

sudo systemctl restart tune-server

或通过 API（若服务器已在运行）：

curl -X POST http://localhost:8888/api/v1/system/restart

第 6 步 — 验证

确认插件已正确加载：

# 已加载插件列表
curl http://localhost:8888/api/v1/system/plugins

预期响应：

{
  "plugins": [
    {
      "name": "hello",
      "version": "0.1.0",
      "status": "running"
    }
  ]
}

测试你的端点：

curl http://localhost:8888/api/v1/hello

{"message": "Bonjour depuis le plugin Hello !"}

然后在 Tune 中播放一首曲目 — 你会在服务器输出中看到日志 Nouvelle piste : Artiste - Titre。

更进一步

Hello 插件刻意做得很简单。下面是插件系统的主要能力。

添加自定义音频输出类型

你可以创建新的输出类型，向原生不支持的设备或服务推流。例如，一个向 Icecast 服务器广播的插件：

from tune_server.plugins.api import OutputTypeSpec
from tune_server.audio.formats import AudioCapabilities


class IcecastPlugin:
    name = "icecast"
    version = "0.1.0"

    async def setup(self, ctx) -> None:
        ctx.register_output_type(OutputTypeSpec(
            name="icecast",
            factory=self._create_output,
            capabilities=AudioCapabilities(
                max_sample_rate=48000,
                max_bit_depth=16,
                formats=["flac", "mp3"],
            ),
            description="Diffusion vers un serveur Icecast",
        ))

    def _create_output(self, zone_id: str | None):
        return IcecastOutput(
            host=os.environ.get("TUNE_ICECAST_HOST", "localhost"),
            port=int(os.environ.get("TUNE_ICECAST_PORT", "8000")),
            mount=os.environ.get("TUNE_ICECAST_MOUNT", "/stream"),
            password=os.environ.get("TUNE_ICECAST_PASSWORD", "hackme"),
        )

    async def teardown(self) -> None:
        pass

插件安装完成后，icecast 输出类型就会出现在每个区域的可用输出列表中。

挂钩播放事件

Tune 的 EventBus 会在每次操作时发出事件。插件可以订阅这些事件来实现 scrobbling（Last.fm、ListenBrainz）、收听统计或通知。

class ScrobblerPlugin:
    name = "lastfm-scrobbler"
    version = "0.1.0"

    async def setup(self, ctx) -> None:
        ctx.event_bus.on("track.started", self._on_track_started)
        ctx.event_bus.on("track.finished", self._on_track_finished)

    async def _on_track_started(self, event) -> None:
        # Signaler "now playing" à Last.fm
        await self.lastfm_client.update_now_playing(
            artist=event.track.artist,
            title=event.track.title,
            album=event.track.album,
        )

    async def _on_track_finished(self, event) -> None:
        # Scrobble si la piste a été écoutée au moins 50%
        if event.progress >= 0.5:
            await self.lastfm_client.scrobble(
                artist=event.track.artist,
                title=event.track.title,
                album=event.track.album,
                timestamp=event.started_at,
            )

    async def teardown(self) -> None:
        await self.lastfm_client.close()

添加 API 路由

每个插件都可以贡献一个完整的 FastAPI APIRouter，路由数量不限。这些路由会自动出现在 Tune 的 Swagger UI 文档中（/docs）。

router = APIRouter(tags=["stats"])

@router.get("/stats/top-artists")
async def top_artists(limit: int = 10):
    """Retourne les artistes les plus écoutés."""
    return await get_top_artists(limit=limit)

@router.get("/stats/listening-time")
async def listening_time(period: str = "week"):
    """Temps d'écoute total sur la période."""
    return await get_listening_time(period=period)

通过环境变量配置

插件可以声明一个基于 pydantic-settings 的配置 schema。环境变量遵循 TUNE_<PLUGIN>_<KEY> 约定：

from pydantic_settings import BaseSettings


class IcecastSettings(BaseSettings):
    host: str = "localhost"
    port: int = 8000
    mount: str = "/stream"
    password: str = "hackme"

    class Config:
        env_prefix = "TUNE_ICECAST_"


class IcecastPlugin:
    name = "icecast"
    version = "0.1.0"

    def config_schema(self) -> type[BaseSettings]:
        return IcecastSettings

    async def setup(self, ctx) -> None:
        self.config = IcecastSettings()
        # self.config.host vaut $TUNE_ICECAST_HOST ou "localhost"
        ...

这些变量从环境或服务器根目录下的 .env 文件中读取。

挂钩 Player 事件

对于需要比 EventBus 更精细控制的场景，插件可以直接在每个区域的 Player 上注册钩子：

async def setup(self, ctx) -> None:
    ctx.register_player_hook("before_track", self._before_track)
    ctx.register_player_hook("after_track", self._after_track)
    ctx.register_player_hook("play", self._on_play)
    ctx.register_player_hook("pause", self._on_pause)

与 EventBus 的区别在于：Player 钩子是在播放管线中同步调用的，因此可以在某个动作发生之前介入（例如，在曲目开始前预先准备缓冲区）。

发布到 PyPI

插件准备好了？把它发布到 PyPI，与社区分享。

命名约定

Tune 插件的 PyPI 包名遵循 tune-plugin-<名称> 约定：

tune-plugin-icecast
tune-plugin-lastfm
tune-plugin-prometheus

插件的内部名称（name 属性）保持简短：icecast、lastfm、prometheus。

打包

确保你的 pyproject.toml 是完整的：

[project]
name = "tune-plugin-icecast"
version = "0.1.0"
description = "Plugin Tune pour diffuser vers Icecast"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.11"
dependencies = [
    "tune-server>=0.7.90",
]

[project.urls]
Homepage = "https://github.com/votre-nom/tune-plugin-icecast"

[project.entry-points."tune_server.plugins"]
icecast = "tune_icecast:IcecastPlugin"

发布：

pip install build twine
python -m build
twine upload dist/*

之后用户只需一条命令即可安装你的插件：

pip install tune-plugin-icecast

重启一次服务器，插件就会被自动发现并启用。

版本兼容性

请始终在依赖中声明兼容的 Tune 版本范围。服务器暴露了一个 PROTOCOL_VERSION，你的插件可以在 setup 中进行校验：

from tune_server.plugins.api import PROTOCOL_VERSION

async def setup(self, ctx) -> None:
    if PROTOCOL_VERSION != "1":
        raise RuntimeError(
            f"Plugin {self.name} requiert PROTOCOL_VERSION 1, "
            f"trouvé {PROTOCOL_VERSION}"
        )

最佳实践

错误处理 — 插件中的异常绝不应导致服务器崩溃。Tune 会捕获加载时和钩子中的错误，但你仍应在业务逻辑中自行处理 try/except。
日志 — 使用 logging.getLogger("tune.plugin.<名称>")，让你的日志与服务器日志保持一致。
异步 — setup、teardown 方法以及事件回调都必须是 async 的。Tune 完全基于异步（asyncio）。
测试 — 使用 pytest 和 pytest-asyncio 独立测试你的插件。单元测试中可对 PluginContext 进行 mock。
前缀 — 如果你的插件在数据库中创建表，请以插件名作为前缀（例如 icecast_sessions）以避免冲突。
资源 — 始终在 teardown 中释放资源（网络连接、已打开的文件、asyncio 任务）。

结语

只需几行 Python 和一个 pyproject.toml，你就能扩展 Tune Server 以满足特定需求：智能家居集成、scrobbling、非常规音频输出、自定义仪表盘……可能性是开放的。

更进一步：

查阅插件 SDK 完整文档
浏览 GitHub 上的示例
在 Mozaik Labs 论坛分享你的作品并提出问题

我们迫不及待想看到你的创造。