Last active
August 20, 2024 06:52
-
-
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)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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 |
Author
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()
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
Uh oh!
There was an error while loading. Please reload this page.