Raspberry Pi Heating Controller

Question

In "most" (my experience) older UK houses, Central Heating and Hot Water both come a gas boiler. They are controlled by a programmer, to come on at different times, which is effectively just three relay switches. (A thermostat is involved but is not relevant in this case.) The ways the relays have to be for different options should be clear in the code.

Note that in my case, I can have heating only, water only, both or none. This isn't the case with some systems. If I go in to any more detail this will be a post about plumbing.

I'm learning Python from scratch, and as will probably become clear, I've stumbled through how to do each thing, rather than approach the whole project from a design perspective. Regardless, I've had some good feedback on here before (the parseFile function below was perfected by @Peilonrayz) so I thought I'd open up what I've done to criticism. I haven't read all of PEP8 yet, but I intend to during the next iteration of this project.

I've tried to at least taste the concept of reusable code, and splitting things up in a way that can be improved in a modular way, but I have a lot to learn there. There are a million things I want to ask your collective opinion on, but maybe its better just to leave it open.

Ignore references to bears. If I'm not sure if a word is reserved (such as "Event" in this case) I stick "bear" on it as a precaution, and I didn't remove this one.

Finally - I'm aware I haven't completely tidied this up yet. I think there might still be the odd variable which is assigned and then never used (I deleted a couple of these as I pasted it here).

This is parsing.py:

import datetime
def parseFile(filePath, startTag, endTag):
 """Parses a file line by line. Yields all lines between two tags
 filePath -- the text file to parse
 startTag -- tag which marks the opening bookend line (not yielded)
 endTag -- tag which marks the closing bookend line (not yielded)
 """
 with open(filePath) as bearFile:
 for bearLine in bearFile: #for each line in the file:
 if startTag not in bearLine: #if it is NOT the start line,
 continue #ignore it, go back to 'for' and read the next line
 for newLine in bearFile: #if it IS the start, then for each line from here on
 if endTag in newLine: #if it is the end line, 
 break #break this 'for' loop and go back to the first one
 yield newLine.strip() #otherwise add the line to the generator
def makeBearEvent(type, line):
 """Parses a line for on and off times, returns a bearEvent
 type -- Heat or hot water
 line -- the line of text containing [ON] and [OFF] times
 """
 on = "[ON]"
 off = "[OFF]"
 ontime = ""
 offtime = ""
 for i in range(len(line)): #for each character in the line
 if line[i:i+len(on)] == on: #if characters from i to i + the length of 'on' are 'on'
 for j in range(i+len(on), len(line)): #search from here to the end
 if line[j:j+len(off)] == off: # if from j to j+the end of 'off' are 'off'
 ontime = line[i+len(on):j].strip() #set ontime to between 'on' and 'off'
 offtime = line[j+len(off):len(line)].strip()#set offtime to between 'off' and the end
 break #stop searching for 'off' (or does this stop searching for 'on'?)
 break #stop searching for 'on' (does this do anything?)
 if ontime != "" and offtime != "":
 return bearEvent(type, ontime, offtime)
class bearEvent:
 def __init__(self, type, startTime, stopTime):
 self.type = type
 self.startTime = startTime
 self.stopTime = stopTime
 def show(self):
 return "Type:" + self.type + " Start:" + self.startTime + " End:" + self.stopTime
def getState(textfile, heatOn, heatOff, waterOn, waterOff):
 lstEvent = []
 genHeat = parseFile(textfile, heatOn, heatOff)
 for f in genHeat:
 event = makeBearEvent("H", f)
 if event is not None:
 lstEvent.append(event)
 genWater = parseFile(textfile, waterOn, waterOff)
 for f in genWater:
 event = makeBearEvent("W", f)
 if event is not None:
 lstEvent.append(event)
 dtnow = datetime.datetime.now()
 dt_now_string = dtnow.strftime("%y%m%d")
 state = {"H":False, "W":False}
 for event in lstEvent:
 st = event.startTime
 strTodayStart = dt_now_string + st
 today_start = datetime.datetime.strptime(dtnow.strftime("%y%m%d") + event.startTime, "%y%m%d%H:%M")
 today_stop = datetime.datetime.strptime(dtnow.strftime("%y%m%d") + event.stopTime, "%y%m%d%H:%M")
 if dtnow > today_start and dtnow < today_stop:
 #print("In effect:", lstEvent.index(event), event.show())
 state[event.type] = True
 if state['H'] == True:
 if state['W'] == True: #Both True
 return "HW"
 elif state['W'] == False: #Heat True, Water False
 return "H"
 elif state['W'] == True: #Heat False, Water True
 return "W"
 else:
 return ""

And this is runme.py - which runs in the background on boot:

import parsing
import RPi.GPIO as io
import datetime
import time
def startup():
 io.setmode(io.BCM)
 io.setup(2, io.OUT)
 io.setup(3, io.OUT)
 io.setup(4, io.OUT)
 io.output(2, io.LOW)
 io.output(3, io.LOW)
 io.output(4, io.LOW)
def turnHeatOn():
 io.output(2, io.HIGH)
 io.output(3, io.LOW) 
 io.output(4, io.HIGH)
def turnWaterOn():
 io.output(2, io.LOW)
 io.output(3, io.HIGH)
 io.output(4, io.LOW)
def turnHeatAndWaterOn():
 io.output(2, io.LOW)
 io.output(3, io.HIGH)
 io.output(4, io.HIGH)
def turnOff():
 io.output(2, io.LOW)
 io.output(3, io.LOW)
 io.output(4, io.LOW)
startup()
while True:
 print("The time is:", datetime.datetime.now().time().replace(microsecond=0))
 state = parsing.getState("/etc/heat/heatsched.txt", "[HEAT]", "[/HEAT]", "[WATER]", "[/WATER]")
 print("state:", state)
 if state == "H":
 print("Heating only, pins 2 and 4")
 turnHeatOn()
 elif state == "W":
 print("Water only, pin 3")
 turnWaterOn()
 elif state == "HW":
 print("Heating and water, pins 3 and 4")
 turnHeatAndWaterOn()
 else:
 print("Nothing, all pins off!")
 turnOff()
 time.sleep(10)

Finally, the file to be parsed, heatsched.txt:

[HEAT]
[ON] 05:00 [OFF] 08:00
[ON] 10:20 [OFF] 10:23
[ON] 17:00 [OFF] 22:20
[/HEAT]
[WATER]
[ON] 05:00 [OFF] 06:00
[ON] 10:20 [OFF] 10:45
[ON] 17:00 [OFF] 20:00
[/WATER]

At the moment this works, I've been using it to control the boiler for a couple of days. Because it parses the file each time, I can change the text file without having to stop python running.

I just use WinSCP to modify the text file remotely, and I added this line to rc.local:

python3 /etc/heat/runme.py

Answer 1

Albeit not being PEP8 compliant, your code is easy to follow and reason with. Please use snake_case instead of pascalCase to name your variables and functions as it is what people should be expecting when reading Python code. You should also put your top-level code under an if __name__ == '__main__' clause and you may benefit from default values for arguments.

Your code would benefit much from using regular expressions to parse the file as it will be easier to:

1. automatically detect sections: using a regex you can identify the begining of a section, extract its name, and build the line corresponding to its end;
2. extract out start and stop times from a single line.

This will also allow you to make your parseFile generate events for each line in the file at once instead of having to call it section by section; hence eliminating the need for lstEvent as iterating over parseFile will be equivalent.

You can also benefit from the right datastructure: instead of using a dict to store your states, use a set and .add event types as you need.

Lastly, datetime has some usefull helpers, such as combine or its format capabilities.

---

Proposed improvements:

import re
from datetime import datetime
from contextlib import suppress
class BearEvent:
 PATTERN = re.compile(r'\[ON\]\s([0-9:]+)\s\[OFF\]\s([0-9:]+)\s')
 def __init__(self, event_type, start_time, stop_time):
 self.type = event_type
 self.start_time = datetime.strptime(start_time, '%H:%M').time()
 self.stop_time = datetime.strptime(stop_time, '%H:%M').time()
 def __str__(self):
 return 'Type: {0.type} Start: {0.start_time:%H:%M} End: {0.stop_time:%H:%M}'.format(self)
 @classmethod
 def from_config_line(cls, event_type, line):
 hours = cls.PATTERN.search(line)
 if not hours:
 raise ValueError('not an event configuration line')
 return cls(event_type, hours.group(1), hours.group(2))
def parse_section(file, section):
 end_section = '[/{}]'.format(section)
 for line in file:
 if line.startswith(end_section):
 return
 with suppress(ValueError):
 yield BearEvent.from_config_line(section, line)
def parse_file(filepath):
 start_section = re.compile(r'^\[([^\]]+)\]')
 with open(filepath) as bear_file:
 for line in bear_file:
 section = start_section.match(line)
 if section:
 section = section.group(1)
 yield from parse_section(bear_file, section)
def get_state(filepath):
 state = set()
 now = datetime.now()
 today = now.date()
 for event in parse_file(filepath):
 start = datetime.combine(today, event.start_time)
 stop = datetime.combine(today, event.stop_time)
 if start < now < stop:
 state.add(event.type)
 return state

import time
import datetime
import RPi.GPIO as io
import parsing
def startup():
 io.setmode(io.BCM)
 io.setup(2, io.OUT)
 io.setup(3, io.OUT)
 io.setup(4, io.OUT)
 io.output(2, io.LOW)
 io.output(3, io.LOW)
 io.output(4, io.LOW)
def turn_heat_on():
 io.output(2, io.HIGH)
 io.output(3, io.LOW)
 io.output(4, io.HIGH)
def turn_water_on():
 io.output(2, io.LOW)
 io.output(3, io.HIGH)
 io.output(4, io.LOW)
def turn_heat_and_water_on():
 io.output(2, io.LOW)
 io.output(3, io.HIGH)
 io.output(4, io.HIGH)
def turn_off():
 io.output(2, io.LOW)
 io.output(3, io.LOW)
 io.output(4, io.LOW)
def main(config_file, seconds_between_state_check=10):
 startup()
 while True:
 print('The time is: {0:%H:%M:%s}'.format(datetime.datetime.now()))
 state = parsing.get_state(config_file)
 if {'HEAT', 'WATER'} <= state:
 print('Heating and water, pins 3 and 4')
 turn_heat_and_water_on()
 elif 'HEAT' in state:
 print('Heating only, pins 2 and 4')
 turn_heat_on()
 elif 'WATER' in state:
 print('Water only, pin 3')
 turn_water_on()
 else:
 turn_off()
 time.sleep(seconds_between_state_check)
if __name__ == '__main__':
 main('/etc/heat/heatsched.txt')