streamlitで画像処理データ管理アプリを作ってみたPart1

# はじめに
**streamlit** の初リリース（2018年）から6年が経過し、現在では社内で業務アプリとして活用されるシーンも多く見られるようになりました。

今回は、<INS>データベースと連携した streamlit によるフルスタック開発に興味のある方</INS>を対象に、**画像処理データ管理アプリ**を作成した内容を共有させていただきたいと思います。

# 対象読者 / 想定シーン
- 業務のデータ管理アプリをstreamlitで開発したいエンジニア
- streamlitの基本は把握しており、ORMやDBはこれから触っていこうと思っているエンジニア
- pythonだけでフルスタックに開発したいエンジニア

# 成果物
　**share.streamlit.io**にデプロイしています。
　※デプロイ先の仕様上、停止している場合があります。
 　停止している場合は、**Yes, get this app back up**のボタンを押下してください🙇
　[https://imaima-image-process-manager.streamlit.app/](https://imaima-image-process-manager.streamlit.app/)

- UI構成
![IPM-1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/892674/66667c20-b41c-d06e-ea50-dbd911d9432a.png)

↓リポジトリ

https://github.com/akihiko-ima/streamlit-image-process-manager

:::note warn
注意
本アプリにはエラーハンドリング処理が完全に実装されておりません。
業務でご使用される場合は、必ずエラーハンドリング処理を適切に実装してください。
:::



# 設計 / 採用技術
- **フロントエンド**: Streamlit
- **ORM**: SQLAlchemy
- **データベース**: SQLite
- **デザイン**: Material Design Icons
- **画像処理**: OpenCV

#### アーキテクチャ
クライアントはstreamlitのUIを経由して、**services** dir. or **database** dir.の関数を呼び出す形でタスクを処理する構成としました。

![ST-IPM-architecture.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/892674/14f981df-8880-a7cd-0ef9-d79088d089f1.png)



# ディレクトリ構成
React(Next.js)の開発経験を元にstreamlitにおいても関数(services, utils)を分離する構成を選択しました。気になる点がありましたら、コメントでお知らせください:grinning:

```bash
    streamlit-image-process-manager/
    ├─ .streamlit/                             # streamlit標準config
    │  └─ config.toml
    ├─ data/                                   # 画像保存先
    │  ├─ processed/
    │  └─ raw/
    ├─ database/                               # データベース設定
    │  ├─ cruds/
    │  │  ├─ image_data.py
    │  │  └─ processed_image_data.py
    │  ├─ database.py
    │  └─ models.py
    ├─ schemas/                                # スキーマディレクトリ
    │  └─ schemas.py
    ├─ services/                               # フロントエンドから呼び出す関数を保存
    │  ├─ delete_image_data.py
    │  ├─ dummy_heavy_image_processing.py
    │  └─ initialize_setting.py
    ├─ utils/                                  # 汎用関数
    │  └─ format_datetime_column.py
    ├─ pages/
    │  ├─ 1_🚀_processed_check_page.py
    │  ├─ 2_📑_bonus_comment_page.py
    │  └─ 3_⚠️_README.py
    ├─ .env                                    # 環境設定
    ├─ app.py                                  # メインアプリケーション
    └─ data.db                                 # データベース(SQLite3)
```
# コード解説

## app.py
- streamlitのメインのソースコードが書かれています。
- インタラクティブで柔軟なデータグリッドを簡単に追加するために、**streamlit-aggrid**を採用しました。

https://github.com/PablocFonseca/streamlit-aggrid

:::note alert
**2025/1/5更新** :
AG Gridを商用で使用する場合は、AG Grid Enterpriseのライセンスが必要となります。詳しくは、以下のリンクをお確かめください。
[https://www.ag-grid.com/license-pricing/](https://www.ag-grid.com/license-pricing/)
:::

- AgGridの設定は、**GridOptionsBuilder** を使用することで効率的に実装することができました。以下のコードが該当する箇所となります。

https://streamlit-aggrid.readthedocs.io/en/docs/GridOptionsBuilder.html

- また、ユーザーが選択したデータ情報を**st.session_state**を使用することで、簡単に状態管理できる状態となっています。

https://docs.streamlit.io/develop/api-reference/caching-and-state/st.session_state


```app.py
image_data_list = get_all_image_data(db=db)

# データフレームの作成
if len(image_data_list) > 0:
    df_raw_img = pd.DataFrame(image_data_list)

    df_raw_img["is_processed"] = df_raw_img["is_processed"].apply(
        lambda x: "済" if x else "未"
    )
    df_raw_img = format_datetime_column(df_raw_img, "uploaded_at")
    df_raw_img = format_datetime_column(df_raw_img, "updated_at")

    # AgGridのsetting
    grid_builder = GridOptionsBuilder.from_dataframe(
        df_raw_img, editable=False, filter=True, resizable=True, sortable=False
    )
    grid_builder.configure_selection(selection_mode="multiple", use_checkbox=True)
    grid_builder.configure_side_bar(filters_panel=True, columns_panel=False)
    grid_builder.configure_default_column(
        enablePivot=True, enableValue=True, enableRowGroup=True
    )
    grid_builder.configure_grid_options(rowHeight=50)
    grid_options = grid_builder.build()
    grid_options["columnDefs"][0]["checkboxSelection"] = True

    response = AgGrid(
        df_raw_img,
        gridOptions=grid_options,
        update_mode=GridUpdateMode.MODEL_CHANGED,
        theme="alpine",
    )

    # 選択したデータ情報を取得
    selected_df = pd.DataFrame(response["selected_rows"])
    if not selected_df.empty:
        selected_id_list = selected_df["id"].tolist()
        st.session_state.selected_id_list = selected_id_list
```


## services/ 
- フロントエンドからの呼び出しに応じる関数を格納します

```
    services/ 
    ├─ delete_image_data.py
    ├─ dummy_heavy_image_processing.py   # ダミーとして重たい画像処理を行います
    └─ initialize_setting.py             # .envを読み込み、st.session_stateにパスを設定
```

- UIから呼び出す重たい画像処理 (今回はdummyとしてtime.sleep(5)で代替) としています。
```services/dummy_heavy_image_processing.py
    def dummy_heavy_image_processing(image: np.ndarray) -> np.ndarray:
        """dummyの重たい画像処理をソーベルフィルターとタイマーで実装"""
        time.sleep(5)  #5sec間、停止
    
        image = cv2.medianBlur(image, 5)
        sobel_x = cv2.Sobel(image, cv2.CV_32F, 1, 0, ksize=3)
        sobel_y = cv2.Sobel(image, cv2.CV_32F, 0, 1, ksize=3)
        sobel_x = cv2.convertScaleAbs(sobel_x)
        sobel_y = cv2.convertScaleAbs(sobel_y)
        sobel_combined = cv2.addWeighted(sobel_x, 0.5, sobel_y, 0.5, 0)
        return sobel_combined
```
## schemas/ 
- データベースへの書き込み時に使用するschemaをdataclassで実装しています。
- dataclassの勉強は、以下の記事が大変参考になります。

https://qiita.com/ttyszk/items/01934dc42cbd4f6665d2


## database/
```bash
    database/
    ├─ cruds/
    │  ├─ image_data.py             # 画像処理前のCRUD操作
    │  └─ processed_image_data.py   # 画像処理後のCRUD操作
    ├─ database.py                  # データベースの接続設定
    └─ models.py                    # テーブルの定義
``` 
- 画像処理前のテーブルと、画像処理後のテーブルを以下のように定義しています。
:::note alert
**file_name**, **file_path** などは本来uniqueであると思いますので、本番環境では注意してください。
:::

```database/models.py
class ImageData(Base):
    __tablename__ = "image_data"

    id = Column(Integer, primary_key=True, index=True)
    file_name = Column(String, index=True)
    file_path = Column(String, index=True)
    file_extension = Column(String, index=True)
    file_size = Column(Float)
    description = Column(String)
    is_processed = Column(Boolean, default=False)
    uploaded_at = Column(
        DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
    )
    updated_at = Column(
        DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
    )

    def to_dict(self):
        return {
            "id": self.id,
            "file_name": self.file_name,
            "file_path": self.file_path,
            "file_extension": self.file_extension,
            "file_size": self.file_size,
            "uploaded_at": self.uploaded_at,
            "updated_at": self.updated_at,
            "is_processed": self.is_processed,
            "description": self.description,
        }


class ProcessedImageData(Base):
    __tablename__ = "processed_image_data"

    id = Column(Integer, primary_key=True, index=True)
    file_name = Column(String, index=True)
    file_path = Column(String, index=True)
    processed_at = Column(
        DateTime, default=pd.Timestamp.now(tz="UTC").tz_convert("Asia/Tokyo").floor("s")
    )

    def to_dict(self):
        return {
            "id": self.id,
            "file_name": self.file_name,
            "file_path": self.file_path,
            "processed_at": self.processed_at,
        }
```


## 追加予定(・・・の可能性あり)
- AI画像認識実行機能

## 最後に
- 今回の構成を参考にし、マイクロサービスアーキテクチャの設計思想に則り、バックエンドをAPI(DRF, Flask, FastAPIなど)で実装してみるのも良いかもしれません。


## 参考
ORMの操作が慣れていないかたは、公式のチュートリアルを実践してみることをお勧めいたします。英語ですが大変わかりやすいです。

https://docs.sqlalchemy.org/en/20/orm/quickstart.html

ICONはstreamlitでもGoogloeFontsが使えるので、是非採用してください。

https://fonts.google.com/icons?icon.set=Material+Symbols&icon.style=Rounded