-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathobserver.py
More file actions
132 lines (96 loc) · 3.29 KB
/
observer.py
File metadata and controls
132 lines (96 loc) · 3.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""
Observer Design Pattern
The Observer pattern defines a one-to-many dependency between objects so that
when one object changes state, all its dependents are notified and updated automatically.
Use cases:
- Event handling systems
- Model-View architectures
- Real-time data feeds
- Notification systems
"""
from abc import ABC, abstractmethod
# Subject (Observable)
class Subject(ABC):
"""Abstract subject class."""
@abstractmethod
def attach(self, observer):
"""Attach an observer."""
pass
@abstractmethod
def detach(self, observer):
"""Detach an observer."""
pass
@abstractmethod
def notify(self):
"""Notify all observers."""
pass
# Concrete subject
class WeatherStation(Subject):
"""Concrete subject: Weather station."""
def __init__(self):
self._observers = []
self._temperature = 0
def attach(self, observer):
"""Attach an observer."""
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer):
"""Detach an observer."""
self._observers.remove(observer)
def notify(self):
"""Notify all observers."""
for observer in self._observers:
observer.update(self._temperature)
def set_temperature(self, temperature):
"""Set temperature and notify observers."""
self._temperature = temperature
self.notify()
# Observer interface
class Observer(ABC):
"""Abstract observer class."""
@abstractmethod
def update(self, temperature):
"""Update observer with new temperature."""
pass
# Concrete observers
class TemperatureDisplay(Observer):
"""Concrete observer: Temperature display."""
def __init__(self, name):
self.name = name
self.temperature = 0
def update(self, temperature):
"""Update temperature."""
self.temperature = temperature
print(f"{self.name}: Temperature is now {self.temperature}°C")
class AlertSystem(Observer):
"""Concrete observer: Alert system."""
def __init__(self, threshold):
self.threshold = threshold
self.temperature = 0
def update(self, temperature):
"""Update and check threshold."""
self.temperature = temperature
if self.temperature > self.threshold:
print(f"ALERT: Temperature {self.temperature}°C exceeds threshold {self.threshold}°C!")
# Example usage
if __name__ == "__main__":
# Create subject
weather_station = WeatherStation()
# Create observers
display1 = TemperatureDisplay("Display 1")
display2 = TemperatureDisplay("Display 2")
alert = AlertSystem(25)
# Attach observers
weather_station.attach(display1)
weather_station.attach(display2)
weather_station.attach(alert)
# Change temperature (observers will be notified)
print("Setting temperature to 20°C:")
weather_station.set_temperature(20)
print("\nSetting temperature to 30°C:")
weather_station.set_temperature(30)
# Detach an observer
print("\nDetaching Display 1:")
weather_station.detach(display1)
print("Setting temperature to 15°C:")
weather_station.set_temperature(15)