aboutsummaryrefslogtreecommitdiffstats
path: root/deckit/modules/music.py
blob: aa30798cfc01fe3a2e0a1b51aaca1a8c3e9deba1 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# deck-it - music module
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Now playing and volume, via playerctl and wpctl."""

import subprocess

from ..module import Module, ModuleError

SEP = "\x1f"

BASE_ACTIONS = {
    "voldown": ["wpctl", "set-volume", "@DEFAULT_AUDIO_SINK@", "5%-"],
    "mute": ["wpctl", "set-mute", "@DEFAULT_AUDIO_SINK@", "toggle"],
    "volup": ["wpctl", "set-volume", "-l", "1.0", "@DEFAULT_AUDIO_SINK@", "5%+"],
    "prev": ["playerctl", "previous"],
    "play": ["playerctl", "play-pause"],
    "next": ["playerctl", "next"],
}


def _run(cmd, timeout=2):
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout.strip()
    except (OSError, subprocess.TimeoutExpired):
        return ""


def _popen(cmd):
    subprocess.Popen(cmd, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def make(cfg: dict) -> Module:
    run = cfg.get("run", _run)
    popen = cfg.get("popen", _popen)

    def state():
        fmt = SEP.join(["{{status}}", "{{title}}", "{{artist}}", "{{album}}", "{{mpris:length}}",
                        "{{position}}", "{{mpris:artUrl}}", "{{playerName}}"])
        out = run(["playerctl", "metadata", "--format", fmt])
        vol = run(["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"])
        volume = {"level": round(float(vol.split()[1]) * 100) if len(vol.split()) > 1 else 0,
                  "muted": "MUTED" in vol}
        if not out:
            return {"status": "Stopped", "volume": volume}
        status, title, artist, album, length, pos, art, name = (out.split(SEP) + [""] * 8)[:8]
        if art.startswith(("http://", "https://")):
            art_src = art
        elif art.startswith("file://"):
            art_src = "/api/art?v=" + str(abs(hash(art)))
        else:
            art_src = None
        return {"status": status, "title": title, "artist": artist, "album": album,
                "length": int(length or 0) / 1e6, "position": int(pos or 0) / 1e6,
                "art": art_src, "player": name.split(".")[0], "volume": volume}

    def seek(body):
        pos = body.get("position")
        if not isinstance(pos, (int, float)) or pos < 0:
            raise ModuleError("position must be a number")
        popen(["playerctl", "position", str(round(float(pos), 1))])

    actions = {name: (lambda cmd: (lambda body: popen(cmd)))(cmd) for name, cmd in BASE_ACTIONS.items()}
    actions["seek"] = seek
    return Module(name="music", title="Music", state=state, actions=actions)