Guys how are you! I wanted to share an integration that I’ve been working on. It’s Called JamsPy, a music recognition integration that uses an external microphone that listens for music and displays it in home assistant. Like Shazam!
The biggest issue I’m having is entities, and I need some big brain’s to help me figure it out. I can share my current code for the integration to see if someone here could help me out.
You’ve shared nothing.
init.py
"""JamSpy Music Recognition Integration."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.const import Platform
_LOGGER = logging.getLogger(__name__)
DOMAIN = "jamspy"
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up JamSpy from a config entry."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {}
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Register service
async def handle_recognize(call):
"""Handle the recognize service call."""
# Get all jamspy sensors
sensors = [
entity for entity in hass.data["entity_registry"].entities.values()
if entity.platform == DOMAIN and entity.domain == "sensor"
]
for sensor_entry in sensors:
sensor = hass.data["entity_registry"].entities[sensor_entry.entity_id]
if sensor:
await hass.services.async_call(
"homeassistant", "update_entity",
{"entity_id": sensor_entry.entity_id},
blocking=True
)
hass.services.async_register(DOMAIN, "recognize", handle_recognize)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
config_flow.py
"""Config flow for JamSpy."""
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
DOMAIN = "jamspy"
class JamSpyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for JamSpy."""
VERSION = 1
async def async_step_user(self, user_input=None) -> FlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
# Check if already configured
await self.async_set_unique_id("jamspy_main")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="JamSpy Music Recognition",
data=user_input
)
schema = vol.Schema({
vol.Optional("name", default="JamSpy"): str,
})
return self.async_show_form(
step_id="user",
data_schema=schema,
errors=errors
)
manifest.json
{
"domain": "jamspy",
"name": "JamSpy Music Recognition",
"version": "1.0.0",
"documentation": "https://github.com/yourname/jamspy",
"dependencies": [],
"codeowners": ["@yourgithub"],
"requirements": ["aiohttp"],
"config_flow": false
}
sensor.py
"""JamSpy Music Recognition Sensor using FFmpeg + PulseAudio."""
import logging
import asyncio
import aiohttp
import tempfile
import os
from datetime import datetime
from typing import Optional, Dict, Any
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
_LOGGER = logging.getLogger(__name__)
AUDD_API_TOKEN = "9ca714cbb140117074722ff227ebb537"
DOMAIN = "jamspy"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
"""Set up JamSpy sensor from a config entry."""
sensor = JamSpySensor()
async_add_entities([sensor], True)
hass.data.setdefault(DOMAIN, {})["sensor"] = sensor
class JamSpySensor(SensorEntity):
"""JamSpy Music Recognition Sensor."""
def __init__(self):
self._attr_name = "JamSpy Music Recognition"
self._attr_unique_id = "jamspy_music_recognition"
self._attr_icon = "mdi:music-note"
self._attr_native_value = "Ready"
self._attr_extra_state_attributes = {
"title": None,
"artist": None,
"album": None,
"last_recognition": None,
"recording_method": "ffmpeg-pulse"
}
self._added_to_hass = False
async def async_added_to_hass(self):
"""Mark that the entity has been added and is ready to write state."""
self._added_to_hass = True
async def async_update(self):
await self.recognize_music()
async def recognize_music(self, duration: int = 8):
try:
_LOGGER.info(f"JamSpy: Starting {duration}s recording")
self._attr_native_value = "Recording..."
await self._safe_write_ha_state()
audio_data = await self._record_with_ffmpeg(duration)
if audio_data:
self._attr_native_value = "Recognizing..."
await self._safe_write_ha_state()
result = await self._recognize_audio(audio_data)
await self._update_state(result)
else:
self._attr_native_value = "Recording failed"
await self._safe_write_ha_state()
except Exception as e:
_LOGGER.error(f"JamSpy: Recognition failed: {e}")
self._attr_native_value = f"Error: {str(e)}"
await self._safe_write_ha_state()
async def _safe_write_ha_state(self):
"""Safely write state only after being added to Home Assistant."""
if self._added_to_hass:
self.async_write_ha_state()
async def _record_with_ffmpeg(self, duration: int) -> Optional[bytes]:
try:
temp_file = f"/tmp/jamspy_{datetime.now().timestamp()}.wav"
cmd = [
"ffmpeg",
"-f", "pulse",
"-i", "alsa_input.usb-C-Media_Electronics_Inc._USB_PnP_Sound_Device-00.mono-fallback",
"-t", str(duration),
"-acodec", "pcm_s16le",
"-ar", "44100",
"-ac", "1",
temp_file,
"-y"
]
_LOGGER.info(f"JamSpy: Recording with: {' '.join(cmd)}")
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
_, stderr = await process.communicate()
if process.returncode == 0 and os.path.exists(temp_file):
with open(temp_file, 'rb') as f:
audio_data = f.read()
os.unlink(temp_file)
return audio_data
else:
_LOGGER.error(f"JamSpy: FFmpeg failed: {stderr.decode()}")
return None
except Exception as e:
_LOGGER.error(f"JamSpy: Recording error: {e}")
return None
async def _recognize_audio(self, audio_data: bytes) -> Optional[Dict[str, Any]]:
try:
_LOGGER.info("JamSpy: Sending to Audd.io for recognition")
async with aiohttp.ClientSession() as session:
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
temp_file.write(audio_data)
temp_file.flush()
data = aiohttp.FormData()
data.add_field('api_token', AUDD_API_TOKEN)
data.add_field('return', 'spotify,apple_music')
with open(temp_file.name, 'rb') as f:
data.add_field('file', f, filename='audio.wav', content_type='audio/wav')
async with session.post("https://api.audd.io/", data=data) as response:
os.unlink(temp_file.name)
if response.status == 200:
return await response.json()
_LOGGER.error(f"JamSpy: Audd.io error: {response.status}")
return None
except Exception as e:
_LOGGER.error(f"JamSpy: Recognition error: {e}")
return None
async def _update_state(self, result: Optional[Dict[str, Any]]):
if result and result.get("status") == "success" and result.get("result"):
song = result["result"]
self._attr_native_value = f"{song.get('title', 'Unknown')} - {song.get('artist', 'Unknown')}"
self._attr_extra_state_attributes.update({
"title": song.get("title"),
"artist": song.get("artist"),
"album": song.get("album"),
"last_recognition": datetime.now().isoformat()
})
else:
self._attr_native_value = "No music detected"
self._attr_extra_state_attributes.update({
"title": None,
"artist": None,
"album": None,
"last_recognition": datetime.now().isoformat()
})
await self._safe_write_ha_state()
services.yaml
recognize:
name: Recognize Music
description: Trigger music recognition
strings.json
{
"config": {
"step": {
"user": {
"title": "JamSpy Setup",
"description": "Set up JamSpy music recognition",
"data": {
"name": "Name"
}
}
}
}
}
configuration.yaml
template:
- sensor:
- name: "JamSpy Last Song"
state: "{{ states('sensor.jamspy_music_recognition') }}"
attributes:
album_art: "{{ state_attr('sensor.jamspy_music_recognition', 'album_art') }}"
availability: >
{{ states('sensor.jamspy_music_recognition') not in ['unknown', '', None] }}:
- sensor:
- name: "JamSpy Active"
state: >
{% if now().timestamp() - as_timestamp(states.sensor.jamspy_last_song.last_changed) < 900 %}
active
{% else %}
idle
{% endif %}
UPDATE : JamsPy is working off of the current config. some small details are that using audd.io free tier will run out of uses after about 25 or so uses so its not a viable option to use long terms