Audio Data Logger

Usage

example_data_logger.sh

python scripts/data_logger.py [-h] -u URL -i INPUT_DIR -l LABEL

Log region-grouped total audio duration to AirTable.

optional arguments:
  -h, --help            show this help message and exit
  -u URL, --url URL     AirTable URL.
  -i INPUT_DIR, --input_dir INPUT_DIR
                        Directory of input audios to log.
  -l LABEL, --label LABEL
                        Log record label. E.g. training/archive.

Example

example_data_logger.sh

export AIRTABLE_API_KEY="AIRTABLE_API_KEY"
export AIRTABLE_URL="AIRTABLE_TABLE_URL"
python scripts/data_logger.py --url $AIRTABLE_URL --input_dir dropbox/ --label archive
python scripts/data_logger.py --url $AIRTABLE_URL --input_dir training/ --label training

---

scripts.data_logger.DataLogger

Source code in scripts/data_logger.py

class DataLogger:
    def parse_args(self, args: List[str]) -> argparse.Namespace:
        """
        Utility argument parser function for data logging to AirTable.

        Args:
            args (List[str]):
                List of arguments.

        Returns:
            argparse.Namespace:
                Objects with arguments values as attributes.
        """
        parser = argparse.ArgumentParser(
            prog="python scripts/data_logger.py",
            description="Log region-grouped total audio duration to AirTable.",
        )

        parser.add_argument(
            "-u", "--url", type=str, required=True, help="AirTable URL."
        )
        parser.add_argument(
            "-i",
            "--input_dir",
            type=str,
            required=True,
            help="Directory of input audios to log.",
        )
        parser.add_argument(
            "-l",
            "--label",
            type=str,
            required=True,
            help="Log record label. E.g. training/archive.",
        )
        return parser.parse_args(args)

    def get_audio_duration(self, audio_path: str) -> float:
        """
        Calculate audio duration via ffprobe.
        Equivalent to:
        ```sh title="example_get_audio_duration.sh"
        ffprobe -v quiet -of csv=p=0 -show_entries format=duration {audio_path}
        ```

        Args:
            audio_path (str):
                Path to audio file.

        Returns:
            float:
                Duration in seconds.
        """
        job = subprocess.run(
            [
                "ffprobe",
                "-v",
                "quiet",
                "-of",
                "csv=p=0",
                "-show_entries",
                "format=duration",
                audio_path,
            ],
            stderr=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stdin=subprocess.PIPE,
        )
        duration = float(job.stdout.decode())
        return duration

    def get_language_total_audio_duration(self, input_dir: str) -> Dict[str, float]:
        """
        Map language folders in `input_dir` to their respective total audio duration.
        Assumes `input_dir` as `{input_dir}/{lang}/{audio}.wav`.

        ### Example
        ```pycon title="example_get_language_total_audio_duration.py"
        >>> logger = DataLogger()
        >>> logger.get_language_total_audio_duration("dropbox/")
        {'en-au': 3.936, 'id-id': 3.797}
        ```

        Args:
            input_dir (str):
                Path to input directory.

        Returns:
            Dict[str, float]:
                Dictionary of language to total audio duration.
        """
        languages = [f.name for f in os.scandir(input_dir) if f.is_dir()]
        language2duration = {}
        for language in languages:
            audios = glob(f"{input_dir}/{language}/*.wav")
            duration = round(sum(p_umap(self.get_audio_duration, audios)), 3)
            language2duration[language] = duration
        return language2duration

    def build_payload(
        self, date: str, label: str, language: str, duration: float
    ) -> Dict[str, Dict[str, Any]]:
        """
        Builds payload for AirTable record.
        AirTable record has the following structure:
        ```pycon
        >>> {
        ...     "date": {YYYY-MM-DD},
        ...     "label": {label},
        ...     "language": {lang},
        ...     "language-code": {lang-country},
        ...     "duration": {duration},
        ... }
        ```

        Args:
            date (str):
                Logging date.
            label (str):
                Audio folder label.
            language (str):
                Language code (lang-country). E.g. `en-us`.
            duration (float):
                Duration in seconds.

        Returns:
            Dict[str, Dict[str, Any]]:
                AirTable record payload.
        """
        return {
            "fields": {
                "date": date,
                "label": label,
                "language": language.split("-")[0],
                "language-code": language,
                "duration": duration,
            }
        }

    def log(self, url: str, input_dir: str, label: str) -> bool:
        """
        Logs region-grouped total audio duration in `input_dir` to AirTable at `url`.

        Args:
            url (str):
                AirTable URL.
            input_dir (str):
                Input directory to log.
            label (str):
                Log record label.

        Returns:
            bool:
                Whether upload was a success.
        """
        airtable = AirTable(url)
        language2duration = self.get_language_total_audio_duration(input_dir)
        records = [
            self.build_payload(
                str(date.today()),
                label,
                language,
                duration,
            )
            for language, duration in language2duration.items()
        ]
        return airtable.batch_add_records(records)

build_payload(self, date, label, language, duration)

Builds payload for AirTable record. AirTable record has the following structure:

>>> {
...     "date": {YYYY-MM-DD},
...     "label": {label},
...     "language": {lang},
...     "language-code": {lang-country},
...     "duration": {duration},
... }

Parameters:

Name	Type	Description	Default
date	str	Logging date.	required
label	str	Audio folder label.	required
language	str	Language code (lang-country). E.g. en-us.	required
duration	float	Duration in seconds.	required

Returns:

Type	Description
Dict[str, Dict[str, Any]]	AirTable record payload.

Source code in scripts/data_logger.py

def build_payload(
    self, date: str, label: str, language: str, duration: float
) -> Dict[str, Dict[str, Any]]:
    """
    Builds payload for AirTable record.
    AirTable record has the following structure:
    ```pycon
    >>> {
    ...     "date": {YYYY-MM-DD},
    ...     "label": {label},
    ...     "language": {lang},
    ...     "language-code": {lang-country},
    ...     "duration": {duration},
    ... }
    ```

    Args:
        date (str):
            Logging date.
        label (str):
            Audio folder label.
        language (str):
            Language code (lang-country). E.g. `en-us`.
        duration (float):
            Duration in seconds.

    Returns:
        Dict[str, Dict[str, Any]]:
            AirTable record payload.
    """
    return {
        "fields": {
            "date": date,
            "label": label,
            "language": language.split("-")[0],
            "language-code": language,
            "duration": duration,
        }
    }

get_audio_duration(self, audio_path)

Calculate audio duration via ffprobe. Equivalent to:

example_get_audio_duration.sh

ffprobe -v quiet -of csv=p=0 -show_entries format=duration {audio_path}

Parameters:

Name	Type	Description	Default
audio_path	str	Path to audio file.	required

Returns:

Type	Description
float	Duration in seconds.

Source code in scripts/data_logger.py

def get_audio_duration(self, audio_path: str) -> float:
    """
    Calculate audio duration via ffprobe.
    Equivalent to:
    ```sh title="example_get_audio_duration.sh"
    ffprobe -v quiet -of csv=p=0 -show_entries format=duration {audio_path}
    ```

    Args:
        audio_path (str):
            Path to audio file.

    Returns:
        float:
            Duration in seconds.
    """
    job = subprocess.run(
        [
            "ffprobe",
            "-v",
            "quiet",
            "-of",
            "csv=p=0",
            "-show_entries",
            "format=duration",
            audio_path,
        ],
        stderr=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stdin=subprocess.PIPE,
    )
    duration = float(job.stdout.decode())
    return duration

get_language_total_audio_duration(self, input_dir)

Map language folders in input_dir to their respective total audio duration. Assumes input_dir as {input_dir}/{lang}/{audio}.wav.

Example

example_get_language_total_audio_duration.py

>>> logger = DataLogger()
>>> logger.get_language_total_audio_duration("dropbox/")
{'en-au': 3.936, 'id-id': 3.797}

Parameters:

Name	Type	Description	Default
input_dir	str	Path to input directory.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of language to total audio duration.

Source code in scripts/data_logger.py

def get_language_total_audio_duration(self, input_dir: str) -> Dict[str, float]:
    """
    Map language folders in `input_dir` to their respective total audio duration.
    Assumes `input_dir` as `{input_dir}/{lang}/{audio}.wav`.

    ### Example
    ```pycon title="example_get_language_total_audio_duration.py"
    >>> logger = DataLogger()
    >>> logger.get_language_total_audio_duration("dropbox/")
    {'en-au': 3.936, 'id-id': 3.797}
    ```

    Args:
        input_dir (str):
            Path to input directory.

    Returns:
        Dict[str, float]:
            Dictionary of language to total audio duration.
    """
    languages = [f.name for f in os.scandir(input_dir) if f.is_dir()]
    language2duration = {}
    for language in languages:
        audios = glob(f"{input_dir}/{language}/*.wav")
        duration = round(sum(p_umap(self.get_audio_duration, audios)), 3)
        language2duration[language] = duration
    return language2duration

log(self, url, input_dir, label)

Logs region-grouped total audio duration in input_dir to AirTable at url.

Parameters:

Name	Type	Description	Default
url	str	AirTable URL.	required
input_dir	str	Input directory to log.	required
label	str	Log record label.	required

Returns:

Type	Description
bool	Whether upload was a success.

Source code in scripts/data_logger.py

def log(self, url: str, input_dir: str, label: str) -> bool:
    """
    Logs region-grouped total audio duration in `input_dir` to AirTable at `url`.

    Args:
        url (str):
            AirTable URL.
        input_dir (str):
            Input directory to log.
        label (str):
            Log record label.

    Returns:
        bool:
            Whether upload was a success.
    """
    airtable = AirTable(url)
    language2duration = self.get_language_total_audio_duration(input_dir)
    records = [
        self.build_payload(
            str(date.today()),
            label,
            language,
            duration,
        )
        for language, duration in language2duration.items()
    ]
    return airtable.batch_add_records(records)

parse_args(self, args)

Utility argument parser function for data logging to AirTable.

Parameters:

Name	Type	Description	Default
args	List[str]	List of arguments.	required

Returns:

Type	Description
argparse.Namespace	Objects with arguments values as attributes.

Source code in scripts/data_logger.py

def parse_args(self, args: List[str]) -> argparse.Namespace:
    """
    Utility argument parser function for data logging to AirTable.

    Args:
        args (List[str]):
            List of arguments.

    Returns:
        argparse.Namespace:
            Objects with arguments values as attributes.
    """
    parser = argparse.ArgumentParser(
        prog="python scripts/data_logger.py",
        description="Log region-grouped total audio duration to AirTable.",
    )

    parser.add_argument(
        "-u", "--url", type=str, required=True, help="AirTable URL."
    )
    parser.add_argument(
        "-i",
        "--input_dir",
        type=str,
        required=True,
        help="Directory of input audios to log.",
    )
    parser.add_argument(
        "-l",
        "--label",
        type=str,
        required=True,
        help="Log record label. E.g. training/archive.",
    )
    return parser.parse_args(args)