|
from __future__ import annotations |
|
from abc import ABC, abstractmethod |
|
from typing import Any, Dict, List, Callable, Literal, Tuple |
|
|
|
import gradio as gr |
|
|
|
class GrComponent(ABC): |
|
""" |
|
コンポーネントのクラス |
|
|
|
全てのコンポーネントを作成してから.render()できるようにするためのクラス |
|
各コンポーネントに対して実行される処理のメソッドを子クラスに追加していく |
|
|
|
Attributes |
|
---------- |
|
comp: gr.Component |
|
コンポーネント |
|
""" |
|
def __init__(self, *param: Any): |
|
""" |
|
コンストラクタ |
|
|
|
createメソッドでコンポーネントを作成する |
|
""" |
|
self.comp: gr.Component = self._create(*param) |
|
|
|
@abstractmethod |
|
def _create(self, *param: Any) -> gr.Component: |
|
""" |
|
コンポーネントの作成 |
|
|
|
Returns |
|
------- |
|
gr.Component |
|
作成されるコンポーネント |
|
""" |
|
pass |
|
|
|
|
|
class GrLayout(ABC): |
|
""" |
|
レイアウトのクラス |
|
|
|
gr.Rowや、gr.Columnを事前に定義しておくためのクラス |
|
|
|
Attributes |
|
---------- |
|
layout_type: gr.BlockContext, default None |
|
gr.Rowやgr.Columnなどの、子要素の配置スタイル |
|
children: (Dict[str, GrLayout | GrComponent] | List[GrLayout | GrComponent]) |
|
子要素 |
|
""" |
|
layout_type: gr.BlockContext = None |
|
|
|
def __init__(self, *param: Any): |
|
""" |
|
コンストラクタ |
|
|
|
子クラス固有のレイアウトタイプにオーバーライドされているか確認し、 |
|
子要素をもつ辞書かリストを作成する |
|
|
|
Raises |
|
------ |
|
NotImplementedError |
|
レイアウトタイプがNoneの場合 |
|
""" |
|
if self.layout_type is None: |
|
raise NotImplementedError('layout_typeを設定してください') |
|
|
|
self.children: ( |
|
Dict[str, GrLayout | GrComponent] | List[GrLayout | GrComponent] |
|
) = self._create(*param) |
|
|
|
@abstractmethod |
|
def _create( |
|
self, *param: Any |
|
) -> Dict[str, GrLayout | GrComponent] | List[GrLayout | GrComponent]: |
|
""" |
|
子要素をまとめた辞書かリストの作成 |
|
|
|
Returns |
|
------- |
|
Dict[str, GrLayout | GrComponent] | List[GrLayout | GrComponent] |
|
子要素が入った辞書かリスト |
|
""" |
|
pass |
|
|
|
|
|
class GrListener: |
|
""" |
|
イベントリスナーのクラス |
|
|
|
GrBlocksで、まとめてイベントリスナーを設定できるように、 |
|
必要な項目を定義しておくクラス |
|
本来、イベントリスナーはwith gr.Blocks():の中で定義するものだが、 |
|
withの外で作成し、後でまとめてブロックに適用することができる |
|
""" |
|
def __init__( |
|
self, |
|
trigger: Callable[[], Any] | List[Callable[[], Any]] = None, |
|
fn: Callable | None | Literal["decorator"] = None, |
|
inputs: GrComponent | list[GrComponent | GrLayout] | None = None, |
|
outputs: GrComponent | list[GrComponent | GrLayout] | None = None, |
|
api_name: str | None | Literal[False] = None, |
|
scroll_to_output: bool = False, |
|
show_progress: Literal["full", "minimal", "hidden"] = 'full', |
|
queue: bool | None = None, |
|
batch: bool = False, |
|
max_batch_size: int = 4, |
|
preprocess: bool = True, |
|
postprocess: bool = True, |
|
cancels: dict[str, Any] | list[dict[str, Any]] | None = None, |
|
every: float | None = None, |
|
trigger_mode: Literal["once", "multiple", "always_last"] | None = None, |
|
js: str | None = None, |
|
concurrency_limit: int | None | Literal["default"] = "default", |
|
concurrency_id: str | None = None, |
|
show_api: bool = True, |
|
thens: GrListener | List[GrListener] | None = None |
|
): |
|
""" |
|
コンストラクタ |
|
|
|
引数についてはGradioのソースコードを参照してください |
|
thenは、あるイベントリスナー実行直後に実行させる処理を指定するものです |
|
詳細は、Gradioの公式サイトを参照してください（https://www.gradio.app/guides/blocks-and-event-listeners#running-events-consecutively） |
|
triggerは、gr.onによる複数のトリガー指定に対応できるよう |
|
List[Callable[[], Any]]に対応している |
|
inputsとoutputsは、GrComponentのオブジェクトをそのまま指定するだけでよく、 |
|
.compにする必要はない |
|
|
|
Raises |
|
------ |
|
NotImplementedError |
|
fnが設定されていない場合 |
|
""" |
|
if fn is None: |
|
raise NotImplementedError('fnを指定してください') |
|
|
|
self._trigger = trigger |
|
self.fn = fn |
|
self.inputs = InputsOutputsMaker.setup_inputs_or_outputs(inputs) |
|
self.outputs = InputsOutputsMaker.setup_inputs_or_outputs(outputs) |
|
self._api_name = api_name |
|
self._scroll_to_output = scroll_to_output |
|
self._show_progress = show_progress |
|
self._queue = queue |
|
self._batch = batch |
|
self._max_batch_size = max_batch_size |
|
self._preprocess = preprocess |
|
self._postprocess = postprocess |
|
self._cancels = cancels |
|
self._every = every |
|
self._trigger_mode = trigger_mode |
|
self._js = js |
|
self._concurrency_limit = concurrency_limit |
|
self._concurrency_id = concurrency_id |
|
self._show_api = show_api |
|
self.thens = thens |
|
|
|
def setup(self) -> None: |
|
""" |
|
イベントリスナーの適用 |
|
|
|
GrBlocksの_set_event_listenerメソッドで実行されることで、 |
|
イベントリスナーが適用される |
|
""" |
|
if self.thens is None: |
|
self._setup_without_then() |
|
|
|
else: |
|
self._setup_with_then() |
|
|
|
def _setup_without_then(self) -> gr.events.Dependency: |
|
""" |
|
thenがないイベントリスナーの適用 |
|
|
|
Returns |
|
------- |
|
gr.events.Dependency |
|
イベントリスナー |
|
""" |
|
if isinstance(self._trigger, list): |
|
dep = gr.on( |
|
triggers=self._trigger, |
|
fn=self.fn, |
|
inputs=self.inputs, |
|
outputs=self.outputs, |
|
api_name=self._api_name, |
|
scroll_to_output=self._scroll_to_output, |
|
show_progress=self._show_progress, |
|
queue=self._queue, |
|
batch=self._batch, |
|
max_batch_size=self._max_batch_size, |
|
preprocess=self._preprocess, |
|
postprocess=self._postprocess, |
|
cancels=self._cancels, |
|
every=self._every, |
|
trigger_mode=self._trigger_mode, |
|
js=self._js, |
|
concurrency_limit=self._concurrency_limit, |
|
concurrency_id=self._concurrency_id, |
|
show_api=self._show_api |
|
) |
|
|
|
else: |
|
dep = self._trigger( |
|
fn=self.fn, |
|
inputs=self.inputs, |
|
outputs=self.outputs, |
|
api_name=self._api_name, |
|
scroll_to_output=self._scroll_to_output, |
|
show_progress=self._show_progress, |
|
queue=self._queue, |
|
batch=self._batch, |
|
max_batch_size=self._max_batch_size, |
|
preprocess=self._preprocess, |
|
postprocess=self._postprocess, |
|
cancels=self._cancels, |
|
every=self._every, |
|
trigger_mode=self._trigger_mode, |
|
js=self._js, |
|
concurrency_limit=self._concurrency_limit, |
|
concurrency_id=self._concurrency_id, |
|
show_api=self._show_api |
|
) |
|
|
|
return dep |
|
|
|
def _setup_with_then(self) -> None: |
|
""" |
|
thenがあるイベントリスナーの適用 |
|
""" |
|
dep = self._setup_without_then() |
|
|
|
all_thens = GrListener._d_f_s_thens(self.thens) |
|
|
|
for listener in all_thens: |
|
dep = listener.setup_then(dep) |
|
|
|
@staticmethod |
|
def _d_f_s_thens( |
|
thens: GrListener | List[GrListener], all_thens: List[GrListener] = [] |
|
) -> List[GrListener]: |
|
""" |
|
全てのthenの取得 |
|
|
|
thenの各要素が呼び出す全てのイベントを深さ優先探索で取得する |
|
入れ子の様になっている全てのイベントを最後まで呼び出せるようにする |
|
|
|
Parameters |
|
---------- |
|
thens : GrListener | List[GrListener] |
|
thensに代入されていたGrListenerか、そのリスト |
|
all_thens : List[GrListener], optional |
|
深さ優先探索で見つかったGrListenerが格納されるリスト, by default [] |
|
|
|
Returns |
|
------- |
|
List[GrListener] |
|
深さ優先探索で見つかった全てのGrListenerが格納されているリスト |
|
""" |
|
if isinstance(thens, list): |
|
for listener in thens: |
|
all_thens.append(listener) |
|
all_thens = GrListener._d_f_s_thens(listener.thens, all_thens) |
|
|
|
elif isinstance(thens, GrListener): |
|
all_thens.append(thens) |
|
|
|
return all_thens |
|
|
|
def setup_then(self, dep: gr.events.Dependency) -> gr.events.Dependency: |
|
""" |
|
thenの追加 |
|
|
|
イベントリスナーにthenを追加する |
|
|
|
Parameters |
|
---------- |
|
dep : gr.events.Dependency |
|
thenを追加されるイベントリスナー |
|
|
|
Returns |
|
------- |
|
gr.events.Dependency |
|
thenが追加されたイベントリスナー |
|
""" |
|
dep.then( |
|
fn=self.fn, |
|
inputs=self.inputs, |
|
outputs=self.outputs, |
|
api_name=self._api_name, |
|
scroll_to_output=self._scroll_to_output, |
|
show_progress=self._show_progress, |
|
queue=self._queue, |
|
batch=self._batch, |
|
max_batch_size=self._max_batch_size, |
|
preprocess=self._preprocess, |
|
postprocess=self._postprocess, |
|
cancels=self._cancels, |
|
every=self._every, |
|
trigger_mode=self._trigger_mode, |
|
js=self._js, |
|
concurrency_limit=self._concurrency_limit, |
|
concurrency_id=self._concurrency_id, |
|
show_api=self._show_api |
|
) |
|
|
|
return dep |
|
|
|
|
|
class GrExamples: |
|
""" |
|
gr.Componentへの入力例のクラス |
|
|
|
詳細は公式ページを参照してください（https://www.gradio.app/docs/examples/） |
|
""" |
|
def __init__( |
|
self, |
|
examples: list[Any] | list[list[Any]] | str, |
|
inputs: GrComponent | list[GrComponent | GrLayout] | None = None, |
|
outputs: GrComponent | list[GrComponent | GrLayout] | None = None, |
|
fn: Callable | None = None, |
|
cache_examples: bool | Literal["lazy"] | None = None, |
|
examples_per_page: int = 10, |
|
label: str | None = "Examples", |
|
elem_id: str | None = None, |
|
run_on_click: bool = False, |
|
preprocess: bool = True, |
|
postprocess: bool = True, |
|
api_name: str | Literal[False] = "load_example", |
|
batch: bool = False, |
|
listener: GrListener | None = None |
|
): |
|
""" |
|
コンストラクタ |
|
|
|
Parameters（gr.Examplesにないもの） |
|
---------- |
|
listener: GrListener | None |
|
インスタンス化したGrListener |
|
これが渡されると、gr.Examplesのinputs,outputs,fnが、 |
|
渡されたリスナーと同じものになる |
|
""" |
|
self._examples = examples |
|
|
|
if listener is None: |
|
self._inputs = InputsOutputsMaker.setup_inputs_or_outputs(inputs) |
|
self._outputs = InputsOutputsMaker.setup_inputs_or_outputs(outputs) |
|
self._fn = fn |
|
else: |
|
self._inputs = listener.inputs |
|
self._outputs = listener.outputs |
|
self._fn = listener.fn |
|
|
|
self._cache_examples = cache_examples |
|
self._examples_per_page = examples_per_page |
|
self._label = label |
|
self._elem_id = elem_id |
|
self._run_on_click = run_on_click |
|
self._preprocess = preprocess |
|
self._postprocess = postprocess |
|
self._api_name = api_name |
|
self._batch = batch |
|
|
|
def set_examples(self) -> None: |
|
""" |
|
Examplesの適用 |
|
""" |
|
gr.Examples( |
|
examples=self._examples, |
|
inputs=self._inputs, |
|
outputs=self._outputs, |
|
fn=self._fn, |
|
cache_examples=self._cache_examples, |
|
examples_per_page=self._examples_per_page, |
|
label=self._label, |
|
elem_id=self._elem_id, |
|
run_on_click=self._run_on_click, |
|
preprocess=self._preprocess, |
|
postprocess=self._postprocess, |
|
api_name=self._api_name, |
|
batch=self._batch |
|
) |
|
|
|
|
|
class InputsOutputsMaker: |
|
""" |
|
inputsとoutputs調整クラス |
|
|
|
GrListenerと、GrExamplesに渡されたinputsとoutputsを |
|
Gradioのイベントリスナーとgr.Examplesに渡せる形にするメソッドを持つクラス |
|
""" |
|
@staticmethod |
|
def setup_inputs_or_outputs( |
|
in_or_outputs: GrComponent | GrLayout | list[GrComponent | GrLayout] | None |
|
) -> gr.Component | List[gr.Component] | None: |
|
""" |
|
inputsとoutputsをgr.Componentへ変換 |
|
|
|
GrComponentから、gr.Componentである.compだけを抜き出す |
|
|
|
Parameters |
|
---------- |
|
in_or_outputs : GrComponent | list[GrComponent | GrLayout] | None |
|
イベントリスナーか、gr.Examplesのinputsかoutputs対象である |
|
gr.Componentを含むGrComponentや、GrComponentやGrLayoutのリスト |
|
|
|
Returns |
|
------- |
|
gr.Component | List[gr.Component] | None |
|
イベントリスナーかgr.Examplesの参照対象であるgr.Component |
|
""" |
|
if in_or_outputs is None: |
|
return in_or_outputs |
|
|
|
else: |
|
if isinstance(in_or_outputs, GrComponent): |
|
return in_or_outputs.comp |
|
|
|
else: |
|
return InputsOutputsMaker._create_flat_comps(in_or_outputs) |
|
|
|
@staticmethod |
|
def _create_flat_comps( |
|
in_or_outputs: GrLayout | List[GrComponent | GrLayout] |
|
) -> List[gr.Component]: |
|
""" |
|
gr.Componentのリストの作成 |
|
|
|
必要であればfind_compsを使って、gr.Componentのリストを作成する |
|
|
|
Parameters |
|
---------- |
|
in_or_outputs : List[GrComponent | GrLayout] |
|
イベントリスナーか、gr.Examplesのinputsかoutputs対象である |
|
gr.Componentを含むGrComponentやGrLayoutのリスト |
|
|
|
Returns |
|
------- |
|
List[gr.Component] |
|
イベントリスナーかgr.Examplesの参照対象であるgr.Componentのリスト |
|
""" |
|
if isinstance(in_or_outputs, GrLayout): |
|
in_or_outputs = in_or_outputs.children |
|
|
|
if isinstance(in_or_outputs, dict): |
|
in_or_outputs = list(in_or_outputs.values()) |
|
|
|
comps = [] |
|
for obj in in_or_outputs: |
|
if isinstance(obj, GrComponent): |
|
comps.append(obj.comp) |
|
|
|
else: |
|
inner_flat_comps = find_comps(obj) |
|
comps.extend([comp.comp for comp in inner_flat_comps]) |
|
|
|
return comps |
|
|
|
|
|
class GrBlocks(ABC): |
|
""" |
|
ブロックのクラス |
|
""" |
|
@classmethod |
|
def create_and_launch(cls, *param: Any, debug: bool = False) -> gr.Blocks: |
|
""" |
|
ブロックの作成と実行 |
|
|
|
Parameters |
|
---------- |
|
debug : bool, optional, default False |
|
デバッグをするならTrue、しないならFalse |
|
デフォルトはFalse |
|
|
|
Returns |
|
------- |
|
gr.Blocks |
|
作成されたブロック |
|
""" |
|
children, listeners = cls._create_children_and_listeners(*param) |
|
|
|
with gr.Blocks() as blocks: |
|
GrBlocks._render_children(children) |
|
|
|
GrBlocks._set_event_listener(listeners) |
|
|
|
blocks.launch(debug=debug) |
|
|
|
return blocks |
|
|
|
@classmethod |
|
@abstractmethod |
|
def _create_children_and_listeners( |
|
cls, *param: Any |
|
) -> Tuple[Dict[str, Any] | List[Any], List[Any]]: |
|
""" |
|
childrenとlistenersの作成 |
|
|
|
Returns |
|
------- |
|
children : Dict[str, Any] | List[Any] |
|
ブロックの子要素をまとめた辞書かリスト |
|
内部の入れ子構造に決まりがないため、Anyとしているが、 |
|
GrLayoutとGrComponentが多重のリストや辞書になったりならなかったりする |
|
listeners : List[Any] |
|
イベントリスナーがまとまったリスト |
|
内部の入れ子構造に決まりがないため、Anyとしているが、 |
|
GrListenerが多重のリストになったりならなかったりする |
|
""" |
|
pass |
|
|
|
@staticmethod |
|
def _render_children(children: Dict[str, Any] | List[Any]) -> None: |
|
""" |
|
子要素のrender |
|
|
|
全ての子要素をまとめてrender()する |
|
コンポーネントをwith gr.Blocks():内で.render()することで、ブロックに実装される |
|
レイアウトはwith child.layout_type():をしてから子要素を.render()する |
|
|
|
Parameters |
|
---------- |
|
children : Dict[str, Any] | List[Any] |
|
子要素の辞書かリスト |
|
""" |
|
if isinstance(children, dict): |
|
children = list(children.values()) |
|
|
|
for child in children: |
|
if isinstance(child, GrLayout): |
|
with child.layout_type(): |
|
GrBlocks._render_children(child.children) |
|
|
|
elif isinstance(child, list): |
|
GrBlocks._render_children(child) |
|
|
|
elif isinstance(child, GrComponent): |
|
child.comp.render() |
|
|
|
elif isinstance(child, GrExamples): |
|
child.set_examples() |
|
|
|
@staticmethod |
|
def _set_event_listener(listeners: List[Any]) -> None: |
|
""" |
|
イベントリスナーの適用 |
|
|
|
全てのイベントリスナーをまとめて適用する |
|
|
|
Parameters |
|
---------- |
|
listeners : List[Any] |
|
イベントリスナーのリスト |
|
""" |
|
for listener in listeners: |
|
if isinstance(listener, list): |
|
GrBlocks._set_event_listener(listener) |
|
|
|
else: |
|
listener.setup() |
|
|
|
def find_comps( |
|
children: Dict[str, GrComponent | GrLayout] | GrLayout | List[Any], |
|
keys: List[str] | str |None = None |
|
) -> List[GrComponent] | GrComponent: |
|
""" |
|
コンポーネントの検索 |
|
|
|
内包する全てのコンポーネントを検索し、 |
|
flatten_compsを用いてフラットなコンポーネントのリストを作成する |
|
keysを渡すなら、childrenは辞書形式、keysを渡さないなら、childrenはGrLayout |
|
childrenがGrLayoutなら内包する全てのコンポーネントをリストにして返す |
|
同種のイベントリスナーをforで簡単に作成するために使う |
|
|
|
Parameters |
|
---------- |
|
children : Dict[str, GrComponent | GrLayout] | GrLayout | List[Any] |
|
子要素の辞書か、GrLayoutか、子要素のリスト |
|
keys : List[str] | str | None, optional, default None |
|
辞書のキーのリストか文字列 |
|
辞書形式の子要素から、順番に深い子要素を取得していく |
|
|
|
Returns |
|
------- |
|
List[GrComponent] | GrComponent |
|
フラットなコンポーネントのリスト |
|
""" |
|
if isinstance(keys, list): |
|
for key in keys: |
|
children: GrLayout | GrComponent = find_dict_value(children, key) |
|
|
|
elif isinstance(keys, str): |
|
children: GrLayout | GrComponent = find_dict_value(children, keys) |
|
|
|
else: |
|
children: List[GrComponent] = flatten_comps(children) |
|
|
|
return children |
|
|
|
def find_dict_value( |
|
children: Dict[str, GrComponent | GrLayout], key: str |
|
) -> GrLayout | GrComponent: |
|
""" |
|
辞書のバリューの取得 |
|
|
|
childrenのkeyのバリューが内包するGrLayoutかGrComponentを取得する |
|
|
|
Parameters |
|
---------- |
|
children : Dict[str, GrComponent | GrLayout] |
|
子要素の辞書 |
|
key : str |
|
子要素の辞書から取得するバリューのキー |
|
|
|
Returns |
|
------- |
|
GrLayout | GrComponent |
|
取得したGrLayoutかGrComponent |
|
""" |
|
if isinstance(children, dict): |
|
children: GrLayout | GrComponent = children[key] |
|
|
|
elif isinstance(children, GrLayout): |
|
children: GrLayout | GrComponent = children.children[key] |
|
|
|
return children |
|
|
|
def flatten_comps(children: GrLayout | List[Any]) -> List[GrComponent]: |
|
""" |
|
フラットなコンポーネントのリストの作成 |
|
|
|
find_compsから呼び出される |
|
一つのGrLayout内で、同種のコンポーネントは |
|
リストに直接入れられている可能性があるため、それらも含めてフラットにする |
|
例：[comp1, comp2, [Comp(i) for i in range(10)], comp13] |
|
|
|
Parameters |
|
---------- |
|
children : GrLayout | List[Any] |
|
GrLayoutか子要素のリスト |
|
|
|
Returns |
|
------- |
|
List[GrComponent] |
|
フラットなコンポーネントのリスト |
|
""" |
|
if isinstance(children, GrLayout): |
|
children = children.children |
|
|
|
if isinstance(children, dict): |
|
children = list(children.values()) |
|
|
|
if isinstance(children, list): |
|
return [ |
|
comp |
|
for grandchildren in children |
|
for comp in flatten_comps(grandchildren) |
|
] |
|
|
|
else: |
|
return [children] |
|
|
|
|
