Feature: move portcheck into python application (#187)

* moved port check from shell script into the python application
* fixed pulseaudio check console output
* portcheck has now 15 chances to get the port before abort

docker-entrypoint.sh

@@ -86,7 +86,7 @@ fi
 CP_MAX_RETRIES=30
 CP_RETRY_DELAY=1
 ### while maybe besser?
-echo "Checking port $PORT..."
+echo "Checking PulseAudio service status..."
 for i in $(seq 1 $CP_MAX_RETRIES); do
   # Check if PulseAudio is running
   if pactl info >/dev/null 2>&1; then
@@ -102,28 +102,7 @@ for i in $(seq 1 $CP_MAX_RETRIES); do
   echo "⏳ PulseAudio not running yet, retrying in $CP_RETRY_DELAY s..."
   sleep $CP_RETRY_DELAY
 done
-
-
-### Check port availability
-# PORT variable is used from env
-PA_MAX_RETRIES=30
-PA_RETRY_DELAY=2
-echo "Checking port $PORT..."
-for i in $(seq 1 $PA_MAX_RETRIES); do
-  # Wait for port to be free (in case of rapid restarts)
-  if ! ss -tln | grep -q ":${PORT} "; then
-      echo "Port $PORT is available"
-      break
-  fi
-
-  if [ $i -eq $PA_MAX_RETRIES ]; then
-      echo "ERROR: Port $PORT still in use after $((PA_MAX_RETRIES * PA_RETRY_DELAY)) seconds"
-      exit 2
-  fi
-
-  echo "Attempt $i/$PA_MAX_RETRIES: Port $PORT in use, waiting ${PA_RETRY_DELAY}s..."
-  sleep $PA_RETRY_DELAY
-done
+echo "✅ PulseAudio is running"
 
 
 ### Start application

linux_voice_assistant/__main__.py

@@ -6,6 +6,7 @@ import logging
 import sys
 import threading
 import time
+import errno
 from pathlib import Path
 from queue import Queue
 from typing import Dict, List, Optional, Set, Union
@@ -340,6 +341,30 @@ async def main() -> None:
     state.music_player.set_volume(initial_volume_percent)
     state.tts_player.set_volume(initial_volume_percent)
 
+    loop = asyncio.get_running_loop()
+    max_attempts = 15
+    attempt = 1
+    server = None
+    
+    while attempt <= max_attempts:
+        try:
+            server = await loop.create_server(
+                lambda: VoiceSatelliteProtocol(state), host=host_ip_address, port=args.port
+            )
+            break  # connect successful, exit the loop
+        except OSError as err:
+            message = err.strerror or str(err)
+            if err.errno == errno.EADDRINUSE:
+                message = "address already in use"
+            
+            if attempt < max_attempts:
+                _LOGGER.warning("Attempt %d/%d failed to bind on address (%s, %s): %s. Retrying in 1 second...", attempt, max_attempts, host_ip_address, args.port, message)
+                await asyncio.sleep(1)
+                attempt += 1
+            else:
+                _LOGGER.exception("All %d attempts failed to bind on address (%s, %s): %s", max_attempts, host_ip_address, args.port, message)
+                sys.exit(1)
+
     process_audio_thread = threading.Thread(
         target=process_audio,
         args=(state, mic, args.audio_input_block_size),
@@ -347,11 +372,6 @@ async def main() -> None:
     )
     process_audio_thread.start()
 
-    loop = asyncio.get_running_loop()
-    server = await loop.create_server(
-        lambda: VoiceSatelliteProtocol(state), host=host_ip_address, port=args.port
-    )
-
     # Auto discovery (zeroconf, mDNS)
     discovery = HomeAssistantZeroconf(port=args.port, name=state.name, mac_address=state.mac_address, host_ip_address=host_ip_address)
     await discovery.register_server()