diff --git a/scripts/marantz-watcher.py b/scripts/marantz-watcher.py new file mode 100644 index 0000000..d37dffc --- /dev/null +++ b/scripts/marantz-watcher.py @@ -0,0 +1,140 @@ +#!/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()