bhetherman/spotify-voice-assistant
1
0 Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Update documentation for v1.0.0 release

Browse Source 
Major documentation updates:
- Streamline README by removing redundant sections
- Link to example files instead of duplicating configuration
- Fix typos and update model recommendations
- Add GPU acceleration documentation link
- Remove outdated "Prompt Optimization" section

Test suite improvements:
- Add comprehensive queue management tests (5 new tests)
- Add shuffle control tests (3 new tests)
- Fix combined query references to match current system prompt
- Update total test count to 53
- Renumber all test sections for consistency
- TODO refresh
This commit is contained in:
Chad Auld committed 2025-11-20 18:14:06 -07:00
1 parent db0303d8a9
commit 4d477fa218
8 files changed
+506 -589
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

custom_components/spotify_search/__init__.py
+108 -122
View File
@@ -1,7 +1,7 @@
"""Spotify Search Integration for Home Assistant."""
import logging
import re
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -17,26 +17,47 @@ _spotify_cache = {
}
def clean_query(query: str, search_type: str) -> str:
"""Remove common command words to improve search accuracy."""
query = query.lower().strip()
# Remove "play" from the start of any query
if query.startswith("play "):
query = query[5:]
# Remove type-specific filler words
if search_type == "artist":
# Remove "artist" prefix if LLM included it
query = query.replace("artist ", "").replace(
"group ", "").replace("band ", "")
elif search_type == "album":
# Remove "album" prefix
query = query.replace("album ", "")
elif search_type == "track":
# Remove "song" or "track" prefix
query = query.replace("song ", "").replace("track ", "")
return " ".join(query.split()).strip()
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Spotify Search component."""
async def get_spotify_client():
"""Get Spotify client with caching and validation."""
# Validate cache if it exists
if _spotify_cache["client"] is not None:
# Fast check: Does entity still exist?
if _spotify_cache["entity_id"] in hass.states.async_entity_ids():
_LOGGER.debug("Using cached Spotify client")
return _spotify_cache["client"]
else:
_LOGGER.info("Cached Spotify entity no longer exists, invalidating cache")
_LOGGER.info(
"Cached Spotify entity no longer exists, invalidating cache")
_spotify_cache["client"] = None
_spotify_cache["entity_id"] = None
_spotify_cache["user_playlists"] = None
# Cache miss or invalidated - do full lookup
_LOGGER.debug("Cache miss, performing Spotify entity lookup")
# Try to find the Spotify media player entity
spotify_entity_id = None
for state in hass.states.async_all("media_player"):
if "spotify" in state.entity_id.lower():
@@ -47,15 +68,11 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
_LOGGER.error("No Spotify media player entity found")
raise LookupError("Spotify not configured")
_LOGGER.debug("Found Spotify entity: %s", spotify_entity_id)
# Get the Spotify integration's data through entity platform
entity_component = hass.data.get("entity_components", {}).get("media_player")
entity_component = hass.data.get(
"entity_components", {}).get("media_player")
if not entity_component:
_LOGGER.error("Media player component not found")
raise LookupError("Media player component not available")
# Find the Spotify entity object
spotify_entity = None
for entity in entity_component.entities:
if entity.entity_id == spotify_entity_id:
@@ -63,24 +80,16 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
break
if not spotify_entity:
_LOGGER.error("Spotify entity %s not found in entities", spotify_entity_id)
raise LookupError("Spotify entity not available")
# Access the Spotify client from the coordinator
if not hasattr(spotify_entity, "coordinator"):
_LOGGER.error("Spotify entity does not have a coordinator")
raise AttributeError("Spotify coordinator not available")
coordinator = spotify_entity.coordinator
if not hasattr(coordinator, "client"):
_LOGGER.error("Spotify coordinator does not have a client attribute")
raise AttributeError("Spotify client not available")
client = coordinator.client
_LOGGER.debug("Spotify client type: %s", type(client).__name__)
# Cache for future calls
_spotify_cache["client"] = client
_spotify_cache["entity_id"] = spotify_entity_id
_LOGGER.info("Cached Spotify client for entity: %s", spotify_entity_id)
@@ -89,18 +98,21 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def search_spotify(call: ServiceCall):
"""Search Spotify and return the first result's URI."""
query = call.data.get("query")
raw_query = call.data.get("query")
search_type = call.data.get("type", "artist")
if not query:
if not raw_query:
_LOGGER.error("No query provided to spotify_search")
return {"error": "No query provided"}
# Validate search type
if search_type not in VALID_SEARCH_TYPES:
_LOGGER.error("Invalid search type: %s", search_type)
return {"error": f"Invalid type. Must be one of: {', '.join(VALID_SEARCH_TYPES)}"}
# --- UPDATE: Clean the query before sending to Spotify ---
query = clean_query(raw_query, search_type)
_LOGGER.info("Searching Spotify (%s) for cleaned query: '%s' (raw: '%s')",
search_type, query, raw_query)
try:
client = await get_spotify_client()
except (LookupError, AttributeError) as err:
@@ -108,156 +120,131 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return {"error": str(err)}
try:
# Search Spotify using the integration's client (async method)
# SpotifyClient.search signature: search(query: str, types: list[SearchType], *, limit: int = 48)
if search_type == "artist":
# For artists, search using the artist type directly to get better matches
_LOGGER.debug("Searching for artist: %s", query)
results = await client.search(query, ["artist"], limit=10)
items_list = results.artists
if items_list and len(items_list) > 0:
# Check if any artist name matches exactly (case-insensitive)
exact_match = None
query_lower = query.lower()
for artist in items_list:
# Defensive attribute check
if hasattr(artist, "name") and artist.name.lower() == query_lower:
exact_match = artist
break
# Use exact match if found, otherwise use first result
selected_artist = exact_match if exact_match else items_list[0]
# Defensive attribute checks
if not hasattr(selected_artist, "uri") or not hasattr(selected_artist, "name"):
_LOGGER.error("Artist result missing required attributes")
if not hasattr(selected_artist, "uri"):
return {"error": "Invalid artist data from Spotify"}
uri = selected_artist.uri
name = selected_artist.name
match_type = "exact match" if exact_match else "first result"
_LOGGER.info("Found Spotify artist: %s (%s) - %s", name, uri, match_type)
return {"uri": uri, "name": name, "type": "artist"}
result = {"uri": uri, "name": name, "type": "artist"}
_LOGGER.info("✅ SEARCH RESULT (%s): %s",
match_type, result)
return result
else:
_LOGGER.warning("No artists found for query: %s", query)
return {"error": f"No artist found for: {query}"}
error_result = {"error": f"No artist found for: {query}"}
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
return error_result
elif search_type == "playlist":
# For playlists, search user's playlists first, then fall back to Spotify
_LOGGER.info("Searching for playlist: %s", query)
# Cleaning is already handled by clean_query logic above,
# but we keep the specific 'playlist' word removal for safety
query_cleaned = query.lower().replace(
"playlist", "").replace("playlists", "").strip()
# Clean query: remove "playlist" and "playlists" from search term
query_cleaned = query.lower()
for word in ["playlist", "playlists"]:
query_cleaned = query_cleaned.replace(word, "")
query_cleaned = " ".join(query_cleaned.split()).strip() # Remove extra spaces
_LOGGER.info("Cleaned query: '%s' (original: '%s')", query_cleaned, query)
# Step 1: Search user's personal playlists first
# 1. Search User Library
try:
# Get user's playlists (with caching)
if _spotify_cache["user_playlists"] is None:
_LOGGER.info("Cache miss, fetching user playlists from Spotify API")
user_playlists_response = await client.get_playlists_for_current_user()
if user_playlists_response and hasattr(user_playlists_response, "items"):
items_list = user_playlists_response.items
_spotify_cache["user_playlists"] = items_list
_LOGGER.info("User playlists fetched and cached: %d playlists", len(items_list) if items_list else 0)
_spotify_cache["user_playlists"] = user_playlists_response.items
else:
_LOGGER.warning("Could not fetch user playlists")
_spotify_cache["user_playlists"] = []
else:
_LOGGER.info("Using cached user playlists")
user_playlists = _spotify_cache["user_playlists"]
if user_playlists and len(user_playlists) > 0:
playlist_names = [p.name if hasattr(p, "name") else "NO_NAME" for p in user_playlists]
_LOGGER.info("User's playlist names: %s", playlist_names)
# Search for exact match
for playlist in user_playlists:
if hasattr(playlist, "name") and playlist.name.lower() == query_cleaned:
uri = playlist.uri if hasattr(playlist, "uri") else None
name = playlist.name
if uri:
_LOGGER.info("Found in user playlists (exact match): %s (%s)", name, uri)
return {"uri": uri, "name": name, "type": "playlist"}
# Exact match in library
for playlist in user_playlists:
if hasattr(playlist, "name") and playlist.name.lower() == query_cleaned:
result = {"uri": playlist.uri,
"name": playlist.name, "type": "playlist"}
_LOGGER.info(
"✅ SEARCH RESULT (user library - exact match): %s", result)
return result
# Search for partial match
for playlist in user_playlists:
if hasattr(playlist, "name") and query_cleaned in playlist.name.lower():
uri = playlist.uri if hasattr(playlist, "uri") else None
name = playlist.name
if uri:
_LOGGER.info("Found in user playlists (partial match): %s (%s)", name, uri)
return {"uri": uri, "name": name, "type": "playlist"}
# Partial match in library
for playlist in user_playlists:
if hasattr(playlist, "name") and query_cleaned in playlist.name.lower():
result = {"uri": playlist.uri,
"name": playlist.name, "type": "playlist"}
_LOGGER.info(
"✅ SEARCH RESULT (user library - partial match): %s", result)
return result
_LOGGER.info("Playlist not found in user's library, searching Spotify public playlists")
except Exception as err:
_LOGGER.warning("Error searching user playlists: %s, falling back to Spotify search", err)
_LOGGER.warning("Error searching user playlists: %s", err)
# Step 2: Fall back to Spotify public playlist search
_LOGGER.info("Searching Spotify public playlists for: %s", query_cleaned)
# 2. Fallback to Public Search
results = await client.search(query_cleaned, ["playlist"], limit=10)
items_list = results.playlists
if items_list and len(items_list) > 0:
# Check if any playlist name matches exactly (case-insensitive)
exact_match = None
for playlist in items_list:
if hasattr(playlist, "name") and playlist.name.lower() == query_cleaned:
exact_match = playlist
break
# Use exact match if found, otherwise use first result
selected_playlist = exact_match if exact_match else items_list[0]
# Defensive attribute checks
if not hasattr(selected_playlist, "uri") or not hasattr(selected_playlist, "name"):
_LOGGER.error("Playlist result missing required attributes")
return {"error": "Invalid playlist data from Spotify"}
uri = selected_playlist.uri
name = selected_playlist.name
match_type = "exact match" if exact_match else "first result"
_LOGGER.info("Found Spotify public playlist: %s (%s) - %s", name, uri, match_type)
return {"uri": uri, "name": name, "type": "playlist"}
selected_playlist = items_list[0]
result = {"uri": selected_playlist.uri,
"name": selected_playlist.name, "type": "playlist"}
_LOGGER.info(
"✅ SEARCH RESULT (public playlist - first result): %s", result)
return result
else:
_LOGGER.warning("No playlists found for query: %s", query_cleaned)
return {"error": f"No playlist found for: {query}"}
error_result = {"error": f"No playlist found for: {query}"}
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
return error_result
else:
# For albums, tracks - search with higher limit and check for exact matches
_LOGGER.debug("Searching for %s: %s", search_type, query)
# Handle Album and Track
results = await client.search(query, [search_type], limit=10)
items_list = getattr(results, f"{search_type}s", None)
if items_list and len(items_list) > 0:
# Check if any item name matches exactly (case-insensitive)
exact_match = None
query_lower = query.lower()
for item in items_list:
# Defensive attribute check
if hasattr(item, "name") and item.name.lower() == query_lower:
exact_match = item
break
# Use exact match if found, otherwise use first result
# Fallback logic for Albums
if not exact_match and search_type == "album" and len(query.split()) >= 2:
_LOGGER.info(
"No exact album match for '%s', trying track search", query)
try:
track_results = await client.search(query, ["track"], limit=10)
track_items = getattr(
track_results, "tracks", None)
if track_items and len(track_items) > 0:
first_track = track_items[0]
result = {"uri": first_track.uri,
"name": first_track.name, "type": "track"}
_LOGGER.info(
"✅ SEARCH RESULT (album→track fallback): %s", result)
return result
except Exception:
pass
selected_item = exact_match if exact_match else items_list[0]
# Defensive attribute checks
if not hasattr(selected_item, "uri") or not hasattr(selected_item, "name"):
_LOGGER.error("%s result missing required attributes", search_type)
return {"error": f"Invalid {search_type} data from Spotify"}
uri = selected_item.uri
name = selected_item.name
match_type = "exact match" if exact_match else "first result"
_LOGGER.info("Found Spotify %s: %s (%s) - %s", search_type, name, uri, match_type)
return {"uri": uri, "name": name, "type": search_type}
result = {"uri": selected_item.uri,
"name": selected_item.name, "type": search_type}
_LOGGER.info("✅ SEARCH RESULT (%s - %s): %s",
search_type, match_type, result)
return result
else:
_LOGGER.warning("No results found for query: %s", query)
return {"error": f"No {search_type} found for: {query}"}
error_result = {
"error": f"No {search_type} found for: {query}"}
_LOGGER.error("❌ SEARCH ERROR: %s", error_result)
return error_result
except AttributeError as err:
_LOGGER.error("Spotify API returned unexpected data structure: %s", err)
return {"error": "Unexpected response from Spotify"}
except Exception as err:
_LOGGER.exception("Unexpected error searching Spotify")
return {"error": "Search failed"}
@@ -265,7 +252,6 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def clear_cache(call: ServiceCall):
"""Clear Spotify client and user playlists cache."""
if _spotify_cache["client"] is not None or _spotify_cache["user_playlists"] is not None:
_LOGGER.info("Manually clearing Spotify cache")
_spotify_cache["client"] = None
_spotify_cache["entity_id"] = None
_spotify_cache["user_playlists"] = None