Skip to content

Instantly share code, notes, and snippets.

@miklos-szel
Last active August 20, 2024 06:52
Show Gist options
  • Select an option

  • Save miklos-szel/5eddf5f0882e502d6a8ad7f290812ba9 to your computer and use it in GitHub Desktop.

Select an option

Save miklos-szel/5eddf5f0882e502d6a8ad7f290812ba9 to your computer and use it in GitHub Desktop.
Wiim soft trigger for amps without 12v trigger (with Tasmota power plug)
#!/bin/bash
## sudo apt install python3-pip
##
# URL and command details
URL="http://192.168.0.65:49152/description.xml"
COMMAND="/usr/local/bin/upnp-client --pprint call-action $URL AVTransport/GetInfoEx InstanceID=0"
POWER_URL="/usr/bin/wget -O /dev/null -o /dev/null http://192.168.0.202/cm?cmnd=Power%20"
INTERVAL=5
# Initialize previous status
previous_status=""
delay_stop=0
# Function to check the status
check_status() {
# Execute the command and capture the output
output=$($COMMAND 2>&1)
# Extract the CurrentTransportState value
current_status=$(echo "$output" | grep -oP '(?<="CurrentTransportState": ")[^"]+')
# Check if the current status is different from the previous status
echo "curr: $current_status" |systemd-cat -t wiim -p info
echo "prev: $previous_status" |systemd-cat -t wiim -p info
if [ "$current_status" != "$previous_status" ]; then
# Update the previous status
# If the current status is "PLAYING", execute the specific command
if [ "$current_status" == "PLAYING" ]; then
echo "Turning on the Amp" |systemd-cat -t wiim -p info
$(${POWER_URL}ON)
previous_status="$current_status"
delay_stop=0
else
#only turn off the power after 5 (60 x 5 sec) minutes without activity
delay_stop=$((delay_stop+1))
echo "delay_stop $delay_stop" | systemd-cat -t wiim -p info
previous_status=""
if [ $delay_stop -eq 60 ]; then
echo "Turning off the Amp" |systemd-cat -t wiim -p info
$(${POWER_URL}OFF)
fi
fi
fi
}
# Infinite loop to run the command every 5 seconds
while true; do
check_status
sleep $INTERVAL
done
@miklos-szel
Copy link
Author

miklos-szel commented May 20, 2024

cat /etc/systemd/system/wiim.service
[Unit]
Description=Wiim power manager
After=network.target
[Service]
ExecStart=/usr/local/bin/wiim_power_control.sh
[Install]
WantedBy=default.target

sudo systemctl daemon-reload
sudo systemctl enable wiim.service
sudo systemctl start  wiim.service

@miklos-szel
Copy link
Author

miklos-szel commented Aug 20, 2024

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()

@miklos-szel
Copy link
Author

/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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment