From 97676ac0e16ce32d1c2d139ddd12889d1185066f Mon Sep 17 00:00:00 2001 From: Pablo Diaz Date: Wed, 3 Jan 2024 22:21:10 -0500 Subject: [PATCH] update to ruff linter and asserts --- LICENSE | 2 +- pipewire_python/_utils.py | 6 ++- pipewire_python/link.py | 82 +++++++++++++++++++++++---------------- pyproject.toml | 4 +- tests/test_interfaces.py | 9 +++-- tests/test_links.py | 13 +++++-- tests/test_playback.py | 1 + tests/test_targets.py | 5 ++- 8 files changed, 74 insertions(+), 48 deletions(-) diff --git a/LICENSE b/LICENSE index 47a7c64..02ac1e7 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ The MIT License (MIT) -Copyright (c) 2023 Pablo Diaz +Copyright (c) 2024 Repository Team Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/pipewire_python/_utils.py b/pipewire_python/_utils.py index deb3aad..94044ea 100644 --- a/pipewire_python/_utils.py +++ b/pipewire_python/_utils.py @@ -111,7 +111,9 @@ def _execute_shell_command( # ) with subprocess.Popen( - command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT # Example ['ls ','l'] + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, # Example ['ls ','l'] ) as terminal_subprocess: # Execute command depending or not in timeout try: @@ -246,7 +248,7 @@ def _generate_dict_interfaces( data = line.replace("\t ", "").split(" ") third_level = str(data[0]) - if type(mydict[first_level]["params"]) != dict: + if not isinstance(mydict[first_level]["params"], dict): mydict[first_level]["params"] = {} mydict[first_level]["params"][third_level] = { "spa": data[1], diff --git a/pipewire_python/link.py b/pipewire_python/link.py index 7261258..f262cea 100644 --- a/pipewire_python/link.py +++ b/pipewire_python/link.py @@ -76,6 +76,7 @@ class InvalidLink(ValueError): """Invalid link configuration.""" + class FailedToLinkPorts(ValueError): """Failed to Link the Specified Ports.""" @@ -139,14 +140,18 @@ def _join_arguments(self, other: "Port", message: str) -> List[str]: def connect(self, other: "Port") -> None: """Connect this channel to another channel.""" - args = self._join_arguments(other=other, message="Cannot connect an {} to another {}.") + args = self._join_arguments( + other=other, message="Cannot connect an {} to another {}." + ) stdout, _ = _execute_shell_command(args) if b"failed to link ports" in stdout: raise FailedToLinkPorts(stdout) def disconnect(self, other: "Port") -> None: """Disconnect this channel from another.""" - args = self._join_arguments(other=other, message="Cannot disconnect an {} from another {}.") + args = self._join_arguments( + other=other, message="Cannot disconnect an {} from another {}." + ) args.append("--disconnect") _ = _execute_shell_command(args) @@ -480,7 +485,7 @@ def list_inputs(pair_stereo: bool = True) -> List[Union[StereoInput, Input]]: """ ports = [] - inputs=_split_id_from_data("--input") + inputs = _split_id_from_data("--input") if len(inputs) == 0: return ports @@ -624,26 +629,33 @@ def list_links() -> List[Link]: port_type=PortType.OUTPUT if direction == "|<-" else PortType.INPUT, ) if side_a_port.port_type == PortType.INPUT: - links.append(Link( - input=side_a_port, - output=side_b_port, - id=int(link_data_lines[i][0]) - )) + links.append( + Link( + input=side_a_port, + output=side_b_port, + id=int(link_data_lines[i][0]), + ) + ) else: - links.append(Link( - input=side_b_port, - output=side_a_port, - id=int(link_data_lines[i][0]) - )) + links.append( + Link( + input=side_b_port, + output=side_a_port, + id=int(link_data_lines[i][0]), + ) + ) i += 1 if i == num_link_lines: break # Determine if Next Line is Associated with this Link - if (not "|->" in link_data_lines[i][1] and - not "|<-" in link_data_lines[i][1]): - break # Continue to Next Link Group + if ( + "|->" not in link_data_lines[i][1] + and "|<-" not in link_data_lines[i][1] + ): + break # Continue to Next Link Group return links + def list_link_groups() -> List[LinkGroup]: """ List the Groped Links Available on System. @@ -691,29 +703,31 @@ def list_link_groups() -> List[LinkGroup]: port_type=PortType.OUTPUT if direction == "|<-" else PortType.INPUT, ) if side_a_port.port_type == PortType.INPUT: - links.append(Link( - input=side_a_port, - output=side_b_port, - id=int(link_data_lines[i][0]) - )) + links.append( + Link( + input=side_a_port, + output=side_b_port, + id=int(link_data_lines[i][0]), + ) + ) else: - links.append(Link( - input=side_b_port, - output=side_a_port, - id=int(link_data_lines[i][0]) - )) + links.append( + Link( + input=side_b_port, + output=side_a_port, + id=int(link_data_lines[i][0]), + ) + ) i += 1 if i == num_link_lines: break # Determine if Next Line is Associated with this Link - if (not "|->" in link_data_lines[i][1] and - not "|<-" in link_data_lines[i][1]): - break # Continue to Next Link Group + if ( + "|->" not in link_data_lines[i][1] + and "|<-" not in link_data_lines[i][1] + ): + break # Continue to Next Link Group link_groups.append( - LinkGroup( - common_device=side_a_device, - common_name=side_a_name, - links=links - ) + LinkGroup(common_device=side_a_device, common_name=side_a_name, links=links) ) return link_groups diff --git a/pyproject.toml b/pyproject.toml index afd7ff2..9570deb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,13 +5,13 @@ build-backend = "flit_core.buildapi" [project] name = "pipewire_python" authors = [ - {name = "Pablo Diaz", email = "pablo.diazv@pucp.edu.pe"}, + {name = "Pablo Diaz"}, {name = "Anna Absi", email = "anna.absi@gmail.com"}, {name = "mrteathyme"}, {name = "Joe Stanley", email = "engineerjoe440@yahoo.com"} ] maintainers = [ - {name = "Pablo Diaz", email = "pablo.diazv@pucp.edu.pe"} + {name = "Pablo Diaz"} ] license = {file = "LICENSE"} home-page = "https://github.com/pablodz/pipewire_python" diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py index 66f3e3a..8134698 100644 --- a/tests/test_interfaces.py +++ b/tests/test_interfaces.py @@ -2,7 +2,6 @@ def test_interfaces(): - # Client audio_controller = Controller() list_interfaces_client = audio_controller.get_list_interfaces( @@ -11,7 +10,9 @@ def test_interfaces(): print(list_interfaces_client) # check if dict - assert type(list_interfaces_client) is dict + assert isinstance( + list_interfaces_client, dict + ), "list_interfaces_client should be of type dict" # not empty dict # empty on CI/CD assert len(list_interfaces_client) >= 0 @@ -23,7 +24,9 @@ def test_interfaces(): ) print(list_interfaces_client) # check if dict - assert type(list_interfaces_client) is dict + assert isinstance( + list_interfaces_client, dict + ), "list_interfaces_client should be of type dict" # not empty dict # empty on CI/CD assert len(list_interfaces_client) >= 0 diff --git a/tests/test_links.py b/tests/test_links.py index 5650986..e024b68 100644 --- a/tests/test_links.py +++ b/tests/test_links.py @@ -1,14 +1,18 @@ from pipewire_python.link import ( - list_inputs, list_outputs, list_links, list_link_groups, - StereoInput, StereoOutput + list_inputs, + list_outputs, + list_links, + list_link_groups, + StereoInput, + StereoOutput, ) def test_list(): """Test that the lists provide some devices and that disconnecting clears them.""" - inputs=list_inputs() + inputs = list_inputs() if len(inputs) == 0: - return # No inputs, no point in testing + return # No inputs, no point in testing assert list_inputs() assert list_outputs() @@ -23,6 +27,7 @@ def test_list(): assert len(list_links()) == 0 + def test_connect_disconnect(): """Test that all points quickly connect then disconnect.""" links = [] diff --git a/tests/test_playback.py b/tests/test_playback.py index e153a80..d8fc7be 100644 --- a/tests/test_playback.py +++ b/tests/test_playback.py @@ -8,6 +8,7 @@ # with open("beers.wav", 'w') as file: # file.write(response.text) + ######################### # PLAYBACK # ######################### diff --git a/tests/test_targets.py b/tests/test_targets.py index 3b0d961..0f6b3d2 100644 --- a/tests/test_targets.py +++ b/tests/test_targets.py @@ -2,14 +2,15 @@ def test_interfaces(): - # Client audio_controller = Controller() list_targets_client = audio_controller.get_list_targets() print(list_targets_client) # check if dict - assert type(list_targets_client) is dict + assert isinstance( + list_targets_client, dict + ), "list_targets_client should be of type dict" # not empty dict # empty on CI/CD assert len(list_targets_client) >= 0