flac_s16_to_u8.py

#!/usr/bin/env python3
"""
Convert FLAC compressed RF files to unsigned 8-bit format.

This script decompresses FLAC (16-bit signed) and converts to u8 format
for use with SECAM decoder and other RF processing tools.
"""

import argparse
import numpy as np
import subprocess
import tempfile
import os
from tqdm import tqdm


def decompress_flac(flac_file, s16_file):
    """
    Decompress FLAC to 16-bit signed raw file.
    
    Args:
        flac_file: Path to input FLAC file
        s16_file: Path to output 16-bit signed raw file
    """
    print(f"Decompressing FLAC...")
    
    cmd = [
        'ffmpeg',
        '-i', flac_file,
        '-f', 's16le',
        '-acodec', 'pcm_s16le',
        s16_file
    ]
    subprocess.run(cmd, check=True)


def convert_s16_to_u8(s16_file, u8_file, chunk_size=20*1024*1024):
    """
    Convert 16-bit signed to 8-bit unsigned.
    
    Args:
        s16_file: Path to 16-bit signed file
        u8_file: Path to output u8 file
        chunk_size: Bytes to process at once (must be divisible by 2)
    """
    # Ensure chunk_size is divisible by 2 (size of int16)
    chunk_size = (chunk_size // 2) * 2
    
    with open(s16_file, 'rb') as fin:
        # Get total file size for progress bar
        fin.seek(0, 2)
        total_size = fin.tell()
        fin.seek(0)
        
        if total_size % 2 != 0:
            raise Exception(f"Input file size ({total_size}) is not divisible by 2 - corrupt s16 file")
        
        with open(u8_file, 'wb') as fout:
            with tqdm(total=total_size, unit='B', unit_scale=True, desc='Converting to u8') as pbar:
                while True:
                    # Read chunk
                    chunk = fin.read(chunk_size)
                    if len(chunk) == 0:
                        break
                    
                    # Convert to numpy array of 16-bit signed samples
                    samples_s16 = np.frombuffer(chunk, dtype=np.int16)
                    
                    # Convert 16-bit signed to 8-bit unsigned
                    # Formula: (value / 256) + 128
                    # This maps -32768..32767 to 0..255
                    samples_u8 = ((samples_s16.astype(np.int32) // 256) + 128).astype(np.uint8)
                    
                    # Write to file
                    fout.write(samples_u8.tobytes())
                    
                    pbar.update(len(chunk))


def main():
    parser = argparse.ArgumentParser(
        description='Convert FLAC compressed RF files to u8 format',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument('--input', required=True, help='Input .flac or .ldf file')
    parser.add_argument('--output', required=True, help='Output .u8 file')
    parser.add_argument('--keep-temp', action='store_true',
                        help='Keep temporary 16-bit file')
    
    args = parser.parse_args()
    
    # Create temporary s16 file
    with tempfile.NamedTemporaryFile(suffix='.s16', delete=False) as tmp:
        temp_s16 = tmp.name
    
    try:
        # Step 1: Decompress FLAC to 16-bit signed
        decompress_flac(args.input, temp_s16)
        
        # Step 2: Convert to u8
        convert_s16_to_u8(temp_s16, args.output)
        
        # Clean up temp file unless requested to keep
        if not args.keep_temp:
            os.remove(temp_s16)
        
        print(f"Conversion complete: {args.output}")
        
        # Show file sizes
        input_size = os.path.getsize(args.input)
        output_size = os.path.getsize(args.output)
        print(f"Input:  {input_size / (1024**3):.2f} GB")
        print(f"Output: {output_size / (1024**3):.2f} GB")
        
    except Exception as e:
        # Clean up temp file on error
        if os.path.exists(temp_s16):
            os.remove(temp_s16)
        raise e


if __name__ == '__main__':
    main()