← Back to branch summary

~jocave/checkbox/hybrid-amd-gpu-mods

« back to all changes in this revision

Viewing changes to providers/plainbox-provider-checkbox/bin/pulse-active-port-change

  • Committer: Zygmunt Krynicki
  • Date: 2013-05-29 07:50:30 UTC
  • mto: This revision was merged to the branch mainline in revision 2153.
  • Revision ID: zygmunt.krynicki@canonical.com-20130529075030-ngwz245hs2u3y6us
checkbox: move current checkbox code into checkbox-old

This patch cleans up the top-level directory of the project into dedicated
sub-project directories. One for checkbox-old (the current checkbox and all the
associated stuff), one for plainbox and another for checkbox-ng.

There are some associated changes, such as updating the 'source' mode of
checkbox provider in plainbox, and fixing paths in various test scripts that we
have.

Signed-off-by: Zygmunt Krynicki <zygmunt.krynicki@canonical.com>

Show diffs side-by-side

added added

removed removed

Lines of Context:
providers/plainbox-provider-checkbox/bin/pulse-active-port-change
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
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
 
 
51
 
 
52
 
class AudioPlugDetection:
53
 
 
54
 
    def __init__(self, timeout, mode):
55
 
        # store parameters
56
 
        self.timeout = timeout
57
 
        self.mode = mode
58
 
        # get the un-localized environment
59
 
        env = dict(os.environb)
60
 
        env[b'LANG'] = b''
61
 
        env[b'LANGUAGE'] = b''
62
 
        env[b'LC_ALL'] = b'C.UTF-8'
63
 
        self.unlocalized_env = env
64
 
        # set SIGALRM handler
65
 
        signal.signal(signal.SIGALRM, self.on_timeout)
66
 
 
67
 
    def get_sound_config(self):
68
 
        text = subprocess.check_output(
69
 
            ["pactl", "list", self.mode],  # either 'sources' or 'sinks'
70
 
            env=self.unlocalized_env, universal_newlines=True)
71
 
        doc = parse_pactl_output(text)
72
 
        cfg = set()
73
 
        for record in doc.record_list:
74
 
             active_port = None
75
 
             port_availability = None
76
 
             # We go through the attribute list once to try to find an active port
77
 
             for attr in record.attribute_list:
78
 
                 if attr.name == "Active Port":
79
 
                     active_port = attr.value
80
 
             # If there is one, we retrieve its availability flag
81
 
             if active_port:
82
 
                 for attr in record.attribute_list:
83
 
                     if attr.name == "Ports":
84
 
                         for port in attr.value:
85
 
                             if port.name == active_port:
86
 
                                 port_availability = port.availability
87
 
                 cfg.add((record.name, active_port, port_availability))
88
 
        return cfg
89
 
 
90
 
    def on_timeout(self, signum, frame):
91
 
        print("Time is up")
92
 
        raise SystemExit(1)
93
 
 
94
 
    @classmethod
95
 
    def main(cls):
96
 
        parser = argparse.ArgumentParser(
97
 
            description=__doc__.split(" ")[0],
98
 
            epilog=__doc__.split(" ")[1],
99
 
            formatter_class=argparse.RawDescriptionHelpFormatter)
100
 
        parser.add_argument(
101
 
            'mode', choices=['sinks', 'sources'],
102
 
            help='Monitor either sinks or sources')
103
 
        parser.add_argument(
104
 
            '-t', '--timeout', type=int, default=30,
105
 
            help='Timeout after which the script fails')
106
 
        ns = parser.parse_args()
107
 
        return cls(ns.timeout, ns.mode).run()
108
 
 
109
 
    def run(self):
110
 
        found = False
111
 
        if self.mode == 'sinks':
112
 
            look_for = "Event 'change' on sink #"
113
 
            look_for2 = "Event 'change' on server #"
114
 
        elif self.mode == 'sources':
115
 
            look_for = "Event 'change' on source #"
116
 
            look_for2 = "Event 'change' on server #"
117
 
        else:
118
 
            assert False
119
 
        # Get the initial / baseline configuration
120
 
        initial_cfg = self.get_sound_config()
121
 
        print("Starting with config: {}".format(initial_cfg))
122
 
        print("You have {} seconds to plug something".format(self.timeout))
123
 
        # Start the timer
124
 
        signal.alarm(self.timeout)
125
 
        # run subscribe in a pty as it doesn't fflush() after every event
126
 
        pid, master_fd = pty.fork()
127
 
        if pid == 0:
128
 
            os.execlpe("pactl", "pactl", "subscribe", self.unlocalized_env)
129
 
        else:
130
 
            child_stream = os.fdopen(master_fd, "rt", encoding='UTF-8')
131
 
            try:
132
 
                for line in child_stream:
133
 
                    if line.startswith(look_for) or line.startswith(look_for2):
134
 
                        new_cfg = self.get_sound_config()
135
 
                        print("Now using config: {}".format(new_cfg))
136
 
                        if new_cfg != initial_cfg:
137
 
                            print("It seems to work!")
138
 
                            found = True
139
 
                            break
140
 
            except KeyboardInterrupt:
141
 
                pass
142
 
            finally:
143
 
                os.kill(pid, signal.SIGTERM)
144
 
                os.close(master_fd)
145
 
        return 0 if found else 1
146
 
 
147
 
 
148
 
if __name__ == "__main__":
149
 
    raise SystemExit(AudioPlugDetection.main())