1#!/usr/bin/env python3

2# This file is part of Checkbox.

3#

4# Copyright 2014 Canonical Ltd.

5# Written by:

6# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>

7#

8# Checkbox is free software: you can redistribute it and/or modify

9# it under the terms of the GNU General Public License version 3,

10# as published by the Free Software Foundation.

11#

12# Checkbox is distributed in the hope that it will be useful,

13# but WITHOUT ANY WARRANTY; without even the implied warranty of

14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

15# GNU General Public License for more details.

16#

17# You should have received a copy of the GNU General Public License

18# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

19"""

20 pulse_active_port_change.py

21========================

22

23 This script checks if the active port on either sinks (speakers or headphones)

24 or sources (microphones, webcams) is changed after an appropriate device is

25 plugged into the DUT. The script is fully automatic and either times out after

26 30 seconds or returns as soon as the change is detected.

27

28 The script monitors pulse audio events with `pactl subscribe`. Any changes to

29 sinks (or sources, depending on the mode) are treated as a possible match. A

30 match is verified by running `pactl list sinks` (or `pactl list sources`) and

31 constructing a set of tuples (sink-source-name, sink-source-active-port,

32 sink-source-availability). Any change to the computed set, as compared to the

33 initially computed set, is considered a match.

34

35 Due to the algorithm used, it will also detect things like USB headsets, HDMI

36 monitors/speakers, webcams, etc.

37

38 The script depends on:

39 python3-checkbox-support

40 Which depends on:

41 python3-pyparsing

42"""

43 import argparse

44 import os

45 import pty

46 import signal

47 import subprocess

48

49 from checkbox_support.parsers.pactl import parse_pactl_output

50 from checkbox_support.snap_utils.system import in_classic_snap

51

52

53 class AudioPlugDetection:

54

55 def __init__(self, timeout, mode):

56 # store parameters

57 self.timeout = timeout

58 self.mode = mode

59 # get the un-localized environment

60 env = dict(os.environb)

61 env[b"LANG"] = b""

62 env[b"LANGUAGE"] = b""

63 env[b"LC_ALL"] = b"C.UTF-8"

64 if in_classic_snap():

65 prp = "/run/user/{}/snap.{}/../pulse".format(

66 os.geteuid(), os.getenv("SNAP_NAME")

67 )

68 env[b"PULSE_RUNTIME_PATH"] = prp

69 self.unlocalized_env = env

70 # set SIGALRM handler

71 signal.signal(signal.SIGALRM, self.on_timeout)

72

73 def get_sound_config(self):

74 text = subprocess.check_output(

75 ["pactl", "list", self.mode], # either 'sources' or 'sinks'

76 env=self.unlocalized_env,

77 universal_newlines=True,

78 )

79 doc = parse_pactl_output(text)

80 cfg = set()

81 for record in doc.record_list:

82 active_port = None

83 port_availability = None

84 # We go through the attribute list once to try to find

85 # an active port

86 for attr in record.attribute_list:

87 if attr.name == "Active Port":

88 active_port = attr.value

89 # If there is one, we retrieve its availability flag

90 if active_port:

91 for attr in record.attribute_list:

92 if attr.name == "Ports":

93 for port in attr.value:

94 if port.name == active_port:

95 port_availability = port.availability

96 cfg.add((record.name, active_port, port_availability))

97 return cfg

98

99 def on_timeout(self, signum, frame):

100 print("Time is up")

101 raise SystemExit(1)

102

103 @classmethod

104 def main(cls):

105 parser = argparse.ArgumentParser(

106 description=__doc__.split(" ")[0],

107 epilog=__doc__.split(" ")[1],

108 formatter_class=argparse.RawDescriptionHelpFormatter,

109 )

110 parser.add_argument(

111 "mode",

112 choices=["sinks", "sources"],

113 help="Monitor either sinks or sources",

114 )

115 parser.add_argument(

116 "-t",

117 "--timeout",

118 type=int,

119 default=30,

120 help="Timeout after which the script fails",

121 )

122 ns = parser.parse_args()

123 return cls(ns.timeout, ns.mode).run()

124

125 def run(self):

126 found = False

127 if self.mode == "sinks":

128 look_for = "Event 'change' on sink #"

129 look_for2 = "Event 'change' on server #"

130 elif self.mode == "sources":

131 look_for = "Event 'change' on source #"

132 look_for2 = "Event 'change' on server #"

133 else:

134 assert False

135 # Get the initial / baseline configuration

136 initial_cfg = self.get_sound_config()

137 print("Starting with config: {}".format(initial_cfg))

138 print(

139 "You have {} seconds to plug the item in".format(self.timeout),

140 flush=True,

141 )

142 # Start the timer

143 signal.alarm(self.timeout)

144 # run subscribe in a pty as it doesn't fflush() after every event

145 pid, master_fd = pty.fork()

146 if pid == 0:

147 os.execlpe("pactl", "pactl", "subscribe", self.unlocalized_env)

148 else:

149 child_stream = os.fdopen(master_fd, "rt", encoding="UTF-8")

150 try:

151 for line in child_stream:

152 if line.startswith(look_for) or line.startswith(look_for2):

153 new_cfg = self.get_sound_config()

154 print("Now using config: {}".format(new_cfg))

155 if new_cfg != initial_cfg:

156 print("It seems to work!")

157 found = True

158 break

159 except KeyboardInterrupt:

160 pass

161 finally:

162 os.kill(pid, signal.SIGTERM)

163 os.close(master_fd)

164 return 0 if found else 1

165

166

167 if __name__ == "__main__":

168 raise SystemExit(AudioPlugDetection.main())