added send ping in server

client.py

@@ -2,21 +2,19 @@ import asyncio
 import websockets
 import requests
 import ssl
-import logging
 
 # Parameters for reading and sending the audio
-AUDIO_FILE_URL = "https://raw.githubusercontent.com/AshDavid12/runpod-serverless-forked/main/test_hebrew.wav"  # Use WAV file
-
-
+#AUDIO_FILE_URL = "https://raw.githubusercontent.com/AshDavid12/runpod-serverless-forked/main/test_hebrew.wav"  # Use WAV file
+AUDIO_FILE_URL = "https://raw.githubusercontent.com/AshDavid12/hugging_face_ivrit_streaming/main/long_hebrew.wav"
 async def send_audio(websocket):
-    buffer_size = 1024  # Buffer audio chunks up to 512KB before sending
+    buffer_size = 512*1024  #HAVE TO HAVE 512!!
     audio_buffer = bytearray()
 
     with requests.get(AUDIO_FILE_URL, stream=True, allow_redirects=False) as response:
         if response.status_code == 200:
             print("Starting to stream audio file...")
 
-            for chunk in response.iter_content(chunk_size=1024):  # Stream in chunks
+            for chunk in response.iter_content(chunk_size=512):  # Stream in chunks
                 if chunk:
                     audio_buffer.extend(chunk)
                     #print(f"Received audio chunk of size {len(chunk)} bytes.")
@@ -24,30 +22,26 @@ async def send_audio(websocket):
                     # Send buffered audio data once it's large enough
                     if len(audio_buffer) >= buffer_size:
                         await websocket.send(audio_buffer)
-                        #print(f"Sent {len(audio_buffer)} bytes of audio data.")
+                        print(f"Sent {len(audio_buffer)} bytes of audio data.")
                         audio_buffer.clear()
-                        await asyncio.sleep(0.01)
+                        await asyncio.sleep(0.001)
 
             print("Finished sending audio.")
         else:
             print(f"Failed to download audio file. Status code: {response.status_code}")
 
-
 async def receive_transcription(websocket):
     while True:
         try:
-            transcription = await websocket.recv()  # Receive transcription from the server
-            # new_segments = process_transcription_results(transcription)
-            # # Now handle only new segments
-            # if new_segments:
-            #     for segment in new_segments:
-            #         print(f"New Segment: {segment['text']}")
+
+            transcription = await asyncio.wait_for(websocket.recv(),timeout=300)
+              # Receive transcription from the server
             print(f"Transcription: {transcription}")
         except Exception as e:
             print(f"Error receiving transcription: {e}")
+            await asyncio.sleep(30)
             break
 
-
 async def send_heartbeat(websocket):
     while True:
         try:
@@ -56,7 +50,7 @@ async def send_heartbeat(websocket):
         except websockets.ConnectionClosed:
             print("Connection closed, stopping heartbeat")
             break
-        await asyncio.sleep(30)  # Send ping every 30 seconds (adjust as needed)
+        await asyncio.sleep(600)  # Send ping every 30 seconds (adjust as needed)
 
 
 async def run_client():
@@ -65,13 +59,11 @@ async def run_client():
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
 
-    async with websockets.connect(uri, ssl=ssl_context, timeout=120) as websocket:
-        print(f"here")
+    async with websockets.connect(uri, ssl=ssl_context, ping_timeout=120,ping_interval=10) as websocket:
         await asyncio.gather(
             send_audio(websocket),
             receive_transcription(websocket),
             send_heartbeat(websocket)
         )
 
-
-asyncio.run(run_client())
+asyncio.run(run_client())

infer.py

@@ -176,7 +176,7 @@ def transcribe_core_ws(audio_file, last_transcribed_time):
             ret['new_segments'].append(seg)
 
             # Update the last transcribed time to the end of the current segment
-            new_last_transcribed_time = max(new_last_transcribed_time, s.end)
+            new_last_transcribed_time = s.end
             logging.debug(f"Updated last transcribed time to: {new_last_transcribed_time} seconds")
 
     #logging.info(f"Returning {len(ret['new_segments'])} new segments and updated last transcribed time.")
@@ -192,6 +192,19 @@ async def websocket_transcribe(websocket: WebSocket):
     await websocket.accept()
     logging.info("WebSocket connection established successfully.")
 
+    async def send_ping():
+        """Function to send periodic ping to keep the connection alive."""
+        while True:
+            try:
+                await websocket.ping()
+                logging.info("Sent keepalive ping to client.")
+                await asyncio.sleep(10)  # Ping every 10 seconds (adjust the interval as needed)
+            except Exception as e:
+                logging.error(f"Error sending ping: {e}")
+                break
+
+    ping_task = asyncio.create_task(send_ping())
+
     try:
         processed_segments = []  # Keeps track of the segments already transcribed
         accumulated_audio_size = 0  # Track how much audio data has been buffered