Get Opus duration and bitrate

get_opus_duration_and_bitrate.py

#!/usr/bin/env python3

"""get_opus_duration_and_bitrate.py

Get the duration and bitrate of an Opus file."""


import os
import sys
import struct
import argparse


def get_opus_duration_and_bitrate(filename):
    """Calculate the duration and bitrate of an Opus file.

    Parameters
    -----------
    filename : str
        the name of the Opus file (i.e. "*.opus")

    Returns
    --------
    duration : float
        the duration of the Opus file is seconds
    bitrate : float
        the bitrate of the Opus file in kbps

    """

    def _read_ogg_page(f):
        # Read the Ogg page header (27 bytes)
        header = f.read(27)
        if len(header) < 27:
            return None  # End of file or error

        # Unpack the header
        try:
            (capture_pattern, version, header_type, granule_position,
             serial_number, page_sequence_no, checksum, page_segments) = \
                struct.unpack('<4sBsqIIIB', header)
        except struct.error:
            raise ValueError("Not a valid Ogg file")

        # Check for the Ogg capture pattern
        if capture_pattern != b'OggS' or version != 0:
            raise ValueError("Not a valid Ogg file")

        # Read the segment table
        segment_table = f.read(page_segments)
        if len(segment_table) < page_segments:
            return None  # End of file or error

        # Read the segment data
        segment_sizes = [seg for seg in segment_table]
        body_size = sum(segment_sizes)
        body = f.read(body_size)

        return {'header': header, 'body': body, 'serial_number': serial_number,
                'granule_position': granule_position, 'header_type': header_type,
                'page_sequence_no': page_sequence_no,
                'page_segments': page_segments}

    def _get_opus_stream_size_and_duration(f):
        f.seek(0)
        header_count = 0
        serial = None
        size = 0
        preskip = 0
        last_granule_position = 0
        final_granule_position = 0
        last_page_size = None
        eos = False

        while True:
            page = _read_ogg_page(f)
            if page is None:
                break  # End of file or error
            # Read pages until first Opus header page (identification header)
            elif header_count == 0: # Identification header
                if page['body'][:8] != b'OpusHead':  # Page from other stream?
                    continue
                else:  # Found it
                    serial = page['serial_number']
                    preskip = struct.unpack('<H', page['body'][10:12])[0]
                    header_count += 1
            # Check for expected second Opus header page (comment header)
            elif header_count == 1 and page['serial_number'] == serial:
                if page['body'][:8] != b'OpusTags':
                    raise ValueError("Not a valid Opus file")
                else:
                    header_count += 1
            # Read pages until first page with audio data
            elif header_count == 2 and page['serial_number'] == serial:
                if int.from_bytes(page['header_type'], 'little') & 0x01:
                    continue  # Contiunation of second Opus header page
                elif page['granule_position'] > 0:  # Found it
                    last_granule_position = page['granule_position']
                    final_granule_position = last_granule_position
                    header_count += 1
                    last_page_size = len(page['body'])
                    size += last_page_size
            # Read remaining pages with audio data
            elif page['granule_position'] > 0 and page['serial_number'] == serial:
                last_page_size = len(page['body'])
                size += last_page_size
                final_granule_position = last_granule_position
                last_granule_position = page['granule_position']
                # If page is last page (end of stream), we are done
                if int.from_bytes(page['header_type'], 'little') & 0x04:
                    eos = True
                    final_granule_position = last_granule_position
                    break

        # If last page did not contain "end of stream" flag, it was incomplete
        if not eos:
            size -= last_page_size

        # Calculate duration
        duration = (final_granule_position - preskip) / 48000.0

        return size, duration

    with open(filename, 'rb') as f:
        size, duration = _get_opus_stream_size_and_duration(f)
        bitrate = round((size * 8 / duration) / 1000, 1)

    return duration, bitrate


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        prog='get_opus_duration_and_bitrate.py',
        description="Print the duration and bitrate of an Opus file")
    parser.add_argument('filename', type=str)
    args = parser.parse_args()
    duration, bitrate = get_opus_duration_and_bitrate(args.filename)
    minutes = int(duration / 60)
    seconds = round(duration % 60)
    print(f"Duration: {minutes:02d}:{seconds:02d} ({duration}s)")
    print(f"Bitrate: {bitrate} kbps")