PubSubTk ライブラリ - リファレンスガイド¶
概要¶
PubSubTk は、Pydantic を用いた型安全な状態管理と、Publish-Subscribe パターンを組み合わせて、Tkinter/ttk を使った GUI アプリケーションをシンプルに構築できる Python ライブラリです。
主な特徴¶
- UIとビジネスロジックの疎結合 ― Publish/Subscribe（Pub/Sub）で部品間を非同期メッセージ連携
- Pydanticモデル による型安全な状態管理。バリデーションや JSON Schema 出力も簡単
- Container / Presentational / Processor 3層分離パターンを標準化（Reactスタイルの設計をTkinterでも）
- Pub/Subによる画面遷移・サブウィンドウ管理とリアクティブUI更新をサポート
- 依存は純正Pythonのみ（
tkinter,pypubsub,pydantic）。Tkテーマ変更用にttkthemesも利用可能
アーキテクチャ概要¶
構造イメージ¶
graph LR
Store[Store]
Processor[Processor]
Container[Container]
View[Presentational View]
Processor -- state変更発行 --> Store
Store -- state変更通知 --> Container
Container -- UI更新 --> View
View -- trigger_event --> Container
Container -- action/イベント --> Processor各コンポーネントの役割¶
- Store: Pydanticモデルでアプリの状態を一元管理。型安全なアクセス＆更新通知が得られます。
- Container: 状態を購読し、UIと連動。ユーザー操作から Processor への橋渡しも担う。
- Presentational: 受け取ったデータを表示するだけの純粋View。状態管理・ロジックは一切持たない。
- Processor: ビジネスロジック/状態変更を集中管理。PubSub経由でContainer/Storeと通信。
🎯 推奨インポートパターン¶
from pubsubtk import (
TkApplication, ThemedApplication, # アプリケーション
ContainerComponentTk, ContainerComponentTtk, # コンテナ
PresentationalComponentTk, PresentationalComponentTtk, # プレゼンテーション
ProcessorBase, # プロセッサ
)
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import tkinter as tk
from tkinter import ttk
主要メソッド一覧¶
| メソッド | 説明・用途 | 主な利用層 |
|---|---|---|
pub_switch_container(cls, kwargs) |
メイン画面（Container）を切り替える | Container / Processor |
pub_switch_slot(slot_name, cls, kwargs) |
テンプレート内の任意スロットのコンポーネントを切り替え | Container / Processor |
pub_open_subwindow(cls, win_id, kwargs) |
サブウィンドウを開く | Container / Processor |
pub_close_subwindow(win_id) |
指定 ID のサブウィンドウを閉じる | Container / Processor |
pub_close_all_subwindows() |
サブウィンドウをすべて閉じる | Container / Processor |
pub_replace_state(new_state) |
状態オブジェクト全体を置き換える | Processor / Container |
pub_update_state(state_path, new_value) |
任意パスの状態を型安全に更新 | Processor / Container |
pub_add_to_list(state_path, item) |
リスト要素を型安全に追加 | Processor / Container |
pub_add_to_dict(state_path, key, value) |
辞書要素を型安全に追加 | Processor / Container |
pub_register_processor(proc, name) |
Processor を動的に登録 | Processor |
pub_delete_processor(name) |
Processor を削除 | Processor |
pub_enable_undo_redo(state_path, max_history) |
Undo/Redo 履歴を有効化 | Processor / Container |
pub_disable_undo_redo(state_path) |
Undo/Redo 履歴を無効化 | Processor / Container |
pub_undo(state_path) |
1つ前の値に戻す | Processor / Container |
pub_redo(state_path) |
Undo を取り消す | Processor / Container |
sub_undo_status(state_path, handler) |
Undo/Redo 状態変化を購読 | Container |
sub_state_changed(state_path, handler) |
指定パスの値変更を購読（old_value, new_value受信） | Container |
sub_for_refresh(state_path, handler) |
状態更新時のUI再描画用シンプル通知を購読（引数なし） | Container |
sub_state_added(state_path, handler) |
リストへの要素追加を購読（item, index受信） | Container |
sub_dict_item_added(state_path, handler) |
辞書への要素追加を購読（key, value受信） | Container |
register_handler(event, cb) |
PresentationalコンポーネントでViewイベントのハンドラ登録 | Container |
trigger_event(event, **kwargs) |
View→Containerへ任意イベント送出 | Presentational |
開発のポイント¶
StateProxyによるIDE連携¶
PubSubTkの最大の価値は、StateProxyによる強力なIDE連携です。
# VSCode/PyCharmで以下が全て効く：
self.store.state.user.name
# ↑ ↑
# Ctrl+Click Ctrl+Click
# で定義へ で定義へ
# ✅ F12: 定義へ移動
# ✅ Shift+F12: すべての参照を検索
# ✅ F2: 安全なリネーム
# ✅ Ctrl+Space: 自動補完
str()が必要なタイミング：
# ✅ 基本的な使用（str()不要）
self.pub_update_state(self.store.state.counter, 42)
self.sub_state_changed(self.store.state.todos, self.on_todos_changed)
# ✅ 文字列操作が必要な場合のみstr()を使用
path = str(self.store.state.user.name) + "_backup"
self.pub_update_state(f"todos.{index}", updated_todo)
コンポーネント設計指針¶
Container - 状態に依存する処理、ユーザー操作のハンドリング
class TodoContainer(ContainerComponentTk[AppState]):
def setup_subscriptions(self):
self.sub_state_changed(self.store.state.todos, self.on_todos_changed)
def add_todo(self):
# 状態更新
self.pub_add_to_list(self.store.state.todos, new_todo)
備考: コンポーネントの __init__ では与えられた *args と **kwargs が
self.args / self.kwargs として保持されます。サブウィンドウを open_subwindow
で開く場合は win_id が self.kwargs に自動追加され、
pub_close_subwindow(self.kwargs["win_id"]) で自身を閉じられます。今後も同様の
デフォルト引数が追加される可能性があります。
Presentational - 純粋な表示、再利用可能な部品
class TodoItemView(PresentationalComponentTk):
def update_data(self, todo_item: TodoItem):
self.label.config(text=todo_item.text)
def on_click(self):
# Container側にイベント通知
self.trigger_event("toggle", todo_id=self.todo_item.id)
Template - レイアウト構造の定義、スロットベースの画面構成
class AppTemplate(TemplateComponentTk[AppState]):
def define_slots(self):
# 各領域を定義・配置
self.header = tk.Frame(self, height=60)
self.header.pack(fill=tk.X)
self.main = tk.Frame(self)
self.main.pack(fill=tk.BOTH, expand=True)
self.sidebar = tk.Frame(self, width=200)
self.sidebar.pack(side=tk.RIGHT, fill=tk.Y)
return {
"header": self.header,
"main": self.main,
"sidebar": self.sidebar
}
# 使用例
app.set_template(AppTemplate)
app.pub_switch_slot("header", HeaderView)
app.pub_switch_slot("sidebar", NavigationPanel)
Processor - ビジネスロジック、複雑な状態操作
class TodoProcessor(ProcessorBase[AppState]):
def setup_subscriptions(self):
self.subscribe("todo.bulk_update", self.handle_bulk_update)
def handle_bulk_update(self, todo_ids: List[int]):
# 複雑なロジック処理
pass
Undo/Redo 機能¶
PubSubTk では、任意の状態パスを指定して Undo/Redo 履歴を管理できます。
pub_enable_undo_redo() で履歴を有効化すると自動的に変更が追跡され、
pub_undo() と pub_redo() で値を戻したりやり直したりできます。履歴数は
max_history 引数で調整してください。
# Undo/Redo の有効化
self.pub_enable_undo_redo(str(self.store.state.counter), max_history=20)
# 操作
self.pub_undo(str(self.store.state.counter))
self.pub_redo(str(self.store.state.counter))
# 履歴追跡の停止
self.pub_disable_undo_redo(str(self.store.state.counter))
sub_undo_status() を使うと、Undo/Redo 可能かどうかや履歴数を購読できます。
カスタムトピック・PubSub拡張¶
AutoNamedTopicによるカスタムトピック作成:
from pubsubtk import AutoNamedTopic
from enum import auto
class MyAppTopic(AutoNamedTopic):
USER_LOGIN = auto() # -> "MyAppTopic.user_login"
DATA_LOADED = auto() # -> "MyAppTopic.data_loaded"
ERROR_OCCURRED = auto() # -> "MyAppTopic.error_occurred"
FILE_EXPORT = auto() # -> "MyAppTopic.file_export"
# 使用例
class MyProcessor(ProcessorBase[AppState]):
def setup_subscriptions(self):
self.subscribe(MyAppTopic.USER_LOGIN, self.handle_user_login)
self.subscribe(MyAppTopic.DATA_LOADED, self.handle_data_loaded)
def some_action(self):
# カスタムトピックでメッセージ送信
self.publish(MyAppTopic.FILE_EXPORT, format="csv", filename="data.csv")
デフォルトトピック vs カスタムトピックの使い分け:
# ✅ デフォルト便利メソッドを使用（推奨）
self.pub_update_state(self.store.state.count, 42) # 状態更新
self.pub_switch_container(NewContainer) # 画面切り替え
self.pub_open_subwindow(DialogContainer) # サブウィンドウ
self.pub_enable_undo_redo(str(self.store.state.count)) # Undo履歴を開始
self.pub_undo(str(self.store.state.count)) # Undo
self.pub_redo(str(self.store.state.count)) # Redo
self.pub_disable_undo_redo(str(self.store.state.count)) # 履歴停止
# ✅ カスタムトピックを使用（ビジネスロジック特有の通信）
self.publish(MyAppTopic.USER_LOGIN, user_id=123) # アプリ固有のイベント
self.subscribe(MyAppTopic.DATA_LOADED, self.on_data) # 複雑なワークフロー
よくある問題と解決法¶
StateProxy使用時のエラー:
# ❌ エラーになる例
path = self.store.state.user.name.replace("old", "new") # AttributeError
# ✅ 正しい使い方
path = str(self.store.state.user.name).replace("old", "new")
便利メソッドの活用:
# ✅ 推奨: 組み込みメソッドを使用
self.pub_update_state(self.store.state.count, 42)
self.pub_switch_container(OtherContainer)
self.pub_enable_undo_redo(str(self.store.state.count))
self.pub_undo(str(self.store.state.count))
self.pub_redo(str(self.store.state.count))
self.pub_disable_undo_redo(str(self.store.state.count))
# ❌ 非推奨: 手動でトピック操作
self.publish(DefaultUpdateTopic.UPDATE_STATE, state_path="count", new_value=42)
実践例¶
全機能を活用したシンプルなカウンターアプリ¶
PubSubDefaultTopicBaseの全メソッドを使用した小規模なデモアプリケーション
"""
PubSubTk 全機能コンパクトデモ
PubSubDefaultTopicBaseの全メソッドを使用した小規模なデモアプリケーション
"""
import asyncio
import json
import tkinter as tk
from enum import auto
from tkinter import filedialog, messagebox, simpledialog
from typing import List
from pydantic import BaseModel
from pubsubtk import (
AutoNamedTopic,
ContainerComponentTk,
PresentationalComponentTk,
ProcessorBase,
TemplateComponentTk,
TkApplication,
make_async_task,
)
# カスタムトピック
class AppTopic(AutoNamedTopic):
INCREMENT = auto()
RESET = auto()
MILESTONE = auto()
# データモデル
class TodoItem(BaseModel):
id: int
text: str
completed: bool = False
class AppState(BaseModel):
counter: int = 0
total_clicks: int = 0
todos: List[TodoItem] = []
next_todo_id: int = 1
settings: dict = {"theme": "default", "auto_save": "true"}
current_view: str = "main"
# =============================================================================
# テンプレート（3スロット構成）
# =============================================================================
class AppTemplate(TemplateComponentTk[AppState]):
def define_slots(self):
# ナビゲーション
self.navbar = tk.Frame(self, height=40, bg="navy")
self.navbar.pack(fill=tk.X)
self.navbar.pack_propagate(False)
# メインコンテンツ
self.main_area = tk.Frame(self)
self.main_area.pack(fill=tk.BOTH, expand=True, side=tk.LEFT)
# サイドバー
self.sidebar = tk.Frame(self, width=200, bg="lightgray")
self.sidebar.pack(fill=tk.Y, side=tk.RIGHT)
self.sidebar.pack_propagate(False)
return {
"navbar": self.navbar,
"main": self.main_area,
"sidebar": self.sidebar,
}
# =============================================================================
# Presentationalコンポーネント（純粋表示）
# =============================================================================
class UndoRedoControlView(PresentationalComponentTk):
"""Undo/Redo操作用のPresentationalコンポーネント"""
def setup_ui(self):
self.configure(relief=tk.RIDGE, borderwidth=1, bg="lightyellow")
# タイトル
title_label = tk.Label(
self, text="🔄 履歴操作", font=("Arial", 10, "bold"), bg="lightyellow"
)
title_label.pack(pady=2)
# ボタンフレーム
btn_frame = tk.Frame(self, bg="lightyellow")
btn_frame.pack(pady=2)
self.undo_btn = tk.Button(btn_frame, text="← Undo", state="disabled", width=8)
self.undo_btn.pack(side=tk.LEFT, padx=2)
self.redo_btn = tk.Button(btn_frame, text="Redo →", state="disabled", width=8)
self.redo_btn.pack(side=tk.LEFT, padx=2)
# ステータス表示
self.status_label = tk.Label(
self, text="無効", font=("Arial", 8), bg="lightyellow", fg="gray"
)
self.status_label.pack(pady=1)
def setup_handlers(self, undo_handler, redo_handler):
"""Undo/Redoハンドラーを設定"""
self.undo_btn.config(command=undo_handler)
self.redo_btn.config(command=redo_handler)
def update_status(
self, can_undo: bool, can_redo: bool, undo_count: int, redo_count: int
):
"""Undo/Redoステータスを更新"""
# ボタンの有効/無効状態
self.undo_btn.config(
state="normal" if can_undo else "disabled",
text=f"← Undo ({undo_count})" if undo_count > 0 else "← Undo",
)
self.redo_btn.config(
state="normal" if can_redo else "disabled",
text=f"Redo ({redo_count}) →" if redo_count > 0 else "Redo →",
)
# ステータステキスト
if can_undo or can_redo:
status = f"Undo:{undo_count} Redo:{redo_count}"
self.status_label.config(text=status, fg="black")
else:
self.status_label.config(text="履歴なし", fg="gray")
class TodoItemView(PresentationalComponentTk):
def setup_ui(self):
self.configure(relief=tk.RAISED, borderwidth=1, padx=5, pady=3)
self.var = tk.BooleanVar()
self.checkbox = tk.Checkbutton(self, variable=self.var, command=self.on_toggle)
self.checkbox.pack(side=tk.LEFT)
self.label = tk.Label(self, text="", anchor="w")
self.label.pack(side=tk.LEFT, fill=tk.X, expand=True)
self.delete_btn = tk.Button(self, text="×", width=3, command=self.on_delete)
self.delete_btn.pack(side=tk.RIGHT)
def update_data(self, todo: TodoItem):
self.todo = todo
self.var.set(todo.completed)
text = f"✓ {todo.text}" if todo.completed else todo.text
self.label.config(text=text, fg="gray" if todo.completed else "black")
def on_toggle(self):
self.trigger_event("toggle", todo_id=self.todo.id)
def on_delete(self):
self.trigger_event("delete", todo_id=self.todo.id)
class StatsView(PresentationalComponentTk):
def setup_ui(self):
self.configure(bg="lightblue", relief=tk.SUNKEN, borderwidth=2)
tk.Label(self, text="📊 統計", font=("Arial", 12, "bold"), bg="lightblue").pack(
pady=5
)
self.stats_label = tk.Label(self, text="", bg="lightblue", justify=tk.LEFT)
self.stats_label.pack(padx=10, pady=5, fill=tk.BOTH, expand=True)
def update_stats(
self,
counter: int,
total_clicks: int,
total_todos: int,
completed_todos: int,
settings_count: int,
current_view: str,
counter_undo_status: dict = None,
todos_undo_status: dict = None,
):
"""純粋な表示コンポーネント - 必要なデータのみを個別に受け取る"""
uncompleted = total_todos - completed_todos
# デフォルト値設定
counter_undo_status = counter_undo_status or {
"can_undo": False,
"can_redo": False,
"undo_count": 0,
"redo_count": 0,
}
todos_undo_status = todos_undo_status or {
"can_undo": False,
"can_redo": False,
"undo_count": 0,
"redo_count": 0,
}
stats = f"""カウンター: {counter}
総クリック: {total_clicks}
Todo統計:
・総数: {total_todos}
・完了: {completed_todos}
・未完了: {uncompleted}
設定数: {settings_count}
現在画面: {current_view}
🔄 履歴状況:
カウンター履歴:
・Undo: {counter_undo_status["undo_count"]}回
・Redo: {counter_undo_status["redo_count"]}回
Todo履歴:
・Undo: {todos_undo_status["undo_count"]}回
・Redo: {todos_undo_status["redo_count"]}回"""
self.stats_label.config(text=stats)
# =============================================================================
# Containerコンポーネント（状態連携）
# =============================================================================
class NavbarContainer(ContainerComponentTk[AppState]):
def setup_ui(self):
self.configure(bg="navy")
tk.Label(
self,
text="🎯 PubSubTk Demo (w/ Undo/Redo)",
fg="white",
bg="navy",
font=("Arial", 14, "bold"),
).pack(side=tk.LEFT, padx=10, pady=5)
nav_frame = tk.Frame(self, bg="navy")
nav_frame.pack(side=tk.RIGHT, padx=10)
self.main_btn = tk.Button(nav_frame, text="メイン", command=self.switch_to_main)
self.main_btn.pack(side=tk.LEFT, padx=2)
self.todo_btn = tk.Button(nav_frame, text="Todo", command=self.switch_to_todo)
self.todo_btn.pack(side=tk.LEFT, padx=2)
def setup_subscriptions(self):
self.sub_state_changed(self.store.state.current_view, self.on_view_changed)
def refresh_from_state(self):
state = self.store.get_current_state()
self.update_buttons(state.current_view)
def on_view_changed(self, old_value, new_value):
self.update_buttons(new_value)
def update_buttons(self, current_view: str):
self.main_btn.config(
bg="lightblue" if current_view == "main" else "SystemButtonFace"
)
self.todo_btn.config(
bg="lightblue" if current_view == "todo" else "SystemButtonFace"
)
def switch_to_main(self):
self.pub_update_state(self.store.state.current_view, "main")
self.pub_switch_slot("main", MainContainer)
def switch_to_todo(self):
self.pub_update_state(self.store.state.current_view, "todo")
self.pub_switch_slot("main", TodoContainer)
class MainContainer(ContainerComponentTk[AppState]):
def setup_ui(self):
tk.Label(self, text="🏠 メインビュー", font=("Arial", 16, "bold")).pack(pady=10)
# カウンター
self.counter_label = tk.Label(self, text="0", font=("Arial", 32))
self.counter_label.pack(pady=20)
# Undo/Redoコントロール（カウンター用）
self.counter_undo_control = UndoRedoControlView(self)
self.counter_undo_control.pack(pady=5)
self.counter_undo_control.setup_handlers(
undo_handler=self.undo_counter, redo_handler=self.redo_counter
)
# ボタン群
btn_frame = tk.Frame(self)
btn_frame.pack(pady=10)
tk.Button(btn_frame, text="カウント", command=self.increment).pack(
side=tk.LEFT, padx=5
)
tk.Button(btn_frame, text="リセット", command=self.reset).pack(
side=tk.LEFT, padx=5
)
tk.Button(btn_frame, text="サブウィンドウ", command=self.open_sub).pack(
side=tk.LEFT, padx=5
)
# Undo/Redo制御ボタン
undo_control_frame = tk.Frame(self)
undo_control_frame.pack(pady=5)
self.enable_undo_btn = tk.Button(
undo_control_frame,
text="履歴記録ON",
command=self.enable_counter_undo,
bg="lightgreen",
)
self.enable_undo_btn.pack(side=tk.LEFT, padx=5)
self.disable_undo_btn = tk.Button(
undo_control_frame,
text="履歴記録OFF",
command=self.disable_counter_undo,
bg="lightcoral",
state="disabled",
)
self.disable_undo_btn.pack(side=tk.LEFT, padx=5)
# ファイル操作
file_frame = tk.Frame(self)
file_frame.pack(pady=10)
tk.Button(file_frame, text="保存", command=self.save_data).pack(
side=tk.LEFT, padx=5
)
tk.Button(file_frame, text="読込", command=self.load_data).pack(
side=tk.LEFT, padx=5
)
# 設定操作（辞書機能テスト）
setting_frame = tk.Frame(self)
setting_frame.pack(pady=10)
tk.Button(setting_frame, text="設定追加", command=self.add_setting).pack(
side=tk.LEFT, padx=5
)
tk.Button(
setting_frame, text="プロセッサー追加", command=self.add_processor
).pack(side=tk.LEFT, padx=5)
# 危険な操作
tk.Button(
self, text="全状態リセット", command=self.reset_all, bg="red", fg="white"
).pack(pady=10)
def setup_subscriptions(self):
self.sub_state_changed(self.store.state.counter, self.on_counter_changed)
self.subscribe(AppTopic.MILESTONE, self.on_milestone)
# カウンターのUndo/Redoステータス変化を購読
self.sub_undo_status(str(self.store.state.counter), self.on_counter_undo_status)
def refresh_from_state(self):
state = self.store.get_current_state()
self.counter_label.config(text=str(state.counter))
def on_counter_changed(self, old_value, new_value):
self.counter_label.config(text=str(new_value))
def on_counter_undo_status(
self, can_undo: bool, can_redo: bool, undo_count: int, redo_count: int
):
"""カウンターのUndo/Redoステータス変化ハンドラー"""
self.counter_undo_control.update_status(
can_undo, can_redo, undo_count, redo_count
)
def enable_counter_undo(self):
"""カウンターのUndo/Redo機能を有効化"""
self.pub_enable_undo_redo(str(self.store.state.counter), max_history=20)
self.enable_undo_btn.config(state="disabled")
self.disable_undo_btn.config(state="normal")
def disable_counter_undo(self):
"""カウンターのUndo/Redo機能を無効化"""
self.pub_disable_undo_redo(str(self.store.state.counter))
self.enable_undo_btn.config(state="normal")
self.disable_undo_btn.config(state="disabled")
def undo_counter(self):
"""カウンターをUndo"""
self.pub_undo(str(self.store.state.counter))
def redo_counter(self):
"""カウンターをRedo"""
self.pub_redo(str(self.store.state.counter))
def increment(self):
self.publish(AppTopic.INCREMENT)
def reset(self):
self.publish(AppTopic.RESET)
def open_sub(self):
self.pub_open_subwindow(SubWindow)
@make_async_task
async def save_data(self):
filename = filedialog.asksaveasfilename(defaultextension=".json")
if filename:
await asyncio.sleep(0.3) # 保存処理シミュレート
state = self.store.get_current_state()
with open(filename, "w") as f:
json.dump(state.model_dump(), f, indent=2)
messagebox.showinfo("完了", "データを保存しました")
@make_async_task
async def load_data(self):
filename = filedialog.askopenfilename(filetypes=[("JSON files", "*.json")])
if filename:
await asyncio.sleep(0.3) # 読込処理シミュレート
with open(filename, "r") as f:
data = json.load(f)
new_state = AppState.model_validate(data)
self.pub_replace_state(new_state)
# 状態リセット後は画面も適切に切り替える
self.pub_switch_slot("main", MainContainer)
messagebox.showinfo("完了", "データを読み込みました")
def add_setting(self):
key = simpledialog.askstring("設定追加", "キーを入力:")
if key:
value = simpledialog.askstring("設定追加", "値を入力:")
if value:
# pub_add_to_dict使用
self.pub_add_to_dict(self.store.state.settings, key, value)
@make_async_task
async def add_processor(self):
await asyncio.sleep(0.5) # プロセッサー初期化シミュレート
try:
# pub_register_processor使用
self.pub_register_processor(DummyProcessor, "dummy")
messagebox.showinfo("成功", "プロセッサーを追加しました")
except Exception as e:
messagebox.showerror("エラー", str(e))
@make_async_task
async def reset_all(self):
if messagebox.askyesno("確認", "全状態をリセットしますか？"):
await asyncio.sleep(1.0) # 重い処理シミュレート
# pub_replace_state使用
self.pub_replace_state(AppState())
# リセット後はメイン画面に戻る
self.pub_switch_slot("main", MainContainer)
messagebox.showinfo("完了", "状態をリセットしました")
def on_milestone(self, value: int):
messagebox.showinfo("マイルストーン!", f"{value}に到達！")
class TodoContainer(ContainerComponentTk[AppState]):
def setup_ui(self):
tk.Label(self, text="📝 Todo管理", font=("Arial", 16, "bold")).pack(pady=10)
# Todo用Undo/Redoコントロール
self.todo_undo_control = UndoRedoControlView(self)
self.todo_undo_control.pack(pady=5)
self.todo_undo_control.setup_handlers(
undo_handler=self.undo_todos, redo_handler=self.redo_todos
)
# Undo/Redo制御
undo_control_frame = tk.Frame(self)
undo_control_frame.pack(pady=5)
self.enable_todo_undo_btn = tk.Button(
undo_control_frame,
text="Todo履歴ON",
command=self.enable_todo_undo,
bg="lightgreen",
)
self.enable_todo_undo_btn.pack(side=tk.LEFT, padx=5)
self.disable_todo_undo_btn = tk.Button(
undo_control_frame,
text="Todo履歴OFF",
command=self.disable_todo_undo,
bg="lightcoral",
state="disabled",
)
self.disable_todo_undo_btn.pack(side=tk.LEFT, padx=5)
# Todo追加
add_frame = tk.Frame(self)
add_frame.pack(fill=tk.X, padx=10, pady=5)
self.entry = tk.Entry(add_frame)
self.entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
self.entry.bind("<Return>", lambda e: self.add_todo())
tk.Button(add_frame, text="追加", command=self.add_todo).pack(side=tk.RIGHT)
# Todoリスト
list_frame = tk.Frame(self, relief=tk.SUNKEN, borderwidth=2)
list_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
# スクロール可能フレーム
canvas = tk.Canvas(list_frame)
scrollbar = tk.Scrollbar(list_frame, orient="vertical", command=canvas.yview)
self.scrollable_frame = tk.Frame(canvas)
self.scrollable_frame.bind(
"<Configure>", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
)
canvas.create_window((0, 0), window=self.scrollable_frame, anchor="nw")
canvas.configure(yscrollcommand=scrollbar.set)
canvas.pack(side="left", fill="both", expand=True)
scrollbar.pack(side="right", fill="y")
self.todo_widgets = {}
def setup_subscriptions(self):
# sub_state_addedとsub_for_refresh使用
self.sub_state_added(self.store.state.todos, self.on_todo_added)
self.sub_for_refresh(self.store.state.todos, self.refresh_todo_list)
# TodoリストのUndo/Redoステータス変化を購読
self.sub_undo_status(str(self.store.state.todos), self.on_todo_undo_status)
def refresh_from_state(self):
self.refresh_todo_list()
def on_todo_undo_status(
self, can_undo: bool, can_redo: bool, undo_count: int, redo_count: int
):
"""TodoリストのUndo/Redoステータス変化ハンドラー"""
self.todo_undo_control.update_status(can_undo, can_redo, undo_count, redo_count)
def enable_todo_undo(self):
"""TodoリストのUndo/Redo機能を有効化"""
self.pub_enable_undo_redo(str(self.store.state.todos), max_history=15)
self.enable_todo_undo_btn.config(state="disabled")
self.disable_todo_undo_btn.config(state="normal")
def disable_todo_undo(self):
"""TodoリストのUndo/Redo機能を無効化"""
self.pub_disable_undo_redo(str(self.store.state.todos))
self.enable_todo_undo_btn.config(state="normal")
self.disable_todo_undo_btn.config(state="disabled")
def undo_todos(self):
"""TodoリストをUndo"""
self.pub_undo(str(self.store.state.todos))
def redo_todos(self):
"""TodoリストをRedo"""
self.pub_redo(str(self.store.state.todos))
def refresh_todo_list(self):
# 既存ウィジェットクリア
for widget in self.todo_widgets.values():
widget.destroy()
self.todo_widgets.clear()
# 新しいリストを描画
state = self.store.get_current_state()
for todo in state.todos:
todo_widget = TodoItemView(self.scrollable_frame)
todo_widget.pack(fill=tk.X, padx=5, pady=2)
todo_widget.update_data(todo)
# イベントハンドラ登録
todo_widget.register_handler("toggle", self.toggle_todo)
todo_widget.register_handler("delete", self.delete_todo)
self.todo_widgets[todo.id] = todo_widget
def on_todo_added(self, item: TodoItem, index: int):
# 新規追加時は全体再描画
self.refresh_todo_list()
def add_todo(self):
text = self.entry.get().strip()
if not text:
return
state = self.store.get_current_state()
new_todo = TodoItem(id=state.next_todo_id, text=text)
# pub_add_to_list使用
self.pub_add_to_list(self.store.state.todos, new_todo)
self.pub_update_state(self.store.state.next_todo_id, state.next_todo_id + 1)
self.entry.delete(0, tk.END)
def toggle_todo(self, todo_id: int):
state = self.store.get_current_state()
updated_todos = []
for todo in state.todos:
if todo.id == todo_id:
updated = todo.model_copy()
updated.completed = not updated.completed
updated_todos.append(updated)
else:
updated_todos.append(todo)
self.pub_update_state(self.store.state.todos, updated_todos)
@make_async_task
async def delete_todo(self, todo_id: int):
if messagebox.askyesno("確認", "このTodoを削除しますか？"):
await asyncio.sleep(0.2) # 削除処理シミュレート
state = self.store.get_current_state()
updated_todos = [t for t in state.todos if t.id != todo_id]
self.pub_update_state(self.store.state.todos, updated_todos)
class SidebarContainer(ContainerComponentTk[AppState]):
def setup_ui(self):
self.configure(bg="lightgray")
# 統計表示（Presentationalコンポーネント使用）
self.stats_view = StatsView(self)
self.stats_view.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# 操作ボタン
btn_frame = tk.Frame(self, bg="lightgray")
btn_frame.pack(fill=tk.X, padx=5, pady=5)
tk.Button(btn_frame, text="全ウィンドウ閉じる", command=self.close_all).pack(
fill=tk.X, pady=2
)
tk.Button(btn_frame, text="プロセッサー削除", command=self.delete_proc).pack(
fill=tk.X, pady=2
)
# Undo/Redo情報保持用
self.counter_undo_status = {
"can_undo": False,
"can_redo": False,
"undo_count": 0,
"redo_count": 0,
}
self.todos_undo_status = {
"can_undo": False,
"can_redo": False,
"undo_count": 0,
"redo_count": 0,
}
def setup_subscriptions(self):
# 複数の状態変更を監視
self.sub_for_refresh(self.store.state.counter, self.refresh_from_state)
self.sub_for_refresh(self.store.state.todos, self.refresh_from_state)
self.sub_for_refresh(self.store.state.settings, self.refresh_from_state)
self.sub_for_refresh(self.store.state.total_clicks, self.refresh_from_state)
self.sub_for_refresh(self.store.state.current_view, self.refresh_from_state)
# sub_dict_item_added使用
self.sub_dict_item_added(self.store.state.settings, self.on_setting_added)
# Undo/Redoステータス監視
self.sub_undo_status(str(self.store.state.counter), self.on_counter_undo_status)
self.sub_undo_status(str(self.store.state.todos), self.on_todos_undo_status)
def on_counter_undo_status(
self, can_undo: bool, can_redo: bool, undo_count: int, redo_count: int
):
"""カウンターのUndo/Redoステータス変化ハンドラー"""
self.counter_undo_status = {
"can_undo": can_undo,
"can_redo": can_redo,
"undo_count": undo_count,
"redo_count": redo_count,
}
self.refresh_from_state()
def on_todos_undo_status(
self, can_undo: bool, can_redo: bool, undo_count: int, redo_count: int
):
"""TodoリストのUndo/Redoステータス変化ハンドラー"""
self.todos_undo_status = {
"can_undo": can_undo,
"can_redo": can_redo,
"undo_count": undo_count,
"redo_count": redo_count,
}
self.refresh_from_state()
def refresh_from_state(self):
state = self.store.get_current_state()
# Containerで状態から必要な値を抽出してPresentationalに渡す
completed_todos = sum(1 for t in state.todos if t.completed)
total_todos = len(state.todos)
self.stats_view.update_stats(
counter=state.counter,
total_clicks=state.total_clicks,
total_todos=total_todos,
completed_todos=completed_todos,
settings_count=len(state.settings),
current_view=state.current_view,
counter_undo_status=self.counter_undo_status,
todos_undo_status=self.todos_undo_status,
)
def on_setting_added(self, key: str, value: str):
messagebox.showinfo("設定追加", f"設定追加: {key} = {value}")
def close_all(self):
# pub_close_all_subwindows使用
self.pub_close_all_subwindows()
def delete_proc(self):
try:
# pub_delete_processor使用
self.pub_delete_processor("dummy")
messagebox.showinfo("成功", "プロセッサーを削除しました")
except KeyError:
messagebox.showerror("エラー", "プロセッサーが見つかりません")
# =============================================================================
# サブウィンドウ
# =============================================================================
class SubWindow(ContainerComponentTk[AppState]):
def setup_ui(self):
tk.Label(self, text="🔢 サブウィンドウ", font=("Arial", 14)).pack(pady=10)
self.value_label = tk.Label(self, text="0", font=("Arial", 20))
self.value_label.pack(pady=10)
btn_frame = tk.Frame(self)
btn_frame.pack(pady=10)
tk.Button(btn_frame, text="+1", command=self.increment).pack(
side=tk.LEFT, padx=5
)
tk.Button(btn_frame, text="+5", command=lambda: self.add_value(5)).pack(
side=tk.LEFT, padx=5
)
tk.Button(self, text="閉じる", command=self.close_window).pack(pady=10)
def setup_subscriptions(self):
self.sub_state_changed(self.store.state.counter, self.on_counter_changed)
def refresh_from_state(self):
state = self.store.get_current_state()
self.value_label.config(text=str(state.counter))
def on_counter_changed(self, old_value, new_value):
self.value_label.config(text=str(new_value))
def increment(self):
self.publish(AppTopic.INCREMENT)
def add_value(self, value: int):
state = self.store.get_current_state()
new_value = state.counter + value
self.pub_update_state(self.store.state.counter, new_value)
self.pub_update_state(self.store.state.total_clicks, state.total_clicks + 1)
def close_window(self):
# pub_close_subwindow使用
self.pub_close_subwindow(self.kwargs["win_id"])
# =============================================================================
# プロセッサー
# =============================================================================
class MainProcessor(ProcessorBase[AppState]):
def setup_subscriptions(self):
self.subscribe(AppTopic.INCREMENT, self.handle_increment)
self.subscribe(AppTopic.RESET, self.handle_reset)
def handle_increment(self):
state = self.store.get_current_state()
new_counter = state.counter + 1
new_total = state.total_clicks + 1
self.pub_update_state(self.store.state.counter, new_counter)
self.pub_update_state(self.store.state.total_clicks, new_total)
# マイルストーン判定
if new_counter % 10 == 0:
self.publish(AppTopic.MILESTONE, value=new_counter)
def handle_reset(self):
self.pub_update_state(self.store.state.counter, 0)
class DummyProcessor(ProcessorBase[AppState]):
def setup_subscriptions(self):
print("DummyProcessor: 初期化されました")
# =============================================================================
# メインアプリケーション
# =============================================================================
if __name__ == "__main__":
app = TkApplication(AppState, title="🎯 PubSubTk Demo", geometry="800x600")
# メインプロセッサー登録
app.pub_register_processor(MainProcessor)
# テンプレート設定
app.set_template(AppTemplate)
# 各スロットにコンテナ配置
app.pub_switch_slot("navbar", NavbarContainer)
app.pub_switch_slot("main", MainContainer) # 初期画面
app.pub_switch_slot("sidebar", SidebarContainer)
# 起動
app.run(use_async=True)
フルソースコード¶
コアPubSubシステム¶
src/pubsubtk/core/pubsub_base.py¶
PubSubパターンの基底クラス
# pubsub_base.py - PubSub 基底クラス
"""Pub/Sub パターンの共通機能をまとめた抽象基底クラス。"""
import logging
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, List
from pubsub import pub
# PubSub専用のロガーを作成
_pubsub_logger = logging.getLogger("pubsubtk.pubsub")
class PubSubBase(ABC):
"""
PubSubパターンの基底クラス。
- setup_subscriptions()で購読設定を行う抽象メソッドを提供
- subscribe()/send_message()/unsubscribe()/unsubscribe_all()で購読管理
- teardown()で全購読解除
- 継承先で購読設定を簡潔に記述可能
- DEBUGレベルでPubSub操作をログ出力
"""
def __init__(self, *args, **kwargs):
self._subscriptions: List[Dict[str, Any]] = []
self.setup_subscriptions()
def subscribe(self, topic: str, handler: Callable, **kwargs) -> None:
pub.subscribe(handler, topic, **kwargs)
self._subscriptions.append({"topic": topic, "handler": handler})
# DEBUGログ：購読登録
_pubsub_logger.debug(
f"SUBSCRIBE: {self.__class__.__name__} -> topic='{topic}', handler={handler.__name__}"
)
def publish(self, topic: str, **kwargs) -> None:
# DEBUGログ：パブリッシュ（引数も表示）
args_str = ", ".join(f"{k}={v}" for k, v in kwargs.items())
_pubsub_logger.debug(
f"PUBLISH: {self.__class__.__name__} -> topic='{topic}'"
+ (f" with args: {args_str}" if args_str else "")
)
pub.sendMessage(topic, **kwargs)
def unsubscribe(self, topic: str, handler: Callable) -> None:
pub.unsubscribe(handler, topic)
self._subscriptions = [
s
for s in self._subscriptions
if not (s["topic"] == topic and s["handler"] == handler)
]
# DEBUGログ：購読解除
_pubsub_logger.debug(
f"UNSUBSCRIBE: {self.__class__.__name__} -> topic='{topic}', handler={handler.__name__}"
)
def unsubscribe_all(self) -> None:
# DEBUGログ：全購読解除
if self._subscriptions:
_pubsub_logger.debug(
f"UNSUBSCRIBE_ALL: {self.__class__.__name__} -> {len(self._subscriptions)} subscriptions"
)
for s in list(self._subscriptions):
pub.unsubscribe(s["handler"], s["topic"])
self._subscriptions.clear()
@abstractmethod
def setup_subscriptions(self) -> None:
"""
継承先で購読設定を行うためのメソッド。
例:
class MyPS(PubSubBase):
def setup_subscriptions(self):
self.subscribe(TopicEnum.STATE_CHANGED, self.on_change)
"""
pass
def teardown(self) -> None:
"""
全ての購読を解除する。
"""
self.unsubscribe_all()
# デバッグログを有効化するユーティリティ関数
def enable_pubsub_debug_logging(level: int = logging.DEBUG) -> None:
"""
PubSubのデバッグログを有効化する。
Args:
level: ログレベル（デフォルト: DEBUG）
使用例:
from pubsubtk.core.pubsub_base import enable_pubsub_debug_logging
enable_pubsub_debug_logging()
"""
_pubsub_logger.setLevel(level)
# ハンドラーが未設定の場合はコンソールハンドラーを追加
if not _pubsub_logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter(
"[%(asctime)s] %(name)s - %(levelname)s - %(message)s", datefmt="%H:%M:%S"
)
handler.setFormatter(formatter)
_pubsub_logger.addHandler(handler)
_pubsub_logger.debug("PubSub debug logging enabled")
def disable_pubsub_debug_logging() -> None:
"""
PubSubのデバッグログを無効化する。
"""
_pubsub_logger.setLevel(logging.WARNING)
_pubsub_logger.debug("PubSub debug logging disabled")
src/pubsubtk/core/default_topic_base.py¶
デフォルトトピック操作をまとめた基底クラス
# default_topic_base.py - デフォルトトピック操作をまとめた基底クラス
"""
src/pubsubtk/core/default_topic_base.py
主要な PubSub トピックに対する便利メソッドを提供します。
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Callable, Optional, Type
from pubsubtk.core.pubsub_base import PubSubBase
from pubsubtk.topic.topics import (
DefaultNavigateTopic,
DefaultProcessorTopic,
DefaultUndoTopic,
DefaultUpdateTopic,
)
if TYPE_CHECKING:
# 型チェック時（mypy や IDE 補完時）のみ読み込む
from pubsubtk.processor.processor_base import ProcessorBase
from pubsubtk.ui.types import ComponentType, ContainerComponentType
class PubSubDefaultTopicBase(PubSubBase):
"""
Built-in convenience methods for common PubSub operations.
**IMPORTANT**: Container and Processor components should use these built-in methods
instead of manually publishing to DefaultTopics. These methods are designed for
ease of use and provide better IDE support.
"""
def pub_switch_container(
self,
cls: ContainerComponentType,
kwargs: dict = None,
) -> None:
"""コンテナを切り替えるPubSubメッセージを送信する。
Args:
cls (ContainerComponentType): 切り替え先のコンテナコンポーネントクラス
kwargs: コンテナに渡すキーワード引数用辞書
Note:
コンテナは、TkApplicationまたはTtkApplicationのコンストラクタで指定された
親ウィジェットの子として配置されます。
"""
self.publish(DefaultNavigateTopic.SWITCH_CONTAINER, cls=cls, kwargs=kwargs)
def pub_switch_slot(
self,
slot_name: str,
cls: ComponentType,
kwargs: dict = None,
) -> None:
"""テンプレートの特定スロットのコンテンツを切り替える。
Args:
slot_name (str): スロット名
cls (ComponentType): コンテナまたはプレゼンテーショナルコンポーネントクラス
kwargs: コンポーネントに渡すキーワード引数用辞書
Note:
ContainerComponentとPresentationalComponentの両方に対応。
テンプレートが設定されていない場合はエラーになります。
"""
self.publish(
DefaultNavigateTopic.SWITCH_SLOT,
slot_name=slot_name,
cls=cls,
kwargs=kwargs,
)
def pub_open_subwindow(
self,
cls: ComponentType,
win_id: Optional[str] = None,
kwargs: dict = None,
) -> None:
"""サブウィンドウを開くPubSubメッセージを送信する。
Args:
cls (ComponentType): サブウィンドウに表示するコンポーネントクラス
win_id (Optional[str], optional): サブウィンドウのID。
指定しない場合は自動生成される。
同じIDを指定すると、既存のウィンドウが再利用される。
kwargs: コンポーネントに渡すキーワード引数用辞書
Note:
サブウィンドウは、Toplevel ウィジェットとして作成されます。
"""
self.publish(
DefaultNavigateTopic.OPEN_SUBWINDOW, cls=cls, win_id=win_id, kwargs=kwargs
)
def pub_close_subwindow(self, win_id: str) -> None:
"""サブウィンドウを閉じるPubSubメッセージを送信する。
Args:
win_id (str): 閉じるサブウィンドウのID
"""
self.publish(DefaultNavigateTopic.CLOSE_SUBWINDOW, win_id=win_id)
def pub_close_all_subwindows(self) -> None:
"""すべてのサブウィンドウを閉じるPubSubメッセージを送信する。"""
self.publish(DefaultNavigateTopic.CLOSE_ALL_SUBWINDOWS)
def pub_replace_state(self, new_state: Any) -> None:
"""状態オブジェクト全体を置き換えるPubSubメッセージを送信する。
Args:
new_state: 新しい状態オブジェクト。
"""
self.publish(DefaultUpdateTopic.REPLACE_STATE, new_state=new_state)
def pub_update_state(self, state_path: str, new_value: Any) -> None:
"""
Storeの状態を更新するPubSubメッセージを送信する。
Args:
state_path (str): 更新する状態のパス（例: "user.name", "items[2].value"）
new_value (Any): 新しい値
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_update_state(str(self.store.state.user.name), "新しい名前")`
The state proxy provides autocomplete and "Go to Definition" functionality.
"""
self.publish(
DefaultUpdateTopic.UPDATE_STATE,
state_path=str(state_path),
new_value=new_value,
)
def pub_add_to_list(self, state_path: str, item: Any) -> None:
"""
Storeの状態（リスト）に要素を追加するPubSubメッセージを送信する。
Args:
state_path (str): 要素を追加するリストの状態パス（例: "items", "user.tasks"）
item (Any): 追加する要素
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_add_to_list(str(self.store.state.items), new_item)`
The state proxy provides autocomplete and "Go to Definition" functionality.
"""
self.publish(
DefaultUpdateTopic.ADD_TO_LIST, state_path=str(state_path), item=item
)
def pub_add_to_dict(self, state_path: str, key: str, value: Any) -> None:
"""Storeの状態(辞書)に要素を追加するPubSubメッセージを送信する。
Args:
state_path: 要素を追加する辞書の状態パス。
key: 追加するキー。
value: 追加する値。
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_add_to_dict(str(self.store.state.mapping), "k", v)`
"""
self.publish(
DefaultUpdateTopic.ADD_TO_DICT,
state_path=str(state_path),
key=key,
value=value,
)
def pub_register_processor(
self,
proc: Type[ProcessorBase],
name: Optional[str] = None,
) -> None:
"""Processorを登録するPubSubメッセージを送信する。
Args:
proc (Type[ProcessorBase]): 登録するProcessorクラス
name (Optional[str], optional): Processorの名前。
省略した場合はクラス名が使用される。Defaults to None.
Note:
登録されたProcessorは、アプリケーションのライフサイクルを通じて有効です。
"""
self.publish(DefaultProcessorTopic.REGISTER_PROCESSOR, proc=proc, name=name)
def pub_delete_processor(self, name: str) -> None:
"""指定した名前のProcessorを削除するPubSubメッセージを送信する。
Args:
name (str): 削除するProcessorの名前
"""
self.publish(DefaultProcessorTopic.DELETE_PROCESSOR, name=name)
# --- Undo/Redo ---------------------------------------------------------
def pub_enable_undo_redo(self, state_path: str, max_history: int = 10) -> None:
"""指定したstate pathに対してUndo/Redo機能を有効化するPubSubメッセージを送信する。
Args:
state_path (str): Undo/Redo対象の状態パス（例: "counter", "user.name"）
max_history (int, optional): 保持する履歴の最大数。デフォルトは10。
メモリ使用量を制御したい場合に調整してください。
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_enable_undo_redo(str(self.store.state.counter), max_history=50)`
The state proxy provides autocomplete and "Go to Definition" functionality.
このメソッドを呼び出すと、指定されたパスの現在の値が初期スナップショットとして
履歴に記録され、以降の変更が追跡されます。
"""
self.publish(
DefaultUndoTopic.ENABLE_UNDO_REDO,
state_path=str(state_path),
max_history=max_history,
)
def pub_disable_undo_redo(self, state_path: str) -> None:
"""指定したstate pathのUndo/Redo機能を無効化するPubSubメッセージを送信する。
Args:
state_path (str): 無効化する状態パス
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_disable_undo_redo(str(self.store.state.counter))`
このメソッドを呼び出すと、指定されたパスの履歴データが完全に削除され、
メモリが解放されます。再度有効化したい場合はpub_enable_undo_redoを
呼び出してください。
"""
self.publish(DefaultUndoTopic.DISABLE_UNDO_REDO, state_path=str(state_path))
def pub_undo(self, state_path: str) -> None:
"""指定したstate pathの状態を1つ前の値に戻すPubSubメッセージを送信する。
Args:
state_path (str): Undoを実行する状態パス
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_undo(str(self.store.state.counter))`
履歴が存在しない場合や、既に最初の状態の場合は何も実行されません。
Undoされた変更はRedoで元に戻すことができます。
"""
self.publish(DefaultUndoTopic.UNDO, state_path=str(state_path))
def pub_redo(self, state_path: str) -> None:
"""指定したstate pathのUndoを取り消すPubSubメッセージを送信する。
Args:
state_path (str): Redoを実行する状態パス
Note:
**RECOMMENDED**: Use store.state proxy for type-safe paths with IDE support:
`self.pub_redo(str(self.store.state.counter))`
Redo可能な履歴が存在しない場合は何も実行されません。
新しい変更が行われるとRedo履歴はクリアされます。
"""
self.publish(DefaultUndoTopic.REDO, state_path=str(state_path))
def sub_undo_status(
self, state_path: str, handler: Callable[[bool, bool, int, int], None]
) -> None:
"""
指定した state path の Undo/Redo 状態変化を購読する。
ハンドラー関数は以下の引数で呼び出されます：
- can_undo (bool): Undo 実行可能かどうか
- can_redo (bool): Redo 実行可能かどうか
- undo_count (int): 実行可能な Undo 回数
- redo_count (int): 実行可能な Redo 回数
Args:
state_path (str): 監視する状態パス
handler (Callable[[bool, bool, int, int], None]): 状態変化時に呼び出される関数
Note:
推奨: store.state プロキシを利用すると IDE 補完や一貫したパス指定が可能です。
例: `self.sub_undo_status(self.store.state.counter, self.on_undo_status_changed)`
"""
self.subscribe(f"{DefaultUndoTopic.STATUS_CHANGED}.{str(state_path)}", handler)
def sub_state_changed(
self, state_path: str, handler: Callable[[Any, Any], None]
) -> None:
"""
状態が変更されたときの通知を購読する。
ハンドラー関数は以下の引数で呼び出されます：
- old_value (Any): 変更前の値
- new_value (Any): 変更後の値
Args:
state_path (str): 監視する状態のパス（例: "user.name", "items[2].value"）
handler (Callable[[Any, Any], None]): 変更時に呼び出される関数
Note:
推奨: store.state プロキシを利用して一貫したパス指定が可能です。
例: `self.sub_state_changed(self.store.state.user.name, self.on_name_changed)`
"""
self.subscribe(f"{DefaultUpdateTopic.STATE_CHANGED}.{str(state_path)}", handler)
def sub_for_refresh(self, state_path: str, handler: Callable[[], None]) -> None:
"""
状態が更新されたときのシンプルな通知（UI再描画用）を購読する。
ハンドラー関数は以下の引数で呼び出されます：
- なし
Args:
state_path (str): 監視する状態のパス
handler (Callable[[], None]): 更新時に呼び出される引数なしの関数
Note:
推奨: store.state プロキシを利用して一貫したパス指定が可能です。
例: `self.sub_for_refresh(self.store.state.user.name, self.refresh_ui)`
"""
self.subscribe(f"{DefaultUpdateTopic.STATE_UPDATED}.{str(state_path)}", handler)
def sub_state_added(
self, state_path: str, handler: Callable[[Any, int], None]
) -> None:
"""
リストに要素が追加されたときの通知を購読する。
ハンドラー関数は以下の引数で呼び出されます：
- item (Any): 追加されたアイテム
- index (int): 追加されたインデックス
Args:
state_path (str): 監視するリスト状態のパス
handler (Callable[[Any, int], None]): 要素追加時に呼び出される関数
Note:
推奨: store.state プロキシを利用して一貫したパス指定が可能です。
例: `self.sub_state_added(self.store.state.items, self.on_item_added)`
"""
self.subscribe(f"{DefaultUpdateTopic.STATE_ADDED}.{str(state_path)}", handler)
def sub_dict_item_added(
self, state_path: str, handler: Callable[[str, Any], None]
) -> None:
"""
辞書に要素が追加されたときの通知を購読する。
ハンドラー関数は以下の引数で呼び出されます：
- key (str): 追加されたキー
- value (Any): 追加された値
Args:
state_path (str): 監視する辞書状態のパス
handler (Callable[[str, Any], None]): 追加されたキーと値を引数に取る関数
Note:
推奨: store.state プロキシを利用して一貫したパス指定が可能です。
例: `self.sub_dict_item_added(self.store.state.mapping, self.on_added)`
"""
self.subscribe(
f"{DefaultUpdateTopic.DICT_ADDED}.{str(state_path)}",
handler,
)
トピックシステム¶
src/pubsubtk/topic/topics.py¶
PubSub トピック列挙型の定義
# topics.py - PubSub トピック列挙型の定義
"""
src/pubsubtk/topic/topics.py
アプリケーションで使用する PubSub トピック列挙型を提供します。
"""
from enum import StrEnum, auto
class AutoNamedTopic(StrEnum):
"""
Enumメンバー名を自動で小文字化し、クラス名のプレフィックス付き文字列を値とする列挙型。
- メンバー値は "ClassName.member" 形式の文字列
- str()や比較でそのまま利用可能
"""
def _generate_next_value_(name, start, count, last_values):
return name.lower()
def __new__(cls, value):
# ここでクラス名プレフィックスを追加
full = f"{cls.__name__}.{value}"
obj = str.__new__(cls, full)
obj._value_ = full
return obj
def __str__(self):
return self.value
class DefaultNavigateTopic(AutoNamedTopic):
"""
標準的な画面遷移・ウィンドウ操作用のPubSubトピック列挙型。
"""
SWITCH_CONTAINER = auto()
SWITCH_SLOT = auto()
OPEN_SUBWINDOW = auto()
CLOSE_SUBWINDOW = auto()
CLOSE_ALL_SUBWINDOWS = auto()
class DefaultUpdateTopic(AutoNamedTopic):
"""
標準的な状態更新通知用のPubSubトピック列挙型。
"""
UPDATE_STATE = auto()
ADD_TO_LIST = auto()
ADD_TO_DICT = auto()
REPLACE_STATE = auto()
STATE_CHANGED = auto()
STATE_ADDED = auto()
STATE_UPDATED = auto()
DICT_ADDED = auto()
class DefaultProcessorTopic(AutoNamedTopic):
"""
標準的なプロセッサ管理のPubSubトピック列挙型。
"""
REGISTER_PROCESSOR = auto()
DELETE_PROCESSOR = auto()
class DefaultUndoTopic(AutoNamedTopic):
"""
Undo/Redo機能用のPubSubトピック列挙型。
"""
ENABLE_UNDO_REDO = auto()
DISABLE_UNDO_REDO = auto()
UNDO = auto()
REDO = auto()
STATUS_CHANGED = auto()
State管理¶
src/pubsubtk/store/store.py¶
Pydantic モデルを用いた型安全な状態管理
# store.py - アプリケーション状態を管理するクラス
"""
src/pubsubtk/store/store.py
Pydantic モデルを用いた型安全な状態管理を提供します。
"""
import copy
from collections import defaultdict
from typing import Any, Generic, Type, TypeVar, cast
from pydantic import BaseModel
from pubsubtk.core.pubsub_base import PubSubBase
from pubsubtk.topic.topics import DefaultUndoTopic, DefaultUpdateTopic
TState = TypeVar("TState", bound=BaseModel)
class StateProxy(Generic[TState]):
"""
Storeのstate属性に対する動的なパスアクセスを提供するプロキシ。
- store.state.foo.bar のようなドット記法でネスト属性へアクセス可能
- 存在しない属性アクセス時は AttributeError を送出
- __repr__ でパス文字列を返す
"""
def __init__(self, store: "Store[TState]", path: str = ""):
"""StateProxy を初期化する。
Args:
store: 値を参照する対象 ``Store``。
path: 現在のパス文字列。
"""
self._store = store
self._path = path
def __getattr__(self, name: str) -> "StateProxy[TState]":
"""属性アクセスを連結した ``StateProxy`` を返す。"""
new_path = f"{self._path}.{name}" if self._path else name
# 存在チェック：TState モデルに new_path が通るか確認
cur = self._store.get_current_state()
for seg in new_path.split("."):
if hasattr(cur, seg):
cur = getattr(cur, seg)
else:
raise AttributeError(f"No such property: store.state.{new_path}")
return StateProxy(self._store, new_path)
def __repr__(self) -> str:
"""State型名を含むパス文字列を返す。"""
prefix = self._store._state_class.__name__
return f"{prefix}.{self._path}" if self._path else prefix
__str__ = __repr__
class Store(PubSubBase, Generic[TState]):
"""
型安全な状態管理を提供するジェネリックなStoreクラス。
- Pydanticモデルを状態として保持し、状態操作を提供
- get_current_state()で状態のディープコピーを取得
- update_state()/add_to_list()/add_to_dict()で状態を更新し、PubSubで通知
- `store.state.count` のようなパスプロキシを使うことで、
`store.update_state(store.state.count, 1)` のようにIDEの「定義へ移動」や補完機能を活用しつつ、
状態更新のパスを安全・明示的に指定できる（従来の文字列パス指定の弱点を解消）
"""
def __init__(self, initial_state_class: Type[TState]):
"""Store を初期化する。
Args:
initial_state_class: 管理対象となる ``BaseModel`` のサブクラス。
"""
self._state_class = initial_state_class
self._state = initial_state_class()
# Undo/Redo 履歴管理用フィールド
self._undo_enabled: set[str] = set() # 追跡対象パス
self._undo_stacks: dict[str, list] = defaultdict(list) # パス別Undoスタック
self._redo_stacks: dict[str, list] = defaultdict(list) # パス別Redoスタック
self._max_histories: dict[str, int] = {} # パス別履歴上限
self._during_ur_op: bool = False # Undo/Redo操作中の再帰抑制フラグ
# PubSubBase.__init__()を呼び出して購読設定を有効化
super().__init__()
def setup_subscriptions(self):
# 既存の状態更新系トピック
self.subscribe(DefaultUpdateTopic.UPDATE_STATE, self.update_state)
self.subscribe(DefaultUpdateTopic.REPLACE_STATE, self.replace_state)
self.subscribe(DefaultUpdateTopic.ADD_TO_LIST, self.add_to_list)
self.subscribe(DefaultUpdateTopic.ADD_TO_DICT, self.add_to_dict)
# Undo/Redo系トピック
self.subscribe(DefaultUndoTopic.ENABLE_UNDO_REDO, self._enable_undo_redo)
self.subscribe(DefaultUndoTopic.DISABLE_UNDO_REDO, self._disable_undo_redo)
self.subscribe(DefaultUndoTopic.UNDO, self._undo)
self.subscribe(DefaultUndoTopic.REDO, self._redo)
@property
def state(self) -> TState:
"""
状態への動的パスアクセス用プロキシを返す。
"""
return cast(TState, StateProxy(self))
def get_current_state(self) -> TState:
"""
現在の状態のディープコピーを返す。
"""
return self._state.model_copy(deep=True)
def replace_state(self, new_state: TState) -> None:
"""状態オブジェクト全体を置き換え、全フィールドに変更通知を送信する。
Args:
new_state: 新しい状態オブジェクト。
"""
if not isinstance(new_state, self._state_class):
raise TypeError(f"new_state must be an instance of {self._state_class}")
old_state = self._state
self._state = new_state.model_copy(deep=True)
# 全フィールドに変更通知を送信
for field_name in self._state_class.model_fields.keys():
old_value = getattr(old_state, field_name)
new_value = getattr(self._state, field_name)
self.publish(
f"{DefaultUpdateTopic.STATE_CHANGED}.{field_name}",
old_value=old_value,
new_value=new_value,
)
self.publish(f"{DefaultUpdateTopic.STATE_UPDATED}.{field_name}")
def update_state(self, state_path: str, new_value: Any) -> None:
"""指定パスの属性を更新し、変更通知を送信する。
Args:
state_path: 変更対象の属性パス（例: ``"foo.bar"``）。
new_value: 新しく設定する値。
"""
try:
target_obj, attr_name, old_value = self._resolve_path(str(state_path))
except ValueError:
return
# Undo履歴をキャプチャ（既存の値を記録）
self._capture_for_undo(str(state_path), old_value)
# 新しい値を設定する前に型チェック
self._validate_and_set_value(target_obj, attr_name, new_value)
# 詳細な変更通知（old_value, new_valueを含む）
self.publish(
f"{DefaultUpdateTopic.STATE_CHANGED}.{state_path}",
old_value=old_value,
new_value=new_value,
)
# シンプルな更新通知（引数なし）
self.publish(f"{DefaultUpdateTopic.STATE_UPDATED}.{state_path}")
def add_to_list(self, state_path: str, item: Any) -> None:
"""リスト属性に要素を追加し、追加通知を送信する。
Args:
state_path: 追加先となるリストの属性パス。
item: 追加する要素。
"""
try:
target_obj, attr_name, current_list = self._resolve_path(str(state_path))
except ValueError:
return
if not isinstance(current_list, list):
raise TypeError(f"Property at '{state_path}' is not a list")
# Undo履歴をキャプチャ（既存のリストを記録）
self._capture_for_undo(str(state_path), current_list)
# リストをコピーして新しい要素を追加
new_list = current_list.copy()
new_list.append(item)
# 新しいリストで更新
self._validate_and_set_value(target_obj, attr_name, new_list)
index = len(new_list) - 1
self.publish(
f"{DefaultUpdateTopic.STATE_ADDED}.{state_path}",
item=item,
index=index,
)
# リスト追加でも更新通知を送信
self.publish(f"{DefaultUpdateTopic.STATE_UPDATED}.{state_path}")
def add_to_dict(self, state_path: str, key: str, value: Any) -> None:
"""辞書属性に要素を追加し、追加通知を送信する。
Args:
state_path: 追加先となる辞書の属性パス。
key: 追加するキー。
value: 追加する値。
"""
try:
target_obj, attr_name, current_dict = self._resolve_path(str(state_path))
except ValueError:
return
if not isinstance(current_dict, dict):
raise TypeError(f"Property at '{state_path}' is not a dict")
# Undo履歴をキャプチャ（既存の辞書を記録）
self._capture_for_undo(str(state_path), current_dict)
new_dict = current_dict.copy()
new_dict[key] = value
self._validate_and_set_value(target_obj, attr_name, new_dict)
self.publish(
f"{DefaultUpdateTopic.DICT_ADDED}.{state_path}",
key=key,
value=value,
)
# 辞書追加でも更新通知を送信
self.publish(f"{DefaultUpdateTopic.STATE_UPDATED}.{state_path}")
# --- Undo/Redo 履歴管理機能 ---
def _enable_undo_redo(self, state_path: str, max_history: int = 10) -> None:
"""指定パスのUndo/Redo機能を有効化し、スタックを作成する。
Args:
state_path: 追跡対象の状態パス
max_history: 保持する履歴の最大数（デフォルト: 10）
"""
if not self._normalize_state_path(state_path)[1]:
return
self._undo_enabled.add(state_path)
self._max_histories[state_path] = max_history
# スタック作成
self._undo_stacks[state_path] = []
self._redo_stacks[state_path].clear()
# ステータス通知を送信
self._emit_ur_status(state_path)
def _disable_undo_redo(self, state_path: str) -> None:
"""指定パスのUndo/Redo機能を無効化し、履歴データを削除する。
Args:
state_path: 無効化する状態パス
"""
if not self._normalize_state_path(state_path)[1]:
return
self._undo_enabled.discard(state_path)
self._undo_stacks.pop(state_path, None)
self._redo_stacks.pop(state_path, None)
self._max_histories.pop(state_path, None)
def _capture_for_undo(self, state_path: str, old_value: Any) -> None:
"""状態変更前に古い値をUndo履歴に記録する。
Args:
state_path: 変更対象の状態パス
old_value: 変更前の値
"""
# Undo/Redo対象でない、またはUndo/Redo操作中の場合はスキップ
if state_path not in self._undo_enabled or self._during_ur_op:
return
stack = self._undo_stacks[state_path]
stack.append(copy.deepcopy(old_value))
# 履歴上限の管理
max_len = self._max_histories.get(state_path, 10)
if len(stack) > max_len:
stack.pop(0) # 最古の履歴を削除
# 新しい変更が発生したのでRedo履歴をクリア
self._redo_stacks[state_path].clear()
# ステータス通知を送信
self._emit_ur_status(state_path)
def _undo(self, state_path: str) -> None:
"""指定パスの状態を 1 つ前の値に戻す。"""
if not self._normalize_state_path(state_path)[1]:
return
if state_path not in self._undo_enabled:
return
undo_stack = self._undo_stacks[state_path]
if len(undo_stack) < 1:
return
# 現在の値を Redo スタックへ退避
try:
_, _, current_value = self._resolve_path(state_path)
self._redo_stacks[state_path].append(copy.deepcopy(current_value))
except (AttributeError, ValueError):
return
# pop() した値こそ「戻すべき直前値」
self._during_ur_op = True
try:
previous_value = undo_stack.pop()
self.update_state(state_path, previous_value)
finally:
self._during_ur_op = False
self._emit_ur_status(state_path)
def _redo(self, state_path: str) -> None:
"""指定パスのUndoを取り消し、Redoを実行する。
Args:
state_path: Redoを実行する状態パス
"""
if not self._normalize_state_path(state_path)[1]:
return
if state_path not in self._undo_enabled:
return
redo_stack = self._redo_stacks[state_path]
if not redo_stack:
return
# 現在の値をUndo履歴に保存
try:
_, _, current_value = self._resolve_path(state_path)
self._undo_stacks[state_path].append(copy.deepcopy(current_value))
except (AttributeError, ValueError):
return
# Redo値を取得して適用
self._during_ur_op = True # 再帰防止フラグを設定
try:
redo_value = redo_stack.pop()
self.update_state(state_path, redo_value)
finally:
self._during_ur_op = False
# ステータス通知を送信
self._emit_ur_status(state_path)
def _emit_ur_status(self, state_path: str) -> None:
"""現在のUndo/Redo可否・スタックサイズを通知する。
Args:
state_path: ステータス通知対象の状態パス
"""
undo_stack = self._undo_stacks.get(state_path, [])
redo_stack = self._redo_stacks.get(state_path, [])
self.publish(
f"{DefaultUndoTopic.STATUS_CHANGED}.{state_path}",
can_undo=len(undo_stack) > 0,
can_redo=len(redo_stack) > 0,
undo_count=max(len(undo_stack), 0),
redo_count=len(redo_stack),
)
def _normalize_state_path(self, state_path: str) -> tuple[str, bool]:
"""State型名プレフィックスを処理して自ストア向けか判定する。
Args:
state_path: 入力された状態パス。
Returns:
正規化後のパスと自ストア向けかどうかのフラグ。
"""
segments = state_path.split(".")
if not segments:
return "", True
first = segments[0]
if first == self._state_class.__name__:
return ".".join(segments[1:]), True
for cls in _stores.keys():
if first == cls.__name__ and cls is not self._state_class:
return "", False
return state_path, True
def _resolve_path(self, path: str) -> tuple[Any, str, Any]:
"""
属性パスを解決し、対象オブジェクト・属性名・現在値を返す。
Args:
path: 解析する属性パス。
Returns:
(対象オブジェクト, 属性名, 現在値)
"""
path, available = self._normalize_state_path(path)
if not available:
raise ValueError("Path does not belong to this store")
segments = path.split(".")
if not segments:
raise ValueError("Empty path")
# 最後のセグメントを取り出し
attr_name = segments[-1]
# 最後のセグメント以外のパスをたどって対象オブジェクトを取得
current = self._state
for segment in segments[:-1]:
if not hasattr(current, segment):
raise AttributeError(f"No such attribute: {segment} in path {path}")
current = getattr(current, segment)
# 現在の値を取得
if not hasattr(current, attr_name):
raise AttributeError(f"No such attribute: {attr_name} in path {path}")
old_value = getattr(current, attr_name)
return current, attr_name, old_value
def _validate_and_set_value(
self, target_obj: Any, attr_name: str, new_value: Any
) -> None:
"""属性値を型検証してから設定する。
Args:
target_obj: 値を設定する対象オブジェクト。
attr_name: 設定する属性名。
new_value: 新しい値。
"""
# Pydanticモデルの場合、フィールドの型情報を取得
if isinstance(target_obj, BaseModel):
model_fields = target_obj.__class__.model_fields
if attr_name in model_fields:
field_info = model_fields[attr_name]
# もし新しい値がPydanticモデルの場合、model_validateを使用
if hasattr(new_value, "model_dump") and hasattr(
field_info.annotation, "model_validate"
):
field_type = field_info.annotation
validated_value = field_type.model_validate(new_value)
setattr(target_obj, attr_name, validated_value)
return
# 通常の属性設定
setattr(target_obj, attr_name, new_value)
# State 型ごとに生成した Store を保持する辞書
_stores: dict[Type[BaseModel], Store[Any]] = {}
def get_store(state_cls: Type[TState]) -> Store[TState]:
"""指定された ``state_cls`` 用の ``Store`` インスタンスを返す。
同じ ``state_cls`` に対しては常に同じ ``Store`` を返し、
初回呼び出し時のみ生成して内部で保持する。
Args:
state_cls: ``Store`` 生成に使用する状態モデルの型。
Returns:
``state_cls`` に対応する ``Store`` インスタンス。
"""
if state_cls not in _stores:
_stores[state_cls] = Store(state_cls)
return cast(Store[TState], _stores[state_cls])
アプリケーションクラス¶
src/pubsubtk/app/application_base.py¶
Tkinter アプリケーション向けの共通基底クラス
# application_base.py - アプリケーションの基底クラスを定義
"""Tkinter アプリケーション向けの共通基底クラスを提供します。
このモジュールでは、Tk および ttk ベースのアプリケーション構築時に
利用する共通メソッドをまとめています。``TkApplication`` と
``ThemedApplication`` の 2 種類のウィンドウクラスを公開しており、
いずれも ``ApplicationCommon`` Mixin を継承して Pub/Sub 機能と
状態管理機能を自動的に組み込みます。
"""
from __future__ import annotations
import asyncio
import tkinter as tk
from typing import TYPE_CHECKING, Dict, Generic, Optional, Tuple, Type, TypeVar
from pydantic import BaseModel
from ttkthemes import ThemedTk
from pubsubtk.core.default_topic_base import PubSubDefaultTopicBase
from pubsubtk.processor.processor_base import ProcessorBase
from pubsubtk.store.store import get_store
from pubsubtk.topic.topics import DefaultNavigateTopic, DefaultProcessorTopic
from pubsubtk.ui.base.container_base import ContainerMixin
from pubsubtk.ui.base.template_base import TemplateMixin
if TYPE_CHECKING:
from pubsubtk.ui.types import (
ComponentType,
ContainerComponentType,
TemplateComponentType,
)
TState = TypeVar("TState", bound=BaseModel)
P = TypeVar("P", bound=ProcessorBase)
def _default_poll(loop: asyncio.AbstractEventLoop, root: tk.Tk, interval: int) -> None:
"""非同期イベントループを ``after`` で定期実行する補助関数。
Args:
loop: 実行対象の ``AbstractEventLoop`` インスタンス。
root: ``after`` を呼び出す Tk ウィジェット（通常はアプリケーション本体）。
interval: ポーリング間隔（ミリ秒）。
"""
try:
loop.call_soon(loop.stop)
loop.run_forever()
except Exception:
pass
root.after(interval, _default_poll, loop, root, interval)
class ApplicationCommon(PubSubDefaultTopicBase, Generic[TState]):
"""Tk/Ttk いずれのウィンドウクラスでも共通の機能を提供する Mixin."""
def __init__(self, state_cls: Type[TState], *args, **kwargs):
"""状態クラスを受け取り、Pub/Sub 機能を初期化する。
Args:
state_cls: アプリケーション状態を表す ``BaseModel`` のサブクラス。
"""
super().__init__(*args, **kwargs)
self.state_cls = state_cls
self.store = get_store(state_cls)
self._processors: Dict[str, ProcessorBase] = {}
def init_common(self, title: str, geometry: str) -> None:
"""ウィンドウタイトルやメインフレームを設定する共通初期化処理。
Args:
title: ウィンドウタイトル。
geometry: ``WIDTHxHEIGHT`` 形式のウィンドウサイズ文字列。
"""
# ウィンドウ基本設定
self.title(title)
self.geometry(geometry)
# コンテナ & アクティブウィジェット
self.main_frame = tk.Frame(self)
self.main_frame.pack(fill=tk.BOTH, expand=True)
self.active: Optional[tk.Widget] = None
# サブウィンドウ管理用辞書
self._subwindows: Dict[str, Tuple[tk.Toplevel, tk.Widget]] = {}
def setup_subscriptions(self) -> None:
"""PubSub の購読設定を行う。
``PubSubBase.__init__`` から自動で呼び出されるメソッドで、
ナビゲーションや Processor 管理に関するトピックを購読します。
"""
self.subscribe(DefaultNavigateTopic.SWITCH_CONTAINER, self.switch_container)
self.subscribe(DefaultNavigateTopic.SWITCH_SLOT, self.switch_slot)
self.subscribe(DefaultNavigateTopic.OPEN_SUBWINDOW, self.open_subwindow)
self.subscribe(DefaultNavigateTopic.CLOSE_SUBWINDOW, self.close_subwindow)
self.subscribe(
DefaultNavigateTopic.CLOSE_ALL_SUBWINDOWS, self.close_all_subwindows
)
self.subscribe(
DefaultProcessorTopic.REGISTER_PROCESSOR, self.register_processor
)
self.subscribe(DefaultProcessorTopic.DELETE_PROCESSOR, self.delete_processor)
def _create_component(
self, cls: ComponentType, parent: tk.Widget, kwargs: dict = None
) -> tk.Widget:
"""コンポーネントを種類に応じて生成する共通メソッド。
Args:
cls: コンポーネントのクラス。
parent: 親ウィジェット。
kwargs: コンポーネント初期化用パラメータ辞書。
Returns:
生成したウィジェットインスタンス。
"""
kwargs = kwargs or {}
# ContainerMixinを継承しているかチェック
is_container = issubclass(cls, ContainerMixin)
if is_container:
# Containerの場合はstoreを渡す
return cls(parent=parent, store=self.store, **kwargs)
else:
# Presentationalの場合はstoreなし
return cls(parent=parent, **kwargs)
def register_processor(self, proc: Type[P], name: Optional[str] = None) -> str:
"""
プロセッサを名前で登録し、登録キーを返します。
Args:
proc: ProcessorBaseを継承したクラス
name: 任意のプロセッサ名。未指定時はクラス名を使用し、重複する場合は接尾辞を追加します。
Returns:
登録に使用したプロセッサ名。
Raises:
KeyError: 既に同名のプロセッサが登録済みの場合。
"""
# ベース名決定
base_key = name or proc.__name__
key = base_key
suffix = 1
# 重複を回避
while key in self._processors:
key = f"{base_key}_{suffix}"
suffix += 1
# インスタンス化して登録
self._processors[key] = proc(store=self.store)
return key
def delete_processor(self, name: str) -> None:
"""登録済みプロセッサを削除し ``teardown`` を実行する。"""
if name not in self._processors:
raise KeyError(f"Processor '{name}' not found.")
self._processors[name].teardown()
del self._processors[name]
def set_template(self, template_cls: TemplateComponentType) -> None:
"""アプリケーションにテンプレートを設定する。
Args:
template_cls: 適用する ``TemplateComponent`` のクラス。
"""
if self.active:
self.active.destroy()
self.active = template_cls(parent=self.main_frame, store=self.store)
self.active.pack(fill=tk.BOTH, expand=True)
def switch_container(
self,
cls: ContainerComponentType,
kwargs: dict = None,
) -> None:
"""メインフレーム内のコンテナを切り替える。
テンプレートが設定されている場合は ``switch_slot`` を使用して
デフォルトスロットのコンテンツを置き換えます。
Args:
cls: 切り替え先のコンテナクラス。
kwargs: コンテナ初期化用のキーワード引数辞書。
"""
# テンプレートが設定されている場合
if self.active and isinstance(self.active, TemplateMixin):
# デフォルトスロット（"main" または "content"）を探す
slots = self.active.get_slots()
if "main" in slots:
self.active.switch_slot_content("main", cls, kwargs)
elif "content" in slots:
self.active.switch_slot_content("content", cls, kwargs)
else:
# デフォルトスロットがない場合は最初のスロットを使用
if slots:
first_slot = list(slots.keys())[0]
self.active.switch_slot_content(first_slot, cls, kwargs)
else:
raise RuntimeError("Template has no slots defined")
else:
# 通常のコンテナ切り替え
if self.active:
self.active.destroy()
kwargs = kwargs or {}
self.active = self._create_component(cls, self.main_frame, kwargs)
self.active.pack(fill=tk.BOTH, expand=True)
def switch_slot(
self,
slot_name: str,
cls: ComponentType,
kwargs: dict = None,
) -> None:
"""テンプレートの特定スロットのコンテンツを切り替える。
Args:
slot_name: 変更対象のスロット名。
cls: 新しく配置するコンポーネントクラス。
kwargs: コンポーネント初期化用のキーワード引数辞書。
"""
if not self.active or not isinstance(self.active, TemplateMixin):
raise RuntimeError("No template is set. Use set_template() first.")
self.active.switch_slot_content(slot_name, cls, kwargs)
def open_subwindow(
self,
cls: ComponentType,
win_id: Optional[str] = None,
kwargs: dict = None,
) -> str:
"""サブウィンドウを開き、生成したウィンドウ ID を返す。
Args:
cls: 表示するコンポーネントクラス。
win_id: 任意のウィンドウ ID。指定しない場合は自動生成される。
kwargs: コンポーネント初期化用のキーワード引数辞書。
Returns:
実際に使用されたウィンドウ ID。
"""
# 既存IDであれば前面に
if win_id and win_id in self._subwindows:
self._subwindows[win_id][0].lift()
return win_id
# キー生成
base_id = win_id or cls.__name__
unique_id = base_id
suffix = 1
while unique_id in self._subwindows:
unique_id = f"{base_id}_{suffix}"
suffix += 1
# ウィンドウ生成
toplevel = tk.Toplevel(self)
kwargs = kwargs or {}
kwargs["win_id"] = unique_id
# 共通メソッドを使用
comp = self._create_component(cls, toplevel, kwargs)
comp.pack(fill=tk.BOTH, expand=True)
def on_close():
self.close_subwindow(unique_id)
toplevel.protocol("WM_DELETE_WINDOW", on_close)
self._subwindows[unique_id] = (toplevel, comp)
return unique_id
def close_subwindow(self, win_id: str) -> None:
"""指定 ID のサブウィンドウを閉じる。"""
if win_id not in self._subwindows:
return
top, comp = self._subwindows.pop(win_id)
try:
comp.destroy()
except Exception:
pass
top.destroy()
def close_all_subwindows(self) -> None:
"""開いているすべてのサブウィンドウを閉じる。"""
for wid in list(self._subwindows):
self.close_subwindow(wid)
def run(
self,
use_async: bool = False,
loop: Optional[asyncio.AbstractEventLoop] = None,
poll_interval: int = 50,
) -> None:
"""アプリケーションのメインループを開始する。
Args:
use_async: ``asyncio`` を併用するかどうか。
loop: 使用するイベントループ。``None`` の場合は ``get_event_loop`` を使用。
poll_interval: ``_default_poll`` を呼び出す間隔（ミリ秒）。
"""
self.protocol("WM_DELETE_WINDOW", self.on_closing)
if not use_async:
self.mainloop()
else:
loop = loop or asyncio.get_event_loop()
self.after(poll_interval, _default_poll, loop, self, poll_interval)
self.mainloop()
try:
loop.run_until_complete(loop.shutdown_asyncgens())
except Exception:
pass
def on_closing(self) -> None:
"""終了時のクリーンアップ処理を行う。
すべてのサブウィンドウを閉じて ``destroy`` を呼び出す。
"""
self.close_all_subwindows()
self.destroy()
class TkApplication(ApplicationCommon[TState], tk.Tk, Generic[TState]):
def __init__(
self,
state_cls: Type[TState],
title: str = "Tk App",
geometry: str = "800x600",
*args,
**kwargs,
):
"""Tk ベースのアプリケーションを初期化する。
Args:
state_cls: アプリケーション状態モデルの型。
title: ウィンドウタイトル。
geometry: ``WIDTHxHEIGHT`` 形式のウィンドウサイズ。
"""
# **first** initialize the actual Tk
tk.Tk.__init__(self, *args, **kwargs)
# **then** initialize the PubSub mixin
ApplicationCommon.__init__(self, state_cls)
# now do your common window setup
self.init_common(title, geometry)
class ThemedApplication(ApplicationCommon[TState], ThemedTk, Generic[TState]):
def __init__(
self,
state_cls: Type[TState],
theme: str = "arc",
title: str = "Themed App",
geometry: str = "800x600",
*args,
**kwargs,
):
"""テーマ対応アプリケーションを初期化する。
Args:
state_cls: アプリケーション状態モデルの型。
theme: 適用する ttk テーマ名。
title: ウィンドウタイトル。
geometry: ``WIDTHxHEIGHT`` 形式のウィンドウサイズ。
"""
# initialize the themed‐Tk
ThemedTk.__init__(self, theme=theme, *args, **kwargs)
# mixin init
ApplicationCommon.__init__(self, state_cls)
# then common setup
self.init_common(title, geometry)
UIコンポーネント¶
src/pubsubtk/ui/base/container_base.py¶
状態連携可能な UI コンテナの基底クラス
"""
src/pubsubtk/ui/base/container_base.py
状態連携可能な UI コンテナの基底クラスを定義します。
"""
import tkinter as tk
from abc import ABC, abstractmethod
from tkinter import ttk
from typing import Any, Generic, TypeVar
from pydantic import BaseModel
from pubsubtk.core.default_topic_base import PubSubDefaultTopicBase
from pubsubtk.store.store import Store
TState = TypeVar("TState", bound=BaseModel)
class ContainerMixin(PubSubDefaultTopicBase, ABC, Generic[TState]):
"""
PubSub連携用のコンテナコンポーネントMixin。
- Storeインスタンスを取得し、購読設定・状態反映を自動実行
- setup_subscriptions()/refresh_from_state()をサブクラスで実装
- destroy時に購読解除(teardown)も自動
**IMPORTANT**: Use built-in pub_* methods for state updates instead of
manually publishing to topics. This provides better IDE support and consistency.
"""
def __init__(self, store: Store[TState], *args, **kwargs: Any):
"""コンテナの初期化を行う。
Args:
store: 使用する ``Store`` インスタンス。
Notes:
渡された ``*args`` と ``**kwargs`` は ``self.args`` / ``self.kwargs``
として保持されます。サブウィンドウを ``open_subwindow`` で開く場合は
``win_id`` が ``self.kwargs`` に自動追加され、
``pub_close_subwindow(self.kwargs["win_id"])`` として自身を閉じることが
できます。将来的に同様のデフォルト引数が増えるかもしれません。
"""
self.args = args
self.kwargs = kwargs
# 型引数付きの Store[TState] を取得
self.store: Store[TState] = store
super().__init__(*args, **kwargs)
self.setup_ui()
self.refresh_from_state()
@abstractmethod
def setup_ui(self) -> None:
"""
ウィジェット構築とレイアウトを行うメソッド。
サブクラスで実装する。
"""
...
@abstractmethod
def refresh_from_state(self) -> None:
"""
購読通知または初期化時にUIを状態で更新するメソッド。
サブクラスで実装する。
"""
...
def destroy(self) -> None:
"""
ウィジェット破棄時に購読を解除してから破棄処理を行う。
"""
self.teardown()
super().destroy()
class ContainerComponentTk(ContainerMixin[TState], tk.Frame, Generic[TState]):
"""
標準tk.FrameベースのPubSub連携コンテナ。
"""
def __init__(self, parent: tk.Widget, store: Store[TState], *args, **kwargs: Any):
"""tk.Frame ベースのコンテナを初期化する。
Args:
parent: 親ウィジェット。
store: 使用する ``Store`` インスタンス。
"""
tk.Frame.__init__(self, master=parent)
ContainerMixin.__init__(self, store=store, *args, **kwargs)
class ContainerComponentTtk(ContainerMixin[TState], ttk.Frame, Generic[TState]):
"""
テーマ対応ttk.FrameベースのPubSub連携コンテナ。
"""
def __init__(self, parent: tk.Widget, store: Store[TState], *args, **kwargs: Any):
"""ttk.Frame ベースのコンテナを初期化する。
Args:
parent: 親ウィジェット。
store: 使用する ``Store`` インスタンス。
"""
ttk.Frame.__init__(self, master=parent)
ContainerMixin.__init__(self, store=store, *args, **kwargs)
src/pubsubtk/ui/base/presentational_base.py¶
イベント発火機能を備えた表示専用 UI コンポーネント基底クラス
"""
src/pubsubtk/ui/base/presentational_base.py
イベント発火機能を備えた表示専用 UI コンポーネント用基底クラス。
"""
import tkinter as tk
from abc import ABC, abstractmethod
from tkinter import ttk
from typing import Any, Callable, Dict
class PresentationalMixin(ABC):
"""
表示専用コンポーネント用のMixin。
- 任意のイベントハンドラ登録・発火機能を持つ
"""
def __init__(self, *args, **kwargs):
"""Mixin の初期化処理。
Notes:
渡された ``*args`` と ``**kwargs`` は ``self.args`` / ``self.kwargs``
として保持されます。サブウィンドウで使用する場合は ``open_subwindow``
が ``win_id`` を自動付与するため、 ``self.kwargs["win_id"]`` を利用して
自身を閉じられます。今後同様のデフォルト引数が追加される可能性があります。
"""
self.args = args
self.kwargs = kwargs
self._handlers: Dict[str, Callable[..., Any]] = {}
self.setup_ui()
@abstractmethod
def setup_ui(self) -> None:
"""
ウィジェット構築とレイアウトを行うメソッド。
サブクラスで実装する。
"""
pass
def register_handler(self, event_name: str, handler: Callable[..., Any]) -> None:
self._handlers[event_name] = handler
def trigger_event(self, event_name: str, **kwargs: Any) -> None:
if handler := self._handlers.get(event_name):
handler(**kwargs)
# tk.Frame ベース の抽象クラス
class PresentationalComponentTk(PresentationalMixin, tk.Frame):
"""
標準tk.Frameベースの表示専用コンポーネント。
"""
def __init__(self, parent: tk.Widget, *args, **kwargs):
"""tk.Frame ベースの表示コンポーネントを初期化する。"""
tk.Frame.__init__(self, master=parent)
PresentationalMixin.__init__(self, *args, **kwargs)
# ttk.Frame ベース の抽象クラス
class PresentationalComponentTtk(PresentationalMixin, ttk.Frame):
"""
テーマ対応ttk.Frameベースの表示専用コンポーネント。
"""
def __init__(self, parent: tk.Widget, *args, **kwargs):
"""ttk.Frame ベースの表示コンポーネントを初期化する。"""
ttk.Frame.__init__(self, master=parent)
PresentationalMixin.__init__(self, *args, **kwargs)
src/pubsubtk/ui/base/template_base.py¶
複数スロットを持つテンプレート UI 基底クラス
# template_base.py - テンプレートコンポーネントの基底クラス
"""複数スロットを持つテンプレート UI を構築するための基底クラス。"""
from __future__ import annotations
import tkinter as tk
from abc import ABC, abstractmethod
from tkinter import ttk
from typing import TYPE_CHECKING, Dict, Generic, TypeVar
from pydantic import BaseModel
from pubsubtk.store.store import Store
from pubsubtk.ui.base.container_base import ContainerMixin
if TYPE_CHECKING:
from pubsubtk.ui.types import ComponentType
TState = TypeVar("TState", bound=BaseModel)
class TemplateMixin(ABC, Generic[TState]):
"""
テンプレートコンポーネント用のMixin。
複数のスロット（区画）を定義し、各スロットに独立してコンポーネントを配置できる。
ヘッダー・フッターなど固定部分と可変部分を分離したレイアウトを実現。
Note:
テンプレート自体は状態を持たず、レイアウト定義とスロット管理のみを行う。
各スロットに配置されるコンポーネントが独自に状態管理を行う。
"""
def __init__(self, store: Store[TState], *args, **kwargs):
"""Mixin の初期化処理。"""
self.store = store
self._slots: Dict[str, tk.Widget] = {}
self._slot_contents: Dict[str, tk.Widget] = {}
# テンプレートのセットアップ
self.setup_template()
self._slots = self.define_slots()
def setup_template(self) -> None:
"""
テンプレート固有の初期化処理（必要に応じてオーバーライド）。
define_slots()の前に呼ばれる。
"""
pass
@abstractmethod
def define_slots(self) -> Dict[str, tk.Widget]:
"""
スロット（区画）を定義する。
Returns:
Dict[str, tk.Widget]: {"スロット名": フレームWidget} の辞書
Example:
# ヘッダー
self.header_frame = tk.Frame(self, height=60, bg='navy')
self.header_frame.pack(fill=tk.X)
# メインコンテンツ
self.main_frame = tk.Frame(self)
self.main_frame.pack(fill=tk.BOTH, expand=True)
# フッター
self.footer_frame = tk.Frame(self, height=30, bg='gray')
self.footer_frame.pack(fill=tk.X)
return {
"header": self.header_frame,
"main": self.main_frame,
"footer": self.footer_frame
}
"""
pass
def switch_slot_content(
self, slot_name: str, cls: ComponentType, kwargs: dict = None
) -> None:
"""
指定スロットのコンテンツを切り替える。
Args:
slot_name: スロット名
cls: コンポーネントクラス（Container/Presentational両対応）
kwargs: コンポーネントに渡す引数
"""
if slot_name not in self._slots:
raise ValueError(f"Unknown slot: {slot_name}")
# 既存のコンテンツを破棄
if slot_name in self._slot_contents:
self._slot_contents[slot_name].destroy()
# 新しいコンテンツを作成
parent_frame = self._slots[slot_name]
content = self._create_component_for_slot(cls, parent_frame, kwargs)
content.pack(fill=tk.BOTH, expand=True)
self._slot_contents[slot_name] = content
def _create_component_for_slot(
self, cls: ComponentType, parent: tk.Widget, kwargs: dict = None
) -> tk.Widget:
"""スロット用のコンポーネント生成"""
kwargs = kwargs or {}
# ContainerMixinを継承しているかチェック
is_container = issubclass(cls, ContainerMixin)
if is_container:
return cls(parent=parent, store=self.store, **kwargs)
else:
return cls(parent=parent, **kwargs)
def get_slots(self) -> Dict[str, tk.Widget]:
"""定義されているスロットの辞書を返す"""
return self._slots.copy()
def get_slot_content(self, slot_name: str) -> tk.Widget | None:
"""指定スロットの現在のコンテンツを返す"""
return self._slot_contents.get(slot_name)
def has_slot(self, slot_name: str) -> bool:
"""指定した名前のスロットが存在するかチェック"""
return slot_name in self._slots
def clear_slot(self, slot_name: str) -> None:
"""指定スロットのコンテンツをクリアする"""
if slot_name in self._slot_contents:
self._slot_contents[slot_name].destroy()
del self._slot_contents[slot_name]
def clear_all_slots(self) -> None:
"""すべてのスロットのコンテンツをクリアする"""
for slot_name in list(self._slot_contents.keys()):
self.clear_slot(slot_name)
class TemplateComponentTk(TemplateMixin[TState], tk.Frame, Generic[TState]):
"""
標準tk.Frameベースのテンプレートコンポーネント。
"""
def __init__(self, parent: tk.Widget, store: Store[TState], *args, **kwargs):
"""tk.Frame ベースのテンプレートを初期化する。"""
tk.Frame.__init__(self, master=parent)
TemplateMixin.__init__(self, store=store, *args, **kwargs)
class TemplateComponentTtk(TemplateMixin[TState], ttk.Frame, Generic[TState]):
"""
テーマ対応ttk.Frameベースのテンプレートコンポーネント。
"""
def __init__(self, parent: tk.Widget, store: Store[TState], *args, **kwargs):
"""ttk.Frame ベースのテンプレートを初期化する。"""
ttk.Frame.__init__(self, master=parent)
TemplateMixin.__init__(self, store=store, *args, **kwargs)
Processorシステム¶
src/pubsubtk/processor/processor_base.py¶
ビジネスロジックを担う Processor 用の抽象基底クラス
# processor_base.py - Processor の基底クラス
"""ビジネスロジックを担う Processor 用の抽象基底クラス。"""
from typing import Generic, TypeVar
from pydantic import BaseModel
from pubsubtk.core.default_topic_base import PubSubDefaultTopicBase
from pubsubtk.store.store import Store
TState = TypeVar("TState", bound=BaseModel)
class ProcessorBase(PubSubDefaultTopicBase, Generic[TState]):
"""Processor の基底クラス。"""
def __init__(self, store: Store[TState], *args, **kwargs) -> None:
"""Store を受け取って初期化します。"""
# 型引数付きの Store[TState] を取得
self.store: Store[TState] = store
super().__init__(*args, **kwargs)
Storybookシステム¶
src/pubsubtk/storybook/app.py¶
Storybookアプリケーションクラス
# storybook/app.py - StorybookApplication
"""ThemedApplicationベースのStorybookApplication。"""
from pubsubtk import ThemedApplication
from .core.state import StorybookState
from .ui.container import StorybookContainer
class StorybookApplication(ThemedApplication[StorybookState]):
"""テーマ対応Storybook専用のアプリケーションクラス"""
def __init__(
self,
theme: str = "arc",
title: str = "PubSubTk Storybook",
geometry: str = "1200x800",
auto_setup: bool = True,
*args,
**kwargs,
):
"""Storybookアプリケーションを初期化する。
Args:
theme: ttkテーマ名（arc, clam, alt, default, classic等）
title: ウィンドウタイトル
geometry: ウィンドウサイズ
auto_setup: 自動でStorybookコンテナを配置するか
"""
super().__init__(
StorybookState, theme=theme, title=title, geometry=geometry, *args, **kwargs
)
if auto_setup:
self._setup_storybook()
def _setup_storybook(self):
"""Storybookコンテナを自動配置"""
sb = StorybookContainer(parent=self.main_frame, store=self.store)
sb.pack(fill="both", expand=True)
src/pubsubtk/storybook/core/decorator.py¶
@story デコレータと Story 登録機能
# storybook/decorator.py - @story デコレータ
"""ストーリーを登録するためのデコレータ。"""
from __future__ import annotations
import re
from typing import Callable
from .meta import StoryMeta
from .registry import StoryRegistry
def _slugify(text: str) -> str:
return re.sub(r"[^a-z0-9_]", "_", text.lower())
def story(path: str | None = None, title: str | None = None):
"""Story 登録用デコレータ。
Args:
path: "Button.Primary" のようなドット区切り階層。
title: 葉ノード名。省略時は path の末尾が使われる。
"""
def decorator(factory: Callable):
# デフォルトパス = <ReturnClass>.<func_name>
comp_name = getattr(factory, "__name__", "Component")
default_path = f"{comp_name}.{factory.__name__}"
full_path = (path or default_path).strip(".")
segments = full_path.split(".")
leaf_title = title or segments[-1]
meta = StoryMeta(
id=_slugify(full_path),
path=segments[:-1],
title=leaf_title,
factory=factory,
)
StoryRegistry.register(meta)
return factory
return decorator
src/pubsubtk/storybook/core/context.py¶
StoryContext - ストーリー実行時のコンテキストオブジェクト
# storybook/context.py - StoryContext 実装
"""StoryFactory に渡すコンテキストオブジェクト。"""
from __future__ import annotations
import tkinter as tk
from typing import Any, Callable, Dict, List, Optional, Type
from pydantic import BaseModel
from ..knobs.store import get_knob_store
from ..knobs.types import KnobSpec, KnobValue
class StoryContext(BaseModel):
"""ストーリー実行時に渡されるコンテキスト"""
parent: tk.Widget
_publish_callback: Callable[[str, dict], None] | None = None
_knob_values: Dict[str, KnobValue] = {}
_story_id: Optional[str] = None
class Config:
arbitrary_types_allowed = True
def set_publish_callback(self, callback: Callable[[str, dict], None]) -> None:
"""PubSub発行用のコールバックを設定"""
self._publish_callback = callback
def set_story_id(self, story_id: str) -> None:
"""ストーリーIDを設定（値の永続化用）"""
self._story_id = story_id
def knob(
self,
name: str,
type_: Type,
default: Any,
desc: str = "",
range_: Optional[tuple] = None,
choices: Optional[List[str]] = None,
multiline: bool = False,
) -> KnobValue:
"""Knobを宣言してKnobValueオブジェクトを返す（値の永続化あり）"""
store = get_knob_store()
story_id = self._story_id or "default"
# グローバルストアから既存のKnobValueを取得
existing_knob = store.get_knob_instance(story_id, name)
if existing_knob:
self._knob_values[name] = existing_knob
return existing_knob
# 新しいKnobSpec作成
spec = KnobSpec(
name=name,
type=type_,
default=default,
desc=desc,
range=range_,
choices=choices,
multiline=multiline,
)
# 保存された値があれば使用、なければデフォルト値
saved_value = store.get_value(story_id, name, default)
# KnobValue作成
knob_value = KnobValue(spec, saved_value)
# 値変更時のコールバックを追加（グローバルストアに保存）
knob_value.add_change_callback(
lambda value: store.set_value(story_id, name, value)
)
# ローカルおよびグローバルストアに保存
self._knob_values[name] = knob_value
store.set_knob_instance(story_id, name, knob_value)
return knob_value
@property
def knob_values(self) -> Dict[str, KnobValue]:
"""登録済みKnobValue一覧"""
return self._knob_values
def clear_knobs(self):
"""Knobをクリア（ストーリー切り替え時）"""
self._knob_values.clear()
def publish(self, topic: str, **kwargs: Any) -> None:
"""Story 空間向け PubSub 発火（名前空間前置き）。"""
if self._publish_callback:
self._publish_callback(f"storybook.{topic}", kwargs)
def on_change(self, var: tk.Variable, cb: Callable[[Any], None]) -> None:
"""tk.Variable にトレーサを張って変更をフック。"""
def _update(*_):
cb(var.get())
var.trace_add("write", _update)
src/pubsubtk/storybook/knobs/types.py¶
Knob動的コントロールの型定義
# storybook/knob/knob_types.py
"""Knobの型定義とデータクラス"""
from __future__ import annotations
from typing import Any, List, Optional, Union
from pydantic import BaseModel, Field
class KnobSpec(BaseModel):
"""Knobの仕様定義"""
name: str
type_: type = Field(..., alias="type")
default: Any
desc: str = ""
range_: Optional[tuple[Union[int, float], Union[int, float]]] = Field(
None, alias="range"
)
choices: Optional[List[str]] = None
multiline: bool = False
class Config:
arbitrary_types_allowed = True
class KnobValue:
"""Knob値の動的オブジェクト（Story内で使用）"""
def __init__(self, spec: KnobSpec, initial_value: Any = None):
self.spec = spec
self._value = initial_value if initial_value is not None else spec.default
self._callbacks: List[callable] = []
@property
def value(self) -> Any:
"""現在の値を取得"""
return self._value
@value.setter
def value(self, new_value: Any) -> None:
"""値を設定し、コールバックを実行"""
if self._value != new_value:
self._value = new_value
for callback in self._callbacks:
callback(new_value)
def add_change_callback(self, callback: callable) -> None:
"""値変更時のコールバックを追加"""
self._callbacks.append(callback)
def __str__(self) -> str:
return str(self._value)
def __repr__(self) -> str:
return f"KnobValue({self.spec.name}={self._value})"
src/pubsubtk/storybook/ui/template.py¶
Storybookレイアウトテンプレート
# storybook/template.py - StorybookTemplate
"""3 スロット (sidebar / preview / knobs) を持つテンプレート。"""
from tkinter import ttk
from pubsubtk import TemplateComponentTtk
from ..core.state import StorybookState
class StorybookTemplate(TemplateComponentTtk[StorybookState]):
"""Storybook の基本レイアウト（テーマ対応）"""
def define_slots(self):
# メインレイアウト設定
self.columnconfigure(1, weight=1)
self.rowconfigure(0, weight=1)
# サイドバー（左側）
sidebar_frame = ttk.Frame(self, width=250)
sidebar_frame.grid(
row=0, column=0, sticky="nsew", padx=(2, 1), pady=2, rowspan=2
)
sidebar_frame.grid_propagate(False) # 幅を固定
# プレビューエリア（中央上）
preview_frame = ttk.LabelFrame(self, text="Preview", padding=5)
preview_frame.grid(row=0, column=1, sticky="nsew", padx=(1, 2), pady=(2, 1))
# Knobパネル（中央下）
knobs_frame = ttk.LabelFrame(self, text="Controls", padding=5)
knobs_frame.grid(row=1, column=1, sticky="nsew", padx=(1, 2), pady=(1, 2))
# 高さ比率設定（Preview 70%, Knobs 30%）
self.grid_rowconfigure(0, weight=7) # Previewエリア
self.grid_rowconfigure(1, weight=3) # Knobエリア
return {
"sidebar": sidebar_frame,
"preview": preview_frame,
"knobs": knobs_frame,
}