teachingAssistant / src /domain /models /processing_result.py
Michael Hu
Create domain service interfaces
55e29e2
Raw
History Blame Contribute Delete
4.2 kB
"""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
@dataclass(frozen=True)
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")
@property
def has_translation(self) -> bool:
"""Check if the result includes translated text."""
return self.translated_text is not None
@property
def has_audio_output(self) -> bool:
"""Check if the result includes audio output."""
return self.audio_output is not None
@property
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)
@classmethod
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
)
@classmethod
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
)