Coverage for src/pythia/utils/message_handlers.py: 52%

1"""Common gstreamer pipeline bus message handlers.""" 

2 

3from typing import Protocol 

4from typing import Tuple 

5 

6from pythia.utils.gst import element_repr 

7from pythia.utils.gst import Gst 

8 

9GOT_EOS_FROM = "Got EOS from" 

10 

11 

12class Stoppable(Protocol): # noqa: R0903 

13 """Interface for classes whihc implement the stop method. 

14 

15 Mainly aimed at (but not restricted to) pythia apps. 

16 

17 """ 

18 

19 def stop(self) -> None: # noqa: C0116 

20 ... 

21 

22 

23def on_message_error( 

24 self: Stoppable, # noqa: W0613 

25 bus: Gst.Bus, # noqa: W0613 

26 message: Gst.Message, 

27) -> Tuple[str, str]: # noqa: W0613 

28 """Stop application on error. 

29 

30 Args: 

31 self: A stoppable instance. 

32 bus: The application's pipeline's bus. 

33 message: The gstreamer error message. 

34 

35 Returns: 

36 Error and debug string. 

37 

38 """ 

39 err, debug = message.parse_error() 

40 src = message.src 

41 if isinstance(src, Gst.Element): 

42 print(f"ERROR: from element {element_repr(src)}:{err}") 

43 else: 

44 print(f"ERROR: from {src}:{err}") 

45 print(f"Additional debug info:\n{debug}") 

46 return err, debug 

47 

48 

49def on_state_change( 

50 self: Stoppable, # noqa: W0613 

51 bus: Gst.Bus, # noqa: W0613 

52 message: Gst.Message, 

53): # noqa: W0613 

54 """Report application state changes. 

55 

56 Args: 

57 self: A stoppable instance, in case an error arises. 

58 bus: The application's pipeline's bus. 

59 message: The gstreamer state change message. 

60 

61 

62 Returns: 

63 Old, new, and pending state 

64 

65 """ 

66 old, new, pending = message.parse_state_changed() 

67 print( 

68 f"STATE CHANGE@{element_repr(message.src)}: {old=}, {new=}, {pending=}" 

69 ) 

70 return old, new, pending 

71 

72 

73def on_message_eos( 

74 self: Stoppable, bus: Gst.Bus, message: Gst.Message # noqa: W0613 

75): 

76 """Stop application on EOS event. 

77 

78 Args: 

79 self: A stoppable instance. 

80 bus: The application's pipeline's bus. 

81 message: The gstreamer state change message. 

82 

83 """ 

84 print(f"{GOT_EOS_FROM} {element_repr(message.src)}") 

85 self.stop()