アダコテック技術ブログ

株式会社アダコテックの技術ブログです。

トップ > Python > Python+Fletで単方向データフローを実現する

Python+Fletで単方向データフローを実現する

Python

こんにちは、AIエンジニアの植草です。 前回、↓の記事でPython+Fletを紹介しました。 techblog.adacotech.co.jp

3行で

  • Python+Flet使うとフロントエンド経験なくてもGUIアプリケーションを作れるよ（前回のブログ）
  • 動くまでは簡単だけど、無計画にコーディングしていくと、コードが複雑化していくよ
  • 単方向データフローを用いて、コードが複雑化するのを防ぐ方法を実際のコード例と共に紹介するよ

前回の記事について

内容の振り返り

Flet は簡単に言うと「良い感じのGUIをお手軽にPythonだけで構築できる」ライブラリです。前回の記事では 学習済みのAIモデル（RemBg）を用いた簡単なアプリケーションをハンズオン形式でプロトタイプを動かすところまで解説しました。

全体のコードはこのようになります。

import base64
from io import BytesIO

import flet as ft
from PIL import Image
from rembg import new_session, remove


def on_image_click(
    e: ft.TapEvent,
    assigned_width: float,
    assigned_height: float,
    image_path: str,
    dist_view: ft.Container,
) -> None:
    """画像クリック時の処理"""
    image = Image.open(image_path)

    # UI上に割り当てられたアスペクト比
    ui_aspect = assigned_width / assigned_height

    # 画像のアスペクト比
    image_aspect = image.width / image.height

    if image_aspect > ui_aspect:
        # 画像が横長
        # widthは割り当てられた領域いっぱい利用し
        # heightはwidthと画像のアスペクト比から計算する
        ui_width = assigned_width
        ui_height = ui_width / image_aspect
    else:
        # 画像が縦長
        # heightは割り当てられた領域いっぱい利用し
        # widthはheightと画像のアスペクト比から計算する
        ui_height = assigned_height
        ui_width = ui_height * image_aspect

    # UI上の座標
    ui_point_x = e.local_x
    ui_point_y = e.local_y

    # 画像上の座標
    image_point_x = image.width * ui_point_x / ui_width
    image_point_y = image.height * ui_point_y / ui_height

    # 背景除去
    print("runnning...")
    res_image = remove_bg(image, int(image_point_x), int(image_point_y))

    # PIL.Imageをbase64エンコード
    buff = BytesIO()
    res_image.save(buff, "png")
    res_base64 = base64.b64encode(buff.getvalue()).decode("utf-8")

    # UIの更新
    res_content = ft.Image(src_base64=res_base64, fit=ft.ImageFit.FIT_HEIGHT)
    dist_view.content = res_content
    dist_view.update()


def remove_bg(image: Image.Image, point_x: int, point_y: int) -> Image.Image:
    """背景除去処理"""
    sam_prompt = [
        {
            "type": "point",
            "data": [point_x, point_y],
            "label": 1,
        }
    ]
    kwargs = {
        "post_process_mask": True,
        "sam_prompt": sam_prompt,
        "bgcolor": (255, 255, 255, 0),
        "session": new_session("sam"),
    }
    return remove(image, **kwargs)


def main(page: ft.Page) -> None:
    """メイン関数"""

    # 画像アップロードボタン
    def upload_image(e: ft.FilePickerResultEvent) -> None:
        """ファイルアップロード時のイベント"""
        if not e.files:
            # キャンセルした時の挙動
            return
        image_path = e.files[0].path
        show_origin_image(image_path)

    file_picker = ft.FilePicker(on_result=upload_image)
    upload_button = ft.ElevatedButton(
        "画像アップロードボタン",
        on_click=lambda _: file_picker.pick_files(
            allow_multiple=False, allowed_extensions=["png", "jpg", "jpeg", "bmp"]
        ),
    )

    # オリジナル画像を配置する領域
    VIEW_IMAGE_WIDTH = 500
    VIEW_IMAGE_HEIGHT = 500
    origin_image_empty_text = ft.Text("画像をアップロードしてください")
    origin_image = ft.Container(
        content=origin_image_empty_text,
        width=VIEW_IMAGE_WIDTH,
        height=VIEW_IMAGE_HEIGHT,
        alignment=ft.alignment.center,
        bgcolor="white",
    )

    # 背景除去画像を配置する領域
    bg_removed_image = ft.Container(
        content=ft.Text("背景除去後の画像", color="black"),
        width=500,
        height=500,
        bgcolor="white",
        alignment=ft.alignment.center,
    )

    def show_origin_image(file_path: str) -> None:
        """画像を表示する"""
        image_content = ft.Image(src=file_path, fit=ft.ImageFit.FIT_HEIGHT)
        origin_image.content = ft.GestureDetector(
            mouse_cursor=ft.MouseCursor.BASIC,
            on_tap_down=lambda e: on_image_click(
                e,
                VIEW_IMAGE_WIDTH,
                VIEW_IMAGE_HEIGHT,
                file_path,
                bg_removed_image,
            ),
            content=image_content,
        )
        origin_image.update()

    main_view = ft.Column(
        controls=[
            upload_button,
            ft.Row(
                controls=[
                    origin_image,
                    bg_removed_image,
                ]
            ),
        ]
    )
    page.add(main_view)
    page.add(file_picker)


if __name__ == "__main__":
    ft.app(target=main)

この程度の長さのコードでAIを使ったGUIアプリケーションを動かせるなんて嬉しいですね！というのが前回のブログの趣旨でした。

さて、一方でこのプロトタイプを通して以下のフィードバックを貰ったらどのように改修をかけるでしょうか？

  • 処理時間が長いのでローディング的なUIが欲しい
  • 画像を連打したらPCがめちゃくちゃ重くなった

具体的な処理の話をすると、これらの原因は「処理時間中」のハンドリングをしていないことです。 on_image_click() 内で連打されても処理が追加で走らないよう、イベントを抑制するハンドリングを追加したり、ローディング的なUIを出す処理を追加したりする必要がありそうです。

問題点

on_image_click() の処理内容を箇条書きにすると以下の処理となっています。

  1. 画像を読み込む
  2. クリックしたUI上の座標を画像上の座標に変換
  3. 画像と座標を背景除去の処理に投げる
  4. 表示を更新する

表示のロジックとコアな計算のロジックが手続き的に入り乱れていて、見通しが悪いです。特に「4. 表示を更新する」ですが、もともと元画像の表示領域に紐づいた関数が、背景除去後の表示領域を直接的に制御していて、直観的ではありません。

たとえば、後輩が「背景除去後の表示はどうやって決まる？」と疑問に思ってコードを追ってみるというシチュエーションを仮定してみましょう。

  • bg_removed_image として定義されていることを確認する
  • bg_removed_imageがon_image_click()の引数として与えられていることを確認する
  • on_image_click() が show_origin_image() で呼ばれていることを確認する
  • show_origin_image() が upload_image() で呼ばれてることを確認する

upload_image()（画像の読み込み処理）まで戻ってきました。つまり、コードをほぼ全て読む必要がある状態です。後輩の方が気の毒です。この調子でコードの規模が大きくなってきた場合、実装した本人であっても把握が厳しくなります。不具合も増えてくるでしょう。仮説検証よりもバグ修正に追われてしまうのは避けたいところです。

リファクタリングする

データフローを整理する

フロントエンジニアの諸先輩方は様々な設計パターンを駆使してこのような問題を解決しています。ここでは単方向データフローという設計パターンを採用することとし、以下の概念に整理された状態で実装すること目指します。

  • view : 表示ロジック。 state を表示するもの
  • action : viewで取得されたイベントを解釈し、ロジックを実行、 state を更新する
  • state : view が表示する内容。変更された view に通知する必要がある
    単方向データフローの概念図。フローの逆転は許されません。

ここから先は以下の先駆者様の記事をかなり参考にしつつ、私なりのアレンジを加えています。とても分かりやすい記事なのでオススメです！ qiita.com

単方向データフローの基盤の実装

Stateの実装

state は変更を検知して view に通知する必要があります。こういった仕組みはFletには現状備わっていないため、自力で実装していきます。

"""states.py"""

from abc import ABC
from typing import Any, Generic, List, Optional, TypeVar, final

T = TypeVar("T")

class State(Generic[T], ABC):
    """状態を管理するオブジェクトのまとめ"""

    def __init__(self) -> None:
        """初期化"""
        self.value: Optional[T] = None
        self.observers: List[Any] = []

    def bind(self, view: Any) -> None:
        """変更通知を行う相手を設定する

        Parameters
        ----------
        view : Any
            変更通知を行う相手。updateメソッドが必要
        """
        if view not in self.observers:
            self.observers.append(view)

    def set_value(self, value: T) -> None:
        """値の設定

        Parameters
        ----------
        value : Optional[T]
            設定する値
        """
        if self.value != value:
            self.value = value
            self.notify()

    def notify(self) -> None:
        """変更通知"""
        for o in self.observers:
            o.update(self)

    @final
    def clear(self) -> None:
        """ステータスを初期化する"""
        if self.value is not None:
            self.value = None
            self.notify()

変更を通知する相手（= view )を bind() しておき、 set_value() により変更が生じた際に view の update() メソッドを呼ぶ仕組みになっています。

ちなみにviewの型ヒントにAnyを使っているのはサイクリックインポートを避けるためです。

Viewの実装

"""view_components.py"""

from abc import ABC, abstractmethod
from typing import List, Optional, final

import flet as ft

from states.base import State

class BaseView(ft.Container, ABC):
    """Stateに依存するUIクラス"""

    def __init__(self, states: Optional[List[State]] = None):
        """初期化

        Parameters
        ----------
        states : List[State]
            依存するState
        """
        if states is not None:
            for state in states:
                state.bind(self)
        super().__init__()

    @final
    def update(self, state: Optional[State] = None) -> None:
        """Stateの変更通知をUIに反映する

        Parameters
        ----------
        state : Optional[State], optional
            変更通知されるState。Noneの場合はfletからの要求
        """
        if state is not None:
            self.apply_state()
        super().update()

    @abstractmethod
    def apply_state(self) -> None:
        """描画処理"""
        pass

ft.Container を継承してControlとしてそのまま使えるように実装しています。コンストラクタの引数となっている states に変更通知をもらう相手をリストで与えておくと、 state に変更が入ったときに自動的に描画を行いUIを更新する仕組みとなっています。

具象クラス実装

Stateの具象クラス

まず State クラスを用いて具象クラスを実装していきます。具体的な State (状態)は以下とします。

  1. ユーザーが選択したオリジナルの画像
  2. 背景除去後の画像
  3. 背景除去処理の処理の状況（追加します）

画像の表示部分に使うStateは共通で良さそうです。 State を継承して実装してみます。

"""states.py"""

import base64
from abc import ABC
from io import BytesIO
from typing import Any, Generic, List, Optional, TypeVar, final

from PIL import Image

T = TypeVar("T")

class State(Generic[T], metaclass=ABC):
    """状態を管理するオブジェクトのまとめ"""

（中略）

class ImageState(State[Image.Image]):
    """画像を管理するState"""

    def get_base64(self) -> str:
        """base64エンコードして取得"""
        if self.value is None:
            raise RuntimeError("画像が選択されていません")
        buff = BytesIO()
        self.value.save(buff, "png")
        res = base64.b64encode(buff.getvalue()).decode("utf-8")
        return res

class ProcessState(State[bool]):
    """実行中"""

    pass

Viewの具象クラス

ViewはUIの部品(Control)に相当すると考えて

  1. オリジナル画像の表示 (背景除去実行中はLoading的な表示をします）
  2. 背景除去後の画像の表示
  3. 画像アップロードボタン (背景除去実行中はdisableにします）

を作成していきます。 各Controlは変更通知を受け取る（監視する）Stateをスーパークラスに投げ込むことで、変更通知を受け取ることができるようになります。具体的な描画処理を apply_state() メソッドに記述します。 なお actions の実装は後で行います。

"""view_components.py"""

from abc import ABC, abstractmethod
from typing import List, Optional, final

import flet as ft

from actions import load_image, remove_bg
from states import ImageState, ProcessState, State

class BaseView(ft.Container, ABC):
    """Stateに依存するUIクラス"""

（中略）

VIEW_IMAGE_WIDTH = 500
VIEW_IMAGE_HEIGHT = 500

class OriginalImageView(BaseView):
    """元画像のView"""

    def __init__(
        self, origin_image_state: ImageState, removed_image_state: ImageState, process_state: ProcessState
    ):
        """初期化"""
        super().__init__(states=[origin_image_state, process_state])
        self.image_state = origin_image_state
        self.remove_bg_state = removed_image_state
        self.process_state = process_state
        self.empty_text = ft.Text("画像をアップロードしてください")

        # ft.Containerの固定の設定
        self.width = VIEW_IMAGE_WIDTH
        self.height = VIEW_IMAGE_HEIGHT
        self.alignment = ft.alignment.center
        self.bgcolor = "white"
        self.apply_state()

    def apply_state(self) -> None:
        """描画処理"""
        if self.image_state.value is None:
            content = self.empty_text
        else:
            image_content = ft.Image(src_base64=self.image_state.get_base64(), fit=ft.ImageFit.FIT_HEIGHT)
            content = ft.GestureDetector(
                mouse_cursor=ft.MouseCursor.BASIC,
                on_tap_down=self.on_image_click,
                content=image_content,
            )
        if self.process_state.value:
            self.content = ft.Stack(
                controls=[content, ft.ProgressRing(color="black")],
                alignment=ft.alignment.center,
            )
        else:
            self.content = content

    def on_image_click(self, e: ft.TapEvent) -> None:
        """クリックイベント"""
        if self.image_state.value is None:
            return
        remove_bg(
            e.local_x,
            e.local_y,
            VIEW_IMAGE_HEIGHT,
            VIEW_IMAGE_WIDTH,
            self.image_state.value,
            self.remove_bg_state,
            self.process_state,
        )

class RemovedBgImageView(BaseView):
    """元画像のView"""

    def __init__(self, removed_image_state: ImageState):
        """初期化"""
        super().__init__(states=[removed_image_state])
        self.image_state = removed_image_state
        self.empty_text = ft.Text("背景除去した画像が表示されます")

        # ft.Containerの固定の設定
        self.width = VIEW_IMAGE_WIDTH
        self.height = VIEW_IMAGE_HEIGHT
        self.alignment = ft.alignment.center
        self.bgcolor = "white"
        self.apply_state()

    def apply_state(self) -> None:
        """描画処理"""
        if self.image_state.value is None:
            content = self.empty_text
        else:
            content = ft.Image(src_base64=self.image_state.get_base64(), fit=ft.ImageFit.FIT_HEIGHT)
        self.content = content

class UploadButton(BaseView):
    def __init__(self, origin_image_state: ImageState, process_state: ProcessState, page: ft.Page):
        """初期化"""
        super().__init__([process_state])
        self.image_state = origin_image_state
        file_picker = ft.FilePicker(on_result=self.upload_image)
        self.content = ft.ElevatedButton(
            "画像アップロードボタン",
            on_click=lambda _: file_picker.pick_files(
                allow_multiple=False, allowed_extensions=["png", "jpg", "jpeg", "bmp"]
            ),
        )
        self.process_state = process_state
        page.add(file_picker)

    def upload_image(self, e: ft.FilePickerResultEvent) -> None:
        """ファイルアップロード時のイベント"""
        if not e.files:
            return
        image_path = e.files[0].path
        load_image(image_path, self.image_state)

    def apply_state(self) -> None:
        """処理中はdisableにする"""
        if self.process_state.value:
            self.content.disabled = True
        else:
            self.content.disabled = False

前回のブログではあちこちに散らばっていた表示に関するロジックが、1か所に固まりましたね！

Actionの実装

イベント発生時に実行するactionsは以下のように実装できます。

"""actions.py"""

from PIL import Image
from rembg import new_session, remove

from states import ImageState, ProcessState

def load_image(file_path: str, origin_image_state: ImageState) -> None:
    """画像の読み込み"""
    image = Image.open(file_path)
    origin_image_state.set_value(image)

def remove_bg(
    ui_x: float,
    ui_y: float,
    assigned_width: float,
    assigned_height: float,
    image: Image.Image,
    remove_bg_state: ImageState,
    process_state: ProcessState,
) -> None:
    """背景除去処理

    Parameters
    ----------
    x_ratio : float
        選択した座標（x / 相対値）
    y_ratio : float
        選択した座標（y / 相対値）
    image : Image.Image
        対象の画像
    """
    process_state.set_value(True)
    point_x, point_y = _get_image_point(ui_x, ui_y, assigned_width, assigned_height, image)
    res_image = _remove_bg(image, point_x, point_y)
    remove_bg_state.set_value(res_image)
    process_state.set_value(False)

def _get_image_point(
    ui_x: float,
    ui_y: float,
    assigned_width: float,
    assigned_height: float,
    image: Image.Image,
) -> tuple[int, int]:
    """相対座標から絶対座標を取得"""
    # UI上に割り当てられたアスペクト比
    ui_aspect = assigned_width / assigned_height

    # 画像のアスペクト比
    image_aspect = image.width / image.height

    if image_aspect > ui_aspect:
        # 画像が横長
        # widthは割り当てられた領域いっぱい利用し
        # heightはwidthと画像のアスペクト比から計算する
        ui_width = assigned_width
        ui_height = ui_width / image_aspect
    else:
        # 画像が縦長
        # heightは割り当てられた領域いっぱい利用し
        # widthはheightと画像のアスペクト比から計算する
        ui_height = assigned_height
        ui_width = ui_height * image_aspect

    # 画像上の座標
    image_point_x = image.width * ui_x / ui_width
    image_point_y = image.height * ui_y / ui_height
    return (int(image_point_x), int(image_point_y))

def _remove_bg(image: Image.Image, point_x: int, point_y: int) -> Image.Image:
    """背景除去処理"""
    sam_prompt = [
        {
            "type": "point",
            "data": [point_x, point_y],
            "label": 1,
        }
    ]
    kwargs = {
        "post_process_mask": True,
        "sam_prompt": sam_prompt,
        "bgcolor": (255, 255, 255, 0),
        "session": new_session("sam"),
    }
    return remove(image, **kwargs)

こちらの実装とあまり変わりませんが、UIを制御するロジックが排除されていて、代わりに State を更新するのみとなっています。

ちなみに前回のブログの実装方針で進めた場合、UIを制御するロジックをこちらに入れることになります。この時、処理実行中は 　「アップロードボタンをdisableにする」「元画像にプログレスリングを表示する」という2つのUIを制御することになります。ユーザーが操作する箇所が増えるほど、この手の処理が増えていきます。すごく大変です。それに比べると、とても「変更しやすい」と言えます。

最後にmain関数を変更しましょう！各クラスを定義に従ってインスタンス化し、 ft.Column と ft.Row を使って並べれば完成です。

import flet as ft

from states import ImageState, ProcessState
from view_components import OriginalImageView, RemovedBgImageView, UploadButton

def main(page: ft.Page) -> None:
    """メイン関数"""

        # Stateのインスタンス化
    origin_image_state = ImageState()
    bg_removed_image_state = ImageState()
    process_state = ProcessState()
    
    # Viewのインスタンス化
    upload_button = UploadButton(origin_image_state, process_state, page)
    origin_image = OriginalImageView(origin_image_state, bg_removed_image_state, process_state)
    removed_bg_image = RemovedBgImageView(bg_removed_image_state)
        
        # UIを構築
    main_view = ft.Column(
        controls=[
            upload_button,
            ft.Row(
                controls=[
                    origin_image,
                    removed_bg_image,
                ]
            ),
        ]
    )
    page.add(main_view)

if __name__ == "__main__":
    ft.app(target=main)

動作しましたか？これにてリファクタリングは完了です。 コード量は増えていますが、変更した時における影響範囲の見通しが立ちやすくなっています。 また、新しい機能を追加する時に「State」「View」「Action」と分けて考えることができるので手が付けやすいなと感じます。

まとめ

Python+Fletでロジックが無秩序になってしまいがちなところを、単方向データフローを用いて秩序だったコーディングを行うTIPSを紹介しました。 これにより、変更に強いコードを実装することができます。

実際に私自身が実装しているPython+Fletのプロトタイプも、この記事に記載している実装を活用することで、フィードバックの反映をスピーディに行うことができています。

メンバー募集中です！！

アダコテックでは画像処理や機械学習を用いたプロダクトを作っています。我々の事業やプロダクトに興味がある方は、ぜひカジュアル面談にいらしてください！ herp.careers