1#!/usr/bin/env python3

2# This file is part of Checkbox.

3#

4# Copyright 2023 Canonical Ltd.

5# Written by:

6# Hanhsuan Lee <hanhsuan.lee@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 from enum import IntEnum

21 import subprocess

22 import argparse

23 import logging

24 import difflib

25 import time

26 import json

27 import sys

28 import re

29 import gi

30

31 gi.require_version("Gst", "1.0")

32 gi.require_version("GLib", "2.0")

33 from gi.repository import Gst # noqa: E402

34 from gi.repository import GLib # noqa: E402

35

36

37 class PipewireTestError(IntEnum):

38 """

39 A class used to define PipewireTest Error code

40

41 :attr NO_ERROR: process success

42 :type NO_ERROR: int

43

44 :attr NOT_DETECTED: couldn't find specific device

45 :type NOT_DETECTED: int

46

47 :attr NO_AVAILABLE_PORT: couldn't find available port

48 :type NO_AVAILABLE_PORT: int

49

50 :attr NO_SPECIFIC_DEVICE: couldn't find specific device

51 :type NO_SPECIFIC_DEVICE: int

52

53 :attr PIPELINE_PROCESS_FAIL: gst pipeline process failed

54 :type PIPELINE_PROCESS_FAIL: int

55

56 :attr NO_CHANGE_DETECTED: couldn't detect audio setting is changed

57 :type NO_CHANGE_DETECTED: int

58 """

59

60 NO_ERROR = 0

61 NOT_DETECTED = -1

62 NO_AVAILABLE_PORT = -2

63 NO_SPECIFIC_DEVICE = -3

64 PIPELINE_PROCESS_FAIL = -4

65 NO_CHANGE_DETECTED = -5

66

67

68 class PipewireTest:

69 """

70 A class used to test pipewire functions

71

72 """

73

74 logger = logging.getLogger()

75

76 def _get_pw_type(self, media_class) -> str:

77 """

78 convert sink to Output and source to Input

79

80 :param media_class: sink(s) or source(s)

81 :type media_class: str

82

83 :returns: "Output", "Input" or "UNKNOWN CLASS"

84 :"rtype": str

85 """

86 if media_class.lower() in ["sink", "sinks"]:

87 return "Output"

88 elif media_class.lower() in ["source", "sources"]:

89 return "Input"

90 else:

91 self.logger.info("Media class:[{}] is unknown".format(media_class))

92 return "UNKNOWN CLASS"

93

94 def _get_pw_dump(self, p_type) -> dict:

95 """

96 Use to convert the json output of pw-dump to dict object

97

98 :param p_type: pipewire object type, such as "Node"

99 :type p_type: str

100

101 :returns: pw-dump in dict data structure

102 :"rtype": dict

103 """

104 pw_dump = subprocess.check_output(

105 "pw-dump {}".format(p_type), shell=True, universal_newlines=True

106 )

107 try:

108 return json.loads(pw_dump)

109 except (json.decoder.JSONDecodeError, TypeError):

110 self.logger.error("pw-dump {} failed !!!".format(p_type))

111 return {}

112

113 def generate_pw_media_class(self, media_type, media_class) -> str:

114 """

115 Combine media_type and media_class to pw-dump format,

116 such as "Audio/Sink".

117

118 :param media_type: For now only support Audio or Video

119 :type media_type: str

120

121 :param media_class: For now only support Sinks* or Sources*

122 :type media_class: str

123

124 :returns: pipewire style media.class or "UNKNOWN CLASS"

125 :"rtype": str

126 """

127 if media_type.lower() == "audio":

128 mtype = "Audio"

129 elif media_type.lower() == "video":

130 mtype = "Video"

131 else:

132 self.logger.info("Media type:[{}] is unknown".format(media_type))

133 return "UNKNOWN TYPE"

134

135 if media_class.lower() in ["sink", "sinks"]:

136 return "{}/Sink".format(mtype)

137 elif media_class.lower() in ["source", "sources"]:

138 return "{}/Source".format(mtype)

139 else:

140 self.logger.info("Media class:[{}] is unknown".format(media_class))

141 return "UNKNOWN CLASS"

142

143 def detect_device(self, media_type, media_class) -> int:

144 """

145 detect specific device is on this system or not.

146 This function parse output of pw-dump to check, the device type

147 equals PipeWire:Interface:Node and media.class equals args.

148

149 :param media_type: For now only support Audio or Video

150 :type media_type: str

151

152 :param media_class: For now only support Sinks* or Sources*

153 :type media_class: str

154

155 :returns: "NOT_DETECTED" or "NO_ERROR"

156 :"rtype": int

157 """

158 mclass = self.generate_pw_media_class(media_type, media_class)

159 if mclass in ["UNKNOWN CLASS", "UNKNOWN TYPE"]:

160 return PipewireTestError.NOT_DETECTED

161 clients = self._get_pw_dump("Node")

162

163 detected_flag = False

164 for client in clients:

165 props = client["info"]["props"]

166 if mclass == props.get("media.class"):

167 self.logger.info(

168 "device id:[{}] media.class:[{}]"

169 " node.name:[{}]".format(

170 client["id"], mclass, props.get("node.name")

171 )

172 )

173 detected_flag = True

174 if detected_flag:

175 return PipewireTestError.NO_ERROR

176 self.logger.info("media.class:[{}] couldn't find".format(mclass))

177 return PipewireTestError.NOT_DETECTED

178

179 def select_device(self, media_type, media_class, device):

180 """

181 Set desired device as default

182 This function parse output of pw-dump to find, the device type

183 equals PipeWire:Interface:Node and media.class equals args.

184

185 :param media_type: For now only support Audio or Video

186 :type media_type: str

187

188 :param media_class: For now only support Sinks* or Sources*

189 :type media_class: str

190

191 :param device: device type, such as hdmi, usb and bluez etc.

192 :type device: str

193 """

194 mclass = self.generate_pw_media_class(media_type, media_class)

195 if mclass in ["UNKNOWN CLASS", "UNKNOWN TYPE"]:

196 return PipewireTestError.NO_AVAILABLE_PORT

197 clients = self._get_pw_dump("Node")

198

199 available_nodes = {}

200 for client in clients:

201 props = client["info"]["props"]

202 name = props.get("node.name")

203 if mclass == props.get("media.class") and device in name:

204 available_nodes[client["id"]] = client

205

206 if len(available_nodes) < 1:

207 self.logger.error("No available {} found".format(mclass))

208 return PipewireTestError.NO_AVAILABLE_PORT

209 self.logger.info("Available {}:".format(mclass))

210 for i in available_nodes:

211 n = available_nodes[i]

212 desc = n["info"]["props"].get("node.description")

213 self.logger.info("Id:[{}], device:[{}]".format(n["id"], desc))

214

215 chosen = False

216 node_id = None

217 while not chosen:

218 self.logger.info(

219 "Which {} would you like to test?"

220 " -1 means don't change".format(mclass)

221 )

222 self.logger.info(" {} id:".format(mclass))

223 node_id = input()

224 try:

225 chosen = int(node_id) in available_nodes

226 except ValueError:

227 chosen = False

228 if chosen:

229 cmd = "wpctl set-default {}".format(node_id)

230 subprocess.check_output(

231 cmd, shell=True, universal_newlines=True

232 )

233 elif node_id == "-1":

234 chosen = True

235 else:

236 self.logger.info(" [{}] isn't existed!".format(node_id))

237 return PipewireTestError.NO_ERROR

238

239 def _check_state(self, device) -> bool:

240 """

241 Checks whether the sink is available for the given device.

242 This function parse output of pw-dump to find the device type

243 equals PipeWire:Interface:Device and media.class equals Audio/Device.

244 For pipewire, the active port will be listed under info.params.Route.

245 Therefore, you could check this object to know the state of it.

246

247 :param device: device you would like to check

248 :type device: str

249 """

250 clients = self._get_pw_dump("Device")

251 try:

252 for client in clients:

253 mclass = client["info"]["props"].get("media.class")

254 if mclass == "Audio/Device": 254 ↛ 252 line 254 didn't jump to line 252 because the condition on line 254 was always true

255 for route in client["info"]["params"]["Route"]:

256 name = route["name"]

257 available = route["available"]

258 if (

259 device.lower() in name.lower()

260 and "output" in route["direction"].lower()

261 and available in ["unknown", "yes"]

262 ):

263 self.logger.info("[ Audio sink ]".center(80, "="))

264 self.logger.info(

265 "Device: [{}] available: [{}]".format(

266 route["description"], available

267 )

268 )

269 return True

270 raise ValueError(

271 "No available output device for {}".format(device)

272 )

273 except (IndexError, ValueError) as e:

274 logging.error(repr(e))

275 return False

276

277 def gst_pipeline(self, pipe, timeout, device) -> int:

278 """

279 Simple GStreamer pipeline player

280

281 :param pipe: Quoted GStreamer pipeline to launch

282 :type pipe: str

283

284 :param timeout: Timeout for running the pipeline

285 :type timeout: int

286

287 :param device: device type, such as hdmi etc.

288 :type device: str

289 """

290 if device: 290 ↛ 294 line 290 didn't jump to line 294 because the condition on line 290 was always true

291 if not self._check_state(device):

292 return PipewireTestError.NO_SPECIFIC_DEVICE

293

294 Gst.init(None)

295 try:

296 self.logger.info(

297 "Attempting to initialize Gstreamer pipeline: {}".format(pipe)

298 )

299 element = Gst.parse_launch(pipe)

300 except GLib.GError as error:

301 self.logger.info("Specified pipeline couldn't be processed.")

302 self.logger.info(

303 "Error when processing pipeline: {}".format(error)

304 )

305 # Exit harmlessly

306 return PipewireTestError.PIPELINE_PROCESS_FAIL

307 self.logger.info("Pipeline initialized, now starting playback.")

308 element.set_state(Gst.State.PLAYING)

309

310 if timeout: 310 ↛ 313 line 310 didn't jump to line 313 because the condition on line 310 was always true

311 time.sleep(timeout)

312

313 element.set_state(Gst.State.NULL)

314

315 return PipewireTestError.NO_ERROR

316

317 def _get_audio_config(self, mode) -> set:

318 """

319 Get simple audio configuration

320 This function parse output of pw-dump to find the device type

321 equals PipeWire:Interface:Device and media.class equals Audio/Device.

322 For pipewire, the active port will be listed under info.params.Route.

323 Therefore, you could check this object to know the state of it.

324

325 :param mode: sink or source

326 :type mode: str

327 """

328 clients = self._get_pw_dump("Device")

329 cfg = set()

330 for client in clients:

331 active_ports = None

332 mclass = client["info"]["props"].get("media.class")

333 if mclass == "Audio/Device": 333 ↛ 335 line 333 didn't jump to line 335 because the condition on line 333 was always true

334 active_ports = client["info"]["params"]["Route"]

335 if active_ports: 335 ↛ 330 line 335 didn't jump to line 330 because the condition on line 335 was always true

336 for p in active_ports:

337 if p["direction"] == self._get_pw_type(mode): 337 ↛ 336 line 337 didn't jump to line 336 because the condition on line 337 was always true

338 cfg.add(

339 (

340 "{} #{}".format(mode, client["id"]),

341 p["name"],

342 p["available"],

343 )

344 )

345 return cfg

346

347 def monitor_active_port_change(self, timeout, mode) -> int:

348 """

349 Monitoring Audio active port changing

350 This script checks if the active port on either sinks

351 (speakers or headphones) or sources (microphones, webcams)

352 is changed after an appropriate device is plugged into the DUT.

353 The script is fully automatic and either times out after or

354 returns as soon as the change is detected.

355

356 :param timeout: Timeout after which the script fails

357 :type timeout: int

358

359 :param mode: Monitor either sinks or sources

360 :type mode: str

361 """

362 initial_cfg = self._get_audio_config(mode)

363 self.logger.info("Starting with config: {}".format(initial_cfg))

364 self.logger.info(

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

366 )

367

368 for _ in range(int(timeout)):

369 new_cfg = self._get_audio_config(mode)

370 if new_cfg != initial_cfg:

371 self.logger.info("Now using config: {}".format(new_cfg))

372 self.logger.info("It seems to work!")

373 return PipewireTestError.NO_ERROR

374 time.sleep(1)

375 self.logger.info("Couldn't detect active port change!")

376 return PipewireTestError.NO_CHANGE_DETECTED

377

378 def go_through_ports(self, cmd, mode):

379 """

380 Go through available ports for testing

381 This script checks if the ports on either sinks

382 (speakers or headphones) or sources (microphones, webcams)

383 is available and working on the DUT.

384

385 :param cmd: command for testing

386 :type cmd: str

387

388 :param mode: Monitor either sinks or sources

389 :type mode: str

390 """

391 clients = self._get_pw_dump("Device")

392 for client in clients:

393 ports = None

394 mclass = client["info"]["props"].get("media.class")

395 if mclass == "Audio/Device": 395 ↛ 397 line 395 didn't jump to line 397 because the condition on line 395 was always true

396 ports = client["info"]["params"]["EnumRoute"]

397 if ports: 397 ↛ 392 line 397 didn't jump to line 392 because the condition on line 397 was always true

398 for p in ports:

399 chosen = None

400 if p["direction"] == self._get_pw_type(mode) and p[ 400 ↛ 398 line 400 didn't jump to line 398 because the condition on line 400 was always true

401 "available"

402 ] in ["yes", "unknown"]:

403 while chosen != "yes":

404 self.logger.info(

405 "Please select [{}] for "

406 "testing (if selected, "

407 "please enter 'yes')".format(p["description"])

408 )

409

410 chosen = input()

411 checked = None

412 while checked != "yes":

413 with subprocess.Popen(

414 cmd,

415 shell=True,

416 stdout=subprocess.PIPE,

417 universal_newlines=True,

418 ) as p:

419 while p.poll() is None:

420 line = p.stdout.readline().strip()

421 self.logger.info(line)

422 p.kill()

423 self.logger.info(

424 "Is working ? "

425 "please enter 'yes' "

426 "to leave"

427 )

428

429 checked = input()

430

431 def _get_node_description(self, properties) -> str:

432 """

433 Get node description from the output of wpctl inspect

434

435 :param properties: output of wpctl inspect

436 :type properties: str

437

438 :returns: the node description

439 :rtype: str

440 """

441 try:

442 for line in properties.splitlines(): 442 ↛ exit line 442 didn't return from function '_get_node_description' because the loop on line 442 didn't complete

443 if "node.description" in line: 443 ↛ 442 line 443 didn't jump to line 442 because the condition on line 443 was always true

444 return line.split("=")[1]

445 except IndexError as e:

446 raise RuntimeError("properties format error {}".format(repr(e)))

447

448 def show_default_device(self, device_type):

449 """

450 show the default device

451

452 :param device_type: audio or video

453 :type device_type: str

454 """

455 device_type = device_type.upper()

456 if device_type not in ["AUDIO", "VIDEO"]:

457 raise ValueError("Only support 'video' and 'audio'")

458 sink_cmd = [

459 "wpctl",

460 "inspect",

461 "@DEFAULT_{}_SINK@".format(device_type),

462 ]

463 source_cmd = [

464 "wpctl",

465 "inspect",

466 "@DEFAULT_{}_SOURCE@".format(device_type),

467 ]

468 self.logger.info("Default input device:")

469 try:

470 source = subprocess.check_output(

471 source_cmd, universal_newlines=True

472 )

473 self.logger.info(self._get_node_description(source))

474 if device_type == "AUDIO":

475 self.logger.info("Default output device:")

476 sink = subprocess.check_output(

477 sink_cmd, universal_newlines=True

478 )

479 self.logger.info(self._get_node_description(sink))

480 self.logger.info(

481 "If these are not you would like to test,"

482 " please change them before testing"

483 )

484 except subprocess.CalledProcessError as e:

485 raise RuntimeError("Show default device error {}".format(repr(e)))

486

487 def _sort_wpctl_status(self, lines: list) -> list:

488 """

489 This method will sort wpctl status for sub-items under catalog only

490

491 :param lines: the wpctl status you would like to sort

492

493 :returns: sorted wpctl status

494 """

495 catalog = [

496 " └─ Clients:",

497 " ├─ Devices:",

498 " ├─ Sinks:",

499 " ├─ Sink endpoints:",

500 " ├─ Sources:",

501 " ├─ Source endpoints:",

502 " └─ Streams:",

503 " └─ Default Configured Node Names:",

504 ]

505 sorted_lines = []

506 sub_items = []

507 for line in lines:

508 clean_line = re.sub(", pid:.*| [0-9]+\\. ", "", line)

509 # only sort needed items

510 if clean_line in catalog:

511 # found next catalog, append sorted sub_items

512 [sorted_lines.append(item) for item in sorted(sub_items)]

513 # clean up after append

514 sub_items = []

515 sorted_lines.append(clean_line)

516 else:

517 sub_items.append(clean_line)

518 [sorted_lines.append(item) for item in sorted(sub_items)]

519 return sorted_lines

520

521 def compare_wpctl_status(self, status_1: str, status_2: str):

522 """

523 This method will sort wpctl status for sub-items under catalog only

524

525 :param status_1: path to first wpctl status

526

527 :param status_2: path to second wpctl status

528 """

529 with open(status_1, "r") as s1, open(status_2, "r") as s2:

530 status_1 = s1.readlines()

531 status_2 = s2.readlines()

532 sorted_status_1 = self._sort_wpctl_status(status_1)

533 sorted_status_2 = self._sort_wpctl_status(status_2)

534 delta = difflib.unified_diff(sorted_status_1, sorted_status_2, n=0)

535 diff = "".join(delta)

536 if diff:

537 self.logger.info("The first status:\n")

538 self.logger.info("".join(status_1))

539 self.logger.info("And the second status:\n")

540 self.logger.info("".join(status_2))

541 self.logger.info(

542 "Differ in the following lines (after sorting):"

543 )

544 self.logger.info(diff)

545 raise SystemExit("The two status don't match !!!")

546

547 def _args_parsing(self, args=sys.argv[1:]):

548 parser = argparse.ArgumentParser(

549 prog="Pipewire validator",

550 description="using for pipewire to valid system functions",

551 )

552

553 subparsers = parser.add_subparsers(dest="test_type")

554 subparsers.required = True

555

556 # Add parser for detecting audio/video function

557 parser_detect = subparsers.add_parser(

558 "detect", help="Detect audio/video devices on this system"

559 )

560 parser_detect.add_argument(

561 "-t",

562 "--type",

563 type=str,

564 default="Audio",

565 help="device type such as Audio "

566 "or Video (default: %(default)s)",

567 )

568 parser_detect.add_argument(

569 "-c",

570 "--clazz",

571 type=str,

572 default="Sink",

573 help="device type such as Sink or "

574 "Source (default: %(default)s)",

575 )

576

577 # Add parser for selecting audio/video function

578 parser_select = subparsers.add_parser(

579 "select", help="Select audio/video devices on this system"

580 )

581 parser_select.add_argument(

582 "-t",

583 "--type",

584 type=str,

585 default="Audio",

586 help="device type such as Audio "

587 "or Video (default: %(default)s)",

588 )

589 parser_select.add_argument(

590 "-c",

591 "--clazz",

592 type=str,

593 default="Sink",

594 help="device type such as Sink or "

595 "Source (default: %(default)s)",

596 )

597 parser_select.add_argument(

598 "-d",

599 "--device",

600 type=str,

601 default="",

602 help="device type such as hdmi or " "bluz (default: %(default)s)",

603 )

604

605 # Add parser for gst pipeline function(Audio only)

606 parser_gst = subparsers.add_parser(

607 "gst", help="Simple GStreamer pipeline player"

608 )

609 parser_gst.add_argument(

610 "PIPELINE", help="Quoted GStreamer pipeline to launch"

611 )

612 parser_gst.add_argument(

613 "-t",

614 "--timeout",

615 type=int,

616 required=True,

617 help="Timeout for running the pipeline",

618 )

619 parser_gst.add_argument(

620 "-d", "--device", type=str, help="Device to check for status"

621 )

622

623 # Add parser for monitor function(Audio only)

624 parser_monitor = subparsers.add_parser(

625 "monitor", help="Monitoring Audio active port changing"

626 )

627 parser_monitor.add_argument(

628 "-t",

629 "--timeout",

630 type=int,

631 required=True,

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

633 )

634 parser_monitor.add_argument(

635 "-m", "--mode", type=str, help="Monitor either sinks or sources"

636 )

637

638 # Add parser for go through function

639 parser_through = subparsers.add_parser(

640 "through", help="Go through available ports for testing"

641 )

642 parser_through.add_argument(

643 "-c",

644 "--command",

645 type=str,

646 required=True,

647 help="command for testing",

648 )

649 parser_through.add_argument(

650 "-m", "--mode", type=str, help="Either sinks or sources"

651 )

652

653 # Add parser for show default device function

654 parser_show = subparsers.add_parser(

655 "show", help="show the default device"

656 )

657 parser_show.add_argument(

658 "-t", "--type", type=str, required=True, help="VIDEO or AUDIO"

659 )

660

661 # Add parser for compare wpctl status function

662 parser_compare = subparsers.add_parser(

663 "compare_wpctl_status", help="compare wpctl status"

664 )

665 parser_compare.add_argument(

666 "-s1",

667 "--status_1",

668 type=str,

669 required=True,

670 help="path to first output of wpctl status",

671 )

672 parser_compare.add_argument(

673 "-s2",

674 "--status_2",

675 type=str,

676 required=True,

677 help="path to second output of wpctl status",

678 )

679

680 return parser.parse_args(args)

681

682 def function_select(self, args):

683 if args.test_type == "detect":

684 # detect_device("audio", "sink")

685 return self.detect_device(args.type, args.clazz)

686 elif args.test_type == "select":

687 # select_device("audio", "sink", "hdmi")

688 return self.select_device(args.type, args.clazz, args.device)

689 elif args.test_type == "gst":

690 # gst_pipeline(PIPELINE, "30", "hdmi")

691 return self.gst_pipeline(args.PIPELINE, args.timeout, args.device)

692 elif args.test_type == "monitor":

693 # monitor_active_port_change("30", "sink")

694 return self.monitor_active_port_change(args.timeout, args.mode)

695 elif args.test_type == "through":

696 # go_through_ports("speaker-test -c 2 -l 1 -t wav", "sink")

697 return self.go_through_ports(args.command, args.mode)

698 elif args.test_type == "show":

699 # show_default_device("AUDIO")

700 return self.show_default_device(args.type)

701 elif args.test_type == "compare_wpctl_status": 701 ↛ exit line 701 didn't return from function 'function_select' because the condition on line 701 was always true

702 # compare_wpctl_status(STATUS_1, STATUS_2)

703 return self.compare_wpctl_status(args.status_1, args.status_2)

704

705

706 if __name__ == "__main__":

707 pw = PipewireTest()

708

709 # create logger formatter

710 log_formatter = logging.Formatter(fmt="%(message)s")

711

712 # set log level

713 pw.logger.setLevel(logging.INFO)

714

715 # create console handler

716 console_handler = logging.StreamHandler()

717 console_handler.setFormatter(log_formatter)

718

719 # Add console handler to logger

720 pw.logger.addHandler(console_handler)

721 sys.exit(pw.function_select(pw._args_parsing()))