Crea il tuo primo plugin Tune in 5 minuti

Tune Server è progettato per essere aperto. Dalla versione 0.7.90, un sistema di plugin Python consente di estendere il server senza modificarne il codice sorgente: aggiungere nuove uscite audio, integrare servizi di terze parti, creare endpoint API personalizzati o reagire agli eventi di riproduzione. In questo articolo creeremo un primo plugin funzionante in pochi minuti.

Prerequisiti

• Python 3.11+ installato sulla tua macchina
• pip (incluso con Python)
• Un server Tune funzionante (versione 0.7.90 o superiore)
• Un editor di testo (VS Code, PyCharm, vim...)

Verifica la tua versione di Python:

python3 --version
# Python 3.11.x o superiore

Il concetto

Il sistema di plugin di Tune si basa su tre elementi:

1. Il Protocol TunePlugin

Ogni plugin è una classe Python che implementa il protocol TunePlugin. Questo protocol definisce due attributi (name e version) e due metodi del ciclo di vita (setup e teardown).

from typing import Protocol, runtime_checkable

@runtime_checkable
class TunePlugin(Protocol):
    name: str       # identificativo univoco, es: "hello"
    version: str    # versione semver, es: "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """Chiamato all'avvio del server."""
        ...

    async def teardown(self) -> None:
        """Chiamato all'arresto del server."""
        ...

2. Il PluginContext

Durante il setup, il tuo plugin riceve un oggetto PluginContext che dà accesso ai servizi interni di Tune in modo sicuro:

• ctx.event_bus — iscriversi agli eventi di riproduzione (traccia avviata, zona creata, ecc.)
• ctx.zone_manager — registrare nuovi tipi di uscita audio
• ctx.api_app — montare route API FastAPI personalizzate
• ctx.settings — leggere la configurazione del server (sola lettura)
• ctx.db — accesso al database

3. Il ciclo di vita

All'avvio, Tune individua automaticamente i plugin installati tramite gli entry point Python, e poi li inizializza nell'ordine:

1. Scoperta — Tune scansiona gli entry_points(group="tune_server.plugins")
2. setup(ctx) — il plugin registra le sue route, hook e tipi di uscita
3. Funzionamento normale — il server gira, il plugin è attivo
4. teardown() — all'arresto, il plugin libera le sue risorse

Nessun ricaricamento a caldo: dopo ogni modifica è necessario riavviare il server. È una scelta deliberata per garantire la stabilità (nessuna corruzione dello stato durante la riproduzione).

Quick start passo passo

Creiamo un plugin "Hello" che aggiunge un endpoint API /api/v1/hello e registra un messaggio a ogni cambio di traccia.

Passo 1 — Creare la struttura del progetto

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

La struttura finale sarà:

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

Passo 2 — Scrivere il pyproject.toml

Il file pyproject.toml dichiara il tuo plugin a Python e a Tune. La sezione [project.entry-points] è la chiave: permette a Tune di individuare il tuo plugin all'avvio.

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

[project]
name = "tune-hello-plugin"
version = "0.1.0"
description = "Il mio primo plugin Tune"
requires-python = ">=3.11"
dependencies = []

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

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

La riga hello = "tune_hello:HelloPlugin" significa: "registra il plugin chiamato 'hello', la cui classe è HelloPlugin nel package tune_hello".

Passo 3 — Scrivere il plugin

Crea src/tune_hello/__init__.py:

from tune_hello.plugin import HelloPlugin

__all__ = ["HelloPlugin"]

Poi src/tune_hello/plugin.py:

import logging
from fastapi import APIRouter

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


class HelloPlugin:
    """Un plugin minimale che aggiunge un endpoint e logga le tracce."""

    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": "Ciao dal 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:
        """Chiamato all'avvio di Tune."""
        # Registriamo le nostre route API
        ctx.register_router(self.router, prefix="/api/v1")

        # Iscriviamoci agli eventi di riproduzione
        ctx.event_bus.on("track.started", self._on_track_started)

        logger.info("Plugin Hello avviato (v%s)", self.version)

    async def teardown(self) -> None:
        """Chiamato all'arresto di Tune."""
        logger.info("Plugin Hello arrestato")

    async def _on_track_started(self, event) -> None:
        """Chiamato a ogni nuova traccia."""
        track = event.track
        logger.info(
            "Nuova traccia: %s - %s",
            track.artist,
            track.title,
        )

Passo 4 — Installare il plugin

Dalla cartella del tuo plugin, installalo in modalità sviluppo nello stesso ambiente Python di Tune:

# Attiva il virtualenv di Tune se necessario
source /percorso/verso/tune-server/venv/bin/activate

# Installa il plugin in modalità editable
pip install -e .

La modalità -e (editable) permette di modificare il codice senza dover reinstallare ogni volta.

Passo 5 — Riavviare Tune

Riavvia il server Tune. Nei log dovresti vedere:

INFO  plugin_loaded  name=hello  version=0.1.0
INFO  Plugin Hello avviato (v0.1.0)

Se usi systemd:

sudo systemctl restart tune-server

Oppure tramite l'API (se il server è già in esecuzione):

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

Passo 6 — Verificare

Conferma che il plugin sia stato caricato correttamente:

# Elenco dei plugin caricati
curl http://localhost:8888/api/v1/system/plugins

Risposta attesa:

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

Testa il tuo endpoint:

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

{"message": "Ciao dal plugin Hello!"}

E avvia una traccia in Tune — vedrai comparire il log Nuova traccia: Artista - Titolo nell'output del server.

Andare oltre

Il plugin Hello è volutamente semplice. Ecco le principali capacità del sistema di plugin.

Aggiungere un tipo di uscita audio personalizzato

Puoi creare nuovi tipi di uscita per trasmettere verso dispositivi o servizi non supportati nativamente. Ad esempio, un plugin per trasmettere verso un server 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="Trasmissione verso un server 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

Una volta installato il plugin, il tipo di uscita icecast compare nell'elenco delle uscite disponibili per ogni zona.

Agganciare gli eventi di riproduzione

L'EventBus di Tune emette eventi a ogni azione. Un plugin può iscriversi per implementare lo scrobbling (Last.fm, ListenBrainz), statistiche di ascolto o notifiche.

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:
        # Segnala "now playing" a 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 se la traccia è stata ascoltata almeno al 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()

Aggiungere route API

Ogni plugin può contribuire con un APIRouter FastAPI completo, con tutte le route necessarie. Le route vengono documentate automaticamente nella Swagger UI di Tune (/docs).

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

@router.get("/stats/top-artists")
async def top_artists(limit: int = 10):
    """Restituisce gli artisti più ascoltati."""
    return await get_top_artists(limit=limit)

@router.get("/stats/listening-time")
async def listening_time(period: str = "week"):
    """Tempo di ascolto totale nel periodo."""
    return await get_listening_time(period=period)

Configurazione tramite variabili d'ambiente

I plugin possono dichiarare uno schema di configurazione basato su pydantic-settings. Le variabili d'ambiente seguono la convenzione TUNE_<PLUGIN>_<CHIAVE>:

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 vale $TUNE_ICECAST_HOST oppure "localhost"
        ...

Le variabili vengono lette dall'ambiente o da un file .env nella radice del server.

Agganciare gli eventi del Player

Per casi d'uso che richiedono un controllo più fine rispetto all'EventBus, i plugin possono registrare hook direttamente sul Player di ogni zona:

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)

La differenza rispetto all'EventBus: gli hook del Player vengono chiamati in modo sincrono nella pipeline di riproduzione, il che consente di intervenire prima che un'azione avvenga (ad esempio, preparare un buffer prima dell'avvio di una traccia).

Pubblicare su PyPI

Il tuo plugin è pronto? Condividilo con la community pubblicandolo su PyPI.

Convenzione di denominazione

I plugin Tune seguono la convenzione tune-plugin-<nome> per il nome del package PyPI:

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

Il nome interno del plugin (attributo name) resta breve: icecast, lastfm, prometheus.

Packaging

Assicurati che il tuo pyproject.toml sia completo:

[project]
name = "tune-plugin-icecast"
version = "0.1.0"
description = "Plugin Tune per trasmettere verso 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"

Pubblica:

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

Gli utenti potranno poi installare il tuo plugin con un solo comando:

pip install tune-plugin-icecast

Basta un riavvio del server perché il plugin venga individuato e attivato automaticamente.

Compatibilità di versione

Dichiara sempre l'intervallo di versioni di Tune compatibili nelle tue dipendenze. Il server espone una PROTOCOL_VERSION che il tuo plugin può verificare nel setup:

from tune_server.plugins.api import PROTOCOL_VERSION

async def setup(self, ctx) -> None:
    if PROTOCOL_VERSION != "1":
        raise RuntimeError(
            f"Il plugin {self.name} richiede PROTOCOL_VERSION 1, "
            f"trovata {PROTOCOL_VERSION}"
        )

Buone pratiche

• Gestione degli errori — Le eccezioni in un plugin non devono mai far crashare il server. Tune cattura gli errori al caricamento e negli hook, ma gestisci i tuoi try/except nella logica applicativa.
• Logging — Usa logging.getLogger("tune.plugin.<nome>") affinché i tuoi log siano coerenti con quelli del server.
• Async — I metodi setup, teardown e le callback degli eventi devono essere async. Tune è interamente asincrono (asyncio).
• Test — Testa il tuo plugin in modo indipendente con pytest e pytest-asyncio. Simula il PluginContext per i test unitari.
• Prefissi — Se il tuo plugin crea tabelle nel database, prefissale con il nome del plugin (es: icecast_sessions) per evitare collisioni.
• Risorse — Libera sempre le tue risorse (connessioni di rete, file aperti, task asyncio) in teardown.

Conclusione

Con poche righe di Python e un pyproject.toml, puoi estendere Tune Server per rispondere alle tue esigenze specifiche: integrazioni domotiche, scrobbling, uscite audio esotiche, dashboard personalizzate... Le possibilità sono aperte.

Per approfondire:

• Consulta la documentazione completa dell'SDK plugin
• Esplora gli esempi su GitHub
• Condividi le tue creazioni e poni le tue domande sul forum Mozaik Labs

Non vediamo l'ora di scoprire cosa costruirai.