Coverage for src/pythia/utils/gst.py: 65%

1"""Deepstream interface, utilities and customization.""" 

2from __future__ import annotations 

3 

4import ctypes 

5from typing import Iterator 

6from typing import Literal 

7 

8import gi 

9 

10gi.require_versions( 

11 { 

12 "Gst": "1.0", 

13 } 

14) 

15from gi.repository import GLib # noqa: C0413 

16from gi.repository import GObject # noqa: C0413 

17from gi.repository import Gst # noqa: C0413 

18 

19 

20def long_to_uint64(long: int) -> int: 

21 """Return the C 64-bit unsigned int datatype. 

22 

23 Args: 

24 long: the long value to convert. 

25 

26 Returns: 

27 The converted unit64 

28 

29 """ 

30 value = ctypes.c_uint64(long & 0xFFFFFFFFFFFFFFFF).value 

31 return value 

32 

33 

34def get_element(gst_bin: Gst.Bin, name) -> Gst.Element: 

35 """Get element from bin. 

36 

37 Args: 

38 gst_bin: parent where the element is to be located. 

39 name: name of the element to locate. 

40 

41 Returns: 

42 The found element. 

43 

44 Raises: 

45 NameError: internal gst getter returned `None`. 

46 

47 """ 

48 element = gst_bin.get_by_name(name) 

49 if element: 49 ↛ 51line 49 didn't jump to line 51, because the condition on line 49 was never false

50 return element 

51 raise NameError( 

52 f"Gst Bin {gst_bin} does not contain an element named '{name}'" 

53 ) 

54 

55 

56PadDirection = Literal["sink", "src"] 

57 

58 

59def get_static_pad(element: Gst.Element, direction: PadDirection) -> Gst.Pad: 

60 """Get static pad from element. 

61 

62 Args: 

63 element: element where the pad is located. 

64 direction: pad direction - "sink" or "source". 

65 

66 Returns: 

67 The found pad. 

68 

69 Raises: 

70 ValueError: Unable to get the pad. 

71 

72 """ 

73 pad = element.get_static_pad(direction) # type: ignore[arg-type] 

74 if pad: 74 ↛ 76line 74 didn't jump to line 76, because the condition on line 74 was never false

75 return pad 

76 raise ValueError(f"Unable to get {direction} pad from {element}") 

77 

78 

79def get_srcpad(gst_bin: Gst.Bin, element_name: str) -> Gst.Pad: 

80 """Get srcpad element from element. 

81 

82 Args: 

83 gst_bin: parent where the element is to be located. 

84 element_name: name of the element to locate the source pad. 

85 

86 Returns: 

87 The found source pad. 

88 

89 """ 

90 return get_static_pad(get_element(gst_bin, element_name), "src") 

91 

92 

93def get_sinkpad(gst_bin: Gst.Bin, element_name: str) -> Gst.Pad: 

94 """Get sinkpad element from element. 

95 

96 Args: 

97 gst_bin: parent where the element is to be located. 

98 element_name: name of the element to locate the sink pad. 

99 

100 Returns: 

101 The found source pad. 

102 

103 """ 

104 return get_static_pad(get_element(gst_bin, element_name), "sink") 

105 

106 

107def gst_iter(iterator: Gst.Iterator) -> Iterator: 

108 """Iterate pythonically over a :class:`Gst.Iterator`. 

109 

110 Args: 

111 iterator: the iterator which produces values. 

112 

113 Yields: 

114 The values from the iterator 

115 

116 See Also: 

117 http://lazka.github.io/pgi-docs/index.html#Gst-1.0/classes/Iterator.html#Gst.Iterator 

118 

119 """ 

120 while iterator is not None: 120 ↛ exitline 120 didn't return from function 'gst_iter', because the condition on line 120 was never false

121 try: 

122 result, elem = iterator.next() 

123 if result is Gst.IteratorResult.OK: 

124 yield elem 

125 elif result is Gst.IteratorResult.DONE: 125 ↛ 128line 125 didn't jump to line 128, because the condition on line 125 was never false

126 break 

127 else: 

128 break 

129 except StopIteration: 

130 break 

131 

132 

133def gst_init() -> None: 

134 """Initialize Gstreamer.""" 

135 if not Gst.is_initialized(): 

136 Gst.init(None) # type: ignore[call-arg] 

137 

138 

139def gst_deinit() -> None: 

140 """Initialize Gstreamer.""" 

141 if Gst.is_initialized(): 

142 Gst.deinit() # type: ignore[call-arg] 

143 

144 

145def element_repr(element: Gst.Object) -> str: 

146 """Compute element strig based on its hierarchy. 

147 

148 Args: 

149 element: The gstreamer element. 

150 

151 Returns: 

152 The element's string representation. 

153 

154 Example: 

155 >>> from gi.repository import Gst 

156 >>> Gst.init() 

157 >>> pipeline = Gst.parse_launch( 

158 ... "bin ( videotestsrc ! identity name=eye ) ! fakesink" 

159 ... ) 

160 >>> element_repr(pipeline.get_by_name("eye")) 

161 '/Pipeline:pipeline0/Bin:bin1/GstIdentity:eye' 

162 

163 """ 

164 if not isinstance(element, Gst.Element): 164 ↛ 165line 164 didn't jump to line 165, because the condition on line 164 was never true

165 return f"{element.__class__.__name__}:{str(element)}" 

166 

167 string = "" 

168 while element: 

169 string = f"/{element.__class__.__name__}:{element.name}{string}" 

170 element = element.parent # type: ignore[assignment] 

171 return string 

172 

173 

174def demote_plugin(element: str): 

175 """Lower a plugin's rank after the last one. 

176 

177 Use this to enforce playback elements (eg `uridecodebin`) to use 

178 another element instead of the one its currently choosing 

179 

180 Args: 

181 element: name of the `Gst.Element` to demote its rank. 

182 

183 Raises: 

184 NameError: Unable to retrieve element factory. 

185 

186 """ 

187 gst_init() 

188 target_element = Gst.ElementFactory.find(element) 

189 if not target_element: 

190 raise NameError(f"Unable to fins {element} factory") 

191 factories = Gst.ElementFactory.list_get_elements( 

192 Gst.ELEMENT_FACTORY_TYPE_ANY, # type: ignore[arg-type] 

193 Gst.Rank.MARGINAL, # type: ignore[arg-type] 

194 ) 

195 lowest_rank = min(factories, key=lambda f: f.get_rank()).get_rank() 

196 target_element.set_rank(lowest_rank - 1) 

197 

198 

199__all__ = [ 

200 "GLib", 

201 "Gst", 

202 "GObject", 

203 "long_to_uint64", 

204 "get_element", 

205 "PadDirection", 

206 "get_static_pad", 

207 "get_srcpad", 

208 "get_sinkpad", 

209 "gst_iter", 

210 "gst_init", 

211 "element_repr", 

212]