refactoring into library

Author: esdandreu

app.py

@@ -0,0 +1,159 @@
+from recorder.settings import DEFAULT_CONFIG_PATH, DEFAULT_OUTPUT_FOLDER
+from recorder.device import Device
+from recorder.io import yaml_dump, yaml_load
+
+import typer
+
+from pathlib import Path
+from typing import Optional
+from PyInquirer import prompt
+
+# Command line application
+app = typer.Typer(
+    add_completion=False,
+    help=__doc__,
+    )
+
+DEFAULT_OUTPUT_FOLDER_OPTION = typer.Option(
+    DEFAULT_OUTPUT_FOLDER,
+    help="""
+    Path to a folder where the recorded data and the recording configuration
+    will be stored."""
+    )
+DEFAULT_CONFIG_PATH_OPTION = typer.Option(
+    DEFAULT_CONFIG_PATH,
+    help="Path to a `yaml` config file. Run `config` command to generate one."
+    )
+
+DEVICE_CLASSES = {cls.__name__.lower(): cls for cls in Device.__subclasses__()}
+DEVICE_CLASS_ARGUMENT = typer.Argument(
+    None,
+    metavar='DEVICE_CLASS',
+    help=(
+        "Case independent name of the device class to use. Available options: "
+        f"{[d for d in DEVICE_CLASSES.keys()]}."
+        )
+    )
+DEVICE_ID_ARGUMENT = typer.Argument(
+    None,
+    metavar='DEVICE_ID',
+    help=(
+        "Numerical id of the device to use. Use the show command to see the available devices for each device class."
+        )
+    )
+VERBOSE_OPTION = typer.Option(False, "--verbose", "-v", help="Verbose output.")
+
+
+def typer_warn(message: str):
+    return typer.secho(message, bg='black', fg='yellow')
+
+
+def choose(message: str, choices: list):
+    # Utility function that asks the user to configure the devices
+    return prompt({
+        'type': 'list',
+        'name': 'choice',
+        'message': message,
+        'choices': choices,
+        })['choice']
+
+
+def wait(seconds: int = 4, label: str = 'Recording...', **kwargs):
+    with typer.progressbar(range(seconds * 10), label=label, **kwargs) as p:
+        for _ in p:
+            sd.sleep(100)
+
+
+@app.command(help='Display the available devices')
+def show(
+        device_class: Optional[str] = DEVICE_CLASS_ARGUMENT,
+        verbose: bool = VERBOSE_OPTION
+    ):
+    if device_class:
+        try:
+            device_classes = [DEVICE_CLASSES[device_class.lower()]]
+        except KeyError:
+            raise RuntimeError(
+                f'Invalid device class `{device_class}`. Available options: '
+                f'{list(DEVICE_CLASSES.keys())}'
+                )
+    else:
+        device_classes = DEVICE_CLASSES.values()
+    for cls in device_classes:
+        devices = cls.find()
+        if devices:
+            if not verbose:
+                devices = {i: d['name'] for i, d in devices.items()}
+            typer.echo(f'{cls}:\n' + yaml_dump(devices))
+        else:
+            typer_warn(f"Could not find {cls} devices")
+
+
+def config_device_class(device: Device) -> dict:
+    choices = []
+    devices = device.find()
+    if not devices:
+        typer_warn(f"Could not find {device} devices")
+        return choices
+    while typer.confirm(
+        f"Add {'another' if choices else 'a'} {device} device?"
+        ):
+        _id = choose(
+            message=f"Select the {device} to add to the configuration:",
+            choices=[{
+                'name': f"{_id} {d['name']}",
+                'value': _id
+                } for _id, d in devices.items()]
+            )
+        choices.append(devices[_id])
+    return {i: c for i, c in enumerate(choices)}
+
+
+@app.command(help='Create a configuration `yaml` file.')
+def config(output: Path = DEFAULT_CONFIG_PATH_OPTION) -> dict:
+    config = {}
+    for name, device_class in DEVICE_CLASSES.items():
+        config[name] = config_device_class(device_class)
+    yaml_dump(config, to_file=output)
+    typer.echo(
+        f"Configuration file written to {output}. Remember that it can be "
+        "edited manually. Check the repository for an explained example file "
+        "https://github.com/AcousticOdometry/VAO-recorder/blob/main/example-config.yaml"
+        )
+    return config
+
+
+def get_config(path: Path = DEFAULT_CONFIG_PATH) -> dict:
+    _config = None
+    if path.exists():
+        _config = yaml_load(path)
+    if not _config:
+        typer_warn(f"No configuration found in {path}. Generate one.")
+        _config = config(path)
+    return _config
+
+
+def record():
+    pass
+
+
+@app.command(help='Test the available devices of a certain [device_class]')
+def test(
+        device_class: Optional[str] = DEVICE_CLASS_ARGUMENT,
+        device_id: Optional[int] = DEVICE_ID_ARGUMENT
+    ):
+    if not device_id:
+        config = get_config()
+        if not device_class:
+            pass
+    else:
+        # Both device_class and device_id are given
+        pass
+
+
+if __name__ == '__main__':
+    # Launch the command line application
+    try:
+        app()
+    except RuntimeError as e:
+        typer.secho(str(e), fg='red')

recorder/__init__.py

@@ -0,0 +1,3 @@
+from .device import Device
+from .microphone import Microphone
+from .realsense import RealSense

recorder/device/__init__.py

@@ -0,0 +1,29 @@
+# def find_cameras(max_index: int = 10) -> dict:
+#     return {
+#         i: {
+#             'cap':
+#                 cap,
+#             'name':
+#                 cv2.videoio_registry.getBackendName(
+#                     int(cap.get(cv2.CAP_PROP_BACKEND))
+#                     ),
+#             'width':
+#                 int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+#             'height':
+#                 int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+#             }
+#         for i in range(max_index) if (cap := cv2.VideoCapture(i)).isOpened()
+#         }
+
+# Camera = Device(find_cameras, 'camera')
+
+# @app.command(help='Display the available cameras')
+# def show_cameras():
+#     # Remove the `cap` attribute from the camera devices
+#     devices = {
+#         i: {k: v
+#             for k, v in camera.items() if k != 'cap'}
+#         for i, camera in Camera.find().items()
+#         }
+#     typer.echo('Real Sense devices:\n' + yaml_dump(devices))
+

recorder/device/camera.py

@@ -0,0 +1,48 @@
+from typing import Dict
+from pathlib import Path
+from abc import ABC, ABCMeta, abstractmethod
+
+
+class MetaDevice(ABCMeta):
+
+    def __str__(cls):
+        return cls.__name__
+
+
+class Device(ABC, metaclass=MetaDevice):
+
+    def __init__(self, _index: int, output_folder: Path, *args, **kwargs):
+        self.name = kwargs.get('name', 'unknown')
+        self.output_file = output_folder / f'{self.__name__.lower()}{_index}'
+
+    def __str__(self) -> str:
+        return self.name
+
+    @classmethod
+    @abstractmethod
+    def find(cls) -> Dict[int, dict]:
+        """Finds devices connected to the current system. 
+
+        Returns:
+            dict: A dictionary of devices with a numeric `id` as key and a
+            the device properties as value.
+        """
+        pass
+
+    @abstractmethod
+    def start(self) -> None:
+        """Starts recording.
+
+        Returns:
+            None
+        """
+        pass
+
+    @abstractmethod
+    def stop(self) -> None:
+        """Stops recording.
+
+        Returns:
+            None
+        """
+        pass

recorder/device/device.py

@@ -0,0 +1,67 @@
+from recorder.device import Device
+
+import sounddevice as sd
+
+from pathlib import Path
+
+
+class Microphone(Device):
+
+    def __init__(
+            self, *args, index: int, samplerate: int, channels: int, **kwargs
+        ):
+        super().__init__(*args, **kwargs)
+        self.index = index
+        self.samplerate = samplerate
+        self.channels = channels
+        self.stream = self._get_stream(
+            device=index,
+            output_path=self.output_file.with_suffix('.wav'),
+            )
+
+    @classmethod
+    def find(cls) -> dict:
+        devices = {}
+        for idx, d in enumerate(sd.query_devices()):
+            if d['max_input_channels'] > 0:
+                devices[idx] = {
+                    'samplerate': int(d['default_samplerate']),
+                    'channels': int(d['max_input_channels']),
+                    'name': d['name'],
+                    'index': idx,
+                    }
+        return devices
+
+    @staticmethod
+    def _get_stream(
+        device: int,  # Device identifier
+        output_path: Path,  # Path to where the audio file should be written
+        samplewidth: int = 2,  # Sample width in bytes
+        samplerate: int = None,  # Sample rate
+        channels: int = None,  # Number of channels
+        **stream_kwargs,  # Additional keyword arguments for sd.RawInputStream
+        # https://python-sounddevice.readthedocs.io/en/0.4.4/api/streams.html#sounddevice.InputStream
+        ) -> sd.InputStream:
+        f = wave.open(str(output_path), 'wb')
+        f.setnchannels(int(channels))
+        f.setframerate(int(samplerate))
+        f.setsampwidth(samplewidth)
+        return sd.RawInputStream(
+            device=device,
+            dtype=f'int{samplewidth*8}',
+            samplerate=samplerate,
+            channels=channels,
+            callback=lambda data, N, t, status: f.writeframesraw(data),
+            finished_callback=f.close,
+            **stream_kwargs
+            )
+
+    def start(self):
+        self.stream.start()
+
+    def stop(self):
+        self.stream.stop()
+        self.stream.close()
+        # Write start timestamp
+        # with open(self.output_folder / 'audio_start.txt', 'w') as f:
+        #     f.write(str(datetime.now().timestamp()))

recorder/device/microphone.py

@@ -0,0 +1,60 @@
+from recorder.device import Device
+
+import pyrealsense2 as rs
+
+
+class RealSense(Device):
+
+    def __init__(self, *args, streams: dict, serial_number: str, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.streams = streams
+        self.serial_number = serial_number
+
+    @classmethod
+    def find(cls) -> dict:
+        devices = {}
+        for d in rs.context().query_devices():
+            sn = d.get_info(rs.camera_info.serial_number)
+            config = rs.config()
+            config.enable_device(sn)
+            config.enable_all_streams()
+            pipeline_profile = config.resolve(
+                rs.pipeline_wrapper(rs.pipeline())
+                )
+            streams = {}
+            for s in pipeline_profile.get_streams():
+                name = s.stream_name()
+                streams[name] = {
+                    'format': str(s.format())[7:],  # remove 'format.'
+                    'framerate': s.fps(),
+                    'type': str(s.stream_type())[7:],  # remove 'stream.'
+                    }
+                if s.is_motion_stream_profile():
+                    streams[name]['intrinsics'] = \
+                        s.as_motion_stream_profile().get_motion_intrinsics().data
+                elif s.is_video_stream_profile():
+                    intrinsics = s.as_video_stream_profile().get_intrinsics()
+                    streams[name].update({
+                        'coeffs': intrinsics.coeffs,
+                        'model': str(intrinsics.model)
+                                 [11:],  # remove 'distortion.'
+                        'fx': intrinsics.fx,
+                        'fy': intrinsics.fy,
+                        'width': intrinsics.width,
+                        'height': intrinsics.height,
+                        'ppx': intrinsics.ppx,
+                        'ppy': intrinsics.ppy,
+                        })
+            devices[sn] = {
+                'name': d.get_info(rs.camera_info.name),
+                'streams': streams,
+                'serial_number': sn,
+                }
+        return devices
+
+
+# RealSense = Device(find_realsense, 'RealSense')
+# RealSense.config_map = {
+#     'streams': 'streams',
+#     **RealSense.config_map,
+#     }