Spaces:
Build error
Build error
File size: 5,825 Bytes
111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 74466cd 1d177ae 74466cd 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d 1d177ae 111538d |
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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 |
"""Processing Result Data Transfer Object"""
from dataclasses import dataclass
from typing import Optional, Dict, Any
from datetime import datetime
@dataclass
class ProcessingResultDto:
"""DTO for pipeline output data
Contains the results of processing audio through the
STT -> Translation -> TTS pipeline.
"""
success: bool
original_text: Optional[str] = None
translated_text: Optional[str] = None
audio_path: Optional[str] = None
processing_time: float = 0.0
error_message: Optional[str] = None
error_code: Optional[str] = None
metadata: Optional[Dict[str, Any]] = None
timestamp: Optional[datetime] = None
def __post_init__(self):
"""Validate and set defaults after initialization"""
self._validate()
if self.metadata is None:
self.metadata = {}
if self.timestamp is None:
self.timestamp = datetime.utcnow()
def _validate(self):
"""Validate processing result data"""
if not isinstance(self.success, bool):
raise ValueError("Success must be a boolean value")
if self.processing_time < 0:
raise ValueError("Processing time cannot be negative")
if self.success:
# For successful processing, we should have some output
if not self.original_text and not self.translated_text and not self.audio_path:
raise ValueError("Successful processing must have at least one output (text or audio)")
else:
# For failed processing, we should have an error message
if not self.error_message:
raise ValueError("Failed processing must include an error message")
# Validate error code format if provided
if self.error_code:
valid_error_codes = [
'STT_ERROR', 'TRANSLATION_ERROR', 'TTS_ERROR',
'AUDIO_FORMAT_ERROR', 'VALIDATION_ERROR', 'SYSTEM_ERROR',
'TYPE_ERROR', 'TIMEOUT_ERROR', 'PERMISSION_ERROR', 'MEMORY_ERROR',
'UNKNOWN_ERROR' # Add missing error code for unexpected errors
]
if self.error_code not in valid_error_codes:
raise ValueError(f"Invalid error code: {self.error_code}. Valid codes: {valid_error_codes}")
# Validate metadata if provided
if self.metadata and not isinstance(self.metadata, dict):
raise ValueError("Metadata must be a dictionary")
@property
def has_text_output(self) -> bool:
"""Check if result has text output"""
return bool(self.original_text or self.translated_text)
@property
def has_audio_output(self) -> bool:
"""Check if result has audio output"""
return bool(self.audio_path)
@property
def is_complete(self) -> bool:
"""Check if processing is complete (success or failure with error)"""
return self.success or bool(self.error_message)
def add_metadata(self, key: str, value: Any) -> None:
"""Add metadata entry"""
if self.metadata is None:
self.metadata = {}
self.metadata[key] = value
def get_metadata(self, key: str, default: Any = None) -> Any:
"""Get metadata value"""
if self.metadata is None:
return default
return self.metadata.get(key, default)
def to_dict(self) -> dict:
"""Convert to dictionary representation"""
return {
'success': self.success,
'original_text': self.original_text,
'translated_text': self.translated_text,
'audio_path': self.audio_path,
'processing_time': self.processing_time,
'error_message': self.error_message,
'error_code': self.error_code,
'metadata': self.metadata or {},
'timestamp': self.timestamp.isoformat() if self.timestamp else None,
'has_text_output': self.has_text_output,
'has_audio_output': self.has_audio_output,
'is_complete': self.is_complete
}
@classmethod
def success_result(cls, original_text: str = None, translated_text: str = None,
audio_path: str = None, processing_time: float = 0.0,
metadata: Dict[str, Any] = None) -> 'ProcessingResultDto':
"""Create a successful processing result"""
return cls(
success=True,
original_text=original_text,
translated_text=translated_text,
audio_path=audio_path,
processing_time=processing_time,
metadata=metadata
)
@classmethod
def error_result(cls, error_message: str, error_code: str = None,
processing_time: float = 0.0, metadata: Dict[str, Any] = None) -> 'ProcessingResultDto':
"""Create a failed processing result"""
return cls(
success=False,
error_message=error_message,
error_code=error_code,
processing_time=processing_time,
metadata=metadata
)
@classmethod
def from_dict(cls, data: dict) -> 'ProcessingResultDto':
"""Create instance from dictionary"""
timestamp = None
if data.get('timestamp'):
timestamp = datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00'))
return cls(
success=data['success'],
original_text=data.get('original_text'),
translated_text=data.get('translated_text'),
audio_path=data.get('audio_path'),
processing_time=data.get('processing_time', 0.0),
error_message=data.get('error_message'),
error_code=data.get('error_code'),
metadata=data.get('metadata'),
timestamp=timestamp
) |