Merge pull request #9 from pablodz/dev

#8
pablodz · Oct 11, 2021 · d6d7f0c · d6d7f0c
2 parents e67224c + 0a48f92
commit d6d7f0c
Show file tree

Hide file tree

Showing 7 changed files with 210 additions and 36 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,10 +1,14 @@
+# v0.0.95 (2021-10-10)
+
+Add feature #8
+
 # v0.0.93-94 (2021-09-02)
 
 Fix issue #5
 
 # v0.0.92 (2021-06-23)
 
-ALL functions of usuall way implemented and documented
+ALL functions of usual way implemented and documented
 
 # v0.0.89 (2021-06-21)
 
@@ -24,7 +28,7 @@ ASYNC still not implemented
 
 ### Features
 
-- ASYNCIO support play
+- Asyncio support play
 - Record default 5 seconds and more selected.
 
 ### Issues

diff --git a/Makefile b/Makefile
@@ -27,26 +27,9 @@ publish:  ## Publish to PyPi
 push:  ## Push code with tags
 	git push && git push --tags
 
-# test:  ## Run tests
-# 	python -m pytest -ra
+test:  ## Run tests [LOCALHOST]
+	python -m pytest -ra
 
 tox:  ## Run tox
 	python3 -m tox
 	ls -la
-
-help:  ## Show help message
-	@IFS=$$'\n' ; \
-	help_lines=(`fgrep -h "##" $(MAKEFILE_LIST) | fgrep -v fgrep | sed -e 's/\\$$//' | sed -e 's/##/:/'`); \
-	printf "%s\n\n" "Usage: make [task]"; \
-	printf "%-20s %s\n" "task" "help" ; \
-	printf "%-20s %s\n" "------" "----" ; \
-	for help_line in $${help_lines[@]}; do \
-		IFS=$$':' ; \
-		help_split=($$help_line) ; \
-		help_command=`echo $${help_split[0]} | sed -e 's/^ *//' -e 's/ *$$//'` ; \
-		help_info=`echo $${help_split[2]} | sed -e 's/^ *//' -e 's/ *$$//'` ; \
-		printf '\033[36m'; \
-		printf "%-20s %s" $$help_command ; \
-		printf '\033[0m'; \
-		printf "%s\n" $$help_info; \
-	done
diff --git a/README.md b/README.md
@@ -24,7 +24,7 @@ Python controller, player and recorder via pipewire's commands.
 
 ## Requirements
 
-1. A pipewire version installed (clean or via pulseaudio) is needed, to check if you have pipewire installed and running, run this command, if the output is different, you'll need to [install pipewire](./docs/INSTALL_PIPEWIRE.md):
+1. A Pipewire version installed (clean or via Pulseaudio) is needed, to check if you have pipewire installed and running, run this command, if the output is different, you'll need to [install pipewire](./docs/INSTALL_PIPEWIRE.md):
 
     1. Pipewire versions supported: 0.3.30, 0.3.32+
 
@@ -60,11 +60,13 @@ pip3 install pipewire_python # or pip
 
 ### Tutorial
 
+#### PLAY AND RECORD
+
 ```python
 from pipewire_python.controller import Controller
 
 # [PLAYBACK]: normal way
-audio_controller = Controller(verbose=True)
+audio_controller = Controller()
 audio_controller.set_config(rate=384000,
                             channels=2,
                             _format='f64',
@@ -73,22 +75,39 @@ audio_controller.set_config(rate=384000,
 audio_controller.playback(audio_filename='docs/beers.wav')
 
 # [RECORD]: normal way
-audio_controller = Controller(verbose=True)
+audio_controller = Controller()
 audio_controller.record(audio_filename='docs/5sec_record.wav',
                         timeout_seconds=5)
 ```
+#### GET INTERFACES
+
+```python
+from pipewire_python.controller import Controller
+
+audio_controller = Controller()
+# Return all Client Interfaces on Pipewire
+audio_controller.get_list_interfaces(
+    type_interfaces="Client",
+    filtered_by_type=True,
+)
+# Return all interfaces
+audio_controller.get_list_interfaces(
+    filtered_by_type=False,
+)
+```
+
 
 ## Documentation
 
-You can check the automatic builded documentation [HERE](https://pablodz.github.io/pipewire_python/html/)
+You can check the automatic build documentation [HERE](https://pablodz.github.io/pipewire_python/html/)
 
 ## Roadmap
 
-Future implementations, next steps, API implementation and Control over pipewire directly from python in the [ROADMAP](docs/ROADMAP.md).
+Future implementations, next steps, API implementation and Control over Pipewire directly from python in the [ROADMAP](docs/ROADMAP.md).
 
 ## Contributions
 
-PR, FR and issues are welcome. Changes with PR in `dev` branch please due documentation runs after each commit in `main` branch. Check more [here](docs/NEW_RELEASE.md)
+PR, FR, and issues are welcome. Changes with PR in `dev` branch please due documentation runs after each commit in `main` branch. Check more [here](docs/NEW_RELEASE.md)
 
 ## License
 

diff --git a/pipewire_python/__init__.py b/pipewire_python/__init__.py
@@ -47,17 +47,15 @@
 from pipewire_python.controller import Controller
 
 # PLAYBACK: normal way
-audio_controller = Controller(verbose=True)
+audio_controller = Controller()
 audio_controller.set_config(rate=384000,
                             channels=2,
                             _format='f64',
                             volume=0.98,
                             quality=4,
                             # Debug
                             verbose=True)
-audio_controller.playback(audio_filename='docs/beers.wav',
-                          # Debug
-                          verbose=True)
+audio_controller.playback(audio_filename='docs/beers.wav')
 
 # RECORD: normal way
 audio_controller = Controller(verbose=True)
@@ -69,7 +67,7 @@
 ```
 """
 
-__version__ = "0.0.94"
+__version__ = "0.0.95"
 
 import sys
 

diff --git a/pipewire_python/_utils.py b/pipewire_python/_utils.py
@@ -196,3 +196,99 @@ def _generate_dict_list_targets(
         print(mydict)
 
     return mydict
+
+
+def _generate_dict_interfaces(
+    longstring: str,  # string output of shell
+    # Debug
+    verbose: bool = False,
+):
+    """
+    Function that transform long string of list interfaces
+    to a `dict`
+    """
+
+    mydict = {}
+    text_in_lines = longstring.split("\n")
+    first_level = "X"
+
+    for line in text_in_lines:
+        try:
+            is_interface = True
+            if "id: " in line:
+                # when interface starts
+                regex_id = r"\tid: ([0-9]*)"
+                results_regex_id = re.findall(regex_id, line)
+                is_interface = False
+
+            if is_interface:
+                if "*" in line[:1]:
+                    # delete * on each line at the beginning
+                    line = line[1:]
+                if "\t\t" in line:
+                    # third level data
+                    data = line.replace("\t\t", "")
+                    data = data.split(" = ")
+                    third_level = str(data[0])
+                    data_to_place = " ".join(data[1:]).replace('"', "")
+
+                    if "properties" not in mydict[first_level]:
+                        mydict[first_level]["properties"] = {}
+                    if third_level not in mydict[first_level]["properties"]:
+                        mydict[first_level]["properties"][third_level] = {}
+                    mydict[first_level]["properties"][third_level] = data_to_place
+
+                elif "\t  " in line:
+                    # second level data: params
+
+                    data = line.replace("\t  ", "").split(" ")
+                    third_level = str(data[0])
+                    if type(mydict[first_level]["params"]) != dict:
+                        mydict[first_level]["params"] = {}
+                    mydict[first_level]["params"][third_level] = {
+                        "spa": data[1],
+                        "permissions": data[2],
+                    }
+
+                elif "\t" in line:
+                    # first level data
+                    data = line.replace("\t", "")
+                    data = data.split(": ")
+                    first_level = str(results_regex_id[0])
+                    second_level = str(data[0])
+                    data_to_place = " ".join(data[1:]).replace('"', "")
+
+                    # to_dict
+                    if first_level not in mydict:
+                        mydict[str(first_level)] = {}
+
+                    mydict[first_level][second_level] = data_to_place
+        except Exception as e:
+            print(e)
+
+    if verbose:
+        print(mydict)
+
+    return mydict
+
+
+def _filter_by_type(
+    dict_interfaces: dict,  # interfaecs dict
+    type_interfaces: str,  # string with type
+    # Debug
+    verbose: bool = False,
+):
+    """
+    Function that filters a `dict` by type of interface
+    """
+
+    dict_filtered = {}
+    for key in dict_interfaces:
+        # Filter
+        if type_interfaces in dict_interfaces[key]["type"]:
+            dict_filtered[key] = dict_interfaces[key]
+
+    if verbose:
+        print(dict_filtered)
+
+    return dict_filtered
diff --git a/pipewire_python/controller.py b/pipewire_python/controller.py
@@ -11,11 +11,13 @@
 from ._utils import (
     _drop_keys_with_none_values,
     _execute_shell_command,
+    _filter_by_type,
     _generate_command_by_dict,
+    _generate_dict_interfaces,
+    _generate_dict_list_targets,
     _get_dict_from_stdout,
     _print_std,
     _update_dict_by_dict,
-    _generate_dict_list_targets,
 )
 
 # Loading constants Constants.py
@@ -483,10 +485,53 @@ def get_list_targets(
         }
         ```
         """
-
         if verbose:
-            print(self._pipewire_list_targets)
-        return self._pipewire_list_targets
+            print(self._pipewire_list_currently_using)
+        return self._pipewire_list_currently_using
+
+    def get_list_interfaces(
+        self,
+        filtered_by_type: str = True,
+        type_interfaces: str = "Client",
+        # Debug
+        verbose: bool = False,
+    ):
+        """Returns a list of applications currently using pipewire on Client.
+        An example of pw-cli usage is the code below:
+
+        ```bash
+        #!/bin/bash
+        pw-cli ls Client
+        ```
+        Args:
+            filtered_by_type : If False, returns all. If not, returns a fitered dict
+            type_interfaces : Set type of Interface ["Client","Link","Node","Factory","Module","Metadata","Endpoint","Session","Endpoint Stream","EndpointLink","Port"]
+
+        Returns:
+            - dict_interfaces_filtered: dictionary with list of interfaces matching conditions
+
+        Examples:
+        ```python
+        >>> Controller().get_list_interfaces()
+
+        ```
+        """
+        mycommand = ["pw-cli", "info", "all"]
+
+        # if verbose:
+        #     print(f"[mycommand]{mycommand}")
+
+        stdout, _ = _execute_shell_command(command=mycommand, timeout=-1, verbose=verbose)
+        dict_interfaces = _generate_dict_interfaces(longstring=stdout.decode(), verbose=verbose)
+
+        if filtered_by_type:
+            dict_interfaces_filtered = _filter_by_type(
+                dict_interfaces=dict_interfaces, type_interfaces=type_interfaces
+            )
+        else:
+            dict_interfaces_filtered = dict_interfaces
+
+        return dict_interfaces_filtered
 
     def playback(
         self,

diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py
@@ -0,0 +1,29 @@
+from pipewire_python.controller import Controller
+
+
+def test_interfaces():
+
+    # Client
+    audio_controller = Controller()
+    list_interfaces_client = audio_controller.get_list_interfaces(
+        type_interfaces="Client", filtered_by_type=True, verbose=True
+    )
+
+    print(list_interfaces_client)
+    # check if dict
+    assert type(list_interfaces_client) == dict
+    # not empty dict
+    # empty on CI/CD
+    assert len(list_interfaces_client) >= 0
+
+    # All
+    audio_controller = Controller()
+    list_interfaces_client = audio_controller.get_list_interfaces(
+        filtered_by_type=False, verbose=True
+    )
+    print(list_interfaces_client)
+    # check if dict
+    assert type(list_interfaces_client) == dict
+    # not empty dict
+    # empty on CI/CD
+    assert len(list_interfaces_client) >= 0