163 lines
4.7 KiB
Python
163 lines
4.7 KiB
Python
"""
|
|
Video Domain Models.
|
|
|
|
Pure business entities and value objects for video operations.
|
|
These models contain no external dependencies and represent core business concepts.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
from enum import Enum
|
|
|
|
|
|
class VideoFormat(Enum):
|
|
"""Supported video formats"""
|
|
AVI = "avi"
|
|
MP4 = "mp4"
|
|
WEBM = "webm"
|
|
|
|
|
|
class VideoStatus(Enum):
|
|
"""Video file status"""
|
|
RECORDING = "recording"
|
|
COMPLETED = "completed"
|
|
PROCESSING = "processing"
|
|
ERROR = "error"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VideoMetadata:
|
|
"""Video metadata value object"""
|
|
duration_seconds: float
|
|
width: int
|
|
height: int
|
|
fps: float
|
|
codec: str
|
|
bitrate: Optional[int] = None
|
|
|
|
@property
|
|
def resolution(self) -> Tuple[int, int]:
|
|
"""Get video resolution as tuple"""
|
|
return (self.width, self.height)
|
|
|
|
@property
|
|
def aspect_ratio(self) -> float:
|
|
"""Calculate aspect ratio"""
|
|
return self.width / self.height if self.height > 0 else 0.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StreamRange:
|
|
"""HTTP range request value object"""
|
|
start: int
|
|
end: Optional[int] = None
|
|
|
|
def __post_init__(self):
|
|
if self.start < 0:
|
|
raise ValueError("Start byte cannot be negative")
|
|
if self.end is not None and self.end < self.start:
|
|
raise ValueError("End byte cannot be less than start byte")
|
|
|
|
@property
|
|
def size(self) -> Optional[int]:
|
|
"""Get range size in bytes"""
|
|
if self.end is not None:
|
|
return self.end - self.start + 1
|
|
return None
|
|
|
|
@classmethod
|
|
def from_header(cls, range_header: str, file_size: int) -> 'StreamRange':
|
|
"""Parse HTTP Range header"""
|
|
if not range_header.startswith('bytes='):
|
|
raise ValueError("Invalid range header format")
|
|
|
|
range_spec = range_header[6:] # Remove 'bytes='
|
|
|
|
if '-' not in range_spec:
|
|
raise ValueError("Invalid range specification")
|
|
|
|
start_str, end_str = range_spec.split('-', 1)
|
|
|
|
if start_str:
|
|
start = int(start_str)
|
|
else:
|
|
# Suffix range (e.g., "-500" means last 500 bytes)
|
|
if not end_str:
|
|
raise ValueError("Invalid range specification")
|
|
suffix_length = int(end_str)
|
|
start = max(0, file_size - suffix_length)
|
|
end = file_size - 1
|
|
return cls(start=start, end=end)
|
|
|
|
if end_str:
|
|
end = min(int(end_str), file_size - 1)
|
|
else:
|
|
end = file_size - 1
|
|
|
|
return cls(start=start, end=end)
|
|
|
|
|
|
@dataclass
|
|
class VideoFile:
|
|
"""Video file entity"""
|
|
file_id: str
|
|
camera_name: str
|
|
filename: str
|
|
file_path: Path
|
|
file_size_bytes: int
|
|
created_at: datetime
|
|
status: VideoStatus
|
|
format: VideoFormat
|
|
metadata: Optional[VideoMetadata] = None
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
machine_trigger: Optional[str] = None
|
|
error_message: Optional[str] = None
|
|
|
|
def __post_init__(self):
|
|
"""Validate video file data"""
|
|
if not self.file_id:
|
|
raise ValueError("File ID cannot be empty")
|
|
if not self.camera_name:
|
|
raise ValueError("Camera name cannot be empty")
|
|
if self.file_size_bytes < 0:
|
|
raise ValueError("File size cannot be negative")
|
|
|
|
@property
|
|
def duration_seconds(self) -> Optional[float]:
|
|
"""Get video duration from metadata"""
|
|
return self.metadata.duration_seconds if self.metadata else None
|
|
|
|
@property
|
|
def is_streamable(self) -> bool:
|
|
"""Check if video can be streamed"""
|
|
return (
|
|
self.status in [VideoStatus.COMPLETED, VideoStatus.RECORDING] and
|
|
self.file_path.exists() and
|
|
self.file_size_bytes > 0
|
|
)
|
|
|
|
@property
|
|
def web_compatible_format(self) -> VideoFormat:
|
|
"""Get web-compatible format for this video"""
|
|
# AVI files should be converted to MP4 for web compatibility
|
|
if self.format == VideoFormat.AVI:
|
|
return VideoFormat.MP4
|
|
return self.format
|
|
|
|
def needs_conversion(self) -> bool:
|
|
"""Check if video needs format conversion for web streaming"""
|
|
return self.format != self.web_compatible_format
|
|
|
|
def get_converted_filename(self) -> str:
|
|
"""Get filename for converted version"""
|
|
if not self.needs_conversion():
|
|
return self.filename
|
|
|
|
# Replace extension with web-compatible format
|
|
stem = Path(self.filename).stem
|
|
return f"{stem}.{self.web_compatible_format.value}"
|