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をプロジェクトに追加します。

https://zenn.dev/singularity/articles/firebase-functions-python-functions

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

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

https://firebase.google.com/docs/functions/firestore-events?hl=ja&gen=2nd

### main.py
```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

```text
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
```


## デプロイ方法

仮想環境の作成
```shell
python -m venv venv
```

アクティベートする
```shell
source venv/bin/activate
```

requirements.txtのパッケージをインストール
```shell
pip install -r requirements.txt
```

デプロイ
```shell
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](https://console.cloud.google.com/welcome/new?hl=ja)を開き、プロジェクトの選択から自分のプロジェクトを選択します。
※`最近のプロジェクト`ではなく`すべて`から探してください

![スクリーンショット 2024-07-04 23.28.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2984024/96928509-d3eb-eaf9-ba6e-e38cf8c5dd13.png)

__② Cloud Storage > バケット__

![スクリーンショット 2024-07-04 23.32.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2984024/ce726032-1c24-6626-be53-b815d35a4641.png)

__③ バケット名とregionを確認__
![スクリーンショット 2024-07-04 23.39.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2984024/a6de47db-6147-552d-d0de-0d31f9338dda.png)

__④ プログラムに以下を追記__
```py
# 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`にしてデプロイしたところ正常にデプロイできました🎉

![スクリーンショット 2024-07-04 23.49.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2984024/65c5b73a-9210-f037-d2e2-9f7538260ef7.png)
![スクリーンショット 2024-07-04 23.49.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2984024/70649c35-92f0-fe44-3557-daae986e3489.png)

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