プロジェクト概要

AcousticSenseは音響信号処理とコンピュータビジョンを統合した音楽ジャンル分類システムです。オーディオをメルスペクトログラムに変換し、Vision Transformerモデルで16種類の音楽ジャンル（ブルース、クラシック、ジャズなど）を識別します。

環境設定と主要コンポーネント

動作要件

• Python 3.10以上
• PyTorchフレームワーク
• Librosaオーディオ処理ライブラリ

中核スクリプト

• app_gradio.py: Webインターフェース
• inference.py: 推論処理コア
• start.sh: サービス起動スクリプト

推論エンジンの主要関数

モデル初期化処理

def initialize_model(weights_path="/data/models/vit_b_16_mel/checkpoint.pt"):
    """ViT-B/16モデルをロード
    引数:
        weights_path: モデル重みファイルのパス
    戻り値:
        初期化済みモデルインスタンス
    """
    vit_model = torchvision.models.vit_b_16(pretrained=False)
    vit_model.heads.head = nn.Linear(vit_model.heads.head.in_features, 16)
    model_weights = torch.load(weights_path, map_location='cpu')
    vit_model.load_state_dict(model_weights['model_state_dict'])
    vit_model.eval()
    return vit_model

音響特徴量変換処理

def transform_audio(input_path, duration=10, sample_rate=22050):
    """オーディオをメルスペクトログラムに変換
    引数:
        input_path: オーディオファイルパス
        duration: 対象長（秒）
        sample_rate: サンプリングレート
    戻り値:
        前処理済みスペクトログラム
    """
    audio_data, _ = librosa.load(input_path, sr=sample_rate)
    if len(audio_data) > sample_rate * duration:
        audio_data = audio_data[:sample_rate * duration]
    else:
        audio_data = np.pad(audio_data, (0, sample_rate * duration - len(audio_data)))
    
    mel_spec = librosa.feature.melspectrogram(y=audio_data, sr=sample_rate, n_mels=128)
    log_mel = librosa.power_to_db(mel_spec, ref=np.max)
    normalized = (log_mel - log_mel.min()) / (log_mel.max() - log_mel.min())
    return np.stack([normalized] * 3, axis=-1)

ジャンル分類処理

def classify_music(input_file, model_instance):
    """オーディオの音楽ジャンルを分類
    引数:
        input_file: オーディオファイルパス
        model_instance: 初期化済みモデル
    戻り値:
        ジャンル予測結果
    """
    spectrogram_data = transform_audio(input_file)
    input_data = torch.tensor(spectrogram_data).permute(2, 0, 1).unsqueeze(0).float()
    
    with torch.no_grad():
        model_output = model_instance(input_data)
        probs = torch.nn.functional.softmax(model_output, dim=1)
    
    top_probs, top_indices = torch.topk(probs, 5)
    genres = ['Blues', 'Classical', 'Jazz', 'Folk', 'Pop', 'Electronic', 
              'Disco', 'Rock', 'Hip-Hop', 'Metal', 'R&B', 'Reggae', 
              'World', 'Latin', 'Country', 'Alternative']
    
    return [{
        'genre': genres[idx.item()], 
        'confidence': round(prob.item() * 100, 2)
    } for prob, idx in zip(top_probs[0], top_indices[0])]

カスタムパス実装例

単一ファイル処理

from inference import initialize_model, classify_music

model_ref = initialize_model()
result = classify_music("/user/data/track.mp3", model_ref)

print("分類結果:")
for idx, item in enumerate(result, 1):
    print(f"{idx}. {item['genre']}: {item['confidence']}%")

ディレクトリ一括処理

import os

def process_directory(music_dir, output="predictions.csv"):
    model_ref = initialize_model()
    with open(output, 'w') as f:
        f.write("ファイル名,主要ジャンル,確信度\n")
        for file in os.listdir(music_dir):
            if file.endswith(('.mp3', '.wav')):
                try:
                    file_path = os.path.join(music_dir, file)
                    prediction = classify_music(file_path, model_ref)[0]
                    f.write(f"{file},{prediction['genre']},{prediction['confidence']}\n")
                except Exception as e:
                    print(f"エラー {file}: {e}")

Web API統合

from flask import Flask, request, jsonify

app = Flask(__name__)
classifier_model = initialize_model()

@app.route('/classify', methods=['POST'])
def classification_handler():
    if 'audio_file' not in request.files:
        return jsonify(error="オーディオファイルがありません"), 400
    
    audio = request.files['audio_file']
    temp_path = f"/tmp/{audio.filename}"
    audio.save(temp_path)
    
    try:
        result = classify_music(temp_path, classifier_model)
        os.remove(temp_path)
        return jsonify(results=result)
    except Exception as e:
        if os.path.exists(temp_path):
            os.remove(temp_path)
        return jsonify(error=str(e)), 500

トラブルシューティング

信頼性向上処理

def validate_and_process(audio_path, model_ref):
    if not os.path.exists(audio_path):
        return {"error": "ファイルが見つかりません"}
    if os.path.getsize(audio_path) > 50 * 1024 * 1024:
        return {"error": "ファイルサイズ超過"}
    return classify_music(audio_path, model_ref)

並列処理最適化

from concurrent.futures import ThreadPoolExecutor

def parallel_processing(file_list, model_ref, workers=4):
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(classify_music, f, model_ref): f for f in file_list}
        return {f: future.result() for future, f in futures.items()}