edpibu/hasspy
edpibu/hasspy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

feat: Add cover art support for user client

Browse source 
This commit adds support for displaying cover art from the currently playing media in the Home Assistant user client.

- The `HassUserClient` now publishes cover art to the `user/image/cover` MQTT topic.
- The `publish_cover` method uses `playerctl` to retrieve the art URL and `pillow` to convert the image to WebP format before publishing.
- The `components` property now includes an `image` component for cover art.
- The `cover` attribute on `HassUserClient` is used to avoid resending the same cover multiple times.
- The `publish_state` method has been extended to spawn a thread for updating the cover.
This commit is contained in:
Edgar P. Burkhart 2025-03-09 18:51:38 +01:00
parent 45f7fc4741
commit e54223e361
Signed by: edpibu
GPG key ID: 9833D3C5A25BD227
6 changed files with 148 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

126
hasspy/mqtt.py
View file

@ -1,11 +1,14 @@
import io
import json
import logging
import re
from subprocess import run
from threading import Timer
from threading import Thread, Timer
from typing import Any, Mapping
from paho.mqtt.client import Client, MQTTMessage, MQTTMessageInfo
from paho.mqtt.enums import CallbackAPIVersion, MQTTErrorCode
from PIL import Image
log = logging.getLogger(__name__)
@ -23,7 +26,9 @@ class HassClient(Client):
self.interval = self.config.get("interval", 60)
self.power_on = True
self.timer = Timer(self.interval, self.publish_state)
self.timer = Timer(0, self.publish_state)
self.cover = ""
self.connect()
@ -58,7 +63,6 @@ class HassClient(Client):
self.message_callback_add(self.command_topic, self.on_command)
def publish_state(self) -> MQTTMessageInfo:
self.timer.cancel()
self.timer = Timer(self.interval, self.publish_state)
self.timer.start()
@ -74,7 +78,9 @@ class HassClient(Client):
self.do_command(*payload.split(":"))
self.publish_state()
self.timer.cancel()
self.timer = Timer(1, self.publish_state)
self.timer.start()
def do_command(self, cmd: str, value: str = "") -> None:
pass
@ -85,7 +91,7 @@ class HassClient(Client):
self.publish_availability()
self.init_subs()
self.publish_state()
self.timer.start()
@property
def state_topic(self) -> str:
@ -139,9 +145,7 @@ class HassSystemClient(HassClient):
def do_command(self, cmd: str, value: str = "") -> None:
if cmd in self.commands:
log.debug(f"Executing command: {cmd}")
proc = run(self.commands[cmd])
if proc.returncode != 0:
log.error(f"Failed to execute command: {cmd}")
run_command(self.commands[cmd])
if cmd == "POWER_ON":
self.power_on = True
@ -179,6 +183,9 @@ class HassSystemClient(HassClient):
class HassUserClient(HassClient):
commands = {
"PLAY_PAUSE": ["playerctl", "play-pause"],
"PLAY_NEXT": ["playerctl", "next"],
"PLAY_PREV": ["playerctl", "previous"],
"PLAY_STOP": ["playerctl", "stop"],
}
def __init__(self, node_id: str, config: Mapping[str, Any]) -> None:
@ -187,14 +194,13 @@ class HassUserClient(HassClient):
def do_command(self, cmd: str, value: str = "") -> None:
if cmd in self.commands:
log.debug(f"Executing command: {cmd}")
proc = run(self.commands[cmd])
if proc.returncode != 0:
log.error(f"Failed to execute command: {cmd}")
run_command(self.commands[cmd])
match [cmd, value]:
case ["VOLUME", value]:
log.debug(f"Executing command: {cmd}:{value}")
proc = run(
run_command(
[
"wpctl",
"set-volume",
@ -202,8 +208,6 @@ class HassUserClient(HassClient):
f"{int(value) / 100:.2f}",
]
)
if proc.returncode != 0:
log.error(f"Failed to set volume: {value}")
@property
def availability_topic(self) -> str:
@ -219,6 +223,44 @@ class HassUserClient(HassClient):
"icon": "mdi:play-pause",
"payload_press": "PLAY_PAUSE",
},
"next": {
"unique_id": f"{self.node_id}_next",
"p": "button",
"name": "Next",
"icon": "mdi:skip-next",
"payload_press": "PLAY_NEXT",
},
"prev": {
"unique_id": f"{self.node_id}_prev",
"p": "button",
"name": "Previous",
"icon": "mdi:skip-previous",
"payload_press": "PLAY_PREV",
},
"stop": {
"unique_id": f"{self.node_id}_stop",
"p": "button",
"name": "Stop",
"icon": "mdi:stop",
"payload_press": "PLAY_STOP",
},
"player": {
"unique_id": f"{self.node_id}_player",
"p": "sensor",
"name": "Player",
"icon": "mdi:music",
"value_template": "{{ value_json.player.value }}",
"json_attributes_topic": self.state_topic,
"json_attributes_template": "{{ value_json.player.attributes | to_json }}",
},
"cover": {
"unique_id": f"{self.node_id}_cover",
"p": "image",
"name": "Cover",
"icon": "mdi:disc-player",
"content_type": "image/webp",
"image_topic": self.cover_topic,
},
"volume": {
"unique_id": f"{self.node_id}_volume",
"p": "number",
@ -237,13 +279,59 @@ class HassUserClient(HassClient):
def state_payload(self) -> dict[str, Any]:
return {
"volume": self.volume_value,
"player": self.player_value,
}
@property
def volume_value(self) -> int:
proc = run(["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"], capture_output=True)
if proc.returncode != 0:
log.error("Failed to get volume")
return 0
vol = run_command(["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"])
return int(float(proc.stdout.decode("utf-8").split(": ")[1]) * 100)
return int(float(vol.split(": ")[1]) * 100)
@property
def player_value(self) -> str | dict[str, str | dict[str, str]]:
return {
"value": run_command(["playerctl", "status"]),
"attributes": {
k: run_command(["playerctl", "metadata", k])
for k in ["title", "album", "artist"]
},
}
def publish_state(self) -> MQTTMessageInfo:
Thread(target=self.publish_cover).start()
return super().publish_state()
@property
def cover_topic(self) -> str:
return f"{self.node_id}/image/cover"
def publish_cover(self) -> None:
log.debug("Publishing cover image")
out = run_command(["playerctl", "metadata"])
artUrl = re.compile(r"mpris:artUrl\s+file://(.*)").search(out)
if not artUrl:
return
art = artUrl.group(1)
if art == self.cover:
return
self.cover = art
by = io.BytesIO()
with Image.open(art) as im:
im.save(by, format="webp")
by.seek(0)
self.publish(self.cover_topic, by.read())
def run_command(cmd: list[str]) -> str:
proc = run(cmd, capture_output=True)
if proc.returncode != 0:
log.error(f"Failed to execute command: {cmd}")
return "null"
return proc.stdout.decode("utf-8")