pyrec - A command line tool to record audio files.

pyrec.py

#!/usr/bin/env python3
"""
pyrec - A command line tool to record audio files.
With silence detection and optional monitoring.
Works on Windows, macOS, and Linux.
https://gist.github.com/psa-jforestier/a8e80d6501fd2ee33725624411027cb9
"""

import argparse
from collections import deque
from datetime import datetime
import os
import sys
import time
import wave
import numpy as np


def list_input_devices():
    """List all available audio input devices on the system."""
    try:
        import sounddevice as sd
    except ImportError:
        print("Error: 'sounddevice' package is required. Install it with: pip install sounddevice", file=sys.stderr)
        sys.exit(1)

    devices = sd.query_devices()
    host_apis = sd.query_hostapis()

    input_devices = [
        (i, d) for i, d in enumerate(devices)
        if d["max_input_channels"] > 0
    ]

    if not input_devices:
        print("No audio input devices found.")
        return

    print(f"{'ID':<5} {'Channels':<10} {'Sample Rate':<14} {'API':<12} Name")
    print("-" * 72)
    for idx, dev in input_devices:
        api_name = host_apis[dev["hostapi"]]["name"]
        default_marker = " *" if idx == sd.default.device[0] else ""
        print(
            f"{idx:<5} {dev['max_input_channels']:<10} {int(dev['default_samplerate']):<14} {api_name:<12} {dev['name']}{default_marker}"
        )

    print()
    default_idx = sd.default.device[0]
    if default_idx is not None and default_idx >= 0:
        print(f"* Default input device: [{default_idx}] {devices[default_idx]['name']}")


def list_output_formats():
    """List all supported audio output formats."""
    formats = {
        "wav8": {
            "description": "8-bit WAVE audio (PCM)",
            "channels": "matches input device",
            "frequency": "matches input device"
        },
        "wav16": {
            "description": "16-bit WAVE audio (PCM)",
            "channels": "matches input device",
            "frequency": "matches input device"
        },
    }

    print("Supported output formats:")
    print(f"{'Format':<12} {'Description':<30} {'Channels':<20} Frequency")
    print("-" * 80)
    for fmt, info in formats.items():
        print(
            f"{fmt:<12} {info['description']:<30} {info['channels']:<20} {info['frequency']}"
        )


def validate_format(format_str):
    """
    Validate and parse the output format string.
    
    Args:
        format_str: Format specification (e.g., 'wav8', 'wav16')
    
    Returns:
        dict: Format configuration with keys: format_type, sample_width, subtype
    """
    if format_str == "wav8":
        return {
            "format_type": "wav",
            "sample_width": 1,  # 8 bits = 1 byte
            "subtype": "PCM_U8"
        }
    elif format_str == "wav16":
        return {
            "format_type": "wav",
            "sample_width": 2,  # 16 bits = 2 bytes
            "subtype": "PCM_16"
        }
    else:
        raise ValueError(f"Unsupported output format: {format_str}. Use 'pyrec -f' to see available formats.")


def get_device_info(device_id):
    """
    Get audio device information.
    
    Args:
        device_id: Device ID number
    
    Returns:
        dict: Device information with channels and sample rate
    """
    try:
        import sounddevice as sd
    except ImportError:
        print("Error: 'sounddevice' package is required.", file=sys.stderr)
        sys.exit(1)
    
    try:
        device = sd.query_devices(device_id)
        if device["max_input_channels"] == 0:
            raise ValueError(f"Device {device_id} is not an input device.")
        
        return {
            "id": device_id,
            "name": device["name"],
            "channels": int(device["max_input_channels"]),
            "samplerate": int(device["default_samplerate"])
        }
    except Exception as e:
        raise ValueError(f"Invalid device ID {device_id}: {e}")


def float_to_pcm(audio_data, sample_width):
    """Convert float32 audio in [-1.0, 1.0] to PCM bytes for WAV writing."""
    clipped = np.clip(audio_data, -1.0, 1.0)

    if sample_width == 1:
        # WAV 8-bit PCM uses unsigned samples: 0..255
        pcm = np.round((clipped + 1.0) * 127.5).astype(np.uint8)
    elif sample_width == 2:
        pcm = np.round(clipped * 32767.0).astype(np.int16)
    else:
        raise ValueError(f"Unsupported sample width: {sample_width}")

    return pcm.tobytes()


def format_elapsed_time(seconds):
    """Format elapsed seconds as hh:mm:ss.sss."""
    total_milliseconds = max(0, int(seconds * 1000))
    total_seconds, milliseconds = divmod(total_milliseconds, 1000)
    hours, remainder = divmod(total_seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    return f"{hours:02d}:{minutes:02d}:{secs:02d}.{milliseconds:03d}"


def format_srt_timestamp(seconds):
    """Format elapsed seconds as SRT timestamp HH:MM:SS,mmm."""
    total_milliseconds = max(0, int(seconds * 1000))
    total_seconds, milliseconds = divmod(total_milliseconds, 1000)
    hours, remainder = divmod(total_seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    return f"{hours:02d}:{minutes:02d}:{secs:02d},{milliseconds:03d}"


def write_srt_file(srt_path, events):
    """Write recording events to an SRT subtitle file."""
    with open(srt_path, "w", encoding="utf-8") as srt_file:
        for idx, event in enumerate(events, start=1):            
            start_ts = format_srt_timestamp(event["offset_seconds"])
            end_ts = format_srt_timestamp(event["offset_seconds"] + 1.0)
            srt_file.write(f"{idx}\n")
            srt_file.write(f"{start_ts} --> {end_ts}\n")
            srt_file.write(f"REC {event['started_at']}\n\n")


def record_audio(device_id, device_info, output_file, format_info, monitor_enabled, level_threshold_db, post_seconds, pre_seconds, srt_enabled):
    """Capture audio from the selected input device and optionally display monitoring."""
    try:
        import sounddevice as sd
    except ImportError:
        print("Error: 'sounddevice' package is required. Install it with: pip install sounddevice", file=sys.stderr)
        sys.exit(1)

    should_record = output_file is not None
    frames = []
    total_frames = 0
    current_level_db = -1 #float("-inf")
    recording_active = level_threshold_db is None
    was_recording_active = recording_active
    post_remaining_seconds = 0.0
    pre_buffer = deque()
    pre_buffer_frames = 0
    max_pre_frames = int(max(0, pre_seconds) * device_info["samplerate"])
    bytes_per_frame = device_info["channels"] * format_info["sample_width"]
    srt_events = []

    def callback(indata, frame_count, time_info, status):
        nonlocal current_level_db, total_frames, recording_active, was_recording_active
        nonlocal post_remaining_seconds, pre_buffer_frames
        if status:
            print(f"Warning: {status}", file=sys.stderr)

        rms = float(np.sqrt(np.mean(np.square(indata), dtype=np.float64)))
        #if rms > 0.0:
        #    current_level_db = 20.0 * np.log10(rms)
        #else:
        #    current_level_db = float("-inf")
        # tweak the the level calculation to be more easy to handle
        if rms > 0.0:
            current_level_db = 100 - abs(int(20.0 * np.log10(rms)))
        else:
            current_level_db = -1

        if level_threshold_db is None:
            recording_active = True
        else:
            current_level_ok = np.isfinite(current_level_db) and current_level_db >= level_threshold_db
            if current_level_ok:
                recording_active = True
                post_remaining_seconds = float(post_seconds)
            elif post_remaining_seconds > 0.0:
                chunk_seconds = frame_count / float(device_info["samplerate"])
                post_remaining_seconds = max(0.0, post_remaining_seconds - chunk_seconds)
                recording_active = True
            else:
                recording_active = False

        if should_record:
            transitioned_to_active = recording_active and not was_recording_active
            if level_threshold_db is not None and max_pre_frames > 0 and not recording_active:
                pre_buffer.append((indata.copy(), frame_count))
                pre_buffer_frames += frame_count
                while pre_buffer_frames > max_pre_frames and pre_buffer:
                    _, dropped_count = pre_buffer.popleft()
                    pre_buffer_frames -= dropped_count

            if recording_active:
                if srt_enabled and transitioned_to_active:
                    event_offset_seconds = total_frames / float(device_info["samplerate"])
                    if level_threshold_db is not None and max_pre_frames > 0:
                        event_offset_seconds += pre_buffer_frames / float(device_info["samplerate"])
                    srt_events.append(
                        {
                            "offset_seconds": event_offset_seconds,
                            "started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                        }
                    )

                if level_threshold_db is not None and max_pre_frames > 0 and not was_recording_active:
                    while pre_buffer:
                        chunk_data, chunk_count = pre_buffer.popleft()
                        frames.append(chunk_data)
                        total_frames += chunk_count
                    pre_buffer_frames = 0

                frames.append(indata.copy())
                total_frames += frame_count

        was_recording_active = recording_active

    if should_record:
        output_dir = os.path.dirname(os.path.abspath(output_file))
        if output_dir and not os.path.exists(output_dir):
            os.makedirs(output_dir, exist_ok=True)

    if should_record:
        print("Recording started. Press Ctrl+C to stop.")
        if level_threshold_db is not None:
            print(f"Recording level threshold: {level_threshold_db} dB")
            print(f"Post-silence hold: {post_seconds} s")
            print(f"Pre-buffer: {pre_seconds} s")

    if monitor_enabled and not should_record:
        print("Monitoring started. Press Ctrl+C to stop.")
    try:
        with sd.InputStream(
            device=device_id,
            channels=device_info["channels"],
            samplerate=device_info["samplerate"],
            dtype="float32",
            callback=callback,
        ):
            if monitor_enabled:
                print("STAT  |    Time     |  Bytes  | Level")
            start_time = time.monotonic()
            while True:
                if monitor_enabled: # Live status display
                    elapsed = time.monotonic() - start_time
                    recorded_bytes = total_frames * bytes_per_frame
                    if current_level_db >= 0:
                        l = current_level_db
                        friendly_level = f"{l:3}" + ' ' + ':' * (l // 5)
                    else:
                        friendly_level = "..."
                    status_text = "REC" if (should_record and recording_active) else "PAUSE"
                    if not should_record:
                        status_text = "MON"
                    print(f"{status_text:5} | {format_elapsed_time(elapsed)} | {recorded_bytes:7} | {friendly_level}")
                sd.sleep(100)
    except KeyboardInterrupt:
        pass
    except Exception as e:
        print(f"Error while recording: {e}", file=sys.stderr)
        sys.exit(1)

    if not should_record:
        if monitor_enabled:
            print("Monitoring stopped.")
        return

    if not frames:
        print("No audio data captured.", file=sys.stderr)
        sys.exit(1)

    audio_data = np.concatenate(frames, axis=0)
    pcm_bytes = float_to_pcm(audio_data, format_info["sample_width"])

    try:
        with wave.open(output_file, "wb") as wav_file:
            wav_file.setnchannels(device_info["channels"])
            wav_file.setsampwidth(format_info["sample_width"])
            wav_file.setframerate(device_info["samplerate"])
            wav_file.writeframes(pcm_bytes)
    except Exception as e:
        print(f"Error writing output file '{output_file}': {e}", file=sys.stderr)
        sys.exit(1)

    duration_seconds = audio_data.shape[0] / float(device_info["samplerate"])
    print(f"Recording saved: {output_file}")
    print(f"Duration: {duration_seconds:.2f} s")

    if srt_enabled:
        srt_path = f"{output_file}.srt"
        try:
            write_srt_file(srt_path, srt_events)
        except Exception as e:
            print(f"Error writing subtitle file '{srt_path}': {e}", file=sys.stderr)
            sys.exit(1)
        print(f"Subtitle file saved: {srt_path}")




def main():
    # Keep standalone listing commands available without requiring mandatory recording args.
    if len(sys.argv) == 2 and sys.argv[1] == "-f":
        list_output_formats()
        return

    if len(sys.argv) == 2 and sys.argv[1] == "-i":
        list_input_devices()
        return

    parser = argparse.ArgumentParser(
        prog="pyrec",
        description="A command line tool to record audio files.",
    )

    parser.add_argument(
        "-i",
        type=int,
        required=True,
        metavar="REC_DEVICE",
        help="Input recording device ID (integer). Use -i to list available devices.",
    )
    
    parser.add_argument(
        "-f",
        nargs="?",
        const="__list__",
        default="wav16",
        metavar="FORMAT",
        help="Output format: use '-f' to list supported formats, or '-f FORMAT' to select one (default: wav16).",
    )
    
    parser.add_argument(
        "-o",
        metavar="OUTPUT_FILE",
        help="Output file path (e.g., /path/to/output.wav). If omitted, no audio file is written.",
    )

    parser.add_argument(
        "-l",
        type=int,
        metavar="LEVEL_DB",
        help="Recording threshold level in dB. Recording writes only while level is >= this value.",
    )

    parser.add_argument(
        "--monitor",
        action="store_true",
        help="Display live monitoring during capture.",
    )

    parser.add_argument(
        "--post",
        type=int,
        default=0,
        metavar="SECONDS",
        help="Number of seconds recording continues after level drops below -l (default: 0).",
    )

    parser.add_argument(
        "--pre",
        type=int,
        default=0,
        metavar="SECONDS",
        help="Number of seconds buffered before level detection and added when recording starts (default: 0).",
    )

    parser.add_argument(
        "--srt",
        action="store_true",
        help="Create an SRT subtitle file with one entry per recording start event.",
    )
    
    args = parser.parse_args()

    # Handle -f flag with no value (display formats)
    if args.f == "__list__":
        list_output_formats()
        return

    if not args.o and not args.monitor:
        print("Error: either -o or --monitor must be provided.", file=sys.stderr)
        sys.exit(1)

    if args.srt and not args.o:
        print("Error: --srt requires -o.", file=sys.stderr)
        sys.exit(1)

    if args.post < 0:
        print("Error: --post must be >= 0.", file=sys.stderr)
        sys.exit(1)

    if args.pre < 0:
        print("Error: --pre must be >= 0.", file=sys.stderr)
        sys.exit(1)

    try:
        device_info = get_device_info(args.i)
    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        print("Available input devices:")
        list_input_devices()
        sys.exit(1)

    try:
        format_info = validate_format(args.f)
    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    print(f"Selected device [{args.i}]: {device_info['name']}")
    print(f"  Channels: {device_info['channels']}")
    print(f"  Sample rate: {device_info['samplerate']} Hz")
    print(f"Output format: {args.f} ({format_info['subtype']})")
    if args.o:
        print(f"Output file: {args.o}")
        if args.l is not None:
            print(f"Recording threshold: {args.l} dB")
            print(f"Post-silence hold: {args.post} s")
            print(f"Pre-buffer: {args.pre} s")
    if args.monitor and not args.o:
        print("Mode: monitoring only")

    record_audio(args.i, device_info, args.o, format_info, args.monitor, args.l, args.post, args.pre, args.srt)




if __name__ == "__main__":
    main()