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       # 고유 식별자, 예: "hello"
    version: str    # semver 버전, 예: "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """서버 시작 시 호출됩니다."""
        ...

    async def teardown(self) -> None:
        """서버 종료 시 호출됩니다."""
        ...

2. PluginContext

setup 시점에 플러그인은 PluginContext 객체를 받으며, 이를 통해 Tune의 내부 서비스에 안전하게 접근할 수 있습니다:

• ctx.event_bus — 재생 이벤트 구독 (트랙 시작, 존 생성 등)
• ctx.zone_manager — 새로운 오디오 출력 유형 등록
• ctx.api_app — 커스텀 FastAPI 라우트 마운트
• ctx.settings — 서버 설정 읽기 (읽기 전용)
• ctx.db — 데이터베이스 접근

3. 생명주기

시작 시 Tune은 Python entry points를 통해 설치된 플러그인을 자동으로 발견한 뒤, 순서대로 초기화합니다:

1. 발견 — Tune이 entry_points(group="tune_server.plugins")를 스캔
2. setup(ctx) — 플러그인이 라우트, 훅, 출력 유형을 등록
3. 정상 동작 — 서버가 실행되고 플러그인이 활성화됨
4. teardown() — 종료 시 플러그인이 리소스를 해제

핫 리로드는 없습니다. 수정 후에는 서버를 재시작해야 합니다. 이는 안정성을 보장하기 위한 의도적인 선택입니다 (재생 중 상태 손상 방지).

단계별 퀵 스타트

/api/v1/hello API 엔드포인트를 추가하고 트랙이 바뀔 때마다 메시지를 로깅하는 "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:
    """엔드포인트를 추가하고 트랙을 로깅하는 최소 플러그인."""

    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": "Hello 플러그인에서 인사드립니다!"}

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

    async def setup(self, ctx) -> None:
        """Tune 시작 시 호출됩니다."""
        # API 라우트 등록
        ctx.register_router(self.router, prefix="/api/v1")

        # 재생 이벤트 구독
        ctx.event_bus.on("track.started", self._on_track_started)

        logger.info("Hello 플러그인 시작됨 (v%s)", self.version)

    async def teardown(self) -> None:
        """Tune 종료 시 호출됩니다."""
        logger.info("Hello 플러그인 중지됨")

    async def _on_track_started(self, event) -> None:
        """새 트랙마다 호출됩니다."""
        track = event.track
        logger.info(
            "새 트랙: %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  Hello 플러그인 시작됨 (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": "Hello 플러그인에서 인사드립니다!"}

그리고 Tune에서 트랙을 재생하면 — 서버 출력에 새 트랙: 아티스트 - 제목 로그가 나타나는 것을 볼 수 있습니다.

더 나아가기

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="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는 모든 동작마다 이벤트를 발행합니다. 플러그인은 이를 구독하여 스크로블링(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:
        # Last.fm에 "now playing" 알리기
        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:
        # 트랙을 최소 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):
    """가장 많이 들은 아티스트를 반환합니다."""
    return await get_top_artists(limit=limit)

@router.get("/stats/listening-time")
async def listening_time(period: str = "week"):
    """해당 기간의 총 청취 시간."""
    return await get_listening_time(period=period)

환경 변수를 통한 설정

플러그인은 pydantic-settings 기반의 설정 스키마를 선언할 수 있습니다. 환경 변수는 TUNE_<PLUGIN>_<키> 규칙을 따릅니다:

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 는 $TUNE_ICECAST_HOST 또는 "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"플러그인 {self.name} 은 PROTOCOL_VERSION 1 이 필요합니다, "
            f"발견된 값: {PROTOCOL_VERSION}"
        )

모범 사례

• 에러 처리 — 플러그인의 예외가 절대 서버를 중단시켜서는 안 됩니다. Tune은 로딩 시점과 훅에서 발생하는 에러를 잡아내지만, 비즈니스 로직에서는 직접 try/except로 처리하세요.
• 로깅 — 로그가 서버 로그와 일관되도록 logging.getLogger("tune.plugin.<이름>")을 사용하세요.
• Async — setup, teardown 메서드와 이벤트 콜백은 async여야 합니다. Tune은 완전히 비동기(asyncio)입니다.
• 테스트 — pytest와 pytest-asyncio로 플러그인을 독립적으로 테스트하세요. 단위 테스트에서는 PluginContext를 모킹하세요.
• 접두사 — 플러그인이 데이터베이스 테이블을 만든다면 충돌을 피하기 위해 플러그인 이름을 접두사로 붙이세요(예: icecast_sessions).
• 리소스 — 네트워크 연결, 열린 파일, asyncio 태스크 등 리소스는 항상 teardown에서 해제하세요.

결론

몇 줄의 Python 코드와 pyproject.toml 하나만으로 Tune Server를 여러분의 특수한 요구에 맞게 확장할 수 있습니다. 홈 오토메이션 연동, 스크로블링, 색다른 오디오 출력, 커스텀 대시보드... 가능성은 열려 있습니다.

더 알아보기:

• 플러그인 SDK 전체 문서를 참고하세요
• GitHub의 예제를 살펴보세요
• Mozaik Labs 포럼에서 작품을 공유하고 질문을 남겨주세요

여러분이 무엇을 만들어낼지 기대됩니다.