Edit on GitHub

pipewire_python.controller

PIPEWIRE's Python controller (wrapper)

In the next pages you'll see documentation of each Python component controller.py.

View Source

  1"""
  2PIPEWIRE's Python controller (wrapper)
  3
  4In the next pages you'll see documentation of each Python component
  5`controller.py`.
  6"""
  7
  8# import warnings
  9
 10# Loading constants Constants.py
 11from pipewire_python._constants import (
 12    MESSAGES_ERROR,
 13    RECOMMENDED_FORMATS,
 14    RECOMMENDED_RATES,
 15)
 16
 17# Loading internal functions
 18from pipewire_python._utils import (
 19    _drop_keys_with_none_values,
 20    _execute_shell_command,
 21    _filter_by_type,
 22    _generate_command_by_dict,
 23    _generate_dict_interfaces,
 24    _generate_dict_list_targets,
 25    _get_dict_from_stdout,
 26)
 27
 28# [DEPRECATED] [FLAKE8] TO_AVOID_F401 PEP8
 29# [DEPRECATED] https://stackoverflow.com/a/31079085/10491422
 30# NOW USED IN DOCUMENTATION
 31# __all__ = [
 32#     # Classes and fucntions to doc
 33#     'Controller',
 34#     # [DEPRECATED] Unused files pylint
 35#     # "_print_std",
 36#     # "_get_dict_from_stdout",
 37#     # "_update_dict_by_dict",
 38#     # "_drop_keys_with_none_values",
 39#     # "_generate_command_by_dict",
 40#     # "_execute_shell_command",
 41# ]
 42
 43
 44class Controller:
 45    """
 46    Class that controls pipewire command line interface
 47    with shell commands, handling outputs, loading default
 48    configs and more.
 49    """
 50
 51    _pipewire_cli = {  # Help
 52        "--help": "--help",  # -h
 53        "--version": "--version",
 54        "--remote": None,  # -r
 55    }
 56
 57    _pipewire_modes = {  # Modes
 58        "--playback": None,  # -p
 59        "--record": None,  # -r
 60        "--midi": None,  # -m
 61    }
 62
 63    _pipewire_list_targets = {  # "--list-targets": None,
 64        "list_playback": None,
 65        "list_record": None,
 66    }
 67
 68    _pipewire_configs = {  # Configs
 69        "--media-type": None,  # *default=Audio
 70        "--media-category": None,  # *default=Playback
 71        "--media-role": None,  # *default=Music
 72        "--target": None,  # *default=auto
 73        "--latency": None,  # *default=100ms (SOURCE FILE if not specified)
 74        "--rate": None,  # *default=48000
 75        "--channels": None,  # [1,2] *default=2
 76        "--channel-map": None,  # ["stereo", "surround-51", "FL,FR"...] *default="FL,FR"
 77        "--format": None,  # [u8|s8|s16|s32|f32|f64] *default=s16
 78        "--volume": None,  # [0.0,1.0] *default=1.000
 79        "--quality": None,  # -q # [0,15] *default=4
 80        "--verbose": None,  # -v
 81    }
 82
 83    _kill_pipewire = {
 84        "all": ["kill", "$(pidof pw-cat)"],
 85        "playback": ["kill", "$(pidof pw-play)"],
 86        "record": ["kill", "$(pidof pw-record)"],
 87    }
 88
 89    def __init__(
 90        self,
 91        # Debug
 92        verbose: bool = False,
 93    ):
 94        """This constructor load default configs from OS executing
 95        the following pipewire command
 96
 97        ```bash
 98        #!/bin/bash
 99        # Get defaults from output of:
100        pw-cat -h
101        ```
102        """
103        # LOAD ALL DEFAULT PARAMETERS
104
105        mycommand = ["pw-cat", "-h"]
106
107        # get default parameters with help
108        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)  # stderr
109        # convert stdout to dictionary
110        dict_default_values = _get_dict_from_stdout(
111            stdout=str(stdout.decode()), verbose=verbose
112        )
113
114        if verbose:
115            print(self._pipewire_configs)
116
117        # Save default system configs to our json
118        self._pipewire_configs.update(
119            ([(key, dict_default_values[key]) for key in dict_default_values])
120        )
121
122        if verbose:
123            print(self._pipewire_configs)
124
125        # Delete keys with None values
126        self._pipewire_configs = _drop_keys_with_none_values(self._pipewire_configs)
127
128        if verbose:
129            print(self._pipewire_configs)
130
131        # Load values of list targets
132        self.load_list_targets(mode="playback", verbose=verbose)
133        self.load_list_targets(mode="record", verbose=verbose)
134
135    def _help_cli(
136        self,
137        # Debug
138        verbose: bool = True,
139    ):
140        """Get pipewire command line help"""
141
142        mycommand = ["pipewire", self._pipewire_cli["--help"]]
143
144        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)  # stderr
145
146        return stdout
147
148    def get_version(
149        self,
150        # Debug
151        verbose: bool = False,
152    ):
153        """Get version of pipewire installed on OS by executing the following
154        code:
155
156        ```bash
157        #!/bin/bash
158        pw-cli --version
159        ```
160
161        Args:
162            verbose (bool) : True enable debug logs. *default=False
163
164        Returns:
165            - versions (list) : Versions of pipewire compiled
166        """
167
168        mycommand = ["pw-cli", "--version"]
169
170        if verbose:
171            print(f"[mycommand]{mycommand}")
172
173        stdout, _ = _execute_shell_command(
174            command=mycommand, timeout=-1, verbose=verbose
175        )
176        versions = stdout.decode().split("\n")[1:]
177
178        self._pipewire_cli["--version"] = versions
179
180        return versions
181
182    def verbose(
183        self,
184        status: bool = True,
185    ):
186        """Get full log of pipewire stream status with the command `pw-cat`
187
188        An example of pw-cli usage is the code below:
189
190        ```bash
191        #!/bin/bash
192        # For example
193        pw-cat --playback beers.wav --verbose
194        ```
195
196        that will generate an output like this:
197
198        ```bash
199        opened file "beers.wav" format 00010002 channels:2 rate:44100
200        using default channel map: FL,FR
201        rate=44100 channels=2 fmt=s16 samplesize=2 stride=4 latency=4410 (0.100s)
202        connecting playback stream; target_id=4294967295
203        stream state changed unconnected -> connecting
204        stream param change: id=2
205        stream properties:
206            media.type = "Audio"
207            ...
208        now=0 rate=0/0 ticks=0 delay=0 queued=0
209        remote 0 is named "pipewire-0"
210        core done
211        stream state changed connecting -> paused
212        stream param change: id=2
213        ...
214        stream param change: id=15
215        stream param change: id=15
216        now=13465394419270 rate=1/48000 ticks=35840 delay=512 queued=0
217        now=13466525228363 rate=1/48000 ticks=90112 delay=512 queued=0
218        ...
219        stream drained
220        stream state changed streaming -> paused
221        stream param change: id=4
222        stream state changed paused -> unconnected
223        stream param change: id=4
224        ```
225        """
226
227        if status:
228            self._pipewire_configs["--verbose"] = "    "
229        else:
230            pass
231
232    def get_config(self):
233        """Return config dictionary with default or setup variables, remember that
234        this object changes only on python-side. Is not updated on real time,
235        For real-time, please create and destroy the class.
236
237        Args:
238            Nothing
239
240        Returns:
241            - _pipewire_configs (`dict`) : dictionary with config values
242
243        """
244
245        return self._pipewire_configs
246
247    def set_config(
248        self,
249        # configs
250        media_type=None,
251        media_category=None,
252        media_role=None,
253        target=None,
254        latency=None,
255        rate=None,
256        channels=None,
257        channels_map=None,
258        _format=None,
259        volume=None,
260        quality=None,
261        # Debug
262        verbose=False,
263    ):
264        """Method that get args as variables and set them
265        to the `json` parameter of the class `_pipewire_configs`,
266        then you can use in other method, such as `playback(...)` or
267        `record(...)`. This method verifies values to avoid wrong
268        settings.
269
270        Args:
271            media_type : Set media type
272            media_category : Set media category
273            media_role : Set media role
274            target : Set node target
275            latency : Set node latency *example=100ms
276            rate : Set sample rate [8000,11025,16000,22050,44100,48000,88200,96000,176400,192000,352800,384000]
277            channels : Numbers of channels [1,2]
278            channels_map : ["stereo", "surround-51", "FL,FR", ...]
279            _format : ["u8", "s8", "s16", "s32", "f32", "f64"]
280            volume : Stream volume [0.000, 1.000]
281            quality : Resampler quality [0, 15]
282            verbose (`bool`): True enable debug logs. *default=False
283
284        Returns:
285            - Nothing
286
287        More:
288            Check all links listed at the beginning of this page
289        """  # 1 - media_type
290        if media_type:
291            self._pipewire_configs["--media-type"] = str(media_type)
292        elif media_type is None:
293            pass
294        else:
295            raise ValueError(
296                f"{MESSAGES_ERROR['ValueError']}[media_type='{media_type}'] EMPTY VALUE"
297            )
298        # 2 - media_category
299        if media_category:
300            self._pipewire_configs["--media-category"] = str(media_category)
301        elif media_category is None:
302            pass
303        else:
304            raise ValueError(
305                f"{MESSAGES_ERROR['ValueError']}[media_category='{media_category}'] EMPTY VALUE"
306            )
307        # 3 - media_role
308        if media_role:
309            self._pipewire_configs["--media-role"] = str(media_role)
310        elif media_role is None:
311            pass
312        else:
313            raise ValueError(
314                f"{MESSAGES_ERROR['ValueError']}[media_role='{media_role}'] EMPTY VALUE"
315            )
316        # 4 - target
317        if target:
318            self._pipewire_configs["--target"] = str(target)
319        elif target is None:
320            pass
321        else:
322            raise ValueError(
323                f"{MESSAGES_ERROR['ValueError']}[target='{target}'] EMPTY VALUE"
324            )
325        # 5 - latency
326        if latency:
327            if any(chr.isdigit() for chr in latency):  # Contain numbers
328                self._pipewire_configs["--latency"] = str(latency)
329            else:
330                raise ValueError(
331                    f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] NO NUMBER IN VARIABLE"
332                )
333        elif latency is None:
334            pass
335        else:
336            raise ValueError(
337                f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] EMPTY VALUE"
338            )
339        # 6 - rate
340        if rate:
341            if rate in RECOMMENDED_RATES:
342                self._pipewire_configs["--rate"] = str(rate)
343            else:
344                raise ValueError(
345                    f"{MESSAGES_ERROR['ValueError']}[rate='{rate}']\
346                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_RATES}"
347                )
348        elif rate is None:
349            pass
350        else:
351            raise ValueError(
352                f"{MESSAGES_ERROR['ValueError']}[rate='{rate}'] EMPTY VALUE"
353            )
354        # 7 - channels
355        if channels:
356            if channels in [1, 2]:  # values
357                self._pipewire_configs["--channels"] = str(channels)
358            else:
359                raise ValueError(
360                    f"{MESSAGES_ERROR['ValueError']}[channels='{channels}']\
361                         WRONG VALUE\n ONLY 1 or 2."
362                )
363        elif channels is None:
364            pass
365        else:
366            raise ValueError(
367                f"{MESSAGES_ERROR['ValueError']}[channels='{channels}'] EMPTY VALUE"
368            )
369        # 8 - channels-map
370        if channels_map:
371            self._pipewire_configs["--channels-map"] = str(channels_map)
372        elif channels_map is None:
373            pass
374        else:
375            raise ValueError(
376                f"{MESSAGES_ERROR['ValueError']}[channels_map='{channels_map}'] EMPTY VALUE"
377            )
378        # 9 - format
379        if _format:
380            if _format in RECOMMENDED_FORMATS:
381                self._pipewire_configs["--format"] = str(_format)
382            else:
383                raise ValueError(
384                    f"{MESSAGES_ERROR['ValueError']}[_format='{_format}']\
385                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_FORMATS}"
386                )
387        elif _format is None:
388            pass
389        else:
390            raise ValueError(
391                f"{MESSAGES_ERROR['ValueError']}[_format='{_format}'] EMPTY VALUE"
392            )
393        # 10 - volume
394        if volume:
395            if 0.0 <= volume <= 1.0:
396                self._pipewire_configs["--volume"] = str(volume)
397            else:
398                raise ValueError(
399                    f"{MESSAGES_ERROR['ValueError']}[volume='{volume}']\
400                         OUT OF RANGE \n [0.000, 1.000]"
401                )
402        elif volume is None:
403            pass
404        else:
405            raise ValueError(
406                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
407            )
408        # 11 - quality
409        if quality:
410            if 0 <= quality <= 15:
411                self._pipewire_configs["--quality"] = str(quality)
412            else:
413                raise ValueError(
414                    f"{MESSAGES_ERROR['ValueError']}[quality='{quality}'] OUT OF RANGE \n [0, 15]"
415                )
416        elif quality is None:
417            pass
418        else:
419            raise ValueError(
420                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
421            )
422
423        # 12 - verbose cli
424        if verbose:  # True
425            self._pipewire_configs["--verbose"] = "    "
426        else:
427            pass
428
429        if verbose:
430            print(self._pipewire_configs)
431
432    def load_list_targets(
433        self,
434        mode,  # playback or record
435        # Debug,
436        verbose: bool = False,
437    ):
438        """Returns a list of targets to playback or record. Then you can use
439        the output to select a device to playback or record.
440        """
441
442        if mode == "playback":
443            mycommand = ["pw-cat", "--playback", "--list-targets"]
444            stdout, _ = _execute_shell_command(
445                command=mycommand, timeout=-1, verbose=verbose
446            )
447            self._pipewire_list_targets["list_playback"] = _generate_dict_list_targets(
448                longstring=stdout.decode(), verbose=verbose
449            )
450        elif mode == "record":
451            mycommand = ["pw-cat", "--record", "--list-targets"]
452            stdout, _ = _execute_shell_command(
453                command=mycommand, timeout=-1, verbose=verbose
454            )
455            self._pipewire_list_targets["list_record"] = _generate_dict_list_targets(
456                longstring=stdout.decode(), verbose=verbose
457            )
458        else:
459            raise AttributeError(MESSAGES_ERROR["ValueError"])
460
461        if verbose:
462            print(f"[mycommand]{mycommand}")
463
464    def get_list_targets(
465        self,
466        # Debug,
467        verbose: bool = False,
468    ):
469        """Returns a list of targets to playback or record. Then you can use
470        the output to select a device to playback or record.
471
472        Returns:
473            - `_pipewire_list_targets`
474
475        Examples:
476        ```python
477        >>> Controller().get_list_targets()
478        {
479        "list_playback": {
480            "86": {
481            "description": "Starship/Matisse HD Audio Controller Pro",
482            "prior": "936"
483            },
484            "_list_nodes": [
485            "86"
486            ],
487            "_node_default": [
488            "86"
489            ],
490            "_alsa_node": [
491            "alsa_output.pci-0000_0a_00.4.pro-output-0"
492            ]
493        },
494        "list_record": {
495            "86": {
496            "description": "Starship/Matisse HD Audio Controller Pro",
497            "prior": "936"
498            },
499            "_list_nodes": [
500            "86"
501            ],
502            "_node_default": [
503            "86"
504            ],
505            "_alsa_node": [
506            "alsa_output.pci-0000_0a_00.4.pro-output-0"
507            ]
508        }
509        }
510        ```
511        """
512        if verbose:
513            print(self._pipewire_list_targets)
514        return self._pipewire_list_targets
515
516    def get_list_interfaces(
517        self,
518        filtered_by_type: str = True,
519        type_interfaces: str = "Client",
520        # Debug
521        verbose: bool = False,
522    ):
523        """Returns a list of applications currently using pipewire on Client.
524        An example of pw-cli usage is the code below:
525
526        ```bash
527        #!/bin/bash
528        pw-cli ls Client
529        ```
530        Args:
531            filtered_by_type : If False, returns all. If not, returns a fitered dict
532            type_interfaces : Set type of Interface
533            ["Client","Link","Node","Factory","Module","Metadata","Endpoint",
534            "Session","Endpoint Stream","EndpointLink","Port"]
535
536        Returns:
537            - dict_interfaces_filtered: dictionary
538            with list of interfaces matching conditions
539
540        Examples:
541        ```python
542        >>> Controller().get_list_interfaces()
543
544        ```
545        """
546        mycommand = ["pw-cli", "info", "all"]
547
548        # if verbose:
549        #     print(f"[mycommand]{mycommand}")
550
551        stdout, _ = _execute_shell_command(
552            command=mycommand, timeout=-1, verbose=verbose
553        )
554        dict_interfaces = _generate_dict_interfaces(
555            longstring=stdout.decode(), verbose=verbose
556        )
557
558        if filtered_by_type:
559            dict_interfaces_filtered = _filter_by_type(
560                dict_interfaces=dict_interfaces, type_interfaces=type_interfaces
561            )
562        else:
563            dict_interfaces_filtered = dict_interfaces
564
565        return dict_interfaces_filtered
566
567    def playback(
568        self,
569        audio_filename: str = "myplayback.wav",
570        # Debug
571        verbose: bool = False,
572    ):
573        """Execute pipewire command to play an audio file with the following
574        command:
575
576        ```bash
577        #!/bin/bash
578        pw-cat --playback {audio_filename} + {configs}
579        # configs are a concatenated params
580        ```
581
582        Args:
583            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
584            verbose (`bool`): True enable debug logs. *default=False
585
586        Returns:
587            - stdout (`str`): Shell response to the command in stdout format
588            - stderr (`str`): Shell response response to the command in stderr format
589        """
590        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
591
592        mycommand = [
593            "pw-cat",
594            "--playback",
595            audio_filename,
596        ] + _generate_command_by_dict(mydict=self._pipewire_configs, verbose=verbose)
597
598        if verbose:
599            print(f"[mycommand]{mycommand}")
600
601        stdout, stderr = _execute_shell_command(
602            command=mycommand, timeout=-1, verbose=verbose
603        )
604        return stdout, stderr
605
606    def record(
607        self,
608        audio_filename: str = "myplayback.wav",
609        timeout_seconds=5,
610        # Debug
611        verbose: bool = False,
612    ):
613        """Execute pipewire command to record an audio file, with a timeout of 5
614        seconds with the following code and exiting the shell when tiomeout is over.
615
616        ```bash
617        #!/bin/bash
618        pw-cat --record {audio_filename}
619        # timeout is managed by python3 (when signal CTRL+C is sended)
620        ```
621
622        Args:
623            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
624            verbose (`bool`): True enable debug logs. *default=False
625
626        Returns:
627            - stdout (`str`): Shell response to the command in stdout format
628            - stderr (`str`): Shell response response to the command in stderr format
629        """
630        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
631
632        mycommand = ["pw-cat", "--record", audio_filename] + _generate_command_by_dict(
633            mydict=self._pipewire_configs, verbose=verbose
634        )
635
636        if verbose:
637            print(f"[mycommand]{mycommand}")
638
639        stdout, stderr = _execute_shell_command(
640            command=mycommand, timeout=timeout_seconds, verbose=verbose
641        )
642        return stdout, stderr
643
644    def clear_devices(
645        self,
646        mode: str = "all",  # ['all','playback','record']
647        # Debug
648        verbose: bool = False,
649    ):
650        """Function to stop process running under pipewire executed by
651        python controller and with default process name of `pw-cat`, `pw-play` or `pw-record`.
652
653        Args:
654            mode (`str`) : string to kill process under `pw-cat`, `pw-play` or `pw-record`.
655
656        Returns:
657            - stdoutdict (`dict`) : a dictionary with keys of `mode`.
658
659        Example with pipewire:
660            pw-cat process
661        """
662
663        mycommand = self._kill_pipewire[mode]
664
665        if verbose:
666            print(f"[mycommands]{mycommand}")
667
668        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)
669
670        return {mode: stdout}

class Controller: View Source

 45class Controller:
 46    """
 47    Class that controls pipewire command line interface
 48    with shell commands, handling outputs, loading default
 49    configs and more.
 50    """
 51
 52    _pipewire_cli = {  # Help
 53        "--help": "--help",  # -h
 54        "--version": "--version",
 55        "--remote": None,  # -r
 56    }
 57
 58    _pipewire_modes = {  # Modes
 59        "--playback": None,  # -p
 60        "--record": None,  # -r
 61        "--midi": None,  # -m
 62    }
 63
 64    _pipewire_list_targets = {  # "--list-targets": None,
 65        "list_playback": None,
 66        "list_record": None,
 67    }
 68
 69    _pipewire_configs = {  # Configs
 70        "--media-type": None,  # *default=Audio
 71        "--media-category": None,  # *default=Playback
 72        "--media-role": None,  # *default=Music
 73        "--target": None,  # *default=auto
 74        "--latency": None,  # *default=100ms (SOURCE FILE if not specified)
 75        "--rate": None,  # *default=48000
 76        "--channels": None,  # [1,2] *default=2
 77        "--channel-map": None,  # ["stereo", "surround-51", "FL,FR"...] *default="FL,FR"
 78        "--format": None,  # [u8|s8|s16|s32|f32|f64] *default=s16
 79        "--volume": None,  # [0.0,1.0] *default=1.000
 80        "--quality": None,  # -q # [0,15] *default=4
 81        "--verbose": None,  # -v
 82    }
 83
 84    _kill_pipewire = {
 85        "all": ["kill", "$(pidof pw-cat)"],
 86        "playback": ["kill", "$(pidof pw-play)"],
 87        "record": ["kill", "$(pidof pw-record)"],
 88    }
 89
 90    def __init__(
 91        self,
 92        # Debug
 93        verbose: bool = False,
 94    ):
 95        """This constructor load default configs from OS executing
 96        the following pipewire command
 97
 98        ```bash
 99        #!/bin/bash
100        # Get defaults from output of:
101        pw-cat -h
102        ```
103        """
104        # LOAD ALL DEFAULT PARAMETERS
105
106        mycommand = ["pw-cat", "-h"]
107
108        # get default parameters with help
109        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)  # stderr
110        # convert stdout to dictionary
111        dict_default_values = _get_dict_from_stdout(
112            stdout=str(stdout.decode()), verbose=verbose
113        )
114
115        if verbose:
116            print(self._pipewire_configs)
117
118        # Save default system configs to our json
119        self._pipewire_configs.update(
120            ([(key, dict_default_values[key]) for key in dict_default_values])
121        )
122
123        if verbose:
124            print(self._pipewire_configs)
125
126        # Delete keys with None values
127        self._pipewire_configs = _drop_keys_with_none_values(self._pipewire_configs)
128
129        if verbose:
130            print(self._pipewire_configs)
131
132        # Load values of list targets
133        self.load_list_targets(mode="playback", verbose=verbose)
134        self.load_list_targets(mode="record", verbose=verbose)
135
136    def _help_cli(
137        self,
138        # Debug
139        verbose: bool = True,
140    ):
141        """Get pipewire command line help"""
142
143        mycommand = ["pipewire", self._pipewire_cli["--help"]]
144
145        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)  # stderr
146
147        return stdout
148
149    def get_version(
150        self,
151        # Debug
152        verbose: bool = False,
153    ):
154        """Get version of pipewire installed on OS by executing the following
155        code:
156
157        ```bash
158        #!/bin/bash
159        pw-cli --version
160        ```
161
162        Args:
163            verbose (bool) : True enable debug logs. *default=False
164
165        Returns:
166            - versions (list) : Versions of pipewire compiled
167        """
168
169        mycommand = ["pw-cli", "--version"]
170
171        if verbose:
172            print(f"[mycommand]{mycommand}")
173
174        stdout, _ = _execute_shell_command(
175            command=mycommand, timeout=-1, verbose=verbose
176        )
177        versions = stdout.decode().split("\n")[1:]
178
179        self._pipewire_cli["--version"] = versions
180
181        return versions
182
183    def verbose(
184        self,
185        status: bool = True,
186    ):
187        """Get full log of pipewire stream status with the command `pw-cat`
188
189        An example of pw-cli usage is the code below:
190
191        ```bash
192        #!/bin/bash
193        # For example
194        pw-cat --playback beers.wav --verbose
195        ```
196
197        that will generate an output like this:
198
199        ```bash
200        opened file "beers.wav" format 00010002 channels:2 rate:44100
201        using default channel map: FL,FR
202        rate=44100 channels=2 fmt=s16 samplesize=2 stride=4 latency=4410 (0.100s)
203        connecting playback stream; target_id=4294967295
204        stream state changed unconnected -> connecting
205        stream param change: id=2
206        stream properties:
207            media.type = "Audio"
208            ...
209        now=0 rate=0/0 ticks=0 delay=0 queued=0
210        remote 0 is named "pipewire-0"
211        core done
212        stream state changed connecting -> paused
213        stream param change: id=2
214        ...
215        stream param change: id=15
216        stream param change: id=15
217        now=13465394419270 rate=1/48000 ticks=35840 delay=512 queued=0
218        now=13466525228363 rate=1/48000 ticks=90112 delay=512 queued=0
219        ...
220        stream drained
221        stream state changed streaming -> paused
222        stream param change: id=4
223        stream state changed paused -> unconnected
224        stream param change: id=4
225        ```
226        """
227
228        if status:
229            self._pipewire_configs["--verbose"] = "    "
230        else:
231            pass
232
233    def get_config(self):
234        """Return config dictionary with default or setup variables, remember that
235        this object changes only on python-side. Is not updated on real time,
236        For real-time, please create and destroy the class.
237
238        Args:
239            Nothing
240
241        Returns:
242            - _pipewire_configs (`dict`) : dictionary with config values
243
244        """
245
246        return self._pipewire_configs
247
248    def set_config(
249        self,
250        # configs
251        media_type=None,
252        media_category=None,
253        media_role=None,
254        target=None,
255        latency=None,
256        rate=None,
257        channels=None,
258        channels_map=None,
259        _format=None,
260        volume=None,
261        quality=None,
262        # Debug
263        verbose=False,
264    ):
265        """Method that get args as variables and set them
266        to the `json` parameter of the class `_pipewire_configs`,
267        then you can use in other method, such as `playback(...)` or
268        `record(...)`. This method verifies values to avoid wrong
269        settings.
270
271        Args:
272            media_type : Set media type
273            media_category : Set media category
274            media_role : Set media role
275            target : Set node target
276            latency : Set node latency *example=100ms
277            rate : Set sample rate [8000,11025,16000,22050,44100,48000,88200,96000,176400,192000,352800,384000]
278            channels : Numbers of channels [1,2]
279            channels_map : ["stereo", "surround-51", "FL,FR", ...]
280            _format : ["u8", "s8", "s16", "s32", "f32", "f64"]
281            volume : Stream volume [0.000, 1.000]
282            quality : Resampler quality [0, 15]
283            verbose (`bool`): True enable debug logs. *default=False
284
285        Returns:
286            - Nothing
287
288        More:
289            Check all links listed at the beginning of this page
290        """  # 1 - media_type
291        if media_type:
292            self._pipewire_configs["--media-type"] = str(media_type)
293        elif media_type is None:
294            pass
295        else:
296            raise ValueError(
297                f"{MESSAGES_ERROR['ValueError']}[media_type='{media_type}'] EMPTY VALUE"
298            )
299        # 2 - media_category
300        if media_category:
301            self._pipewire_configs["--media-category"] = str(media_category)
302        elif media_category is None:
303            pass
304        else:
305            raise ValueError(
306                f"{MESSAGES_ERROR['ValueError']}[media_category='{media_category}'] EMPTY VALUE"
307            )
308        # 3 - media_role
309        if media_role:
310            self._pipewire_configs["--media-role"] = str(media_role)
311        elif media_role is None:
312            pass
313        else:
314            raise ValueError(
315                f"{MESSAGES_ERROR['ValueError']}[media_role='{media_role}'] EMPTY VALUE"
316            )
317        # 4 - target
318        if target:
319            self._pipewire_configs["--target"] = str(target)
320        elif target is None:
321            pass
322        else:
323            raise ValueError(
324                f"{MESSAGES_ERROR['ValueError']}[target='{target}'] EMPTY VALUE"
325            )
326        # 5 - latency
327        if latency:
328            if any(chr.isdigit() for chr in latency):  # Contain numbers
329                self._pipewire_configs["--latency"] = str(latency)
330            else:
331                raise ValueError(
332                    f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] NO NUMBER IN VARIABLE"
333                )
334        elif latency is None:
335            pass
336        else:
337            raise ValueError(
338                f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] EMPTY VALUE"
339            )
340        # 6 - rate
341        if rate:
342            if rate in RECOMMENDED_RATES:
343                self._pipewire_configs["--rate"] = str(rate)
344            else:
345                raise ValueError(
346                    f"{MESSAGES_ERROR['ValueError']}[rate='{rate}']\
347                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_RATES}"
348                )
349        elif rate is None:
350            pass
351        else:
352            raise ValueError(
353                f"{MESSAGES_ERROR['ValueError']}[rate='{rate}'] EMPTY VALUE"
354            )
355        # 7 - channels
356        if channels:
357            if channels in [1, 2]:  # values
358                self._pipewire_configs["--channels"] = str(channels)
359            else:
360                raise ValueError(
361                    f"{MESSAGES_ERROR['ValueError']}[channels='{channels}']\
362                         WRONG VALUE\n ONLY 1 or 2."
363                )
364        elif channels is None:
365            pass
366        else:
367            raise ValueError(
368                f"{MESSAGES_ERROR['ValueError']}[channels='{channels}'] EMPTY VALUE"
369            )
370        # 8 - channels-map
371        if channels_map:
372            self._pipewire_configs["--channels-map"] = str(channels_map)
373        elif channels_map is None:
374            pass
375        else:
376            raise ValueError(
377                f"{MESSAGES_ERROR['ValueError']}[channels_map='{channels_map}'] EMPTY VALUE"
378            )
379        # 9 - format
380        if _format:
381            if _format in RECOMMENDED_FORMATS:
382                self._pipewire_configs["--format"] = str(_format)
383            else:
384                raise ValueError(
385                    f"{MESSAGES_ERROR['ValueError']}[_format='{_format}']\
386                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_FORMATS}"
387                )
388        elif _format is None:
389            pass
390        else:
391            raise ValueError(
392                f"{MESSAGES_ERROR['ValueError']}[_format='{_format}'] EMPTY VALUE"
393            )
394        # 10 - volume
395        if volume:
396            if 0.0 <= volume <= 1.0:
397                self._pipewire_configs["--volume"] = str(volume)
398            else:
399                raise ValueError(
400                    f"{MESSAGES_ERROR['ValueError']}[volume='{volume}']\
401                         OUT OF RANGE \n [0.000, 1.000]"
402                )
403        elif volume is None:
404            pass
405        else:
406            raise ValueError(
407                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
408            )
409        # 11 - quality
410        if quality:
411            if 0 <= quality <= 15:
412                self._pipewire_configs["--quality"] = str(quality)
413            else:
414                raise ValueError(
415                    f"{MESSAGES_ERROR['ValueError']}[quality='{quality}'] OUT OF RANGE \n [0, 15]"
416                )
417        elif quality is None:
418            pass
419        else:
420            raise ValueError(
421                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
422            )
423
424        # 12 - verbose cli
425        if verbose:  # True
426            self._pipewire_configs["--verbose"] = "    "
427        else:
428            pass
429
430        if verbose:
431            print(self._pipewire_configs)
432
433    def load_list_targets(
434        self,
435        mode,  # playback or record
436        # Debug,
437        verbose: bool = False,
438    ):
439        """Returns a list of targets to playback or record. Then you can use
440        the output to select a device to playback or record.
441        """
442
443        if mode == "playback":
444            mycommand = ["pw-cat", "--playback", "--list-targets"]
445            stdout, _ = _execute_shell_command(
446                command=mycommand, timeout=-1, verbose=verbose
447            )
448            self._pipewire_list_targets["list_playback"] = _generate_dict_list_targets(
449                longstring=stdout.decode(), verbose=verbose
450            )
451        elif mode == "record":
452            mycommand = ["pw-cat", "--record", "--list-targets"]
453            stdout, _ = _execute_shell_command(
454                command=mycommand, timeout=-1, verbose=verbose
455            )
456            self._pipewire_list_targets["list_record"] = _generate_dict_list_targets(
457                longstring=stdout.decode(), verbose=verbose
458            )
459        else:
460            raise AttributeError(MESSAGES_ERROR["ValueError"])
461
462        if verbose:
463            print(f"[mycommand]{mycommand}")
464
465    def get_list_targets(
466        self,
467        # Debug,
468        verbose: bool = False,
469    ):
470        """Returns a list of targets to playback or record. Then you can use
471        the output to select a device to playback or record.
472
473        Returns:
474            - `_pipewire_list_targets`
475
476        Examples:
477        ```python
478        >>> Controller().get_list_targets()
479        {
480        "list_playback": {
481            "86": {
482            "description": "Starship/Matisse HD Audio Controller Pro",
483            "prior": "936"
484            },
485            "_list_nodes": [
486            "86"
487            ],
488            "_node_default": [
489            "86"
490            ],
491            "_alsa_node": [
492            "alsa_output.pci-0000_0a_00.4.pro-output-0"
493            ]
494        },
495        "list_record": {
496            "86": {
497            "description": "Starship/Matisse HD Audio Controller Pro",
498            "prior": "936"
499            },
500            "_list_nodes": [
501            "86"
502            ],
503            "_node_default": [
504            "86"
505            ],
506            "_alsa_node": [
507            "alsa_output.pci-0000_0a_00.4.pro-output-0"
508            ]
509        }
510        }
511        ```
512        """
513        if verbose:
514            print(self._pipewire_list_targets)
515        return self._pipewire_list_targets
516
517    def get_list_interfaces(
518        self,
519        filtered_by_type: str = True,
520        type_interfaces: str = "Client",
521        # Debug
522        verbose: bool = False,
523    ):
524        """Returns a list of applications currently using pipewire on Client.
525        An example of pw-cli usage is the code below:
526
527        ```bash
528        #!/bin/bash
529        pw-cli ls Client
530        ```
531        Args:
532            filtered_by_type : If False, returns all. If not, returns a fitered dict
533            type_interfaces : Set type of Interface
534            ["Client","Link","Node","Factory","Module","Metadata","Endpoint",
535            "Session","Endpoint Stream","EndpointLink","Port"]
536
537        Returns:
538            - dict_interfaces_filtered: dictionary
539            with list of interfaces matching conditions
540
541        Examples:
542        ```python
543        >>> Controller().get_list_interfaces()
544
545        ```
546        """
547        mycommand = ["pw-cli", "info", "all"]
548
549        # if verbose:
550        #     print(f"[mycommand]{mycommand}")
551
552        stdout, _ = _execute_shell_command(
553            command=mycommand, timeout=-1, verbose=verbose
554        )
555        dict_interfaces = _generate_dict_interfaces(
556            longstring=stdout.decode(), verbose=verbose
557        )
558
559        if filtered_by_type:
560            dict_interfaces_filtered = _filter_by_type(
561                dict_interfaces=dict_interfaces, type_interfaces=type_interfaces
562            )
563        else:
564            dict_interfaces_filtered = dict_interfaces
565
566        return dict_interfaces_filtered
567
568    def playback(
569        self,
570        audio_filename: str = "myplayback.wav",
571        # Debug
572        verbose: bool = False,
573    ):
574        """Execute pipewire command to play an audio file with the following
575        command:
576
577        ```bash
578        #!/bin/bash
579        pw-cat --playback {audio_filename} + {configs}
580        # configs are a concatenated params
581        ```
582
583        Args:
584            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
585            verbose (`bool`): True enable debug logs. *default=False
586
587        Returns:
588            - stdout (`str`): Shell response to the command in stdout format
589            - stderr (`str`): Shell response response to the command in stderr format
590        """
591        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
592
593        mycommand = [
594            "pw-cat",
595            "--playback",
596            audio_filename,
597        ] + _generate_command_by_dict(mydict=self._pipewire_configs, verbose=verbose)
598
599        if verbose:
600            print(f"[mycommand]{mycommand}")
601
602        stdout, stderr = _execute_shell_command(
603            command=mycommand, timeout=-1, verbose=verbose
604        )
605        return stdout, stderr
606
607    def record(
608        self,
609        audio_filename: str = "myplayback.wav",
610        timeout_seconds=5,
611        # Debug
612        verbose: bool = False,
613    ):
614        """Execute pipewire command to record an audio file, with a timeout of 5
615        seconds with the following code and exiting the shell when tiomeout is over.
616
617        ```bash
618        #!/bin/bash
619        pw-cat --record {audio_filename}
620        # timeout is managed by python3 (when signal CTRL+C is sended)
621        ```
622
623        Args:
624            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
625            verbose (`bool`): True enable debug logs. *default=False
626
627        Returns:
628            - stdout (`str`): Shell response to the command in stdout format
629            - stderr (`str`): Shell response response to the command in stderr format
630        """
631        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
632
633        mycommand = ["pw-cat", "--record", audio_filename] + _generate_command_by_dict(
634            mydict=self._pipewire_configs, verbose=verbose
635        )
636
637        if verbose:
638            print(f"[mycommand]{mycommand}")
639
640        stdout, stderr = _execute_shell_command(
641            command=mycommand, timeout=timeout_seconds, verbose=verbose
642        )
643        return stdout, stderr
644
645    def clear_devices(
646        self,
647        mode: str = "all",  # ['all','playback','record']
648        # Debug
649        verbose: bool = False,
650    ):
651        """Function to stop process running under pipewire executed by
652        python controller and with default process name of `pw-cat`, `pw-play` or `pw-record`.
653
654        Args:
655            mode (`str`) : string to kill process under `pw-cat`, `pw-play` or `pw-record`.
656
657        Returns:
658            - stdoutdict (`dict`) : a dictionary with keys of `mode`.
659
660        Example with pipewire:
661            pw-cat process
662        """
663
664        mycommand = self._kill_pipewire[mode]
665
666        if verbose:
667            print(f"[mycommands]{mycommand}")
668
669        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)
670
671        return {mode: stdout}

Class that controls pipewire command line interface with shell commands, handling outputs, loading default configs and more.

Controller(verbose: bool = False) View Source

 90    def __init__(
 91        self,
 92        # Debug
 93        verbose: bool = False,
 94    ):
 95        """This constructor load default configs from OS executing
 96        the following pipewire command
 97
 98        ```bash
 99        #!/bin/bash
100        # Get defaults from output of:
101        pw-cat -h
102        ```
103        """
104        # LOAD ALL DEFAULT PARAMETERS
105
106        mycommand = ["pw-cat", "-h"]
107
108        # get default parameters with help
109        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)  # stderr
110        # convert stdout to dictionary
111        dict_default_values = _get_dict_from_stdout(
112            stdout=str(stdout.decode()), verbose=verbose
113        )
114
115        if verbose:
116            print(self._pipewire_configs)
117
118        # Save default system configs to our json
119        self._pipewire_configs.update(
120            ([(key, dict_default_values[key]) for key in dict_default_values])
121        )
122
123        if verbose:
124            print(self._pipewire_configs)
125
126        # Delete keys with None values
127        self._pipewire_configs = _drop_keys_with_none_values(self._pipewire_configs)
128
129        if verbose:
130            print(self._pipewire_configs)
131
132        # Load values of list targets
133        self.load_list_targets(mode="playback", verbose=verbose)
134        self.load_list_targets(mode="record", verbose=verbose)

This constructor load default configs from OS executing the following pipewire command

#!/bin/bash
# Get defaults from output of:
pw-cat -h

def get_version(self, verbose: bool = False): View Source

149    def get_version(
150        self,
151        # Debug
152        verbose: bool = False,
153    ):
154        """Get version of pipewire installed on OS by executing the following
155        code:
156
157        ```bash
158        #!/bin/bash
159        pw-cli --version
160        ```
161
162        Args:
163            verbose (bool) : True enable debug logs. *default=False
164
165        Returns:
166            - versions (list) : Versions of pipewire compiled
167        """
168
169        mycommand = ["pw-cli", "--version"]
170
171        if verbose:
172            print(f"[mycommand]{mycommand}")
173
174        stdout, _ = _execute_shell_command(
175            command=mycommand, timeout=-1, verbose=verbose
176        )
177        versions = stdout.decode().split("\n")[1:]
178
179        self._pipewire_cli["--version"] = versions
180
181        return versions

Get version of pipewire installed on OS by executing the following code:

#!/bin/bash
pw-cli --version

Arguments:

verbose (bool) : True enable debug logs. *default=False

Returns:

versions (list) : Versions of pipewire compiled

def verbose(self, status: bool = True): View Source

183    def verbose(
184        self,
185        status: bool = True,
186    ):
187        """Get full log of pipewire stream status with the command `pw-cat`
188
189        An example of pw-cli usage is the code below:
190
191        ```bash
192        #!/bin/bash
193        # For example
194        pw-cat --playback beers.wav --verbose
195        ```
196
197        that will generate an output like this:
198
199        ```bash
200        opened file "beers.wav" format 00010002 channels:2 rate:44100
201        using default channel map: FL,FR
202        rate=44100 channels=2 fmt=s16 samplesize=2 stride=4 latency=4410 (0.100s)
203        connecting playback stream; target_id=4294967295
204        stream state changed unconnected -> connecting
205        stream param change: id=2
206        stream properties:
207            media.type = "Audio"
208            ...
209        now=0 rate=0/0 ticks=0 delay=0 queued=0
210        remote 0 is named "pipewire-0"
211        core done
212        stream state changed connecting -> paused
213        stream param change: id=2
214        ...
215        stream param change: id=15
216        stream param change: id=15
217        now=13465394419270 rate=1/48000 ticks=35840 delay=512 queued=0
218        now=13466525228363 rate=1/48000 ticks=90112 delay=512 queued=0
219        ...
220        stream drained
221        stream state changed streaming -> paused
222        stream param change: id=4
223        stream state changed paused -> unconnected
224        stream param change: id=4
225        ```
226        """
227
228        if status:
229            self._pipewire_configs["--verbose"] = "    "
230        else:
231            pass

Get full log of pipewire stream status with the command pw-cat

An example of pw-cli usage is the code below:

#!/bin/bash
# For example
pw-cat --playback beers.wav --verbose

that will generate an output like this:

opened file "beers.wav" format 00010002 channels:2 rate:44100
using default channel map: FL,FR
rate=44100 channels=2 fmt=s16 samplesize=2 stride=4 latency=4410 (0.100s)
connecting playback stream; target_id=4294967295
stream state changed unconnected -> connecting
stream param change: id=2
stream properties:
    media.type = "Audio"
    ...
now=0 rate=0/0 ticks=0 delay=0 queued=0
remote 0 is named "pipewire-0"
core done
stream state changed connecting -> paused
stream param change: id=2
...
stream param change: id=15
stream param change: id=15
now=13465394419270 rate=1/48000 ticks=35840 delay=512 queued=0
now=13466525228363 rate=1/48000 ticks=90112 delay=512 queued=0
...
stream drained
stream state changed streaming -> paused
stream param change: id=4
stream state changed paused -> unconnected
stream param change: id=4

def get_config(self): View Source

233    def get_config(self):
234        """Return config dictionary with default or setup variables, remember that
235        this object changes only on python-side. Is not updated on real time,
236        For real-time, please create and destroy the class.
237
238        Args:
239            Nothing
240
241        Returns:
242            - _pipewire_configs (`dict`) : dictionary with config values
243
244        """
245
246        return self._pipewire_configs

Return config dictionary with default or setup variables, remember that this object changes only on python-side. Is not updated on real time, For real-time, please create and destroy the class.

Arguments:

Nothing

Returns:

_pipewire_configs (dict) : dictionary with config values

def set_config( self, media_type=None, media_category=None, media_role=None, target=None, latency=None, rate=None, channels=None, channels_map=None, _format=None, volume=None, quality=None, verbose=False): View Source

248    def set_config(
249        self,
250        # configs
251        media_type=None,
252        media_category=None,
253        media_role=None,
254        target=None,
255        latency=None,
256        rate=None,
257        channels=None,
258        channels_map=None,
259        _format=None,
260        volume=None,
261        quality=None,
262        # Debug
263        verbose=False,
264    ):
265        """Method that get args as variables and set them
266        to the `json` parameter of the class `_pipewire_configs`,
267        then you can use in other method, such as `playback(...)` or
268        `record(...)`. This method verifies values to avoid wrong
269        settings.
270
271        Args:
272            media_type : Set media type
273            media_category : Set media category
274            media_role : Set media role
275            target : Set node target
276            latency : Set node latency *example=100ms
277            rate : Set sample rate [8000,11025,16000,22050,44100,48000,88200,96000,176400,192000,352800,384000]
278            channels : Numbers of channels [1,2]
279            channels_map : ["stereo", "surround-51", "FL,FR", ...]
280            _format : ["u8", "s8", "s16", "s32", "f32", "f64"]
281            volume : Stream volume [0.000, 1.000]
282            quality : Resampler quality [0, 15]
283            verbose (`bool`): True enable debug logs. *default=False
284
285        Returns:
286            - Nothing
287
288        More:
289            Check all links listed at the beginning of this page
290        """  # 1 - media_type
291        if media_type:
292            self._pipewire_configs["--media-type"] = str(media_type)
293        elif media_type is None:
294            pass
295        else:
296            raise ValueError(
297                f"{MESSAGES_ERROR['ValueError']}[media_type='{media_type}'] EMPTY VALUE"
298            )
299        # 2 - media_category
300        if media_category:
301            self._pipewire_configs["--media-category"] = str(media_category)
302        elif media_category is None:
303            pass
304        else:
305            raise ValueError(
306                f"{MESSAGES_ERROR['ValueError']}[media_category='{media_category}'] EMPTY VALUE"
307            )
308        # 3 - media_role
309        if media_role:
310            self._pipewire_configs["--media-role"] = str(media_role)
311        elif media_role is None:
312            pass
313        else:
314            raise ValueError(
315                f"{MESSAGES_ERROR['ValueError']}[media_role='{media_role}'] EMPTY VALUE"
316            )
317        # 4 - target
318        if target:
319            self._pipewire_configs["--target"] = str(target)
320        elif target is None:
321            pass
322        else:
323            raise ValueError(
324                f"{MESSAGES_ERROR['ValueError']}[target='{target}'] EMPTY VALUE"
325            )
326        # 5 - latency
327        if latency:
328            if any(chr.isdigit() for chr in latency):  # Contain numbers
329                self._pipewire_configs["--latency"] = str(latency)
330            else:
331                raise ValueError(
332                    f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] NO NUMBER IN VARIABLE"
333                )
334        elif latency is None:
335            pass
336        else:
337            raise ValueError(
338                f"{MESSAGES_ERROR['ValueError']}[latency='{latency}'] EMPTY VALUE"
339            )
340        # 6 - rate
341        if rate:
342            if rate in RECOMMENDED_RATES:
343                self._pipewire_configs["--rate"] = str(rate)
344            else:
345                raise ValueError(
346                    f"{MESSAGES_ERROR['ValueError']}[rate='{rate}']\
347                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_RATES}"
348                )
349        elif rate is None:
350            pass
351        else:
352            raise ValueError(
353                f"{MESSAGES_ERROR['ValueError']}[rate='{rate}'] EMPTY VALUE"
354            )
355        # 7 - channels
356        if channels:
357            if channels in [1, 2]:  # values
358                self._pipewire_configs["--channels"] = str(channels)
359            else:
360                raise ValueError(
361                    f"{MESSAGES_ERROR['ValueError']}[channels='{channels}']\
362                         WRONG VALUE\n ONLY 1 or 2."
363                )
364        elif channels is None:
365            pass
366        else:
367            raise ValueError(
368                f"{MESSAGES_ERROR['ValueError']}[channels='{channels}'] EMPTY VALUE"
369            )
370        # 8 - channels-map
371        if channels_map:
372            self._pipewire_configs["--channels-map"] = str(channels_map)
373        elif channels_map is None:
374            pass
375        else:
376            raise ValueError(
377                f"{MESSAGES_ERROR['ValueError']}[channels_map='{channels_map}'] EMPTY VALUE"
378            )
379        # 9 - format
380        if _format:
381            if _format in RECOMMENDED_FORMATS:
382                self._pipewire_configs["--format"] = str(_format)
383            else:
384                raise ValueError(
385                    f"{MESSAGES_ERROR['ValueError']}[_format='{_format}']\
386                         VALUE NOT IN RECOMMENDED LIST \n{RECOMMENDED_FORMATS}"
387                )
388        elif _format is None:
389            pass
390        else:
391            raise ValueError(
392                f"{MESSAGES_ERROR['ValueError']}[_format='{_format}'] EMPTY VALUE"
393            )
394        # 10 - volume
395        if volume:
396            if 0.0 <= volume <= 1.0:
397                self._pipewire_configs["--volume"] = str(volume)
398            else:
399                raise ValueError(
400                    f"{MESSAGES_ERROR['ValueError']}[volume='{volume}']\
401                         OUT OF RANGE \n [0.000, 1.000]"
402                )
403        elif volume is None:
404            pass
405        else:
406            raise ValueError(
407                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
408            )
409        # 11 - quality
410        if quality:
411            if 0 <= quality <= 15:
412                self._pipewire_configs["--quality"] = str(quality)
413            else:
414                raise ValueError(
415                    f"{MESSAGES_ERROR['ValueError']}[quality='{quality}'] OUT OF RANGE \n [0, 15]"
416                )
417        elif quality is None:
418            pass
419        else:
420            raise ValueError(
421                f"{MESSAGES_ERROR['ValueError']}[volume='{volume}'] EMPTY VALUE"
422            )
423
424        # 12 - verbose cli
425        if verbose:  # True
426            self._pipewire_configs["--verbose"] = "    "
427        else:
428            pass
429
430        if verbose:
431            print(self._pipewire_configs)

Method that get args as variables and set them to the json parameter of the class _pipewire_configs, then you can use in other method, such as playback(...) or record(...). This method verifies values to avoid wrong settings.

Arguments:

media_type : Set media type
media_category : Set media category
media_role : Set media role
target : Set node target
latency : Set node latency *example=100ms
rate : Set sample rate [8000,11025,16000,22050,44100,48000,88200,96000,176400,192000,352800,384000]
channels : Numbers of channels [1,2]
channels_map : ["stereo", "surround-51", "FL,FR", ...]
_format : ["u8", "s8", "s16", "s32", "f32", "f64"]
volume : Stream volume [0.000, 1.000]
quality : Resampler quality [0, 15]
verbose (bool): True enable debug logs. *default=False

Returns:

Nothing

Check all links listed at the beginning of this page

def load_list_targets(self, mode, verbose: bool = False): View Source

433    def load_list_targets(
434        self,
435        mode,  # playback or record
436        # Debug,
437        verbose: bool = False,
438    ):
439        """Returns a list of targets to playback or record. Then you can use
440        the output to select a device to playback or record.
441        """
442
443        if mode == "playback":
444            mycommand = ["pw-cat", "--playback", "--list-targets"]
445            stdout, _ = _execute_shell_command(
446                command=mycommand, timeout=-1, verbose=verbose
447            )
448            self._pipewire_list_targets["list_playback"] = _generate_dict_list_targets(
449                longstring=stdout.decode(), verbose=verbose
450            )
451        elif mode == "record":
452            mycommand = ["pw-cat", "--record", "--list-targets"]
453            stdout, _ = _execute_shell_command(
454                command=mycommand, timeout=-1, verbose=verbose
455            )
456            self._pipewire_list_targets["list_record"] = _generate_dict_list_targets(
457                longstring=stdout.decode(), verbose=verbose
458            )
459        else:
460            raise AttributeError(MESSAGES_ERROR["ValueError"])
461
462        if verbose:
463            print(f"[mycommand]{mycommand}")

Returns a list of targets to playback or record. Then you can use the output to select a device to playback or record.

def get_list_targets(self, verbose: bool = False): View Source

465    def get_list_targets(
466        self,
467        # Debug,
468        verbose: bool = False,
469    ):
470        """Returns a list of targets to playback or record. Then you can use
471        the output to select a device to playback or record.
472
473        Returns:
474            - `_pipewire_list_targets`
475
476        Examples:
477        ```python
478        >>> Controller().get_list_targets()
479        {
480        "list_playback": {
481            "86": {
482            "description": "Starship/Matisse HD Audio Controller Pro",
483            "prior": "936"
484            },
485            "_list_nodes": [
486            "86"
487            ],
488            "_node_default": [
489            "86"
490            ],
491            "_alsa_node": [
492            "alsa_output.pci-0000_0a_00.4.pro-output-0"
493            ]
494        },
495        "list_record": {
496            "86": {
497            "description": "Starship/Matisse HD Audio Controller Pro",
498            "prior": "936"
499            },
500            "_list_nodes": [
501            "86"
502            ],
503            "_node_default": [
504            "86"
505            ],
506            "_alsa_node": [
507            "alsa_output.pci-0000_0a_00.4.pro-output-0"
508            ]
509        }
510        }
511        ```
512        """
513        if verbose:
514            print(self._pipewire_list_targets)
515        return self._pipewire_list_targets

Returns a list of targets to playback or record. Then you can use the output to select a device to playback or record.

Returns:

_pipewire_list_targets

Examples:

>>> Controller().get_list_targets()
{
"list_playback": {
    "86": {
    "description": "Starship/Matisse HD Audio Controller Pro",
    "prior": "936"
    },
    "_list_nodes": [
    "86"
    ],
    "_node_default": [
    "86"
    ],
    "_alsa_node": [
    "alsa_output.pci-0000_0a_00.4.pro-output-0"
    ]
},
"list_record": {
    "86": {
    "description": "Starship/Matisse HD Audio Controller Pro",
    "prior": "936"
    },
    "_list_nodes": [
    "86"
    ],
    "_node_default": [
    "86"
    ],
    "_alsa_node": [
    "alsa_output.pci-0000_0a_00.4.pro-output-0"
    ]
}
}

def get_list_interfaces( self, filtered_by_type: str = True, type_interfaces: str = 'Client', verbose: bool = False): View Source

517    def get_list_interfaces(
518        self,
519        filtered_by_type: str = True,
520        type_interfaces: str = "Client",
521        # Debug
522        verbose: bool = False,
523    ):
524        """Returns a list of applications currently using pipewire on Client.
525        An example of pw-cli usage is the code below:
526
527        ```bash
528        #!/bin/bash
529        pw-cli ls Client
530        ```
531        Args:
532            filtered_by_type : If False, returns all. If not, returns a fitered dict
533            type_interfaces : Set type of Interface
534            ["Client","Link","Node","Factory","Module","Metadata","Endpoint",
535            "Session","Endpoint Stream","EndpointLink","Port"]
536
537        Returns:
538            - dict_interfaces_filtered: dictionary
539            with list of interfaces matching conditions
540
541        Examples:
542        ```python
543        >>> Controller().get_list_interfaces()
544
545        ```
546        """
547        mycommand = ["pw-cli", "info", "all"]
548
549        # if verbose:
550        #     print(f"[mycommand]{mycommand}")
551
552        stdout, _ = _execute_shell_command(
553            command=mycommand, timeout=-1, verbose=verbose
554        )
555        dict_interfaces = _generate_dict_interfaces(
556            longstring=stdout.decode(), verbose=verbose
557        )
558
559        if filtered_by_type:
560            dict_interfaces_filtered = _filter_by_type(
561                dict_interfaces=dict_interfaces, type_interfaces=type_interfaces
562            )
563        else:
564            dict_interfaces_filtered = dict_interfaces
565
566        return dict_interfaces_filtered

Returns a list of applications currently using pipewire on Client. An example of pw-cli usage is the code below:

#!/bin/bash
pw-cli ls Client

Arguments:

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:

>>> Controller().get_list_interfaces()

def playback(self, audio_filename: str = 'myplayback.wav', verbose: bool = False): View Source

568    def playback(
569        self,
570        audio_filename: str = "myplayback.wav",
571        # Debug
572        verbose: bool = False,
573    ):
574        """Execute pipewire command to play an audio file with the following
575        command:
576
577        ```bash
578        #!/bin/bash
579        pw-cat --playback {audio_filename} + {configs}
580        # configs are a concatenated params
581        ```
582
583        Args:
584            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
585            verbose (`bool`): True enable debug logs. *default=False
586
587        Returns:
588            - stdout (`str`): Shell response to the command in stdout format
589            - stderr (`str`): Shell response response to the command in stderr format
590        """
591        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
592
593        mycommand = [
594            "pw-cat",
595            "--playback",
596            audio_filename,
597        ] + _generate_command_by_dict(mydict=self._pipewire_configs, verbose=verbose)
598
599        if verbose:
600            print(f"[mycommand]{mycommand}")
601
602        stdout, stderr = _execute_shell_command(
603            command=mycommand, timeout=-1, verbose=verbose
604        )
605        return stdout, stderr

Execute pipewire command to play an audio file with the following command:

#!/bin/bash
pw-cat --playback {audio_filename} + {configs}
# configs are a concatenated params

Arguments:

audio_filename (str): Path of the file to be played. *default='myplayback.wav'
verbose (bool): True enable debug logs. *default=False

Returns:

stdout (str): Shell response to the command in stdout format

stderr (str): Shell response response to the command in stderr format

def record( self, audio_filename: str = 'myplayback.wav', timeout_seconds=5, verbose: bool = False): View Source

607    def record(
608        self,
609        audio_filename: str = "myplayback.wav",
610        timeout_seconds=5,
611        # Debug
612        verbose: bool = False,
613    ):
614        """Execute pipewire command to record an audio file, with a timeout of 5
615        seconds with the following code and exiting the shell when tiomeout is over.
616
617        ```bash
618        #!/bin/bash
619        pw-cat --record {audio_filename}
620        # timeout is managed by python3 (when signal CTRL+C is sended)
621        ```
622
623        Args:
624            audio_filename (`str`): Path of the file to be played. *default='myplayback.wav'
625            verbose (`bool`): True enable debug logs. *default=False
626
627        Returns:
628            - stdout (`str`): Shell response to the command in stdout format
629            - stderr (`str`): Shell response response to the command in stderr format
630        """
631        # warnings.warn("The name of the function may change on future releases", DeprecationWarning)
632
633        mycommand = ["pw-cat", "--record", audio_filename] + _generate_command_by_dict(
634            mydict=self._pipewire_configs, verbose=verbose
635        )
636
637        if verbose:
638            print(f"[mycommand]{mycommand}")
639
640        stdout, stderr = _execute_shell_command(
641            command=mycommand, timeout=timeout_seconds, verbose=verbose
642        )
643        return stdout, stderr

Execute pipewire command to record an audio file, with a timeout of 5 seconds with the following code and exiting the shell when tiomeout is over.

#!/bin/bash
pw-cat --record {audio_filename}
# timeout is managed by python3 (when signal CTRL+C is sended)

Arguments:

audio_filename (str): Path of the file to be played. *default='myplayback.wav'
verbose (bool): True enable debug logs. *default=False

Returns:

stdout (str): Shell response to the command in stdout format

stderr (str): Shell response response to the command in stderr format

def clear_devices(self, mode: str = 'all', verbose: bool = False): View Source

645    def clear_devices(
646        self,
647        mode: str = "all",  # ['all','playback','record']
648        # Debug
649        verbose: bool = False,
650    ):
651        """Function to stop process running under pipewire executed by
652        python controller and with default process name of `pw-cat`, `pw-play` or `pw-record`.
653
654        Args:
655            mode (`str`) : string to kill process under `pw-cat`, `pw-play` or `pw-record`.
656
657        Returns:
658            - stdoutdict (`dict`) : a dictionary with keys of `mode`.
659
660        Example with pipewire:
661            pw-cat process
662        """
663
664        mycommand = self._kill_pipewire[mode]
665
666        if verbose:
667            print(f"[mycommands]{mycommand}")
668
669        stdout, _ = _execute_shell_command(command=mycommand, verbose=verbose)
670
671        return {mode: stdout}

Function to stop process running under pipewire executed by python controller and with default process name of pw-cat, pw-play or pw-record.

Arguments:

mode (str) : string to kill process under pw-cat, pw-play or pw-record.

Returns:

stdoutdict (dict) : a dictionary with keys of mode.

Example with pipewire:

pw-cat process