Heating and Ventilation Control
HvcMain
Based on RaspberryPi, written in python, check the comments in the source code.
""" Main control file of my Heating and Ventilation Control. Infinite loop controlls execution of cyclic tasks. Dr. Arne Jachens 2020-06-17 To automatically start the HVC at boot time, do: sudo cp HVC.service /etc/systemd/system/HVC.service sudo chmod +x /etc/systemd/system/HVC.service sudo systemctl enable HVC.service sudo systemctl start HVC.service sudo systemctl status HVC.service Then you may do: sudo service HVC start/stop alternatively, edit sudo /etc/rc.local and add the line: cd /var/www/html/HVC && sudo -u arne python3.8 HvcMain.py& """ import os import time, threading from datetime import datetime from HvcControl import HvcControl # main logic from HvcTables import * from HvcSetGPIO import HvcSetGPIO from HvcReadSPI import HvcReadSPI from HvcSendI2C import HvcSendI2C from HvcRaw2phys import HvcRaw2phys from HvcOperationMode import HvcOperationMode from HvcWeather import HvcWeather from HvcHCSR04ultrasonic import HvcHCSR04ultrasonic from HvcOneWire import HvcOneWire_DS1820 def initialize(path="./"): """ Initialization of my HeatingVentilationControl * Objects to access the hardware are created * initial sensor values are read * initial actuators are set to off * protocol of start-up is returned """ now = datetime.now() message = "HvcMain.initialize \n" OM = HvcOperationMode() operationMode,parameters,changed = OM.read() message = message + "initial operationMode \n" message = message + str(operationMode) + "\n" message = message + "load HvcTables \n" tables = HvcTables() #first reading of sensors SPI = HvcReadSPI() ADC,msg = SPI.readADCverbose() message = message + msg r2p = HvcRaw2phys() #check for floating inputs sensors,msg = r2p.ADCverbose(ADC,1,tables) #message = message + msg ADC,msg = SPI.readADCverbose() sensors,msg = r2p.ADCverbose(ADC,1,tables) message = message + msg #initialize and read 1-wire temperature sensors w1 = HvcOneWire_DS1820() T,msg = w1.read() message = message + msg sensors["fire"] = T["Toven"] sensors["erde"] = T["Terde"] sensors["folu"] = T["Tfolu"] sensors["walu"] = T["Twalu"] sensors["haus1"] = T["Thaus1"] #wohnzimmer sensors["haus2"] = T["Thaus2"] #AZ sensors["haus3"] = T["Thaus3"] #MZ #read H2O distance US = HvcHCSR04ultrasonic() dist = US.distance() distFlt = US.filter(dist) sensors['H2O'] = US.volume(distFlt) #read actual weather # sensors['TAmb'] = 3.14 Weather = HvcWeather("") today = now.strftime("%Y-%m-%d") hour = now.strftime("%H") minutes = now.strftime("%M") weather = Weather.actual(today,hour,minutes) sensors['TAmb'] = weather['TAmb'] #initialize Relays via GPIO GPIO = HvcSetGPIO() actuators,msg = GPIO.setMode() for key in actuators: HvcMain.actuatorsLast[key] = actuators[key] HvcMain.actuatorsFilter[key] = actuators[key] message = message + msg GPIO.set(actuators) #initialize ventialtion via I2C I2C = HvcSendI2C() actuators['vent']=0 actuators['solar']=0 I2C.setPWM([0,0]) #open sensor logfile for appending data today = now.strftime("%Y-%m-%d") initLogFiles(today,sensors,actuators) logFile = open(HvcMain.Log['sensors'], "a") message = message + "sensor logFile: \t "+HvcMain.Log['sensors']+" \n" #myString = "#time" #for s in sensors: # myString = myString + ('\t{}'.format(s)) #logFile.write( myString + "\n") myString = now.strftime("%H:%M") for s in sensors: myString = myString + ('\t{:.1f}'.format(sensors[s])) logFile.write( myString + "\n") logFile.close() #open actuator logfile for appending data logFile = open(HvcMain.Log['actuators'], "a") message = message + "actuators logFile: \t "+HvcMain.Log['actuators']+" \n" #myString = "#time" #for a in actuators: # myString = myString + ('\t{}'.format(a)) #logFile.write( myString + "\n") myString = now.strftime("%H:%M") for a in actuators: myString = myString + ('\t{:.1f}'.format(actuators[a])) logFile.write( myString + "\n") logFile.close() hvc = HvcControl() return sensors, actuators, message, tables, GPIO, SPI, I2C, r2p, OM, US, hvc, w1 def initLogFiles(today,sensors, actuators): #create logDirectory, if it does not exist yet path = today[0:7] try: os.mkdir(path) message = message + "directory "+path+" created \n" except FileExistsError: pass HvcMain.Log['sensors'] = path+"/sensorsLog_" +today+".txt" HvcMain.Log['actuators'] = path+"/actuatorsLog_"+today+".txt" #write sensor names logFile = open(HvcMain.Log['sensors'], "a") myString = " #time" for s in sensors: myString = myString + ('\t{}'.format(s)) logFile.write( myString + "\n") logFile.close() #write actuator names logFile = open(HvcMain.Log['actuators'], "a") myString = " #time" for a in actuators: myString = myString + ('\t{}'.format(a)) logFile.write( myString + "\n") logFile.close() def secondContainer(sensors, actuators, tables, GPIO, SPI, I2C, r2p, w1, OM, hvc, hour, minutes): """ Execute these functions each second. """ #check operationMode operationMode,parameters,changed = OM.read() # reading of sensors ADC = SPI.readADC() #nur die neuen beschreiben! mySensors = r2p.ADC(ADC, 0.01, tables) for k in mySensors.keys(): sensors[k] = mySensors[k] """ if operationMode['byPs']=="shut": sensors['folu'] = sensors['ablu'] - (sensors['zulu']-sensors['erde']) else: sensors['folu'] = sensors['ablu'] """ #filter sensors # HVC logic actuators = hvc.execute(operationMode, sensors, actuators, parameters, hour, minutes) GPIO.set(actuators) I2C.setPWM([ actuators['vent'], actuators['solar'] ]) for key in actuators: HvcMain.actuatorsFilter[key] = HvcMain.actuatorsFilter[key]+actuators[key] return sensors,actuators def minuteContainer(sensors,actuators, US, secondCounter, w1): """ Execute these functions each minute. """ debug=0 dist = US.distance() distFlt = US.filter(dist) sensors['H2O'] = US.volume(distFlt) #write sensors logFile now = datetime.now() Weather = HvcWeather("") today = now.strftime("%Y-%m-%d") hour = now.strftime("%H") minutes = now.strftime("%M") weather = Weather.actual(today,hour,minutes) sensors['TAmb'] = weather['TAmb'] #re-initialize once per day and have new log files #this does not re-init the filters :-) actTime = now.strftime("%H:%M") if actTime=="00:00": today = now.strftime("%Y-%m-%d") initLogFiles(today,sensors,actuators) #regularly write to logs myString = actTime for s in sensors: myString = myString + ('\t{:.1f}'.format(sensors[s])) logFile = open(HvcMain.Log['sensors'], "a") logFile.write( myString + "\n") logFile.close() if debug: print("wrote sensors:\t ",myString) #take average over logging period for key in actuators: HvcMain.actuatorsFilter[key] = HvcMain.actuatorsFilter[key]/float(secondCounter) myString = actTime for a in actuators: myString = myString + ('\t{:.2f}'.format(HvcMain.actuatorsFilter[a])) logFile = open(HvcMain.Log['actuators'], "a") logFile.write( myString + "\n") logFile.close() for key in actuators: HvcMain.actuatorsFilter[key] = actuators[key] if debug: print("wrote actuators:\t ",myString) return sensors def HvcMain(): """ The main infinite while-loop, time-dependent functions are called each second or minute. """ now = datetime.now() month = now.strftime("%Y-%m") hour = now.strftime("%H") minutes = now.strftime("%M") path = "/var/www/html/HVC/"+month+"/" sensors, actuators, message, tables, GPIO, SPI, I2C, r2p, OM, US, hvc, w1 = initialize(path) print(message) message = "" Delta_t = 5 secondCounter = 0 minuteCounter = 0 for key in actuators: HvcMain.actuatorsFilter[key] = 0 next_call = time.time() while True: secondCounter = secondCounter + 1 secondContainer(sensors, actuators, tables, GPIO, SPI, I2C, r2p, w1, OM, hvc, hour, minutes) #read DS1820 Temperatures if secondCounter==1: T,msg = w1.read("Toven") elif secondCounter==2: T,msg = w1.read("Terde") elif secondCounter==3: T,msg = w1.read("Tfolu") elif secondCounter==4: T,msg = w1.read("Twalu") elif secondCounter==5: T,msg = w1.read("Thaus1") elif secondCounter==6: T,msg = w1.read("Thaus2") elif secondCounter==7: T,msg = w1.read("Thaus3") if secondCounter*Delta_t >= 60: now = datetime.now() hour = now.strftime("%H") minutes = now.strftime("%M") #if w1.T["Toven"]>-900: sensors["fire"] = w1.T["Toven"] sensors["erde"] = w1.T["Terde"] sensors["folu"] = w1.T["Tfolu"] sensors["walu"] = w1.T["Twalu"] sensors["haus1"] = w1.T["Thaus1"] sensors["haus2"] = w1.T["Thaus2"] sensors["haus3"] = w1.T["Thaus3"] sensors = minuteContainer(sensors,actuators, US, secondCounter, w1) secondCounter = 0 minuteCounter = minuteCounter +1 next_call = next_call+Delta_t; if next_call - time.time()<0: secondCounter = secondCounter + 1 next_call = next_call+Delta_t; time.sleep(next_call - time.time()) #go for it! HvcMain.Log={} HvcMain.actuatorsLast={} HvcMain.actuatorsFilter={} HvcMain()
Index of Library
Der gesamte Sourcecode darf gemäß GNU General Public License weiterverbreitet werden.