Spaces:
Build error
Build error
"""ProcessingResult value object for audio processing pipeline results.""" | |
from dataclasses import dataclass | |
from typing import Optional | |
from .text_content import TextContent | |
from .audio_content import AudioContent | |
class ProcessingResult: | |
"""Value object representing the result of audio processing pipeline.""" | |
success: bool | |
original_text: Optional[TextContent] | |
translated_text: Optional[TextContent] | |
audio_output: Optional[AudioContent] | |
error_message: Optional[str] | |
processing_time: float | |
def __post_init__(self): | |
"""Validate processing result after initialization.""" | |
self._validate() | |
def _validate(self): | |
"""Validate processing result properties.""" | |
if not isinstance(self.success, bool): | |
raise TypeError("Success must be a boolean") | |
if self.original_text is not None and not isinstance(self.original_text, TextContent): | |
raise TypeError("Original text must be a TextContent instance or None") | |
if self.translated_text is not None and not isinstance(self.translated_text, TextContent): | |
raise TypeError("Translated text must be a TextContent instance or None") | |
if self.audio_output is not None and not isinstance(self.audio_output, AudioContent): | |
raise TypeError("Audio output must be an AudioContent instance or None") | |
if self.error_message is not None and not isinstance(self.error_message, str): | |
raise TypeError("Error message must be a string or None") | |
if not isinstance(self.processing_time, (int, float)): | |
raise TypeError("Processing time must be a number") | |
if self.processing_time < 0: | |
raise ValueError("Processing time cannot be negative") | |
# Business rule validations | |
if self.success: | |
if self.error_message is not None: | |
raise ValueError("Successful result cannot have an error message") | |
if self.original_text is None: | |
raise ValueError("Successful result must have original text") | |
else: | |
if self.error_message is None or not self.error_message.strip(): | |
raise ValueError("Failed result must have a non-empty error message") | |
def has_translation(self) -> bool: | |
"""Check if the result includes translated text.""" | |
return self.translated_text is not None | |
def has_audio_output(self) -> bool: | |
"""Check if the result includes audio output.""" | |
return self.audio_output is not None | |
def is_complete_pipeline(self) -> bool: | |
"""Check if the result represents a complete pipeline execution.""" | |
return (self.success and | |
self.original_text is not None and | |
self.translated_text is not None and | |
self.audio_output is not None) | |
def success_result( | |
cls, | |
original_text: TextContent, | |
translated_text: Optional[TextContent] = None, | |
audio_output: Optional[AudioContent] = None, | |
processing_time: float = 0.0 | |
) -> 'ProcessingResult': | |
"""Create a successful processing result.""" | |
return cls( | |
success=True, | |
original_text=original_text, | |
translated_text=translated_text, | |
audio_output=audio_output, | |
error_message=None, | |
processing_time=processing_time | |
) | |
def failure_result( | |
cls, | |
error_message: str, | |
processing_time: float = 0.0, | |
original_text: Optional[TextContent] = None, | |
translated_text: Optional[TextContent] = None, | |
audio_output: Optional[AudioContent] = None | |
) -> 'ProcessingResult': | |
"""Create a failed processing result.""" | |
return cls( | |
success=False, | |
original_text=original_text, | |
translated_text=translated_text, | |
audio_output=audio_output, | |
error_message=error_message, | |
processing_time=processing_time | |
) |