Refactor diarization configuration and unify transcription provider interfaces with networking fixes

backends/advanced/src/advanced_omi_backend/config.py

@@ -0,0 +1,111 @@
+"""
+Configuration management for Friend-Lite backend.
+
+Currently contains diarization settings because they were used in multiple places 
+causing circular imports. Other configurations can be moved here as needed.
+"""
+
+import json
+import logging
+import shutil
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# Default diarization settings
+DEFAULT_DIARIZATION_SETTINGS = {
+    "diarization_source": "pyannote",
+    "similarity_threshold": 0.15,
+    "min_duration": 0.5,
+    "collar": 2.0,
+    "min_duration_off": 1.5,
+    "min_speakers": 2,
+    "max_speakers": 6
+}
+
+# Global cache for diarization settings
+_diarization_settings = None
+
+
+def get_diarization_config_path():
+    """Get the path to the diarization config file."""
+    # Try different locations in order of preference
+    # 1. Data directory (for persistence across container restarts)
+    data_path = Path("/app/data/diarization_config.json")
+    if data_path.parent.exists():
+        return data_path
+
+    # 2. App root directory
+    app_path = Path("/app/diarization_config.json")
+    if app_path.parent.exists():
+        return app_path
+
+    # 3. Local development path
+    local_path = Path("diarization_config.json")
+    return local_path
+
+
+def load_diarization_settings_from_file():
+    """Load diarization settings from file or create from template."""
+    global _diarization_settings
+
+    config_path = get_diarization_config_path()
+    template_path = Path("/app/diarization_config.json.template")
+
+    # If no template, try local development path
+    if not template_path.exists():
+        template_path = Path("diarization_config.json.template")
+
+    # If config doesn't exist, try to copy from template
+    if not config_path.exists():
+        if template_path.exists():
+            try:
+                # Ensure parent directory exists
+                config_path.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copy(template_path, config_path)
+                logger.info(f"Created diarization config from template at {config_path}")
+            except Exception as e:
+                logger.warning(f"Could not copy template to {config_path}: {e}")
+
+    # Load from file if it exists
+    if config_path.exists():
+        try:
+            with open(config_path, 'r') as f:
+                _diarization_settings = json.load(f)
+                logger.info(f"Loaded diarization settings from {config_path}")
+                return _diarization_settings
+        except Exception as e:
+            logger.error(f"Error loading diarization settings from {config_path}: {e}")
+
+    # Fall back to defaults
+    _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy()
+    logger.info("Using default diarization settings")
+    return _diarization_settings
+
+
+def save_diarization_settings_to_file(settings):
+    """Save diarization settings to file."""
+    global _diarization_settings
+
+    config_path = get_diarization_config_path()
+
+    try:
+        # Ensure parent directory exists
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Write settings to file
+        with open(config_path, 'w') as f:
+            json.dump(settings, f, indent=2)
+
+        # Update cache
+        _diarization_settings = settings
+
+        logger.info(f"Saved diarization settings to {config_path}")
+        return True
+    except Exception as e:
+        logger.error(f"Error saving diarization settings to {config_path}: {e}")
+        return False
+
+
+# Initialize settings on module load
+_diarization_settings = load_diarization_settings_from_file()

backends/advanced/src/advanced_omi_backend/controllers/system_controller.py

@@ -19,6 +19,7 @@
 from wyoming.audio import AudioChunk
 
 from advanced_omi_backend.client_manager import generate_client_id
+from advanced_omi_backend.config import load_diarization_settings_from_file, save_diarization_settings_to_file
 from advanced_omi_backend.database import chunks_col
 from advanced_omi_backend.job_tracker import FileStatus, JobStatus, get_job_tracker
 from advanced_omi_backend.processors import AudioProcessingItem, get_processor_manager
@@ -748,103 +749,7 @@ async def process_files_with_content(
         await job_tracker.update_job_status(job_id, JobStatus.FAILED, error_msg)
 
 
-# Default diarization settings
-DEFAULT_DIARIZATION_SETTINGS = {
-    "diarization_source": "pyannote",
-    "similarity_threshold": 0.15,
-    "min_duration": 0.5,
-    "collar": 2.0,
-    "min_duration_off": 1.5,
-    "min_speakers": 2,
-    "max_speakers": 6
-}
-
-# Global cache for diarization settings
-_diarization_settings = None
-
-
-def get_diarization_config_path():
-    """Get the path to the diarization config file."""
-    # Try different locations in order of preference
-    # 1. Data directory (for persistence across container restarts)
-    data_path = Path("/app/data/diarization_config.json")
-    if data_path.parent.exists():
-        return data_path
-
-    # 2. App root directory
-    app_path = Path("/app/diarization_config.json")
-    if app_path.parent.exists():
-        return app_path
-
-    # 3. Local development path
-    local_path = Path("diarization_config.json")
-    return local_path
-
-
-def load_diarization_settings_from_file():
-    """Load diarization settings from file or create from template."""
-    global _diarization_settings
-
-    config_path = get_diarization_config_path()
-    template_path = Path("/app/diarization_config.json.template")
-
-    # If no template, try local development path
-    if not template_path.exists():
-        template_path = Path("diarization_config.json.template")
-
-    # If config doesn't exist, try to copy from template
-    if not config_path.exists():
-        if template_path.exists():
-            try:
-                # Ensure parent directory exists
-                config_path.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copy(template_path, config_path)
-                logger.info(f"Created diarization config from template at {config_path}")
-            except Exception as e:
-                logger.warning(f"Could not copy template to {config_path}: {e}")
-
-    # Load from file if it exists
-    if config_path.exists():
-        try:
-            with open(config_path, 'r') as f:
-                _diarization_settings = json.load(f)
-                logger.info(f"Loaded diarization settings from {config_path}")
-                return _diarization_settings
-        except Exception as e:
-            logger.error(f"Error loading diarization settings from {config_path}: {e}")
-
-    # Fall back to defaults
-    _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy()
-    logger.info("Using default diarization settings")
-    return _diarization_settings
-
-
-def save_diarization_settings_to_file(settings):
-    """Save diarization settings to file."""
-    global _diarization_settings
-
-    config_path = get_diarization_config_path()
-
-    try:
-        # Ensure parent directory exists
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-
-        # Write settings to file
-        with open(config_path, 'w') as f:
-            json.dump(settings, f, indent=2)
-
-        # Update cache
-        _diarization_settings = settings
-
-        logger.info(f"Saved diarization settings to {config_path}")
-        return True
-    except Exception as e:
-        logger.error(f"Error saving diarization settings to {config_path}: {e}")
-        return False
-
-
-# Initialize settings on module load
-_diarization_settings = load_diarization_settings_from_file()
+# Configuration functions moved to config.py to avoid circular imports
 
 
 async def get_diarization_settings():
@@ -884,34 +789,36 @@ async def save_diarization_settings(settings: dict):
                     return JSONResponse(
                         status_code=400, content={"error": f"Invalid value for {key}: must be integer 1-20"}
                     )
+            elif key == "diarization_source":
+                if not isinstance(value, str) or value not in ["pyannote", "deepgram"]:
+                    return JSONResponse(
+                        status_code=400, content={"error": f"Invalid value for {key}: must be 'pyannote' or 'deepgram'"}
+                    )
             else:
                 if not isinstance(value, (int, float)) or value < 0:
                     return JSONResponse(
                         status_code=400, content={"error": f"Invalid value for {key}: must be positive number"}
                     )
 
-        # Update global settings with new values
-        global _diarization_settings
-        if _diarization_settings is None:
-            _diarization_settings = DEFAULT_DIARIZATION_SETTINGS.copy()
-
-        _diarization_settings.update(settings)
+        # Get current settings and merge with new values
+        current_settings = load_diarization_settings_from_file()
+        current_settings.update(settings)
 
         # Save to file
-        if save_diarization_settings_to_file(_diarization_settings):
+        if save_diarization_settings_to_file(current_settings):
             logger.info(f"Updated and saved diarization settings: {settings}")
 
             return {
                 "message": "Diarization settings saved successfully",
-                "settings": _diarization_settings,
+                "settings": current_settings,
                 "status": "success"
             }
         else:
             # Even if file save fails, we've updated the in-memory settings
             logger.warning("Settings updated in memory but file save failed")
             return {
                 "message": "Settings updated (file save failed)",
-                "settings": _diarization_settings,
+                "settings": current_settings,
                 "status": "partial"
             }
 

backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py

@@ -63,7 +63,7 @@ async def diarize_identify_match(
             logger.info(f"Diarizing, identifying, and matching words for {audio_path}")
 
             # Read diarization source from existing config system
-            from advanced_omi_backend.controllers.system_controller import load_diarization_settings_from_file
+            from advanced_omi_backend.config import load_diarization_settings_from_file
             config = load_diarization_settings_from_file()
             diarization_source = config.get("diarization_source", "pyannote")
 
@@ -78,54 +78,42 @@ async def diarize_identify_match(
                     )
 
                     if diarization_source == "deepgram":
-                        # DEEPGRAM PATH: Use /v1/listen with structured config
-                        logger.info("Using Deepgram diarization path")
+                        # DEEPGRAM DIARIZATION PATH: We EXPECT transcript has speaker info from Deepgram
+                        # Only need speaker identification of existing segments
+                        logger.info("Using Deepgram diarization path - transcript should have speaker segments, identifying speakers")
 
-                        # Structure config for Deepgram diarization
-                        diarization_config = {"diarization": {"provider": "deepgram"}}
+                        # TODO: Implement proper speaker identification for Deepgram segments
+                        # For now, use diarize-identify-match as fallback until we implement segment identification
+                        logger.warning("Deepgram segment identification not yet implemented, using diarize-identify-match as fallback")
 
-                        # Log warning if pyannote params provided
-                        pyannote_params = ["min_speakers", "max_speakers", "collar", "min_duration_off"]
-                        provided_params = [p for p in pyannote_params if config.get(p) is not None]
-                        if provided_params:
-                            logger.warning("Ignoring pyannote parameters for Deepgram diarization: %s", provided_params)
-
-                        # Add structured diarization config
-                        form_data.add_field("diarization_config", json.dumps(diarization_config))
-
-                        # Add speaker identification params
-                        form_data.add_field("enhance_speakers", "true")
+                        form_data.add_field("transcript_data", json.dumps(transcript_data))
                         form_data.add_field("user_id", "1")  # TODO: Implement proper user mapping
-                        form_data.add_field("speaker_confidence_threshold", str(config.get("similarity_threshold", 0.15)))
+                        form_data.add_field("similarity_threshold", str(config.get("similarity_threshold", 0.15)))
+                        form_data.add_field("min_duration", str(config.get("min_duration", 0.5)))
 
-                        # Use /v1/listen endpoint
-                        endpoint = "/v1/listen"
+                        # Use /v1/diarize-identify-match endpoint as fallback
+                        endpoint = "/v1/diarize-identify-match"
 
                     else:  # pyannote (default)
-                        # PYANNOTE PATH: Use /v1/listen with structured config including pyannote parameters
-                        logger.info("Using Pyannote diarization path")
-
-                        # Structure config for Pyannote diarization  
-                        diarization_config = {
-                            "diarization": {
-                                "provider": "pyannote",
-                                "min_speakers": config.get("min_speakers"),
-                                "max_speakers": config.get("max_speakers"),
-                                "collar": config.get("collar", 2.0),
-                                "min_duration_off": config.get("min_duration_off", 1.5)
-                            }
-                        }
+                        # PYANNOTE PATH: Backend has transcript, need diarization + speaker identification
+                        logger.info("Using Pyannote path - diarizing backend transcript and identifying speakers")
 
-                        # Add structured diarization config
-                        form_data.add_field("diarization_config", json.dumps(diarization_config))
-
-                        # Add speaker identification params
-                        form_data.add_field("enhance_speakers", "true")
+                        # Send existing transcript for diarization and speaker matching
+                        form_data.add_field("transcript_data", json.dumps(transcript_data))
                         form_data.add_field("user_id", "1")  # TODO: Implement proper user mapping
-                        form_data.add_field("speaker_confidence_threshold", str(config.get("similarity_threshold", 0.15)))
+                        form_data.add_field("similarity_threshold", str(config.get("similarity_threshold", 0.15)))
+
+                        # Add pyannote diarization parameters
+                        form_data.add_field("min_duration", str(config.get("min_duration", 0.5)))
+                        form_data.add_field("collar", str(config.get("collar", 2.0)))
+                        form_data.add_field("min_duration_off", str(config.get("min_duration_off", 1.5)))
+                        if config.get("min_speakers"):
+                            form_data.add_field("min_speakers", str(config.get("min_speakers")))
+                        if config.get("max_speakers"):
+                            form_data.add_field("max_speakers", str(config.get("max_speakers")))
 
-                        # Use /v1/listen endpoint (now supports both providers)
-                        endpoint = "/v1/listen"
+                        # Use /v1/diarize-identify-match endpoint for backend integration
+                        endpoint = "/v1/diarize-identify-match"
 
                     # Make the request to the consolidated endpoint
                     async with session.post(

backends/advanced/src/advanced_omi_backend/transcription.py

@@ -13,6 +13,7 @@
     BaseTranscriptionProvider,
     get_transcription_provider,
 )
+from advanced_omi_backend.config import load_diarization_settings_from_file
 
 # ASR Configuration
 TRANSCRIPTION_PROVIDER = os.getenv("TRANSCRIPTION_PROVIDER")  # Optional: 'deepgram' or 'parakeet'
@@ -117,7 +118,18 @@ async def _get_transcript(self, audio_duration_seconds: Optional[float] = None):
                 logger.warning("No audio data or sample rate available for transcription")
                 return None
 
-            return await self.provider.transcribe(combined_audio, sample_rate)
+            # Check if we should request diarization based on configuration
+            config = load_diarization_settings_from_file()
+            diarization_source = config.get("diarization_source", "pyannote")
+
+            # Request diarization if using Deepgram as diarization source
+            should_diarize = (diarization_source == "deepgram" and 
+                            self.provider.name in ["Deepgram", "Deepgram-Streaming"])
+
+            if should_diarize:
+                logger.info(f"Requesting diarization from {self.provider.name} (diarization_source=deepgram)")
+
+            return await self.provider.transcribe(combined_audio, sample_rate, diarize=should_diarize)
 
         except Exception as e:
             logger.error(f"Error getting transcript from {self.provider.name}: {e}")