Spaces:
Running
on
A100
Running
on
A100
File size: 2,328 Bytes
ae238b3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import logging
import threading
from datetime import datetime
from typing import Dict
logger = logging.getLogger(__name__)
class TranscriptionStatus:
"""Simple transcription status tracker"""
def __init__(self):
self.is_busy = False
self.current_operation = None
self.current_filename = None
self.started_at = None
self.progress = 0.0
self.lock = threading.Lock()
self.total_completed = 0
def start_transcription(self, operation_type: str, filename: str = None):
"""Mark transcription as started"""
with self.lock:
self.is_busy = True
self.current_operation = operation_type
self.current_filename = filename
self.started_at = datetime.now()
self.progress = 0.0
logger.info(f"Started {operation_type} transcription for {filename or 'unknown file'}")
def update_progress(self, progress: float):
"""Update transcription progress (0.0 to 1.0)"""
with self.lock:
self.progress = max(0.0, min(1.0, progress))
def finish_transcription(self):
"""Mark transcription as finished"""
with self.lock:
self.is_busy = False
self.current_operation = None
self.current_filename = None
self.started_at = None
self.progress = 0.0
self.total_completed += 1
logger.info("Transcription finished")
def get_status(self) -> Dict:
"""Get current status for API response"""
with self.lock:
status = {"is_busy": self.is_busy, "total_completed": self.total_completed}
if self.is_busy:
duration = (
(datetime.now() - self.started_at).total_seconds()
if self.started_at
else 0
)
status.update(
{
"current_operation": self.current_operation,
"current_filename": self.current_filename,
"progress": self.progress,
"duration_seconds": round(duration, 1),
}
)
return status
# Global status instance
transcription_status = TranscriptionStatus()
|