Updated tests, CLI WIP, now started on interactive mode.

Author: zeyus

bluebox/__init__.py

@@ -0,0 +1,23 @@
+import typing as t
+from .freqs import BaseMF, DTMF, MF
+
+_MF: t.Dict[str, t.Type[BaseMF]] = {}
+
+
+def register_mf(name: str, backend: t.Type[BaseMF]) -> None:
+    """Register a MF set."""
+    _MF[name] = backend
+
+
+def get_mf(name: str) -> t.Type[BaseMF]:
+    """Get a MF set."""
+    return _MF[name]
+
+
+def list_mf() -> t.List[str]:
+    """List the available MF sets."""
+    return list(_MF.keys())
+
+
+register_mf('dtmf', DTMF)
+register_mf('mf', MF)

bluebox/backends/__init__.py

@@ -1,2 +1,23 @@
+import typing as t
 from .base import BlueboxBackend as BlueboxBackend  # noqa: F401
 from .backend_pyaudio import PyAudioBackend as PyAudioBackend  # noqa: F401
+
+_BACKENDS: t.Dict[str, t.Type[BlueboxBackend]] = {}
+
+
+def register_backend(name: str, backend: t.Type[BlueboxBackend]) -> None:
+    """Register a backend."""
+    _BACKENDS[name] = backend
+
+
+def get_backend(name: str) -> t.Type[BlueboxBackend]:
+    """Get a backend."""
+    return _BACKENDS[name]
+
+
+def list_backends() -> t.List[str]:
+    """List the available backends."""
+    return list(_BACKENDS.keys())
+
+
+register_backend('pyaudio', PyAudioBackend)

bluebox/backends/backend_dummy.py

@@ -0,0 +1,72 @@
+"""backend_dummy.py
+
+This file contains the dummy backend for bluebox.
+This is used for testing and instead of generating
+sound, it can print the data to the console, or return
+it as a list.
+"""
+
+import typing as t
+import logging
+from .base import BlueboxBackend
+
+
+class DummyBackend(BlueboxBackend):
+    """DummyBackend class for the dummy backend."""
+
+    _data: t.List[float]
+
+    def __init__(
+                self,
+                sample_rate: float = 44100.0,
+                channels: int = 1,
+                amplitude: float = 1.0,
+                logger: t.Optional[logging.Logger] = None,
+                mode: str = 'print') -> None:
+        """Initialize the dummy backend."""
+        super().__init__(sample_rate, channels, amplitude, logger)
+        self._mode = mode
+        self._data = []
+
+    def _to_bytes(self, data: t.Iterator[float]) -> t.List[float]:
+        """Wrap the data in a buffer."""
+        _data = []
+        while True:
+            try:
+                d = next(data)
+                _data.append(d)
+            except StopIteration:
+                break
+
+        return _data
+
+    def play(self, data: t.Iterator[float], close=True) -> None:
+        """Play the given data."""
+        d = self._to_bytes(data)
+        if self._mode == 'print':
+            print(d)
+        elif self._mode == 'list':
+            self._data += d
+        else:
+            raise ValueError(f'Invalid mode: {self._mode}')
+
+    def play_all(self, queue: t.Iterator[t.Iterator[float]]) -> None:
+        """Play the given data and then stop."""
+        for data in queue:
+            self.play(data, close=False)
+
+    def stop(self) -> None:
+        """Stop playing the data."""
+        pass
+
+    def close(self) -> None:
+        """Close the backend."""
+        pass
+
+    def __del__(self) -> None:
+        """Delete the backend."""
+        self.close()
+
+    def get_data(self) -> t.List[float]:
+        """Get the data."""
+        return self._data

bluebox/box.py

@@ -35,7 +35,9 @@ def __init__(
                 channels: int = 1,
                 stop_on_error: bool = False,
                 logger: t.Optional[logging.Logger] = None,
-                backend: t.Optional[BlueboxBackend] = None) -> None:
+                backend: t.Optional[
+                        t.Union[BlueboxBackend, t.Type[BlueboxBackend]]
+                    ] = None) -> None:
         """Initialize the Sequencer object.
 
         Args:
@@ -54,7 +56,7 @@ def __init__(
         """
 
         self._mf = mf
-        self._wave = SineWave(sr=sample_rate, ch=channels)
+        self._wave = SineWave(sample_rate=sample_rate, channels=channels)
         self._length = length
         self._amplitude = amplitude
         self._pause = pause
@@ -75,7 +77,7 @@ def __init__(
                 channels=channels,
                 amplitude=1.0,
                 logger=self._logger)
-        self._backend = backend  # type: ignore        
+        self._backend = backend  # type: ignore
 
     def sequence(self, codes: str) -> t.Iterator[float]:
         """Generate a sequence of waveforms."""
@@ -101,7 +103,6 @@ def sequence(self, codes: str) -> t.Iterator[float]:
                     t2 = next(tone2)
                     yield t1 + t2
                 except StopIteration:
-                    print('stop')
                     break
 
             i += 1

bluebox/cli.py

@@ -3,11 +3,12 @@
 This file can be used to interactively generate tone sequences.
 """
 
+from pathlib import Path
 import argparse
 import logging
 import sys
 from .box import Sequencer
-from .freqs import DTMF, MF
+from . import get_mf, list_mf
 
 
 def parse_args() -> argparse.Namespace:
@@ -39,27 +40,58 @@ def parse_args() -> argparse.Namespace:
             type=float,
             default=44100.0,
             help='The sample rate of the waveforms.')
-    parser.add_argument(
-            '-c', '--channels',
-            type=int,
-            default=1,
-            help='The number of channels in the waveforms.')
     parser.add_argument(
             '-m', '--mf',
             type=str,
-            default='DTMF',
-            help='The MF to use e.g. DTMF.')
+            default='dtmf',
+            help='The MF to use e.g. dtmf, mf.')
     parser.add_argument(
             '-d', '--debug',
             action='store_true',
             help='Enable debug logging.')
-    parser.add_argument(
+    # we can have sequence or file,pipe,stdin OR interactive
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument(
+            '-f', '--file',
+            type=Path,
+            help='The file to read the sequence from.')
+    group.add_argument(
+            '-P', '--pipe',
+            type=Path,
+            help='Read the sequence from a pipe.')
+    group.add_argument(
+            '-S', '--stdin',
+            action='store_true',
+            help='Read the sequence from stdin.')
+    group.add_argument(
+            '-i', '--interactive',
+            action='store_true',
+            help='Enter interactive mode.')
+    group.add_argument(
             'sequence',
             type=str,
-            help='The sequence to generate.')
+            nargs='?',
+            help='The sequence of tones to generate.')
+
     return parser.parse_args()
 
 
+def bluebox_interactive(seq: Sequencer) -> None:
+    """Enter interactive mode."""
+
+    print('Entering interactive mode. Type "exit" to quit.')
+    while True:
+        try:
+            seq(input('Sequence: '))
+        except KeyboardInterrupt:
+            print('Exiting...')
+            break
+        except Exception as e:
+            logging.error(e)
+            if seq._stop_on_error:
+                break
+
+
 def bluebox() -> None:
     """Generate a tone sequence.
 
@@ -76,12 +108,11 @@ def bluebox() -> None:
         stop_on_error = False
         logging.basicConfig(level=logging.INFO)
 
-    if args.mf == 'DTMF':
-        mf = DTMF()
-    elif args.mf == 'MF':
-        mf = MF()
+    if args.mf in list_mf():
+        mf = get_mf(args.mf)()
     else:
         logging.error('Invalid MF: %s', args.mf)
+        logging.error('Valid MFs: %s', ', '.join(list_mf()))
         sys.exit(1)
 
     seq = Sequencer(
@@ -90,9 +121,11 @@ def bluebox() -> None:
             length=args.length,
             pause=args.pause,
             sample_rate=args.sample_rate,
-            channels=args.channels,
+            channels=1,
             stop_on_error=stop_on_error)
-
+    if args.interactive:
+        bluebox_interactive(seq)
+        return
     try:
         seq(args.sequence)
     except Exception as e:

bluebox/freqs.py

@@ -8,6 +8,7 @@
 from abc import ABC, abstractmethod
 import math
 
+
 class BaseMF(ABC):
     """BaseMF class for defining MF frequencies."""
 

bluebox/wave.py

@@ -13,18 +13,18 @@
 class SineWave:
     """SineWave class for generating waveform arrays."""
 
-    _sample_rate: float = 44100.0
-    _channels: int = 1
+    _sr: float = 44100.0
+    _ch: int = 1
 
     def __init__(
                 self,
-                sr: t.Optional[float] = None,
-                ch: t.Optional[int] = None) -> None:
+                sample_rate: t.Optional[float] = None,
+                channels: t.Optional[int] = None) -> None:
         """Initialize the Wave object."""
-        if sr is not None:
-            self._sample_rate = sr
-        if ch is not None:
-            self._channels = ch
+        if sample_rate is not None:
+            self._sr = sample_rate
+        if channels is not None:
+            self._ch = channels
 
     def sine(
             self,
@@ -47,14 +47,14 @@ def sine(
 
         # silence / pauses
         if freq == 0.0 or amplitude == 0.0:
-            for i in range(math.ceil(length * self._sample_rate / 1000)):
+            for i in range(math.ceil(length * self._sr / 1000)):
                 yield 0.0
             return
 
         # sine wave
-        for i in range(math.ceil(length * self._sample_rate / 1000)):
+        for i in range(math.ceil(length * self._sr / 1000)):
             yield amplitude * math.sin(
-                    2 * math.pi * freq * (i / self._sample_rate) + phase)
+                    2 * math.pi * freq * (i / self._sr) + phase)
 
     def __call__(
                 self,
@@ -81,4 +81,4 @@ def __call__(
 
     def __repr__(self) -> str:
         """Get the representation of the Wave."""
-        return f'{self.__class__.__name__}(Sample Rate: {self._sample_rate}, Channels: {self._channels})'  # noqa: E501
+        return f'{self.__class__.__name__}(Sample Rate: {self._sr}, Channels: {self._ch})'  # noqa: E501