14
14
15
15
import json
16
16
import math
17
- from typing import Optional
17
+ from typing import Dict , Tuple , List , Optional
18
18
from threading import Lock
19
19
import yaml
20
20
from websocket import create_connection , WebSocket # type: ignore
@@ -47,7 +47,7 @@ def __init__(self, host: str, port: int):
47
47
self ._host = host
48
48
self ._port = int (port )
49
49
self ._ws : Optional [WebSocket ] = None
50
- self .cdsp_version : Optional [tuple [str , str , str ]] = None
50
+ self .cdsp_version : Optional [Tuple [str , str , str ]] = None
51
51
self ._lock = Lock ()
52
52
53
53
def query (self , command : str , arg = None ):
@@ -217,39 +217,39 @@ def range_decibel(self) -> float:
217
217
range_decibel = - 1000
218
218
return range_decibel
219
219
220
- def capture_rms (self ) -> list [float ]:
220
+ def capture_rms (self ) -> List [float ]:
221
221
"""
222
222
Get capture signal level rms in dB for the last processed chunk.
223
223
Full scale is 0 dB. Returns a list with one element per channel.
224
224
"""
225
225
sigrms = self .client .query ("GetCaptureSignalRms" )
226
226
return sigrms
227
227
228
- def playback_rms (self ) -> list [float ]:
228
+ def playback_rms (self ) -> List [float ]:
229
229
"""
230
230
Get playback signal level rms in dB for the last processed chunk.
231
231
Full scale is 0 dB. Returns a list with one element per channel.
232
232
"""
233
233
sigrms = self .client .query ("GetPlaybackSignalRms" )
234
234
return sigrms
235
235
236
- def capture_peak (self ) -> list [float ]:
236
+ def capture_peak (self ) -> List [float ]:
237
237
"""
238
238
Get capture signal level peak in dB for the last processed chunk.
239
239
Full scale is 0 dB. Returns a list with one element per channel.
240
240
"""
241
241
sigpeak = self .client .query ("GetCaptureSignalPeak" )
242
242
return sigpeak
243
243
244
- def playback_peak (self ) -> list [float ]:
244
+ def playback_peak (self ) -> List [float ]:
245
245
"""
246
246
Get playback signal level peak in dB for the last processed chunk.
247
247
Full scale is 0 dB. Returns a list with one element per channel.
248
248
"""
249
249
sigpeak = self .client .query ("GetPlaybackSignalPeak" )
250
250
return sigpeak
251
251
252
- def playback_peak_since (self , interval : float ) -> list [float ]:
252
+ def playback_peak_since (self , interval : float ) -> List [float ]:
253
253
"""
254
254
Get playback signal level peak in dB for the last `interval` seconds.
255
255
Full scale is 0 dB. Returns a list with one element per channel.
@@ -260,7 +260,7 @@ def playback_peak_since(self, interval: float) -> list[float]:
260
260
sigpeak = self .client .query ("GetPlaybackSignalPeakSince" , arg = float (interval ))
261
261
return sigpeak
262
262
263
- def playback_rms_since (self , interval : float ) -> list [float ]:
263
+ def playback_rms_since (self , interval : float ) -> List [float ]:
264
264
"""
265
265
Get playback signal level rms in dB for the last `interval` seconds.
266
266
Full scale is 0 dB. Returns a list with one element per channel.
@@ -271,7 +271,7 @@ def playback_rms_since(self, interval: float) -> list[float]:
271
271
sigrms = self .client .query ("GetPlaybackSignalRmsSince" , arg = float (interval ))
272
272
return sigrms
273
273
274
- def capture_peak_since (self , interval : float ) -> list [float ]:
274
+ def capture_peak_since (self , interval : float ) -> List [float ]:
275
275
"""
276
276
Get capture signal level peak in dB for the last `interval` seconds.
277
277
Full scale is 0 dB. Returns a list with one element per channel.
@@ -282,7 +282,7 @@ def capture_peak_since(self, interval: float) -> list[float]:
282
282
sigpeak = self .client .query ("GetCaptureSignalPeakSince" , arg = float (interval ))
283
283
return sigpeak
284
284
285
- def capture_rms_since (self , interval : float ) -> list [float ]:
285
+ def capture_rms_since (self , interval : float ) -> List [float ]:
286
286
"""
287
287
Get capture signal level rms in dB for the last `interval` seconds.
288
288
Full scale is 0 dB. Returns a list with one element per channel.
@@ -293,39 +293,39 @@ def capture_rms_since(self, interval: float) -> list[float]:
293
293
sigrms = self .client .query ("GetCaptureSignalRmsSince" , arg = float (interval ))
294
294
return sigrms
295
295
296
- def playback_peak_since_last (self ) -> list [float ]:
296
+ def playback_peak_since_last (self ) -> List [float ]:
297
297
"""
298
298
Get playback signal level peak in dB since the last read by the same client.
299
299
Full scale is 0 dB. Returns a list with one element per channel.
300
300
"""
301
301
sigpeak = self .client .query ("GetPlaybackSignalPeakSinceLast" )
302
302
return sigpeak
303
303
304
- def playback_rms_since_last (self ) -> list [float ]:
304
+ def playback_rms_since_last (self ) -> List [float ]:
305
305
"""
306
306
Get playback signal level rms in dB since the last read by the same client.
307
307
Full scale is 0 dB. Returns a list with one element per channel.
308
308
"""
309
309
sigrms = self .client .query ("GetPlaybackSignalRmsSinceLast" )
310
310
return sigrms
311
311
312
- def capture_peak_since_last (self ) -> list [float ]:
312
+ def capture_peak_since_last (self ) -> List [float ]:
313
313
"""
314
314
Get capture signal level peak in dB since the last read by the same client.
315
315
Full scale is 0 dB. Returns a list with one element per channel.
316
316
"""
317
317
sigpeak = self .client .query ("GetCaptureSignalPeakSinceLast" )
318
318
return sigpeak
319
319
320
- def capture_rms_since_last (self ) -> list [float ]:
320
+ def capture_rms_since_last (self ) -> List [float ]:
321
321
"""
322
322
Get capture signal level rms in dB since the last read by the same client.
323
323
Full scale is 0 dB. Returns a list with one element per channel.
324
324
"""
325
325
sigrms = self .client .query ("GetCaptureSignalRmsSinceLast" )
326
326
return sigrms
327
327
328
- def levels (self ) -> dict [str , list [float ]]:
328
+ def levels (self ) -> Dict [str , List [float ]]:
329
329
"""
330
330
Get all signal levels in dB for the last processed chunk.
331
331
Full scale is 0 dB.
@@ -336,7 +336,7 @@ def levels(self) -> dict[str, list[float]]:
336
336
siglevels = self .client .query ("GetSignalLevels" )
337
337
return siglevels
338
338
339
- def levels_since (self , interval : float ) -> dict [str , list [float ]]:
339
+ def levels_since (self , interval : float ) -> Dict [str , List [float ]]:
340
340
"""
341
341
Get all signal levels in dB for the last `interval` seconds.
342
342
Full scale is 0 dB.
@@ -350,7 +350,7 @@ def levels_since(self, interval: float) -> dict[str, list[float]]:
350
350
siglevels = self .client .query ("GetSignalLevelsSince" , arg = float (interval ))
351
351
return siglevels
352
352
353
- def levels_since_last (self ) -> dict [str , list [float ]]:
353
+ def levels_since_last (self ) -> Dict [str , List [float ]]:
354
354
"""
355
355
Get all signal levels in dB since the last read by the same client.
356
356
Full scale is 0 dB.
@@ -361,7 +361,7 @@ def levels_since_last(self) -> dict[str, list[float]]:
361
361
siglevels = self .client .query ("GetSignalLevelsSinceLast" )
362
362
return siglevels
363
363
364
- def peaks_since_start (self ) -> dict [str , list [float ]]:
364
+ def peaks_since_start (self ) -> Dict [str , List [float ]]:
365
365
"""
366
366
Get the playback and capture peak level since processing started.
367
367
The values are returned as a json object with keys `playback` and `capture`.
@@ -420,31 +420,31 @@ def set_active_raw(self, config_string: str):
420
420
"""
421
421
self .client .query ("SetConfig" , arg = config_string )
422
422
423
- def active (self ) -> Optional [dict ]:
423
+ def active (self ) -> Optional [Dict ]:
424
424
"""
425
425
Get the active configuration as a Python object.
426
426
427
427
Returns:
428
- dict | None: Current config as a Python dict, or None.
428
+ Dict | None: Current config as a Python dict, or None.
429
429
"""
430
430
config_string = self .active_raw ()
431
431
if config_string is None :
432
432
return None
433
433
config_object = yaml .safe_load (config_string )
434
434
return config_object
435
435
436
- def previous (self ) -> Optional [dict ]:
436
+ def previous (self ) -> Optional [Dict ]:
437
437
"""
438
438
Get the previously active configuration as a Python object.
439
439
440
440
Returns:
441
- dict | None: Previous config as a Python dict, or None.
441
+ Dict | None: Previous config as a Python dict, or None.
442
442
"""
443
443
config_string = self .client .query ("GetPreviousConfig" )
444
444
config_object = yaml .safe_load (config_string )
445
445
return config_object
446
446
447
- def parse_yaml (self , config_string : str ) -> dict :
447
+ def parse_yaml (self , config_string : str ) -> Dict :
448
448
"""
449
449
Parse a config from yaml string and return the contents
450
450
as a Python object, with defaults filled out with their default values.
@@ -453,47 +453,47 @@ def parse_yaml(self, config_string: str) -> dict:
453
453
config_string (str): A config as raw yaml string.
454
454
455
455
Returns:
456
- dict | None: Parsed config as a Python dict.
456
+ Dict | None: Parsed config as a Python dict.
457
457
"""
458
458
config_raw = self .client .query ("ReadConfig" , arg = config_string )
459
459
config_object = yaml .safe_load (config_raw )
460
460
return config_object
461
461
462
- def read_and_parse_file (self , filename : str ) -> dict :
462
+ def read_and_parse_file (self , filename : str ) -> Dict :
463
463
"""
464
464
Read and parse a config file from disk and return the contents as a Python object.
465
465
466
466
Args:
467
467
filename (str): Path to a config file.
468
468
469
469
Returns:
470
- dict | None: Parsed config as a Python dict.
470
+ Dict | None: Parsed config as a Python dict.
471
471
"""
472
472
config_raw = self .client .query ("ReadConfigFile" , arg = filename )
473
473
config = yaml .safe_load (config_raw )
474
474
return config
475
475
476
- def set_active (self , config_object : dict ):
476
+ def set_active (self , config_object : Dict ):
477
477
"""
478
478
Upload and apply a new configuration from a Python object.
479
479
480
480
Args:
481
- config_object (dict ): A configuration as a Python dict.
481
+ config_object (Dict ): A configuration as a Python dict.
482
482
"""
483
483
config_raw = yaml .dump (config_object )
484
484
self .set_active_raw (config_raw )
485
485
486
- def validate (self , config_object : dict ) -> dict :
486
+ def validate (self , config_object : Dict ) -> Dict :
487
487
"""
488
488
Validate a configuration object.
489
489
Returns the validated config with all optional fields filled with defaults.
490
490
Raises a CamillaError on errors.
491
491
492
492
Args:
493
- config_object (dict ): A configuration as a Python dict.
493
+ config_object (Dict ): A configuration as a Python dict.
494
494
495
495
Returns:
496
- dict | None: Validated config as a Python dict.
496
+ Dict | None: Validated config as a Python dict.
497
497
"""
498
498
config_string = yaml .dump (config_object )
499
499
validated_string = self .client .query ("ValidateConfig" , arg = config_string )
@@ -773,14 +773,14 @@ def reload(self):
773
773
"""
774
774
self .client .query ("Reload" )
775
775
776
- def supported_device_types (self ) -> tuple [ list [str ], list [str ]]:
776
+ def supported_device_types (self ) -> Tuple [ List [str ], List [str ]]:
777
777
"""
778
778
Read what device types the running CamillaDSP process supports.
779
779
Returns a tuple with two lists of device types,
780
780
the first for playback and the second for capture.
781
781
782
782
Returns:
783
- tuple[list [str], list [str]]: A tuple containing two lists,
783
+ Tuple[List [str], List [str]]: A tuple containing two lists,
784
784
with the supported playback and capture device types.
785
785
"""
786
786
(playback , capture ) = self .client .query ("GetSupportedDeviceTypes" )
@@ -806,29 +806,29 @@ def state_file_updated(self) -> bool:
806
806
updated = self .client .query ("GetStateFileUpdated" )
807
807
return updated
808
808
809
- def list_playback_devices (self , value : str ) -> list [ tuple [str , str ]]:
809
+ def list_playback_devices (self , value : str ) -> List [ Tuple [str , str ]]:
810
810
"""
811
811
List the available playback devices for a given backend.
812
812
Returns a list of tuples. Returns the system name and
813
813
a descriptive name for each device.
814
814
For some backends, those two names are identical.
815
815
816
816
Returns:
817
- list[tuple [str, str]: A list containing tuples of two strings,
817
+ List[Tuple [str, str]: A list containing tuples of two strings,
818
818
with system device name and a descriptive name.
819
819
"""
820
820
devs = self .client .query ("GetAvailablePlaybackDevices" , arg = value )
821
821
return devs
822
822
823
- def list_capture_devices (self , value : str ) -> list [ tuple [str , str ]]:
823
+ def list_capture_devices (self , value : str ) -> List [ Tuple [str , str ]]:
824
824
"""
825
825
List the available capture devices for a given backend.
826
826
Returns a list of tuples. Returns the system name and
827
827
a descriptive name for each device.
828
828
For some backends, those two names are identical.
829
829
830
830
Returns:
831
- list[tuple [str, str]: A list containing tuples of two strings,
831
+ List[Tuple [str, str]: A list containing tuples of two strings,
832
832
with system device name and a descriptive name.
833
833
"""
834
834
devs = self .client .query ("GetAvailableCaptureDevices" , arg = value )
@@ -840,25 +840,26 @@ class Versions(_CommandGroup):
840
840
Version info
841
841
"""
842
842
843
- def camilladsp (self ) -> Optional [tuple [str , str , str ]]:
843
+ def camilladsp (self ) -> Optional [Tuple [str , str , str ]]:
844
844
"""
845
845
Read CamillaDSP version.
846
846
847
847
Returns:
848
- tuple[list [str], list [str]] | None: A tuple containing the CamillaDSP version,
848
+ Tuple[List [str], List [str]] | None: A tuple containing the CamillaDSP version,
849
849
as (major, minor, patch).
850
850
"""
851
851
return self .client .cdsp_version
852
852
853
- def library (self ) -> tuple [str , str , str ]:
853
+ def library (self ) -> Tuple [str , str , str ]:
854
854
"""
855
855
Read pyCamillaDSP library version.
856
856
857
857
Returns:
858
- tuple[list [str], list [str]] | None: A tuple containing the pyCamillaDSP version,
858
+ Tuple[List [str], List [str]] | None: A tuple containing the pyCamillaDSP version,
859
859
as (major, minor, patch).
860
860
"""
861
- return VERSION .split ("." )
861
+ v = VERSION .split ("." )
862
+ return (v [0 ], v [1 ], v [2 ])
862
863
863
864
864
865
class CamillaClient (_CamillaWS ):
0 commit comments