Coverage for src / lexigram / contracts / multimedia / protocols.py: 100%

1"""Structural protocols for multimedia generation providers. 

2 

3Four separate Protocols rather than one shared base class — Python's 

4structural typing makes a base class unnecessary, and each media type's 

5request/response shape differs enough that a shared abstract method 

6signature would need to be typed loosely (``Any``) to fit all four. 

7""" 

8 

9from __future__ import annotations 

10 

11from collections.abc import Callable 

12from typing import TYPE_CHECKING, Protocol, runtime_checkable 

13 

14if TYPE_CHECKING: 

15 from lexigram.contracts.core.result import Result 

16 from lexigram.contracts.multimedia.exceptions import ( 

17 MultimediaError, 

18 VideoGenerationError, 

19 ) 

20 from lexigram.contracts.multimedia.types import ( 

21 BeatAnalysisRequest, 

22 BeatAnalysisResult, 

23 ImageRequest, 

24 InterpolationRequest, 

25 MediaAsset, 

26 MusicRequest, 

27 TTSRequest, 

28 UpscaleRequest, 

29 VideoOperation, 

30 VideoRequest, 

31 ) 

32 

33 

34@runtime_checkable 

35class TTSProvider(Protocol): 

36 """Protocol for text-to-speech generation backends.""" 

37 

38 async def generate( 

39 self, request: TTSRequest 

40 ) -> Result[MediaAsset, MultimediaError]: ... 

41 

42 

43@runtime_checkable 

44class InterpolationProvider(Protocol): 

45 """Protocol for two-frame midpoint interpolation backends.""" 

46 

47 async def interpolate( 

48 self, request: InterpolationRequest 

49 ) -> Result[MediaAsset, MultimediaError]: ... 

50 

51 

52@runtime_checkable 

53class MusicProvider(Protocol): 

54 """Protocol for music/sound generation backends.""" 

55 

56 async def generate( 

57 self, request: MusicRequest 

58 ) -> Result[MediaAsset, MultimediaError]: ... 

59 

60 

61@runtime_checkable 

62class VideoProvider(Protocol): 

63 """Protocol for video generation backends.""" 

64 

65 async def generate( 

66 self, request: VideoRequest 

67 ) -> Result[MediaAsset, MultimediaError]: ... 

68 

69 

70@runtime_checkable 

71class VideoProcessor(Protocol): 

72 """Protocol for ffmpeg-backed video processing/editing backends. 

73 

74 `progress_callback` receives `0.0 <= pct <= 1.0`; implementations 

75 should emit `1.0` exactly once when processing finishes successfully. 

76 """ 

77 

78 async def process( 

79 self, 

80 operation: VideoOperation, 

81 *, 

82 progress_callback: Callable[[float], None] | None = None, 

83 ) -> Result[MediaAsset, VideoGenerationError]: ... 

84 

85 async def extract_frames( 

86 self, asset: MediaAsset, *, fps: float | None = None 

87 ) -> Result[list[MediaAsset], VideoGenerationError]: ... 

88 

89 async def assemble_frames( 

90 self, frames: list[MediaAsset], *, fps: float 

91 ) -> Result[MediaAsset, VideoGenerationError]: ... 

92 

93 

94@runtime_checkable 

95class BeatAnalysisProvider(Protocol): 

96 """Protocol for audio tempo/beat-detection backends.""" 

97 

98 async def analyze( 

99 self, request: BeatAnalysisRequest 

100 ) -> Result[BeatAnalysisResult, MultimediaError]: ... 

101 

102 

103@runtime_checkable 

104class ImageProvider(Protocol): 

105 """Protocol for still-image generation backends.""" 

106 

107 async def generate( 

108 self, request: ImageRequest 

109 ) -> Result[MediaAsset, MultimediaError]: ... 

110 

111 

112@runtime_checkable 

113class UpscaleProvider(Protocol): 

114 """Protocol for single-image super-resolution backends.""" 

115 

116 async def upscale( 

117 self, request: UpscaleRequest 

118 ) -> Result[MediaAsset, MultimediaError]: ... 

119 

120 

121__all__ = [ 

122 "BeatAnalysisProvider", 

123 "ImageProvider", 

124 "InterpolationProvider", 

125 "MusicProvider", 

126 "TTSProvider", 

127 "UpscaleProvider", 

128 "VideoProcessor", 

129 "VideoProvider", 

130]