141 lines
4.8 KiB
Python
141 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
# THIS IS AI-GENERATED, USE AT YOUR OWN RISK
|
|
# Simple script to turn on, set volume and source when Plexamp starts playing using Marantz API. Tested on a Marantz M-CR510.
|
|
# To be run on same machine that outputs audio (where Plexamp Headless is installed)
|
|
import glob
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import requests
|
|
|
|
# ================= CONFIG =================
|
|
RECEIVER_IP = "10.X.X.X"
|
|
INPUT_PLAYING = "SIAUXD"
|
|
INPUT_FALLBACK = "SISERVER" # optional fallback input when music stops
|
|
FALLBACK_BEHAVIOR = "nothing" # "standby" or "input" or "nothing"
|
|
CHECK_INTERVAL = 5 # seconds between ALSA checks
|
|
POWER_ON_TIMEOUT = 5 # seconds for power-on request
|
|
SOURCE_SWITCH_DELAY = 10 # seconds to wait after power-on before switching input
|
|
# Denon volume range
|
|
VOL_MIN_DB = -79.5
|
|
VOL_MAX_DB = 18.0
|
|
|
|
|
|
# ==========================================
|
|
|
|
def parse_status(file_path):
|
|
"""Parse ALSA PCM state."""
|
|
try:
|
|
with open(file_path, "r") as f:
|
|
content = f.read()
|
|
if "RUNNING" in content:
|
|
return "playing"
|
|
elif "PREPARED" in content:
|
|
return "paused"
|
|
elif "closed" in content:
|
|
return "off"
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def get_audio_state():
|
|
"""Return overall audio state based on ALSA status."""
|
|
states = []
|
|
for f in glob.glob("/proc/asound/card*/pcm*/sub*/status"):
|
|
state = parse_status(f)
|
|
if state:
|
|
states.append(state)
|
|
if "playing" in states:
|
|
return "playing"
|
|
if "paused" in states:
|
|
return "paused"
|
|
return "off"
|
|
|
|
|
|
def is_receiver_on():
|
|
"""Check if Marantz is powered on via /formMainZone_MainZoneXml.xml."""
|
|
try:
|
|
r = requests.get(f"http://{RECEIVER_IP}/goform/formMainZone_MainZoneXml.xml", timeout=5)
|
|
r.raise_for_status()
|
|
root = ET.fromstring(r.text)
|
|
power_node = root.find(".//Power/value")
|
|
if power_node is not None:
|
|
power_state = power_node.text.strip().upper()
|
|
print(f"Receiver power state: {power_state}")
|
|
return power_state != "STANDBY"
|
|
except Exception as e:
|
|
print("⚠️ Error checking receiver power:", e)
|
|
return False
|
|
|
|
|
|
def set_volume(decimal_volume: float):
|
|
"""
|
|
Set Denon volume.
|
|
decimal_volume: 0.0 = min, 1.0 = max
|
|
"""
|
|
decimal_volume = max(0.0, min(1.0, decimal_volume))
|
|
denon_db = decimal_volume * (VOL_MAX_DB - VOL_MIN_DB) + VOL_MIN_DB
|
|
try:
|
|
print(f"http://{RECEIVER_IP}/goform/formiPhoneAppVolume.xml?1+{denon_db:.0f}")
|
|
requests.get(f"http://{RECEIVER_IP}/goform/formiPhoneAppVolume.xml?1+{denon_db:.0f}", timeout=5)
|
|
print(f"➡️ Volume set to {denon_db:.2f} dB ({decimal_volume * 100:.0f}%)")
|
|
except requests.exceptions.RequestException as e:
|
|
print("⚠️ Error setting volume:", e)
|
|
|
|
|
|
def marantz_power_on():
|
|
if not is_receiver_on():
|
|
try:
|
|
requests.get(f"http://{RECEIVER_IP}/goform/formiPhoneAppPower.xml?1+PowerOn", timeout=POWER_ON_TIMEOUT)
|
|
print("➡️ Power on request sent")
|
|
time.sleep(SOURCE_SWITCH_DELAY)
|
|
except requests.exceptions.RequestException as e:
|
|
print("⚠️ Error powering on:", e)
|
|
# Always switch input after ensuring receiver is on
|
|
try:
|
|
requests.get(f"http://{RECEIVER_IP}/goform/formiPhoneAppInputSelect.xml?{INPUT_PLAYING}", timeout=5)
|
|
print(f"➡️ Input switched to {INPUT_PLAYING}")
|
|
except requests.exceptions.RequestException as e:
|
|
print("⚠️ Error switching input:", e)
|
|
|
|
|
|
def marantz_power_off():
|
|
try:
|
|
requests.get(f"http://{RECEIVER_IP}/goform/formiPhoneAppPower.xml?1+PowerStandby", timeout=5)
|
|
print("➡️ Power standby request sent")
|
|
except requests.exceptions.RequestException as e:
|
|
print("⚠️ Error sending power off:", e)
|
|
|
|
|
|
def marantz_fallback_input():
|
|
try:
|
|
requests.get(f"http://{RECEIVER_IP}/goform/formiPhoneAppInputSelect.xml?{INPUT_FALLBACK}", timeout=5)
|
|
print(f"➡️ Input switched to fallback {INPUT_FALLBACK}")
|
|
except requests.exceptions.RequestException as e:
|
|
print("⚠️ Error switching to fallback input:", e)
|
|
|
|
|
|
def main():
|
|
last_state = None
|
|
while True:
|
|
state = get_audio_state()
|
|
if state != last_state:
|
|
print(f"🔄 State changed: {last_state} -> {state}")
|
|
if state == "playing":
|
|
marantz_power_on()
|
|
set_volume(0.35)
|
|
elif state == "off":
|
|
set_volume(0.1)
|
|
if FALLBACK_BEHAVIOR == "standby":
|
|
marantz_power_off()
|
|
elif FALLBACK_BEHAVIOR == "input":
|
|
marantz_fallback_input()
|
|
# paused -> do nothing
|
|
last_state = state
|
|
time.sleep(CHECK_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|