refactored into library

Author: esdandreu

app.py

@@ -1,11 +1,13 @@
 microphone:
   # A microphone configuration should match the call signature of
-  # `sounddevice.InputStream` as it will be used on the call, except for two
-  # optional parameters: `name` and `samplewidth`
+  # `sounddevice.InputStream` as it will be used on the call, except for three
+  # additional parameters: `name`, `index` and `samplewidth`
   # https://python-sounddevice.readthedocs.io/en/0.4.4/api/streams.html#sounddevice.InputStream
-  17: # device_id
+
+  0: # Device index
     name: Realtek HD Audio Mic input # Optional, human-readable device_id
+    index: 17 # device_id
     channels: 2
     samplerate: 44100.0
     samplewidth: 2 # Sample width in bytes when stored in the `wav` file
-camera: {}
+realsense: {}

example-config.yaml

@@ -0,0 +1,2 @@
+from .recorder import Recorder
+from .config import Config

recorder/__init__.py

@@ -0,0 +1,57 @@
+from recorder.io import yaml_load
+from recorder.device import Device, DEVICE_CLASSES
+
+from typing import List
+from pathlib import Path
+from datetime import datetime
+
+DEFAULT_CONFIG_PATH = Path.cwd() / 'vao-recorder.yaml'
+
+DEFAULT_OUTPUT_FOLDER = Path.cwd() / 'output' 
+
+
+class Config(dict):
+
+    @classmethod
+    def from_yaml(cls, path: Path = DEFAULT_CONFIG_PATH) -> 'Config':
+        if not path.exists():
+            raise AttributeError(f'Configuration file {path} does not exist')
+        return cls(yaml_load(path))
+
+    def devices(
+        self,
+        output_folder: Path = DEFAULT_OUTPUT_FOLDER,
+        ) -> List[Device]:
+        """Initializes all devices present in the dictionary with the provided
+        output folder.
+
+        Args:
+            output_folder (Path, optional): Devices output_folder. Defaults to
+            DEFAULT_OUTPUT_FOLDER.
+
+        Raises:
+            RuntimeError: _description_
+
+        Returns:
+            List[Device]: List of initialized devices.
+
+        """
+        devices = []
+        for _device_class, _devices in self.items():
+            try:
+                device_class = DEVICE_CLASSES[_device_class]
+            except KeyError:
+                raise RuntimeError(
+                    f'Invalid configuration file, {_device_class} is not a '
+                    'valid device class. Available options: '
+                    f'{list(DEVICE_CLASSES.keys())}'
+                    )
+            for index, device_config in _devices.items():
+                devices.append(
+                    device_class(
+                        index=index,
+                        output_folder=output_folder,
+                        config=device_config,
+                        )
+                    )
+        return devices

recorder/config.py

@@ -1,3 +1,5 @@
 from .device import Device
 from .microphone import Microphone
 from .realsense import RealSense
+
+DEVICE_CLASSES = {repr(cls): cls for cls in Device.__subclasses__()}

recorder/device/__init__.py

@@ -1,23 +1,40 @@
+from recorder.io import yaml_dump
+
 from typing import Dict
 from pathlib import Path
+from warnings import warn
+from datetime import datetime
 from abc import ABC, ABCMeta, abstractmethod
 
 
 class MetaDevice(ABCMeta):
 
-    def __str__(cls):
+    def __str__(cls) -> str:
         return cls.__name__
 
+    def __repr__(cls) -> str:
+        return cls.__name__.lower()
+
 
 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 __init__(self, index: int, output_folder: Path, config: dict):
+        self.config = config
+        assert 'start_timestamp' not in self.config
+        assert 'end_timestamp' not in self.config
+        self.index = index
+        self.name = self.config.get('name', 'unknown')
+        self.output_file = output_folder / f'{repr(self)}'
 
     def __str__(self) -> str:
         return self.name
 
+    def __repr__(self) -> str:
+        return f'{repr(type(self))}{self.index}'
+
+    def __del__(self):
+        yaml_dump(self.config, to_file=self.output_file.with_suffix('.yaml'))
+
     @classmethod
     @abstractmethod
     def find(cls) -> Dict[int, dict]:
@@ -29,20 +46,27 @@ def find(cls) -> Dict[int, dict]:
         """
         pass
 
-    @abstractmethod
     def start(self) -> None:
-        """Starts recording.
+        # Starts recording storing the start timestamp in configuration
+        self.config['start_timestamp'] = datetime.now().timestamp()
+        self._start()
 
-        Returns:
-            None
-        """
+    @abstractmethod
+    def _start(self) -> None:
+        # Starts recording.
         pass
 
-    @abstractmethod
-    def stop(self) -> None:
-        """Stops recording.
+    def stop(self):
+        # Stops recording storing the end timestamp in configuration
+        self._stop()
+        self.config['end_timestamp'] = datetime.now().timestamp()
 
-        Returns:
-            None
-        """
+    @abstractmethod
+    def _stop(self):
+        # Stops recording
         pass
+
+    @classmethod
+    @abstractmethod
+    def show_results(cls, from_folder):
+        pass