Introduction
Today I worked a little on Observer pattern. I'd like to note down all that I have done today in this post.
Firstly I want to cite a short introduction of Observer pattern copied from this website:
Observer pattern is used when there is one-to-many relationship between objects such as if one object is modified, its dependent objects are to be notified automatically. Observer pattern falls under behavioral pattern category.
I am reading this chinese book about Design patterns and my coding work has been inspired by this book:
The scheme of Observer pattern presented in the book is like this:
Scenario
Imagine a common scenario this year: John has unfortunately been infected with Covid-19 and he is being monitored closely and carefully in clinic by medical workers. They mounted several medical devices to monitor his temperature, heart rate, oxygen saturation, etc. These smart devices can transmit an alert signal if the measured value surpasses or goes below a certain threshold. Today I want to simulate this scenario using Observer pattern in Python.
Start coding
Abstract classes
Firstly write the abstract classes: abstractSubject and abstractObserver.
Class abstractSubject is the abstract base class for all observable subjects; it has a list member which contains all attached observers; it has methods addObs()/removeObs() for adding/removing an observer in its list. It has another method notifyObservers() for communicating its state change to observers.
Class abstractObserver is the abstract base class for all observers. It has only one empty method update() which shall be overridden and realized by its derived classes.
class abstractSubject():
"""
Abstract Subject - Abstract patient in this demo
"""
def __init__(self):
self.__observers = []
def addObs(self, observer):
self.__observers.append(observer)
def removeObs(self, observer):
self.__observers.remove(observer)
def notifyObservers(self, arg=0):
for o in self.__observers:
o.update(self, arg)
class abstractObserver():
"""
Abstract Observer - Abstract medical device in this demo
"""
def __init__(self):
pass
def update(self): ## shall be overridden
pass
Patient to be observed
Then I create covidPatient class which is a concrete class derived from abstractSubject class. This class has a public member "name" and a private dict-type member containing its real-time physiological parameters. Its method setValue() serves for updating a chosen parameter and calling the inherited method notifyObservers(). Method setValue() returns the real-time value of a chosen physiologic parameter.
class covidPatient(abstractSubject):
"""
Concrete Subject - Patient in this demo
"""
def __init__(self, name):
super().__init__()
self.name = name
self.__physioParams = {"temperature": 0.0, "heartrate": 0.0, "oxygen": 0.0, "respiration": 0.0}
## function to the observed subject's state
def setValue(self, measureType, val):
if measureType in self.__physioParams:
self.__physioParams[measureType] = val
# print("{}'s {} set to: {}".format(self.name, measureType, str(val)) )
self.notifyObservers()
else:
print("Parameter type \"{}\" not yet supported.".format(measureType))
def getValue(self, measureType):
if measureType in self.__physioParams:
return self.__physioParams[measureType]
else:
return None
Observers
Now define two observers as below, each of them implements method update() which obtains and processes a certain physiological parameter of the covid patient.
class thermometer(abstractObserver):
"""
Concrete Observer - Thermometer
"""
def __init__(self):
super().__init__()
def update(self, tt, obj):
if tt.__class__ == covidPatient:
temp = tt.getValue("temperature")
if temp > 37.8:
print("EMCY - " "Temperature too high: " str(temp))
elif temp < 36.0:
print("EMCY - " "Temperature too slow: " str(temp))
else:
print("INFO - " "Temperature normal: " str(temp))
else:
pass
class heartbeatMonitor(abstractObserver):
"""
Concrete Observer - heartbeat monitor
"""
def __init__(self):
super().__init__()
def update(self, tt, obj):
if tt.__class__ == covidPatient:
hr = tt.getValue("heartrate")
if hr > 120:
print("EMCY - " "Heart rate too high: " str(hr))
elif hr < 35:
print("EMCY - " "Heart rate too slow: " str(hr))
else:
print("INFO - " "Heart rate normal: " str(hr))
else:
pass
Simulation kicks off
OK, it's time to start the simulation. I create an instance of covidPatient whose name is John, an instance of thermometer and an instance of heartbeatMonitor. This main function below is clear enough so I will do no more explanation.
import time
if __name__ == "__main__":
h = covidPatient("John")
o1 = thermometer()
o2 = heartbeatMonitor()
## now kick off the simulation
for i in range(0, 15):
time.sleep(1.5)
print("====== Time step {} =======".format(i 1))
# At rount #3: thermometer is added for monitoring temperature
# At rount #5: heartbeatMonitor is added for monitoring heart rate
# At rount #10: thermometer is removed
if i == 3:
h.addObs(o1)
elif i == 5:
h.addObs(o2)
elif i == 10:
h.removeObs(o1)
# simulating the variation of patient's physiological parameters
if i%3 ==0:
h.setValue("temperature", 35.5 0.5*i)
elif i%3 == 1:
h.setValue("heartrate", 30 10*i)
else:
h.setValue("oxygen", 5.0 0.05*i)
Execution output:
As one can see, the patient's physiological parameters vary in each time step. The two observers have been mounted one by one during the simulation and one of them has then been removed in step #10. Both observers have sent emergency signals as soon as they have detected an abnormal value.
May John recover soon from this annoying coronavirus...
Top comments (1)
I see clean code, I press ❤️