Fixed it so far - not perfect (still clipping a little - but easy to understand now)
import appdaemon.plugins.hass.hassapi as hass
import requests
from requests.auth import HTTPBasicAuth
from requests.exceptions import RequestException
import subprocess
from io import BytesIO
from time import sleep, time
class DoorbirdException(Exception):
"""An exception for doorbirds"""
class Doorbird:
CHUNK_SIZE = 4096 # Moderate chunk size in bytes
RATE_LIMIT = 8192 # Bytes per second (8KB per second)
CHUNK_INTERVAL = 0.5 # Interval in seconds (500 ms for larger chunks)
def __init__(self, device_ip, username, password):
"""
Connect to a Doorbird
Args:
device_ip (str): Doorbird device IP address.
username (str): Doorbird HTTP username.
password (str): Doorbird HTTP password.
Raises:
DoorbirdException
"""
self.device_ip = device_ip
self.username = username
self.password = password
self.session_id = self._get_session_id()
def _get_session_id(self):
"""
Get session ID from Doorbird device.
Raises:
DoorbirdException: If unable to obtain session ID.
"""
get_session_url = f"http://{self.device_ip}/bha-api/getsession.cgi"
auth = (self.username, self.password)
try:
response = requests.get(get_session_url, auth=auth)
response.raise_for_status()
data = response.json()
return data["BHA"]["SESSIONID"]
except RequestException as e:
raise DoorbirdException(f"Failed to obtain session ID: {e}")
def _convert_audio(self, input_file):
"""
Convert audio to 8000Hz mono PCM mu-law format.
Args:
input_file (str): Path to the input audio file.
Returns:
bytes: Converted audio data.
"""
try:
process = subprocess.run(
[
'ffmpeg', '-y', '-i', input_file, '-ar', '8000', '-ac', '1', '-f', 'mulaw', 'pipe:1'
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True
)
return process.stdout
except subprocess.CalledProcessError as e:
raise DoorbirdException(f"FFmpeg conversion failed: {e.stderr.decode()}")
def _generate_audio_chunks(self, audio_data):
"""
Generator function to yield chunks of audio data from a file.
Doorbird is rate-limited to 8KB per second.
Args:
audio_data: Converted audio data
Yields:
bytes: Chunks of audio data.
"""
stream = BytesIO(audio_data)
next_chunk_time = time()
while True:
start_time = time()
chunk = stream.read(self.CHUNK_SIZE)
if not chunk:
break
yield chunk
elapsed_time = time() - start_time
next_chunk_time += self.CHUNK_INTERVAL
sleep_time = max(0, next_chunk_time - time())
self._log_timing(self.CHUNK_SIZE, elapsed_time, sleep_time)
sleep(sleep_time)
def _log_timing(self, chunk_size, elapsed_time, sleep_time):
"""
Log the timing details for each chunk.
Args:
chunk_size (int): Size of the chunk.
elapsed_time (float): Time taken to process the chunk.
sleep_time (float): Time to sleep before sending the next chunk.
"""
print(f"Chunk Size: {chunk_size}, Elapsed Time: {elapsed_time:.6f}, Sleep Time: {sleep_time:.6f}")
def send_audio(self, audio_url):
"""
Send audio to Doorbird device.
Args:
audio_url (str): URL of the audio file.
Raises:
DoorbirdException: If any step in the process fails.
"""
try:
# Download the audio file
audio_response = requests.get(audio_url)
audio_response.raise_for_status()
audio_file_path = '/config/downloaded_audio.mp3'
with open(audio_file_path, 'wb') as audio_file:
audio_file.write(audio_response.content)
# Convert the audio file
audio_data = self._convert_audio(audio_file_path)
# Transmit the audio file in chunks
audio_transmit_url = f"http://{self.device_ip}/bha-api/audio-transmit.cgi?sessionid={self.session_id}"
auth = HTTPBasicAuth(self.username, self.password)
def audio_stream():
for chunk in self._generate_audio_chunks(audio_data):
yield chunk
response = requests.post(
audio_transmit_url,
headers={"Content-Type": "audio/basic", "Connection": "Keep-Alive", "Cache-Control": "no-cache"},
data=audio_stream(),
auth=auth,
timeout=60
)
response.raise_for_status()
print("Audio transmission completed successfully.")
except RequestException as e:
raise DoorbirdException(f"Failed to send audio: {e}")
class DoorbirdAudio(hass.Hass):
"""AppDaemon app to handle Doorbird audio events."""
def initialize(self):
"""Initialize the AppDaemon app."""
self.listen_event(self.doorbird_audio, "doorbird_audio")
def doorbird_audio(self, event_name, data, kwargs):
"""Handle Doorbird audio event."""
try:
self.log(f"Received event: {event_name} with data: {data}")
doorbird = Doorbird(data["device_ip"], data["username"], data["password"])
doorbird.send_audio(data["audio_url"])
self.log("Audio transmission completed successfully.")
except DoorbirdException as e:
self.log(f"Failed to send audio: {e}")