FP:疎結合は通知と変化のしくみにある
このワークシートはMath by Codeの一部です。
オブザーバーパタンの親子ガモは疎結合で親は子に細かい指示をしないことがうまくいくパタンでした。
メディエータパタンは同僚が密に結合してトラブルになることを、仏のDさんが結合を切り相談を一元対応してました。
今回は、疎結合、イベントドリブンという２つの視点にかかわる２パタンを見ていこう。
１．イベントドリブンにみる究極の疎結合がFPのオブザーバーパタン
＜OOP:Observer＞
import math
# ==========================================
# 【クラスO】子ガモ（親を凝視追従する義務がある）
# ==========================================
class ObserverContract:
def update(self, updated_value):
"""親に追従すべしのメソッド"""
raise NotImplementedError()
# ==========================================
# 【クラスS】被凝視者（親ガモのシステム）
# ==========================================
class Subject:
"""親ガモは子ガモリストを管理するが、子ガモに変化を通知して、追従を促すが、
子ガモのことを何も知らない。疎結合"""
def __init__(self):
self._observers = [] # 子ガモのリスト
def attach(self, observer: ObserverContract):
"""子ガモの追加"""
self._observers.append(observer)
def notify(self, new_value):
"""自分の変化を背中で子ガモへの追従を促す。"""
for observer in self._observers:
observer.update(new_value)
# ==========================================
# 【カスタマイズ】ユーザーが作る必死な子ガモたち（Oのサブクラス化）
# ==========================================
class RealObserver(ObserverContract):
"""子ガモ1号"""
def update(self, updated_value):
print(f"1号：親は {updated_value}㎜進みました。急ごう！")
class AboutObserver(ObserverContract):
"""子ガモ2号"""
def update(self, updated_value):
print(f"2号：親が {math.floor(updated_value/10)}cm ぐらい進んだみたい。少し動こう。 ")
class LazyObserver(ObserverContract):
"""子ガモ3号"""
def update(self, updated_value):
print(f"3号：約 {math.floor(updated_value/100 + 0.5)}ｍすすんだ。でも、なんとかなるさ。 ")
# ==========================================
# 親鳥の具体的なデータ（計算コア）
# ==========================================
class MathDataCore(Subject):
def __init__(self):
super().__init__()
self._g = 0
def change_value(self, new_g):
self._g = new_g
print(f"[親データ] は{self._g}になった。すると、 ")
self.notify(self._g)
# １. 親子ガモ登場と登録
parent_data = MathDataCore()
parent_data.attach(RealObserver())
parent_data.attach(AboutObserver())
parent_data.attach(LazyObserver())
# ２．親データの値に子ガモが追従するはず ---
parent_data.change_value(15)
parent_data.change_value(111)
parent_data.change_value(199)
[OUT]
[親データ] は15になった。すると子ガモたちは、
1号：親は 15㎜進みました。急ごう！
2号：親が 1cm ぐらい進んだみたい。少し動こう。
3号：約 0ｍすすんだ。でも、なんとかなるさ。
[親データ] は111になった。すると子ガモたちは、
1号：親は 111㎜進みました。急ごう！
2号：親が 11cm ぐらい進んだみたい。少し動こう。
3号：約 1ｍすすんだ。でも、なんとかなるさ。
[親データ] は199になった。すると子ガモたちは、
1号：親は 199㎜進みました。急ごう！
2号：親が 19cm ぐらい進んだみたい。少し動こう。
3号：約 2ｍすすんだ。でも、なんとかなるさ。
＜FP:Observer＞
毎度のことですが、クラスはすべて関数にしましょう。 子ガモはただのメッセージ（発言・行動）をする関数たちになります。 親ガモのparent_data.attach(...) という処理も不要です。 子ガモを関数リストとしてかくだけで登録完了です。
OOPでは通知（ イベント駆動における親子の疎結合を維持しています。
ただ、リストに入っている関数（子ガモたち）に値を渡して実行しているだけです。
import math
# ==========================================
# 1. サブクラスの代わり（通知された時に動く純粋な処理関数群）
# ==========================================
# 子ガモはそれぞれ単一の関数になります。
def real_observer_action(updated_value):
print(f"1号：親は {updated_value}㎜進みました。急ごう！")
def about_observer_action(updated_value):
print(f"2号：親が {math.floor(updated_value / 10)}cm ぐらい進んだみたい。少し動こう。")
def lazy_observer_action(updated_value):
print(f"3号：約 {math.floor(updated_value / 100 + 0.5)}ｍすすんだ。でも、なんとかなるさ。")
# ==========================================
# 2. フレームワーク側（高階関数によるイベント通知ロジック）
# ==========================================
# 登録されている関数（子ガモたち）のリストに、一斉に値を通知して実行する関数
def notify_observers(observers: list, new_value):
for observer_func in observers:
observer_func(new_value)
# 親データの状態を変更し、同時に通知を行う関数
def change_parent_value(observers: list, new_g):
print(f"[親データ] は{new_g}になった。すると、 ")
# 保持している関数リストに対して値を通知（実行）する
notify_observers(observers, new_g)
# ==========================================
# 3. 実行
# ==========================================
# 1. 親子ガモの登場と登録（子ガモは単なる関数のリスト）
duck_observers = [
real_observer_action,
about_observer_action,
lazy_observer_action
]
# 2. 親データの値に子ガモ（関数）が追従する
change_parent_value(duck_observers, 15)
change_parent_value(duck_observers, 111)
change_parent_value(duck_observers, 199)
関数になった親に子ガモたち（関数リスト）と値を渡すだけ。子ガモの管理という操作がなくても、投げた関数リストによって、関数リストが反応するだけ。
子ガモ３号を加えても、親ガモへの渡し方も、親ガモの行動も何も変わらない。
これが疎結合だね。
ObserverContract）を定義し毎回 new してましたが、FPでは親Subject クラスは関数の入ったリスト（duck_observers）をループで回して直接実行する関数になります。change_parent_value 関数は、自分が呼び出している関数（real_observer_action など）の具体的な中身を一切知りません。２、メディエータDさんはFPによって、タイムトラベラーになった
＜OOP:Mediator＞
メディエータでは、同僚（部品）たちを疎結合にして、仏のDさんに紐づけ、部品の相談を一元管理することで、部品同士の衝突を防止していました。
# ==========================================
# 【クラスMediator】仏のDさんの契約書
# ==========================================
class MediatorContract:
"""手の平に乗せる契約の仕方は、名前のサインと対象イベントだけでOK"""
def consult(self, component_name, event_type, value=None):
raise NotImplementedError()
# ==========================================
# 【クラスControl or Colleague：画面の部品】互いに会話しない部品たち
# ==========================================
class ControlComponent:
"""部品同士は会話しない。分断されている"""
def __init__(self, name: str, mediator: MediatorContract):
self.name = name
self.mediator = mediator
self.visible = True
self.value = None
"""ユーザーの要求が来たら、自分の変化をDさんに伝えるだけ"""
def changed(self, event_type, value=None):
self.value = value
self.mediator.consult(self, event_type, value)
# ==========================================
# 【カスタマイズ】 ユーザが詳細すべてを実装する。
# ==========================================
class BuddhaD(MediatorContract):
def __init__(self):
# 調停する対象のオブジェクトたちを紐づける
self.A = ControlComponent("CheckboxA", self)
self.B = ControlComponent("DropdownB", self)
self.C = ControlComponent("TextInputC", self)
"""【Dは「調停」という名のもとに、A,B,Cを手の平にのせて一元管理する】"""
def consult(self, component, event_type, value=None):
# 1. チェックボックスAが動いたとき
if component.name == "CheckboxA" and event_type == "TOGGLE":
if self.A.value == "ON":
print("[Dさん調停] AさんがONになりましたね。BさんはリストXを表示してください。Cさんは出番待ちです。")
self.B.visible = True
self.B.value = "リストX表示中"
self.C.visible = False
elif self.A.value == "OFF":
print("[Dさん調停] AさんがOFFになりました。Bさんは身を隠し、Cさん（テキスト入力）が表に出てください。")
self.B.visible = False
self.C.visible = True
# 2. ドロップダウンリストBが動いたとき
elif component.name == "DropdownB" and event_type == "SELECT":
if self.B.value == "Y":
print("[Dさん調停] Bさんで'Y'が選ばれました。Cさんの入力欄に'Y'を自動代入します。")
self.C.value = "Y"
elif self.B.value == "Z":
print("[Dさん調停] Bさんで'Z'が選ばれました。Cさんは空欄にして入力待ちにしてください。")
self.C.value = ""
# 3. テキスト入力Cに文字が打ち込まれたとき
elif component.name == "TextInputC" and event_type == "INPUT":
if self.C.value == "P":
print("[Dさん調停] Cさんに特異点'P'が入力されました！大局的判断により、Aさんを非表示にします。")
self.A.visible = False
# 仏のDさん（場）が誕生。手の平にのる部品に略名をつける。
d_san = BuddhaD()
A=d_san.A
B=d_san.B
C=d_san.C
# ここから、仏のDさんの大活躍がスタート！
print("--- 1. ユーザーがチェックボックスAを『ON』にした ---")
A.changed("TOGGLE", "ON")
print(f" -> [その結果] Bの表示:{B.visible}, Cの表示:{C.visible}\n")
print("--- 2. ユーザーがドロップダウンBで『Y』を選択した ---")
B.changed("SELECT", "Y")
print(f" -> [その結果] Cの自動入力値: '{C.value}'\n")
print("--- 3. ユーザーがチェックボックスAを『OFF』にした ---")
A.changed("TOGGLE", "OFF")
print(f" -> [その結果] Bの表示:{B.visible}, Cの表示:{C.visible}\n")
print("--- 4. テキストCに、禁断の文字『P』が入力された ---")
C.changed("INPUT", "P")
print(f" -> [その結果] Aの表示:{A.visible}")
[OUT]
--- 1. ユーザーがチェックボックスAを『ON』にした ---
[Dさん調停] AさんがONになりましたね。BさんはリストXを表示してください。Cさんは出番待ちです。
-> [その結果] Bの表示:True, Cの表示:False
--- 2. ユーザーがドロップダウンBで『Y』を選択した ---
[Dさん調停] Bさんで'Y'が選ばれました。Cさんの入力欄に'Y'を自動代入します。
-> [その結果] Cの自動入力値: 'Y'
--- 3. ユーザーがチェックボックスAを『OFF』にした ---
[Dさん調停] AさんがOFFになりました。Bさんは身を隠し、Cさん（テキスト入力）が表に出てください。
-> [その結果] Bの表示:False, Cの表示:True
--- 4. テキストCに、禁断の文字『P』が入力された ---
[Dさん調停] Cさんに特異点'P'が入力されました！大局的判断により、Aさんを非表示にします。
-> [その結果] Aの表示:False
＜FP：Mediator＞
FPではこれまで構築してきた「イミュータブルなバケツリレー」と「パターンマッチング」の考え方を組み合わせます。
そして、「すべての部品の状態をまとめた1つの辞書（State）」と「イベントを受け取って新しいStateを返す純粋な調停関数consult_mediator」に分離します。
３つの部品というよりもStateというオブジェクトがconsult_mediatorの処理で変化する。
だから、ステートパタンにもなっているといえます。
ステートの１つ１つ別名でイミュータブルな生成なので、ステートトラベルが過去に遡っても可能になります。ステート名を入れると、その状態に飛べますからね。
状態変化についてはReactでよく使うスプレッド構文を使います。
# Pythonでのイミュータブルな辞書更新
updated_state = {
**state,
component_name: {
**state[component_name],
"value": value
}
}とあるのは、
Reactでの…の展開上書きのスプレッド構文と同じ仕様で、記号が…が**に変わっただけです。
setSingleState(prevState => ({
...prevState,
[component_name]: {
...prevState[component_name],
value: value
}
}));
この仕様によって、一度展開されて、変更した部品の変更した要素の状態だけ上書きされる。
それが、別の状態変数で別名でバケツリレーされるので、状態変化履歴ができますね。
仏のDさんはFPによってタイムトラベラーになったのです。
#Mdiator FP
# ==========================================
# 1. 初期状態の定義（イミュータブルな辞書データ）
# ==========================================
# 各コンポーネントの状態（visible, value）を一つの「状態（State）」として一元管理します。
initial_state = {
"A": {"visible": True, "value": None},
"B": {"visible": True, "value": None},
"C": {"visible": True, "value": None}
}
# ==========================================
# 2. 調停者側（純粋関数による状態遷移ロジック）
# ==========================================
# 現在の状態（state）とイベント情報を受け取り、不変性を保ったまま「新状態」を新状態名で作成して返します。
def consult_mediator(state: dict, component_name: str, event_type: str, value=None) -> dict:
# ユーザーの入力を現在の状態に反映したベースとなる状態を作成
updated_state = {
**state,
component_name: {**state[component_name], "value": value}
}
# マッチングのために条件をタプルにまとめる
match (component_name, event_type, value):
# 1. チェックボックスAが動いたとき
case ("A", "TOGGLE", "ON"):
print("[Dさん調停] AさんがONになりましたね。BさんはリストXを表示してください。Cさんは出番待ちです。")
return {**updated_state,"B": {"visible": True, "value": "リストX表示中"},"C": {**updated_state["C"], "visible": False}}
case ("A", "TOGGLE", "OFF"):
print("[Dさん調停] AさんがOFFになりました。Bさんは身を隠し、Cさん（テキスト入力）が表に出てください。")
return {**updated_state,"B": {**updated_state["B"], "visible": False},"C": {**updated_state["C"], "visible": True}}
# 2. ドロップダウンリストBが動いたとき
case ("B", "SELECT", "Y"):
print("[Dさん調停] Bさんで'Y'が選ばれました。Cさんの入力欄に'Y'を自動代入します。")
return {**updated_state,"C": {**updated_state["C"], "value": "Y"}}
case ("B", "SELECT", "Z"):
print("[Dさん調停] Bさんで'Z'が選ばれました。Cさんは空欄にして入力待ちにしてください。")
return {**updated_state,"C": {**updated_state["C"], "value": ""}}
# 3. テキスト入力Cに文字が打ち込まれたとき
case ("C", "INPUT", "P"):
print("[Dさん調停] Cさんに特異点'P'が入力されました！大局的判断により、Aさんを非表示にします。")
return {**updated_state,"A": {**updated_state["A"], "visible": False}}
# 該当しないイベントは状態をそのまま返す
case _:
return updated_state
# ==========================================
# 3. 実行（不変なバケツリレー）
# ==========================================
# アプリケーション全体の「状態の履歴」が定数のように積み重なっていきます
state_v0 = initial_state
print("--- 1. ユーザーがチェックボックスAを『ON』にした ---")
state_v1 = consult_mediator(state_v0, "A", "TOGGLE", "ON")
print(f" -> [その結果] Bの表示:{state_v1['B']['visible']}, Cの表示:{state_v1['C']['visible']}\n")
print("--- 2. ユーザーがドロップダウンBで『Y』を選択した ---")
# 常に「直前の状態」を入力として引き渡します
state_v2 = consult_mediator(state_v1, "B", "SELECT", "Y")
print(f" -> [その結果] Cの自動入力値: '{state_v2['C']['value']}'\n")
print("--- 3. ユーザーがチェックボックスAを『OFF』にした ---")
state_v3 = consult_mediator(state_v2, "A", "TOGGLE", "OFF")
print(f" -> [その結果] Bの表示:{state_v3['B']['visible']}, Cの表示:{state_v3['C']['visible']}\n")
print("--- 4. テキストCに、禁断の文字『P』が入力された ---")
state_v4 = consult_mediator(state_v3, "C", "INPUT", "P")
print(f" -> [その結果] Aの表示:{state_v4['A']['visible']}")
＜振り返り＞
通知の疎結合（Observer）
親ガモは「管理・お世話する」という執着を捨て
背中を追いかけていることになっている子ガモ「関数リスト」に値を通知して各個に反応のきっかけを
与える。子ガモたちはもちろん、なんの干渉も交流も予想されていない。
変化の疎結合（Mediator）
部品を疎結合にして、相談者と部品すらも疎な関係にしたFPの疎結合。
状態変化からの流れ作業となり、通知というイベントさえ希薄になる関係性。
しかし、それだけではなかった。
「その場の状態を直接書き換える」というイベントドリブンの罠、情報の更新という名の破壊をやめる。
古い状態とイベントを掛け合わせて「新築の歴史（State）」を別名で積み重ねていく。
ただ上書きされて、忘れ去らるのではなく古きを訪ねることが可能な歴史としての状態の累積。
疎結合が、疎結合のよさが、
FPによる通知と変化のしくみによって、
くっきりと浮かび上がりましたね。
これらがイミュータブルな変数の存在に関係があることがわかりましたね。
課題：Geogebraの変数がイミュータブルかどうかを確認するにはどうしますか。
数式にa=3などと入力します。
ボタンツールの見出しに[a=a+3]とかき、
スクリプトとしてsetValue(a, a+3)とかきます。
さて、ボタンを押すとどうなるでしょうか。
数式のaの値は3ずつ増えた値に上書きされています。
これは同名のaの値の別のメモリが作られたわけではないはずですね。
geogebraは同姓同名をゆるしませんから。
だから、同名のaが可変ということは、
geogebraの変数は可変型であるということがわかりますね。