Storageにデータが追加されたときにトリガーされるFirebase FunctionsをPythonで実装し, デプロイする際に苦労した備忘録

はじめに

Pythonで書かれた音声編集プログラムをFlutterから動かしたいと考え、方法を模索していたところ、FunctionsがPythonにも対応していることを思い出し、早速実装してみました。しかし、情報が少ないこともあり、かなり苦労したので記録しておきたいと思います。

メインポイント

想像以上に記事が長くなったため、メインポイントのみ最初にまとめておきます。
pythonで実装したFirebaseのStorageに関するFunctionsをデプロイする際には主に以下2つの点に注意してください。

1. pythonのバージョンが3.12.4以上であること
2. StorageのバケットのリージョンとFunctionsのリージョンが一致していること

開発環境

flutter: 3.19.0
firebase: 13.12.0
python: 3.12.4
pyenv: 2.4.4

Functionsの追加

以下の記事を参考にfunctionsをプロジェクトに追加します。

デプロイしようとしたコード

プログラムの内容を簡潔に説明します。
これは、Firebase Functionsで指定したバケットを監視し、m4aファイルがアップロードされた際に自動的にトリガーされ、音声をmp3に変換し、特定周波数帯を増幅して編集後の音声をm4a形式で保存するプログラムです。

コードの書き方は公式サイトを参考にしました。

main.py

from firebase_functions import firestore_fn, https_fn, storage_fn
from firebase_admin import initialize_app, storage, firestore
import os
import pathlib
import librosa
import numpy as np
import soundfile as sf
import ffmpeg
import google.cloud.firestore
from firebase_functions import storage_fn
from firebase_admin import initialize_app, storage, credentials

# Firebaseアプリの初期化。ストレージバケットを明示的に指定
cred = credentials.ApplicationDefault()
initialize_app(cred, {
    'storageBucket': 'cocomakers-sound-classify-app.appspot.com'  # バケット名を指定
})

@storage_fn.on_object_finalized(bucket="cocomakers-sound-classify-app.appspot.com", region='asia-northeast1')

# 取得した音を編集するコード
def process_audio(event: storage_fn.CloudEvent[storage_fn.StorageObjectData]):
    bucket_name = event.data.bucket
    file_path = pathlib.PurePath(event.data.name)
    content_type = event.data.content_type
    if not content_type or not content_type.startswith("audio/x-m4a"):
        print(f"Unsupported content type: {content_type}")
        return

    if file_path.name.startswith("edited_"):
        print("Already edited.")
        return

    bucket = storage.bucket(bucket_name)

    input_blob = bucket.blob(str(file_path))
    input_path = f"/tmp/{file_path.name}"
    input_blob.download_to_filename(input_path)

    # Convert m4a to mp3
    mp3_filename = f"{file_path.stem}.mp3"
    mp3_path = f"/tmp/{mp3_filename}"
    try:
        ffmpeg.input(input_path).output(mp3_path).run()
    except ffmpeg.Error as e:
        print(f"Error converting m4a to mp3: {e}")
        return

    # Verify that the mp3 file was created successfully
    if not os.path.exists(mp3_path):
        print(f"mp3 file not found at {mp3_path}")
        return
    else:
        print(f"mp3 file successfully created at {mp3_path}")

    # Audio processing
    try:
        y, sr = librosa.load(mp3_path, sr=48000)
    except Exception as e:
        print(f"Error loading mp3 file with librosa: {e}")
        return
    
    D = librosa.stft(y)
    D_magnitude, D_phase = librosa.magphase(D)
    D_magnitude_db = librosa.amplitude_to_db(D_magnitude, ref=np.max)

    # Amplify specific frequency band
    target_low = 2000
    target_high = 4000
    amplification_factor = 2.5
    freqs = librosa.fft_frequencies(sr=sr)
    target_freqs_mask = (freqs >= target_low) & (freqs <= target_high)
    D_magnitude_db[target_freqs_mask, :] += amplification_factor

    D_amplified = librosa.db_to_amplitude(D_magnitude_db) * D_phase
    y_amplified = librosa.istft(D_amplified)

    # Save the edited file
    output_filename = f"edited_{file_path.stem}.wav"
    output_path = f"/tmp/{output_filename}"
    sf.write(output_path, y_amplified, sr)

    # Convert wav to m4a
    output_m4a_filename = f"edited_{file_path.stem}.m4a"
    output_m4a_path = f"/tmp/{output_m4a_filename}"
    try:
        ffmpeg.input(output_path).output(output_m4a_path, acodec="aac").run()
    except ffmpeg.Error as e:
        print(f"Error converting wav to m4a: {e}")
        return

    # Upload the processed file to Cloud Storage
    output_blob = bucket.blob(f"edited/{output_m4a_filename}")
    output_blob.upload_from_filename(output_m4a_path)

    # Clean up temporary files
    os.remove(input_path)
    os.remove(mp3_path)
    os.remove(output_path)
    os.remove(output_m4a_path)

    print(f"Processed file uploaded to edited/{output_m4a_filename}")

requirements.txt

firebase_functions~=0.1.0
librosa==0.10.2.post1
numpy==2.0.0
soundfile==0.12.1
ffmpeg-python==0.2.0
google-cloud-storage==2.17.0
firebase-admin==6.5.0

デプロイ方法

仮想環境の作成

python -m venv venv

アクティベートする

source venv/bin/activate

requirements.txtのパッケージをインストール

pip install -r requirements.txt

デプロイ

firebase deploy --only functions:process_audio  

※process_audio の部分には各々の関数名を記載してください。

発生したエラー

1. Failed to find location of Firebase Functions SDK. Did you forget to run" ~ && python3.12 -m pip install -r requirements.txt?

[解決方法]
仮想環境のpythonのバージョンが3.10になってたせいでした。3.12にしたら解決しました。

2.Can't find the storage bucket region

公式サイトにもロケーションが一致しないとデプロイが失敗することもあります。と書かれている通り、StorageとFunctionsとでregionの設定が異なる場合に起きるエラーです。
Functionsのrigionがデフォルトでus-central1に設定されている一方で、StorageのrigionはFirestoreを設定する際などに設定したregion(asia-northeast1)になっているのが原因でした。

GCP(Google Cloud Platform)にアクセスし、バケットのリージョンを確認します。

[解決方法]
① GCLでプロジェクトを開く

GCLを開き、プロジェクトの選択から自分のプロジェクトを選択します。
※最近のプロジェクトではなくすべてから探してください

② Cloud Storage > バケット

③ バケット名とregionを確認

④ プログラムに以下を追記

# Firebaseアプリの初期化。ストレージバケットを明示的に指定します。
cred = credentials.ApplicationDefault()
initialize_app(cred, {
    'storageBucket': 'cocomakers-sound-classify-app.appspot.com'  # バケット名を指定します
})

@storage_fn.on_object_finalized(bucket="cocomakers-sound-classify-app.appspot.com", region='asia-northeast1')　　# バケット名とリージョン名を指定します

3. さらにエラーが出た

pythonのバージョンをpython3.12.4にしてデプロイしたところ正常にデプロイできました🎉

最後に

まだ情報の少ない新しい技術だからこそたくさんのエラーに出会いますよね。
私はインターンのメンターさんや仲間に夜遅くまで手伝っていただき解決することができました。本当に感謝しています。
これからも、アプリ開発者として新しい技術にも怖がらずに手を出していく開拓者を目指していきたいと思います。