black

Author: 0xrushi

backends/advanced/scripts/create_plugin.py

@@ -7,6 +7,7 @@
 Usage:
     uv run python scripts/create_plugin.py my_awesome_plugin
 """
+
 import argparse
 import os
 import shutil

backends/advanced/scripts/delete_all_conversations_api.py

@@ -4,11 +4,12 @@
 This uses the proper API authentication and endpoints.
 """
 
+import argparse
 import asyncio
 import os
 import sys
-import argparse
 from pathlib import Path
+
 import aiohttp
 from dotenv import load_dotenv
 
@@ -20,113 +21,111 @@ async def get_auth_token():
     """Get admin authentication token."""
     admin_email = os.getenv("ADMIN_EMAIL")
     admin_password = os.getenv("ADMIN_PASSWORD")
-    
+
     if not admin_email or not admin_password:
         print("Error: ADMIN_EMAIL and ADMIN_PASSWORD must be set in .env file")
         sys.exit(1)
-    
+
     base_url = "http://localhost:8000"
-    
+
     async with aiohttp.ClientSession() as session:
         # Login to get token
-        login_data = {
-            "username": admin_email,
-            "password": admin_password
-        }
-        
+        login_data = {"username": admin_email, "password": admin_password}
+
         async with session.post(
-            f"{base_url}/auth/jwt/login",
-            data=login_data
+            f"{base_url}/auth/jwt/login", data=login_data
         ) as response:
             if response.status != 200:
                 print(f"Failed to login: {response.status}")
                 text = await response.text()
                 print(f"Response: {text}")
                 sys.exit(1)
-            
+
             result = await response.json()
             return result["access_token"]
 
 
 async def delete_all_conversations(skip_prompt=False):
     """Delete all conversations using the API."""
-    
+
     base_url = "http://localhost:8000"
-    
+
     # Get auth token
     print("Getting admin authentication token...")
     token = await get_auth_token()
-    
-    headers = {
-        "Authorization": f"Bearer {token}"
-    }
-    
+
+    headers = {"Authorization": f"Bearer {token}"}
+
     async with aiohttp.ClientSession() as session:
         # First, get all conversations
         print("Fetching all conversations...")
         async with session.get(
-            f"{base_url}/api/conversations",
-            headers=headers
+            f"{base_url}/api/conversations", headers=headers
         ) as response:
             if response.status != 200:
                 print(f"Failed to fetch conversations: {response.status}")
                 text = await response.text()
                 print(f"Response: {text}")
                 return
-            
+
             data = await response.json()
-            
+
         # Extract conversations from nested structure
         conversations_dict = data.get("conversations", {})
         conversations = []
         for client_id, client_conversations in conversations_dict.items():
             conversations.extend(client_conversations)
-            
+
         print(f"Found {len(conversations)} conversations")
-        
+
         if len(conversations) == 0:
             print("No conversations to delete")
             return
-        
+
         # Confirm deletion unless --yes flag is used
         if not skip_prompt:
-            response = input(f"Are you sure you want to delete ALL {len(conversations)} conversations? (yes/no): ")
+            response = input(
+                f"Are you sure you want to delete ALL {len(conversations)} conversations? (yes/no): "
+            )
             if response.lower() != "yes":
                 print("Deletion cancelled")
                 return
-        
+
         # Delete each conversation
         deleted_count = 0
         failed_count = 0
-        
+
         for conv in conversations:
             audio_uuid = conv.get("audio_uuid")
             if not audio_uuid:
                 print(f"Skipping conversation without audio_uuid: {conv.get('_id')}")
                 continue
-            
+
             # Delete the conversation
             async with session.delete(
-                f"{base_url}/api/conversations/{audio_uuid}",
-                headers=headers
+                f"{base_url}/api/conversations/{audio_uuid}", headers=headers
             ) as response:
                 if response.status == 200:
                     deleted_count += 1
-                    print(f"Deleted conversation {audio_uuid} ({deleted_count}/{len(conversations)})")
+                    print(
+                        f"Deleted conversation {audio_uuid} ({deleted_count}/{len(conversations)})"
+                    )
                 else:
                     failed_count += 1
                     text = await response.text()
                     print(f"Failed to delete {audio_uuid}: {response.status} - {text}")
-        
+
         print(f"\nDeletion complete:")
         print(f"  Successfully deleted: {deleted_count}")
         print(f"  Failed: {failed_count}")
 
 
 if __name__ == "__main__":
     # Parse command line arguments
-    parser = argparse.ArgumentParser(description='Delete all conversations')
-    parser.add_argument('--yes', '-y', action='store_true', help='Skip confirmation prompt')
+    parser = argparse.ArgumentParser(description="Delete all conversations")
+    parser.add_argument(
+        "--yes", "-y", action="store_true", help="Skip confirmation prompt"
+    )
     args = parser.parse_args()
-    
-    asyncio.run(delete_all_conversations(skip_prompt=args.yes))
+
+    asyncio.run(delete_all_conversations(skip_prompt=args.yes))

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

@@ -6,10 +6,7 @@
 
 from fastapi.responses import JSONResponse
 
-from advanced_omi_backend.client_manager import (
-    ClientManager,
-    get_user_clients_active,
-)
+from advanced_omi_backend.client_manager import ClientManager, get_user_clients_active
 from advanced_omi_backend.users import User
 
 logger = logging.getLogger(__name__)
@@ -37,7 +34,9 @@ async def get_active_clients(user: User, client_manager: ClientManager):
 
             # Filter to only the user's clients
             user_clients = [
-                client for client in all_clients if client["client_id"] in user_active_clients
+                client
+                for client in all_clients
+                if client["client_id"] in user_active_clients
             ]
 
             return {

backends/advanced/src/advanced_omi_backend/models/__init__.py

@@ -7,4 +7,4 @@
 
 # Models can be imported directly from their files
 # e.g. from .job import TranscriptionJob
-# e.g. from .conversation import Conversation, create_conversation
+# e.g. from .conversation import Conversation, create_conversation

backends/advanced/src/advanced_omi_backend/routers/modules/user_routes.py

@@ -24,13 +24,17 @@ async def get_users(current_user: User = Depends(current_superuser)):
 
 
 @router.post("")
-async def create_user(user_data: UserCreate, current_user: User = Depends(current_superuser)):
+async def create_user(
+    user_data: UserCreate, current_user: User = Depends(current_superuser)
+):
     """Create a new user. Admin only."""
     return await user_controller.create_user(user_data)
 
 
 @router.put("/{user_id}")
-async def update_user(user_id: str, user_data: UserUpdate, current_user: User = Depends(current_superuser)):
+async def update_user(
+    user_id: str, user_data: UserUpdate, current_user: User = Depends(current_superuser)
+):
     """Update a user. Admin only."""
     return await user_controller.update_user(user_id, user_data)
 
@@ -43,4 +47,6 @@ async def delete_user(
     delete_memories: bool = False,
 ):
     """Delete a user and optionally their associated data. Admin only."""
-    return await user_controller.delete_user(user_id, delete_conversations, delete_memories)
+    return await user_controller.delete_user(
+        user_id, delete_conversations, delete_memories
+    )

backends/advanced/src/advanced_omi_backend/services/audio_service.py

@@ -42,9 +42,9 @@ def __init__(self, redis_url: Optional[str] = None):
         self.memory_events_stream = "memory:events"
 
         # Consumer group names (action verbs - what they DO)
-        self.audio_writer = "audio-file-writer"            # Writes audio chunks to WAV files
-        self.memory_enqueuer = "memory-job-enqueuer"       # Enqueues memory extraction jobs
-        self.event_listener = "event-listener"             # Listens for completion events
+        self.audio_writer = "audio-file-writer"  # Writes audio chunks to WAV files
+        self.memory_enqueuer = "memory-job-enqueuer"  # Enqueues memory extraction jobs
+        self.event_listener = "event-listener"  # Listens for completion events
 
     async def connect(self):
         """Connect to Redis with connection pooling."""
@@ -55,7 +55,7 @@ async def connect(self):
             max_connections=20,  # Allow multiple concurrent operations
             socket_keepalive=True,
             socket_connect_timeout=5,
-            retry_on_timeout=True
+            retry_on_timeout=True,
         )
         logger.info(f"Audio stream service connected to Redis at {self.redis_url}")
 
@@ -84,7 +84,7 @@ async def publish_audio_chunk(
         user_email: str,
         audio_chunk: AudioChunk,
         audio_uuid: Optional[str] = None,
-        timestamp: Optional[int] = None
+        timestamp: Optional[int] = None,
     ) -> str:
         """
         Publish audio chunk to Redis Stream.
@@ -135,12 +135,11 @@ async def publish_audio_chunk(
         # Ensure consumer group exists for this stream
         try:
             await self.redis.xgroup_create(
-                stream_name,
-                self.audio_writer,
-                id="0",
-                mkstream=True
+                stream_name, self.audio_writer, id="0", mkstream=True
+            )
+            audio_logger.debug(
+                f"Created consumer group {self.audio_writer} for {stream_name}"
             )
-            audio_logger.debug(f"Created consumer group {self.audio_writer} for {stream_name}")
         except aioredis.ResponseError as e:
             if "BUSYGROUP" not in str(e):
                 raise
@@ -152,7 +151,7 @@ async def publish_transcript_event(
         audio_uuid: str,
         conversation_id: str,
         status: str,
-        error: Optional[str] = None
+        error: Optional[str] = None,
     ):
         """
         Publish transcript completion event.
@@ -189,7 +188,7 @@ async def publish_transcript_event(
                 self.transcript_events_stream,
                 self.memory_enqueuer,
                 id="0",
-                mkstream=True
+                mkstream=True,
             )
         except aioredis.ResponseError as e:
             if "BUSYGROUP" not in str(e):
@@ -200,7 +199,7 @@ async def publish_memory_event(
         conversation_id: str,
         status: str,
         memory_count: int = 0,
-        error: Optional[str] = None
+        error: Optional[str] = None,
     ):
         """
         Publish memory processing event.
@@ -234,21 +233,14 @@ async def publish_memory_event(
         # Ensure consumer group exists
         try:
             await self.redis.xgroup_create(
-                self.memory_events_stream,
-                self.event_listener,
-                id="0",
-                mkstream=True
+                self.memory_events_stream, self.event_listener, id="0", mkstream=True
             )
         except aioredis.ResponseError as e:
             if "BUSYGROUP" not in str(e):
                 raise
 
     async def consume_audio_stream(
-        self,
-        consumer_name: str,
-        callback,
-        block_ms: int = 5000,
-        count: int = 10
+        self, consumer_name: str, callback, block_ms: int = 5000, count: int = 10
     ):
         """
         Consume audio chunks from all client streams.
@@ -288,7 +280,7 @@ async def consume_audio_stream(
                 consumer_name,
                 streams_dict,
                 count=count,
-                block=block_ms
+                block=block_ms,
             )
 
             for stream_name, stream_messages in messages:
@@ -299,15 +291,13 @@ async def consume_audio_stream(
 
                         # Acknowledge message
                         await self.redis.xack(
-                            stream_name,
-                            self.audio_writer,
-                            message_id
+                            stream_name, self.audio_writer, message_id
                         )
 
                     except Exception as e:
                         logger.error(
                             f"Error processing audio message {message_id.decode()}: {e}",
-                            exc_info=True
+                            exc_info=True,
                         )
 
         except Exception as e: