spotifydとは

https://github.com/Spotifyd/spotifyd
SpotifyのPlayer Daemon
Linuxの公式クライアントもあるが、タイル型WMユーザーにとってはspotifyd + spotify-tuiの組み合わせが至高

Nanamiiiii 2023/03/08

SpotifyのCredentialをシステムのKeyringに入れる

Spotifydの認証は以下のいずれかで行う

spotifyd.confに username/password 直書き
Discoveryを有効にしてローカルネットワークからの接続待ち
stdout経由でコマンドから username/password を取得
spotifyd.confに username のみ記載し、passwordはSystem Keyringから取得

1は言わずもがな却下。直書きはNG。
2は公式Docsにも記載のとおり複数のアカウントから利用可能な利点があるが、API経由でデバイスを検出するspotify-tuiとは使えないのでNG
選択肢は3,4のいずれかになるが、コマンド直書きで環境依存性が生じるのが嫌だったので4を選択
4はDBus Secret Service APIを実装するKeyringサービスに接続して取得するので、この条件さえ満たしていればgnome-keyringだろうとkdewalletだろうと関係ない

公式Docsに従い、指定のスキーマを持たせてKeyringに格納する

Your keyring entry needs to have the following attributes set:
application: rust-keyring
service: spotifyd
username: <your-spotify-username>

Linuxだとsecret-toolでシステムデフォルトのKeyringに接続し、CLI経由でlookup/storeなどの操作ができる

secret-tool store --label='spotifyd' application rust-keyring service spotifyd username <your-username>

spotifyd.confには以下のように記載

username = "username"
use_keyring = true

user_keyring=trueの指定でKeyringを見に行ってくれるようになる

ただしSpotifydの起動時にKeyringがロックされていると取得できないので、ログインに連動してKeyringをアンロックするなどの設定をしておくと便利

Nanamiiiii 2023/03/08

公式クライアントみたいにいい感じにNotificationする

モチベ

公式クライアントだと、再生・選曲時に通知を出せる
Spotifydでも似たようなことをしたい

手段

Spotifydには選曲イベント時にコマンドをhookする機能がある

# A command that gets executed in your shell after each song changes.
on_song_change_hook = "command_to_run_on_playback_events"

この機能を使ってNotificationを発生させるスクリプトをhookすればいい

できたもの

最終的に落ち着いたもの
こんな感じの通知が出る（通知サーバはdeadd-notification-center）

これになるまでにshellscriptのものだったり色々試行した
何をしてるかは後述

import gi
gi.require_version('Playerctl', '2.0')
gi.require_version('Secret', '1')
gi.require_version('Notify', '0.7')
gi.require_version('GdkPixbuf', '2.0')
from gi.repository import Playerctl, Secret, Notify, GdkPixbuf
import logging
import os
import pathlib
import requests
import spotipy
from spotipy.oauth2 import SpotifyClientCredentials

logger = logging.getLogger(__name__)

CRED_SCHEMA = Secret.Schema.new("org.freedesktop.Secret.Generic",
    Secret.SchemaFlags.DONT_MATCH_NAME,
    {
        "application": Secret.SchemaAttributeType.STRING,
        "type": Secret.SchemaAttributeType.STRING,
    }
)


def notify_info(info, track_id):
    # Retrive cached coverimage
    cover_path = "/tmp/spotifyd-event/{}".format(track_id)
    if not os.path.exists(cover_path) and info["cover_url"] != None:
        # if not cached, retrive using url from playerctl
        pathlib.Path(cover_path).parent.mkdir(parents=True, exist_ok=True)
        response = requests.get(info["cover_url"])
        if response.status_code == 200:
            with open(cover_path, 'wb') as f:
                f.write(response.content)
            logger.info("-CACHED- coverimage track_id:%s path:%s", track_id, cover_path)
        else:
            logger.error("Failed to cache coverimage track_id:%s", track_id)

    cover_buf = GdkPixbuf.Pixbuf.new_from_file(cover_path)

    Notify.init("Spotify")
    notification = Notify.Notification.new(
        "{} - {}".format(info["title"], ",".join(info["artists"])),
        info["album"],
        "spotify"
    )
    notification.set_image_from_pixbuf(cover_buf)
    notification.set_urgency(Notify.Urgency.NORMAL)

    # Pickup notification id
    id_path = "/tmp/spotifyd-event/notify-id"
    if os.path.exists(id_path):
        with open(id_path, 'r') as f:
            notify_id = f.readline()
        notification.props.id = int(notify_id)

    # Show notification
    notification.show()

    # Write notification id
    notify_id = notification.props.id
    with open(id_path, 'w') as f:
        f.write(str(notify_id))


def get_api_cred() -> dict:
    # Lookup keyring
    client_id = Secret.password_lookup_sync(CRED_SCHEMA, { "application": "spotifyd_event", "type" : "client_id" }, None) 
    client_secret = Secret.password_lookup_sync(CRED_SCHEMA, { "application": "spotifyd_event", "type" : "client_secret" }, None) 

    credential = {
        "client_id" : client_id,
        "client_secret" : client_secret
    }
    return credential


def cache_cover(track_id: str):
    # get api credential from OS keyring
    credential = get_api_cred()

    # Spotify API
    spotify_cred = SpotifyClientCredentials(client_id=credential["client_id"], client_secret=credential["client_secret"])
    spotify = spotipy.Spotify(client_credentials_manager=spotify_cred)

    # Get track info & cover image url
    track_info = spotify.track(track_id)
    image_url = track_info["album"]["images"][0]["url"]

    # Download & save image to /tmp
    image_path = "/tmp/spotifyd-event/{}".format(track_id)
    pathlib.Path(image_path).parent.mkdir(parents=True, exist_ok=True)
    response = requests.get(image_url)
    if response.status_code == 200:
        with open(image_path, 'wb') as f:
            f.write(response.content)
        logger.info("-CACHED- coverimage track_id:%s path:%s", track_id, image_path)
    else:
        logger.error("Failed to cache coverimage track_id:%s", track_id)


def get_playing_info(player) -> dict:
    # Collect information
    metadata = player.props.metadata
    keys = metadata.keys()
    if "xesam:artist" in keys:
        artists = metadata["xesam:artist"]
    else:
        artists = "unknown"
    if "xesam:title" in keys:
        title = metadata["xesam:title"]
    else:
        title = "unknown"
    if "xesam:album" in keys:
        album = metadata["xesam:album"]
    else:
        album = "unknown"
    if "mpris:artUrl" in keys:
        cover_url = metadata["mpris:artUrl"]
    else:
        cover_url = None
    
    playing_info = {
        "artists" : artists,
        "title" : title,
        "album" : album,
        "cover_url" : cover_url
    }
    return playing_info

    
def main():
    # Environment vers
    try:
        TRACK_ID = os.getenv("TRACK_ID")
        PLAYER_EVENT = os.getenv("PLAYER_EVENT")
    except KeyError as e:
        logger.error("Missing required environment variables. Aborted.")
        logger.error("Please launch this from spotifyd event hook.")
        exit(1)

    # Process event
    if PLAYER_EVENT == "preload" or PLAYER_EVENT == "preloading" or PLAYER_EVENT == "load":
        # Cache coverimage
        cache_cover(TRACK_ID)
    elif PLAYER_EVENT == "start" or PLAYER_EVENT == "change" or PLAYER_EVENT == "play":
        # Collect track info & send notification
        manager = Playerctl.PlayerManager()
        for name in manager.props.player_names:
            if name.name == "spotifyd":
                player_name = name
                break
        player = Playerctl.Player.new_from_name(player_name)
        playing_info = get_playing_info(player)
        notify_info(playing_info, TRACK_ID)


if __name__ == "__main__":
    main()

Nanamiiiii 2023/03/18

Spotifydは環境変数にいくつか情報を入れてscriptをhookする
以下のソースで定義されている

src/process.rs

pub(crate) fn spawn_program_on_event(
    shell: &str,
    cmd: &str,
    event: PlayerEvent,
) -> Result<Child, Error> {
    let mut env = HashMap::new();
    match event {
        PlayerEvent::Changed {
            old_track_id,
            new_track_id,
        } => {
            env.insert("OLD_TRACK_ID", old_track_id.to_base62());
            env.insert("PLAYER_EVENT", "change".to_string());
            env.insert("TRACK_ID", new_track_id.to_base62());
        }
        PlayerEvent::Started {
            track_id,
            play_request_id,
            position_ms,
        } => {
            env.insert("PLAYER_EVENT", "start".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
            env.insert("POSITION_MS", position_ms.to_string());
        }
        PlayerEvent::Stopped {
            track_id,
            play_request_id,
        } => {
            env.insert("PLAYER_EVENT", "stop".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
        }
        PlayerEvent::Loading {
            track_id,
            play_request_id,
            position_ms,
        } => {
            env.insert("PLAYER_EVENT", "load".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
            env.insert("POSITION_MS", position_ms.to_string());
        }
        PlayerEvent::Playing {
            track_id,
            play_request_id,
            position_ms,
            duration_ms,
        } => {
            env.insert("PLAYER_EVENT", "play".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
            env.insert("POSITION_MS", position_ms.to_string());
            env.insert("DURATION_MS", duration_ms.to_string());
        }
        PlayerEvent::Paused {
            track_id,
            play_request_id,
            position_ms,
            duration_ms,
        } => {
            env.insert("PLAYER_EVENT", "pause".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
            env.insert("POSITION_MS", position_ms.to_string());
            env.insert("DURATION_MS", duration_ms.to_string());
        }
        PlayerEvent::TimeToPreloadNextTrack {
            track_id,
            play_request_id,
        } => {
            env.insert("PLAYER_EVENT", "preload".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
        }
        PlayerEvent::EndOfTrack {
            track_id,
            play_request_id,
        } => {
            env.insert("PLAYER_EVENT", "endoftrack".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
        }
        PlayerEvent::VolumeSet { volume } => {
            env.insert("PLAYER_EVENT", "volumeset".to_string());
            env.insert("VOLUME", volume.to_string());
        }
        PlayerEvent::Unavailable {
            play_request_id,
            track_id,
        } => {
            env.insert("PLAYER_EVENT", "unavailable".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
            env.insert("PLAY_REQUEST_ID", play_request_id.to_string());
        }
        PlayerEvent::Preloading { track_id } => {
            env.insert("PLAYER_EVENT", "preloading".to_string());
            env.insert("TRACK_ID", track_id.to_base62());
        }
    }
    spawn_program(shell, cmd, env)
}

ここでは以下の環境変数を使用する

ENV Name	Value
`PLAYER_EVENT`	イベントタイプ
`TRACK_ID`	トラックID

Nanamiiiii 2023/03/18に更新

PLAYER_EVENTでイベントタイプを判別し，処理を分けている

    # Process event
    if PLAYER_EVENT == "preload" or PLAYER_EVENT == "preloading" or PLAYER_EVENT == "load":
        # Cache coverimage
        cache_cover(TRACK_ID)
    elif PLAYER_EVENT == "start" or PLAYER_EVENT == "change" or PLAYER_EVENT == "play":
        # Collect track info & send notification
        manager = Playerctl.PlayerManager()
        for name in manager.props.player_names:
            if name.name == "spotifyd":
                player_name = name
                break
        player = Playerctl.Player.new_from_name(player_name)
        playing_info = get_playing_info(player)
        notify_info(playing_info, TRACK_ID)

`preload` / `preloading` / `load`

アルバムジャケットをSpotify API経由で取得し，tmpfsにキャッシュする

`start` / `change` / `play`

Playerctl経由で曲の詳細情報を取得
通知を送信

Nanamiiiii 2023/03/18

ジャケットのキャッシュ

再生・曲変更時に取得すると通知までにラグが出るかも？と勝手に思って，プリロードイベントの時に裏でAPIでとってくるようにした
プリロードは前のトラックがだいたい半分以降に到達したくらいで発生するみたい

def cache_cover(track_id: str):
    # get api credential from OS keyring
    credential = get_api_cred()

    # Spotify API
    spotify_cred = SpotifyClientCredentials(client_id=credential["client_id"], client_secret=credential["client_secret"])
    spotify = spotipy.Spotify(client_credentials_manager=spotify_cred)

    # Get track info & cover image url
    track_info = spotify.track(track_id)
    image_url = track_info["album"]["images"][0]["url"]

    # Download & save image to /tmp
    image_path = "/tmp/spotifyd-event/{}".format(track_id)
    pathlib.Path(image_path).parent.mkdir(parents=True, exist_ok=True)
    response = requests.get(image_url)
    if response.status_code == 200:
        with open(image_path, 'wb') as f:
            f.write(response.content)
        logger.info("-CACHED- coverimage track_id:%s path:%s", track_id, image_path)
    else:
        logger.error("Failed to cache coverimage track_id:%s", track_id)

track_info APIで取得できる情報に，アルバムジャケットのURLがあるのでこれを取得．
このURLからrequestsで画像データを取得し，/tmp/spotifyd-eventに保存する．
API CredentialはSystem Keyringから取得．