HEOS Source Implementation Plan¶


Status	Draft - Revised after peer review
Author	Claude Code
Created	2025-01-09
Revised	2025-01-09 (incorporated GPT peer review feedback)
Related	music-display, music-control

Problem Statement¶

The music-display shows album art from Navidrome's "now playing" API, which relies on scrobbles. When multiple browser tabs have music-control open, each tab independently scrobbles every 30 seconds, causing the now-playing data to flip-flop between tracks.

Goal¶

Add a HEOS/Denon source to music-display that gets "now playing" info directly from the Denon receiver, eliminating the scrobble flip-flopping issue.

Current Architecture¶

config.json
    └── navidrome: { url, username, password }

display.py
    └── MusicWidget(config, source_config=navidrome_config)
            └── self.source = NavidromeSource(url, username, password)

NavidromeSource
    ├── get_now_playing() → dict | None
    ├── get_cover_art(cover_id, size) → PIL.Image | None
    └── ping() → bool

Proposed Architecture¶

config.json
    ├── music_source: "heos" | "navidrome"  (NEW - default: "navidrome")
    ├── heos: { host, port }                 (NEW)
    └── navidrome: { url, username, password }

display.py
    └── MusicWidget(config, source=<ISource>)  # Source injected based on config

sources/
    ├── base.py        (NEW) - Abstract interface
    ├── navidrome.py   (MODIFY) - Implement interface
    └── heos.py        (NEW) - HEOS + Navidrome for cover art

Implementation Steps¶

Step 1: Create Base Source Interface¶

File: sources/base.py

from abc import ABC, abstractmethod
from typing import Optional
from PIL import Image

class BaseSource(ABC):
    """Abstract base class for music sources.

    All sources must return tracks with these required keys:
    - id: str - Unique identifier for the track
    - title: str - Track title
    - artist: str - Artist name
    - album: str - Album name
    - cover_id: str - Stable identifier for cover art lookup (REQUIRED)
                      Format: "{source}:{artist}|{album}" e.g. "heos:Zedd|Clarity"
    """

    @abstractmethod
    def get_now_playing(self) -> Optional[dict]:
        """Get currently playing track.

        Returns:
            dict with required keys: id, title, artist, album, cover_id
            None if nothing playing or paused/stopped
        """
        pass

    @abstractmethod
    def get_cover_art(self, cover_id: str, size: int = 64) -> Optional[Image.Image]:
        """Fetch album cover art.

        Args:
            cover_id: Stable identifier from get_now_playing() - DO NOT rely on
                      internal state, use only the cover_id parameter
            size: Desired size in pixels

        Returns:
            PIL Image or None
        """
        pass

    @abstractmethod
    def ping(self) -> bool:
        """Test connection to source."""
        pass

Step 2: Update NavidromeSource¶

File: sources/navidrome.py

Add from sources.base import BaseSource
Change class NavidromeSource: to class NavidromeSource(BaseSource):
Add search_album(artist: str, album: str) -> Optional[dict] method for HEOS fallback

def search_album(self, artist: str, album: str) -> Optional[dict]:
    """Search for album by artist and album name."""
    query = f"{artist} {album}"
    resp = self._request("search3", {
        "query": query,
        "albumCount": "1",
        "songCount": "0",
        "artistCount": "0",
    })
    if resp and resp.get("status") == "ok":
        albums = resp.get("searchResult3", {}).get("album", [])
        if albums:
            return albums[0]
    return None

Step 3: Create HEOS Source¶

File: sources/heos.py

"""HEOS/Denon source - gets now playing directly from receiver."""

import logging
import socket
import json
import time
from typing import Optional
from PIL import Image

from sources.base import BaseSource
from sources.navidrome import NavidromeSource

logger = logging.getLogger("heos")


class HeosSource(BaseSource):
    """Gets now-playing from Denon/HEOS, cover art from Navidrome.

    Paused state behavior: Returns None (drops to idle widget).
    Only actively playing tracks are shown.
    """

    def __init__(
        self,
        host: str,
        port: int = 1255,
        player: Optional[str] = None,  # Player name or pid to use
        navidrome: Optional[NavidromeSource] = None,
        itunes_fallback: bool = True,  # Enable/disable iTunes API
    ):
        self.host = host
        self.port = port
        self.player_filter = player  # Name or pid string to match
        self.navidrome = navidrome
        self.itunes_fallback = itunes_fallback

        self.player_id: Optional[int] = None

        # Cover art caching (keyed by cover_id)
        self._cover_cache: dict[str, Optional[Image.Image]] = {}

        # Negative cache for iTunes failures (don't retry for 1 hour)
        self._itunes_failures: dict[str, float] = {}  # cover_id -> timestamp
        self._itunes_failure_ttl = 3600  # 1 hour

        # Store HEOS image URLs by cover_id for fallback
        self._heos_urls: dict[str, str] = {}

    def _send_command(self, command: str) -> Optional[dict]:
        """Send HEOS command and get response."""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            sock.connect((self.host, self.port))
            sock.sendall(f"heos://{command}\r\n".encode())

            data = b""
            while b"\r\n" not in data:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk

            sock.close()
            return json.loads(data.decode().strip())
        except Exception as e:
            logger.warning(f"HEOS connection error: {e}")
            return None

    def _get_player_id(self) -> Optional[int]:
        """Get player ID, optionally matching configured player name/pid."""
        if self.player_id:
            return self.player_id

        resp = self._send_command("player/get_players")
        if not resp or resp.get("heos", {}).get("result") != "success":
            return None

        players = resp.get("payload", [])
        if not players:
            return None

        # If player filter configured, try to match
        if self.player_filter:
            for p in players:
                # Match by name or pid
                if (p.get("name") == self.player_filter or
                    str(p.get("pid")) == self.player_filter):
                    self.player_id = p["pid"]
                    logger.info(f"Matched HEOS player: {p.get('name')} (pid={p['pid']})")
                    return self.player_id
            # No match - warn and fall back to first
            logger.warning(f"HEOS player '{self.player_filter}' not found, using first")

        # Default to first player
        self.player_id = players[0]["pid"]
        logger.info(f"Using HEOS player: {players[0].get('name')} (pid={self.player_id})")
        return self.player_id

    def ping(self) -> bool:
        """Test connection to HEOS."""
        return self._get_player_id() is not None

    def get_now_playing(self) -> Optional[dict]:
        """Get currently playing track from HEOS.

        Returns None if paused/stopped (display will show idle widget).
        """
        pid = self._get_player_id()
        if not pid:
            return None

        # Get now playing media (includes play state in some responses)
        resp = self._send_command(f"player/get_now_playing_media?pid={pid}")
        if not resp or resp.get("heos", {}).get("result") != "success":
            return None

        payload = resp.get("payload", {})
        if not payload or not payload.get("song"):
            return None

        # Check play state - only return if actively playing
        state_resp = self._send_command(f"player/get_play_state?pid={pid}")
        if state_resp:
            msg = state_resp.get("heos", {}).get("message", "")
            if "state=play" not in msg:
                logger.debug(f"HEOS not playing (state={msg})")
                return None

        artist = payload.get("artist", "Unknown")
        album = payload.get("album", "Unknown")

        # Generate stable cover_id for caching and art lookup
        cover_id = f"heos:{artist}|{album}"

        # Store HEOS image URL for this cover_id (used in get_cover_art)
        if payload.get("image_url"):
            self._heos_urls[cover_id] = payload["image_url"]

        track = {
            "id": payload.get("mid", ""),
            "title": payload.get("song", "Unknown"),
            "artist": artist,
            "album": album,
            "cover_id": cover_id,  # Stable identifier for cover art
        }

        return track

    def get_cover_art(self, cover_id: str, size: int = 64) -> Optional[Image.Image]:
        """Get cover art with multi-tier fallback.

        Uses cover_id parameter only - does NOT rely on internal _last_track state.
        This prevents race conditions when tracks change during art fetch.
        """
        # Check positive cache first
        if cover_id in self._cover_cache:
            cached = self._cover_cache[cover_id]
            if cached is not None:
                logger.debug(f"Cover art from cache: {cover_id}")
                return cached

        # Parse artist/album from cover_id (format: "heos:artist|album")
        try:
            _, artist_album = cover_id.split(":", 1)
            artist, album = artist_album.split("|", 1)
        except ValueError:
            logger.warning(f"Invalid cover_id format: {cover_id}")
            return None

        img = None

        # 1. Try HEOS image_url (may not be accessible)
        heos_url = self._heos_urls.get(cover_id)
        if heos_url:
            img = self._try_fetch_url(heos_url)
            if img:
                logger.info(f"Cover art from HEOS: {artist} - {album}")
                img = img.resize((size, size), Image.Resampling.LANCZOS)
            else:
                logger.debug(f"HEOS image_url failed: {heos_url}")

        # 2. Try Navidrome album search
        if not img and self.navidrome:
            album_info = self.navidrome.search_album(artist, album)
            if album_info and album_info.get("coverArt"):
                img = self.navidrome.get_cover_art(album_info["coverArt"], size)
                if img:
                    logger.info(f"Cover art from Navidrome: {artist} - {album}")
            else:
                # Not in Navidrome - log for discovery
                self._log_discovery(artist, album)

        # 3. Try iTunes Search API (if enabled and not in negative cache)
        if not img and self.itunes_fallback:
            if not self._is_itunes_failed(cover_id):
                img = self._search_itunes_art(artist, album, size)
                if img:
                    logger.info(f"Cover art from iTunes: {artist} - {album}")
                else:
                    # Add to negative cache
                    self._itunes_failures[cover_id] = time.time()
                    logger.debug(f"iTunes lookup failed, cached: {artist} - {album}")

        if not img:
            logger.warning(f"No cover art found: {artist} - {album}")

        # Cache result (even None to avoid repeated lookups)
        self._cover_cache[cover_id] = img

        # Limit cache size
        if len(self._cover_cache) > 50:
            oldest = next(iter(self._cover_cache))
            del self._cover_cache[oldest]

        return img

    def _is_itunes_failed(self, cover_id: str) -> bool:
        """Check if cover_id is in iTunes negative cache."""
        if cover_id not in self._itunes_failures:
            return False
        age = time.time() - self._itunes_failures[cover_id]
        if age > self._itunes_failure_ttl:
            del self._itunes_failures[cover_id]
            return False
        return True

    def _search_itunes_art(self, artist: str, album: str, size: int) -> Optional[Image.Image]:
        """Search iTunes for album art (free, no API key)."""
        try:
            import requests
            from io import BytesIO
            from urllib.parse import quote

            query = quote(f"{artist} {album}")
            url = f"https://itunes.apple.com/search?term={query}&entity=album&limit=1"

            resp = requests.get(url, timeout=5)
            if resp.status_code != 200:
                return None

            data = resp.json()
            results = data.get("results", [])
            if not results:
                return None

            art_url = results[0].get("artworkUrl100", "")
            if not art_url:
                return None

            # Replace 100x100 with desired size
            art_url = art_url.replace("100x100", f"{min(size * 2, 600)}x{min(size * 2, 600)}")

            img_resp = requests.get(art_url, timeout=5)
            if img_resp.status_code == 200:
                img = Image.open(BytesIO(img_resp.content))
                return img.resize((size, size), Image.Resampling.LANCZOS)

        except Exception as e:
            logger.debug(f"iTunes art search error: {e}")

        return None

    def _try_fetch_url(self, url: str) -> Optional[Image.Image]:
        """Try to fetch image from URL."""
        try:
            import requests
            from io import BytesIO
            resp = requests.get(url, timeout=3)
            if resp.status_code == 200 and "image" in resp.headers.get("content-type", ""):
                return Image.open(BytesIO(resp.content))
        except Exception:
            pass
        return None

    def _log_discovery(self, artist: str, album: str):
        """Log tracks not found in Navidrome (implementation in Logging section)."""
        pass  # See Logging & Discovery section

Step 4: Update sources/init.py¶

"""Data sources for music-display widgets."""

from sources.base import BaseSource
from sources.mbta import MBTASource, MBTA_COLORS
from sources.navidrome import NavidromeSource
from sources.heos import HeosSource

__all__ = ["BaseSource", "MBTASource", "MBTA_COLORS", "NavidromeSource", "HeosSource"]

Step 5: Update display.py¶

Modify source creation based on config:

# In main():

# Note: config is a Config wrapper object with .get(section, key, default=) method
# and .data dict for direct access

# Determine music source type
music_config = config.data.get("music", {})
music_source_type = music_config.get("source", "navidrome")

# Always create Navidrome source (needed for cover art fallback)
navidrome_config = config.data.get("navidrome", {})
navidrome_source = NavidromeSource(
    url=navidrome_config.get("url", ""),
    username=navidrome_config.get("username", ""),
    password=navidrome_config.get("password", ""),
)

# Create the active source
if music_source_type == "heos":
    heos_config = config.data.get("heos", {})
    cover_config = music_config.get("cover_fallback", {})

    music_source = HeosSource(
        host=heos_config.get("host", "10.89.97.15"),
        port=int(heos_config.get("port", 1255)),
        player=heos_config.get("player"),  # Optional: name or pid
        navidrome=navidrome_source,
        itunes_fallback=cover_config.get("itunes", True),
    )
    print(f"Music source: HEOS at {heos_config.get('host', '10.89.97.15')}")
else:
    music_source = navidrome_source
    print("Music source: Navidrome")

# Create music widget with the source
music_widget = MusicWidget(
    config={...},
    source=music_source,  # Pass source directly instead of source_config
)

Step 6: Update MusicWidget¶

File: widgets/music.py

Change constructor to accept source directly:

# Before:
def __init__(self, config: dict, source_config: dict):
    super().__init__(config)
    self.source = NavidromeSource(
        url=source_config.get("url", ""),
        username=source_config.get("username", ""),
        password=source_config.get("password", ""),
    )

# After:
def __init__(self, config: dict, source: BaseSource):
    super().__init__(config)
    self.source = source

Step 7: Update config.example.json¶

{
  "music": {
    "source": "heos",
    "show_text": true,
    "text_position": "bottom",
    "text_background": "gradient",
    "text_style": "popup",
    "popup_duration": 8,
    "cover_fallback": {
      "itunes": true
    }
  },
  "heos": {
    "host": "10.89.97.15",
    "port": 1255,
    "player": "Denon AVR-X1700H"
  },
  "navidrome": {
    "url": "http://your-navidrome-server:4533",
    "username": "your-username",
    "password": "your-password"
  },
  "display": {
    "brightness": 60,
    "rotation": 90,
    "idle_mode": "clock"
  }
}

New config options: - music.cover_fallback.itunes - Enable/disable iTunes API fallback (default: true) - heos.player - Optional player name or pid to use (defaults to first player)

Step 8: Update Web UI¶

Add source selection to web.py settings:

<div class="flex items-center justify-between">
  <label class="text-sm font-medium">Music Source</label>
  <select name="music_source" class="...">
    <option value="navidrome" ${source === 'navidrome' ? 'selected' : ''}>Navidrome (user activity)</option>
    <option value="heos" ${source === 'heos' ? 'selected' : ''}>HEOS/Denon (receiver)</option>
  </select>
</div>

Testing Plan¶

Unit test HeosSource with mocked socket responses
Mock _send_command() to return canned HEOS responses
Verify cover_id generation is stable
Verify caching works correctly
Integration test on Pi:
Set music.source: "heos" in config
Play music via any source (DLNA, Navidrome, etc.)
Verify display shows correct track
Verify cover art loads (via Navidrome fallback)
Paused state behavior
Play a track, verify it displays
Pause playback, verify display drops to idle widget
Resume playback, verify track reappears
Decision: Paused = idle (not showing last track static)
Cover art race condition prevention
Rapidly switch tracks (skip 5+ times quickly)
Verify displayed art always matches the title/artist shown
This validates that cover_id-based lookup prevents races
Player selection
Configure heos.player: "Wrong Name", verify warning logged and fallback to first
Configure correct name, verify match logged
Negative caching
Play track not on iTunes (obscure artist)
Verify iTunes lookup attempted once
Verify subsequent refreshes don't retry (check logs)
Wait >1 hour (or adjust TTL for test), verify retry
Regression test Navidrome source still works
Test switching between sources via web UI

Rollback Plan¶

If issues arise, simply change config back to:

{ "music": { "source": "navidrome" } }

Files Changed¶

File	Action
`sources/base.py`	CREATE
`sources/heos.py`	CREATE
`sources/navidrome.py`	MODIFY (add BaseSource, search_album)
`sources/__init__.py`	MODIFY (export new classes)
`widgets/music.py`	MODIFY (accept source in constructor)
`display.py`	MODIFY (source factory based on config)
`web.py`	MODIFY (add source selector to UI)
`config.example.json`	MODIFY (add heos config)

Benefits¶

Authoritative source: HEOS knows exactly what's playing on the receiver
No scrobble confusion: Doesn't matter how many tabs/clients are open
Source agnostic: Works with DLNA, Spotify, Navidrome, etc.
Backwards compatible: Default is still Navidrome source
Cover art reliability: Multi-tier fallback (HEOS → Navidrome → iTunes)

Performance Considerations¶

Cover Art Caching (implemented in HeosSource above): - Positive cache: _cover_cache[cover_id] stores fetched images - Negative cache: _itunes_failures[cover_id] tracks failed iTunes lookups (1hr TTL) - Cache keyed by stable cover_id (e.g., heos:Zedd|Clarity), not mutable state - LRU eviction at 50 entries

Socket Efficiency: - Current design opens/closes TCP socket per command (simple, reliable) - At 5-second polling interval, this is acceptable - Future optimization: persistent connection with reconnect logic (if needed)

Logging & Discovery¶

Cover Art Source Logging¶

Logging is implemented in HeosSource.get_cover_art() (see Step 3 above):

INFO  heos: Cover art from HEOS: Zedd - Clarity
INFO  heos: Cover art from Navidrome: Arcade Fire - Funeral
INFO  heos: Cover art from iTunes: NewJeans - Get Up
WARN  heos: No cover art found: Obscure Artist - Unknown Album
DEBUG heos: HEOS image_url failed: http://10.89.97.168:8015/...
DEBUG heos: iTunes lookup failed, cached: Obscure Artist - Unknown Album

Music Discovery Log¶

Track music played on HEOS that isn't in your Navidrome library:

File: discovery.json (auto-created in config dir)

[
  {
    "timestamp": "2024-01-15T20:30:00",
    "artist": "NewJeans",
    "album": "Get Up",
    "title": "Super Shy",
    "source": "spotify",
    "cover_source": "itunes"
  }
]

Implementation:

import json
import os
from datetime import datetime

DISCOVERY_LOG = os.path.join(os.path.dirname(CONFIG_PATH), "discovery.json")

def _log_discovery(self, artist: str, album: str, title: str):
    """Log tracks not found in Navidrome for music discovery."""
    if not artist or not album:
        return

    entry = {
        "timestamp": datetime.now().isoformat(),
        "artist": artist,
        "album": album,
        "title": title,
    }

    # Load existing
    try:
        with open(DISCOVERY_LOG, "r") as f:
            discoveries = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        discoveries = []

    # Dedupe by artist+album (only log each album once)
    key = f"{artist}|{album}"
    existing_keys = {f"{d['artist']}|{d['album']}" for d in discoveries}

    if key not in existing_keys:
        discoveries.append(entry)
        logger.info(f"Discovery logged: {artist} - {album}")

        # Write back (keep last 500 entries)
        with open(DISCOVERY_LOG, "w") as f:
            json.dump(discoveries[-500:], f, indent=2)

Web UI Discovery View (Future Enhancement)¶

Could add a /discoveries page to web.py showing: - List of artists/albums played but not in Navidrome - Links to add them (Navidrome search, Spotify, etc.) - "Dismiss" button to ignore

This turns the display into a passive music discovery tool!

Peer Review Feedback (2025-01-09)¶

External peer review identified the following issues, which have been incorporated:

Fixed Issues¶

Issue	Fix Applied
Race condition: `get_cover_art()` relied on mutable `_last_track` state	Now uses stable `cover_id` parameter; parses artist/album from it
Interface mismatch: `cover_id` parameter was ignored by HeosSource	`get_now_playing()` now returns `cover_id: "heos:{artist}\\|{album}"`
"First player wins": Would break with multiple HEOS zones	Added `heos.player` config option (name or pid)
Config access patterns: Pseudocode used invalid Python syntax	Fixed to use proper nested `.get()` calls
iTunes has no kill-switch: External API with no disable option	Added `music.cover_fallback.itunes` config option
No negative caching: Failed iTunes lookups would retry every refresh	Added `_itunes_failures` cache with 1-hour TTL
Paused behavior undocumented: Unclear what happens when paused	Documented: paused = idle (drops to idle widget)

Deferred Issues¶

Issue	Rationale
Socket per call: Opens/closes TCP socket for each HEOS command	Acceptable at 5-second polling; optimize later if needed
Track dataclass: Suggested using `@dataclass` instead of dicts	Over-engineering; documented required keys instead
print() vs logging: Codebase uses print() throughout	Out of scope; HeosSource uses logging, rest is separate refactor

Additional Test Cases Added¶

Paused state behavior verification
Cover art race condition prevention (rapid track switching)
Player selection matching and fallback
Negative caching validation