Python中如何实现观察者模式 事件驱动编程的设计思路

import abc

class Observer(abc.ABC):
    @abc.abstractmethod
    def update(self, subject):
        pass

class Subject(abc.ABC):
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self):
        for observer in self._observers:
            observer.update(self)

class ConcreteSubject(Subject):
    def __init__(self):
        super().__init__()
        self._state = None

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, value):
        self._state = value
        self.notify()

class ConcreteObserverA(Observer):
    def update(self, subject):
        print(f"ConcreteObserverA: Subject's state is now {subject.state}")

class ConcreteObserverB(Observer):
    def update(self, subject):
        print(f"ConcreteObserverB: Subject's state is now {subject.state}")

if __name__ == "__main__":
    subject = ConcreteSubject()

    observer_a = ConcreteObserverA()
    observer_b = ConcreteObserverB()

    subject.attach(observer_a)
    subject.attach(observer_b)

    subject.state = "State 1"
    subject.state = "State 2"

    subject.detach(observer_a)

    subject.state = "State 3"

from pysignals import Signal

class Button:
    clicked = Signal()  # 定义一个名为clicked的信号

    def __init__(self, name):
        self.name = name

    def press(self):
        print(f"Button {self.name} pressed")
        self.clicked.emit()  # 发射信号

class Handler:
    def __init__(self, name):
        self.name = name

    def on_button_clicked(self):
        print(f"Handler {self.name}: Button clicked!")

button = Button("MyButton")
handler = Handler("MyHandler")

button.clicked.connect(handler.on_button_clicked)  # 连接信号和槽

button.press()

import time
import random
import threading

class TemperatureSensor(Subject):
    def __init__(self):
        super().__init__()
        self._temperature = 20

    @property
    def temperature(self):
        return self._temperature

    @temperature.setter
    def temperature(self, value):
        self._temperature = value
        self.notify()

    def run(self):
        while True:
            self.temperature = random.randint(15, 30)
            time.sleep(1) # 模拟温度变化

class TemperatureDisplay(Observer):
    def __init__(self, sensor):
        self.sensor = sensor
        self.update(sensor) # 初始显示

    def update(self, subject):
        print(f"Temperature: {self.sensor.temperature}°C")

if __name__ == "__main__":
    sensor = TemperatureSensor()
    display = TemperatureDisplay(sensor)
    sensor.attach(display)

    # 启动一个线程模拟温度变化
    thread = threading.Thread(target=sensor.run)
    thread.daemon = True  # 设置为守护线程，主线程退出时自动退出
    thread.start()

    # 主线程保持运行
    try:
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        print("Exiting...")