cover.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# pylint: disable=invalid-name

from time import sleep
import requests
import asyncio
import json
import sys
import time
import xmltodict
from datetime import datetime
from typing import Any, Optional, Sequence, Tuple, Union, cast
from collections import OrderedDict
import urllib.request
from async_upnp_client.advertisement import SsdpAdvertisementListener
from async_upnp_client.aiohttp import AiohttpNotifyServer, AiohttpRequester
from async_upnp_client.client import UpnpDevice, UpnpService, UpnpStateVariable
from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.const import NS, AddressTupleVXType, SsdpHeaders
from async_upnp_client.exceptions import UpnpResponseError
from async_upnp_client.profiles.dlna import dlna_handle_notify_last_change
from async_upnp_client.search import async_search as async_ssdp_search
from async_upnp_client.ssdp import SSDP_IP_V4, SSDP_IP_V6, SSDP_PORT, SSDP_ST_ALL
from async_upnp_client.utils import get_local_ip

items = {}
art = ""

pprint_indent = 4

event_handler = None
playing = False

async def create_device(description_url: str) -> UpnpDevice:
    """Create UpnpDevice."""
    timeout = 60
    non_strict = True
    requester = AiohttpRequester(timeout)
    factory = UpnpFactory(requester, non_strict=non_strict)
    return await factory.async_create_device(description_url)


def get_timestamp() -> Union[str, float]:
    """Timestamp depending on configuration."""
    return time.time()


def service_from_device(device: UpnpDevice, service_name: str) -> Optional[UpnpService]:
    """Get UpnpService from UpnpDevice by name or part or abbreviation."""
    for service in device.all_services:
        part = service.service_id.split(":")[-1]
        abbr = "".join([c for c in part if c.isupper()])
        if service_name in (service.service_type, part, abbr):
            return service

    return None

def on_event(
    service: UpnpService, service_variables: Sequence[UpnpStateVariable]
) -> None:
    """Handle a UPnP event."""
    obj = {
        "timestamp": get_timestamp(),
        "service_id": service.service_id,
        "service_type": service.service_type,
        "state_variables": {sv.name: sv.value for sv in service_variables},
    }
    global playing

    # special handling for DLNA LastChange state variable
    if len(service_variables) == 1 and service_variables[0].name == "LastChange":
        last_change = service_variables[0]
        dlna_handle_notify_last_change(last_change)
    else:
        for sv in service_variables:
            ### PAUSED, PLAYING, STOPPED, etc
            #print(sv.name,sv.value)
            if sv.name == "TransportState":
                print(sv.value)

            ### Grab and print the metadata
            if sv.name == "CurrentTrackMetaData" or sv.name == "AVTransportURIMetaData":
                ### Convert the grubby XML to beautiful JSON, because we HATE XML!
                items = xmltodict.parse(sv.value)["DIDL-Lite"]["item"]
                ### Print the entire mess
                print(json.dumps(items,indent=4))
                imgURL=items["upnp:albumArtURI"]
                urllib.request.urlretrieve(imgURL, "/root/ha/config/www/640.jpg")

async def subscribe(description_url: str, service_names: Any) -> None:
    """Subscribe to service(s) and output updates."""
    global event_handler  # pylint: disable=global-statement

    device = await create_device(description_url)

    # start notify server/event handler
    source = (get_local_ip(device.device_url), 0)
    server = AiohttpNotifyServer(device.requester, source=source)
    await server.async_start_server()

    # gather all wanted services
    if "*" in service_names:
        service_names = device.services.keys()

    services = []

    for service_name in service_names:
        service = service_from_device(device, service_name)
        if not service:
            print(f"Unknown service: {service_name}")
            sys.exit(1)
        service.on_event = on_event
        services.append(service)

    # subscribe to services
    event_handler = server.event_handler
    for service in services:
       try:
            await event_handler.async_subscribe(service)
       except UpnpResponseError as ex:
            print("Unable to subscribe to %s: %s", service, ex)

    s = 0
    while True:
        await asyncio.sleep(10)
        s = s + 1
        if s >= 12:
          await event_handler.async_resubscribe_all()
          s = 0

async def async_main() -> None:
    """Async main."""

    ####  NOTICE!!!! #####################################
    ####  Your WiiM Mini's IP and port go here
    device = "http://192.168.0.65:49152/description.xml"
    ####             #####################################
    service = ["AVTransport"]

    await subscribe(device, service)


def main() -> None:
    """Set up async loop and run the main program."""
    loop = asyncio.get_event_loop()

    try:
        loop.run_until_complete(async_main())
    except KeyboardInterrupt:
        if event_handler:
            loop.run_until_complete(event_handler.async_unsubscribe_all())
    finally:
        loop.close()


if __name__ == "__main__":
    main()

/etc/systemd/system
root@ha:/etc/systemd/system# cat cover.service
[Unit]
Description=Update cover art
After=network.target
StartLimitIntervalSec=33
StartLimitBurst=5

[Service]
ExecStart=/usr/local/bin/cover.py
StandardOutput=inherit
StandardError=inherit
User=ha

[Install]
WantedBy=default.target