Raspberry Pi Auto FAN Control

fan.cfg

limit=50
cooldown=7
interval=5

fan.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

#########################################################
# Schematics:
#   - NPN Transistor 2N5551 or any suitable
#   - Resistor ~300 Ohms
#   - 3-pin connector on pins 4,6,8
#   - Resistor to Transistor base
#   - Fan+ to device 5V pin (pin 4)
#   - Fan- to transistor collector
#   - Transistor emitter to ground (pin 6)
#   - Resistor (connected to transistor base) to control pin (pin 8, BCM14)
#
# Usage: 
#  1. Make it autostarted.
#  2. Place fan.cfg next to fan.py (optional)
#  3. Edit config in fan.cfg
#    - limit - which temperature should start cooling, default 50
#    - cooldown - how many degrees down should be cooled to stop the fan, default 10
#    - interval - interval in seconds to check temperature, default 5
#    - controlPin - pin where transistor base is connected, default 14 (pin 8)
#
# Overheating:
#   If fan works for 5 minutes and temperature can't reach cooldown value ("limit" minus "cooldown" degrees)
#   script calls initiateShutdown() method, which is empty here.
#   It's up to you to decide how exactly correct shutdown should be implemented.
#########################################################

import RPi.GPIO as GPIO
import sys, traceback
import os, datetime

from time import sleep

def fileAtScript(name):
    currentDir = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(currentDir, name)

class Log:

    filename = '{date:%Y-%m-%d_%H%M}.log'.format(date=datetime.datetime.now())
    _file = open(fileAtScript(filename), 'a')

    @classmethod
    def info(self, text, *args):
        self.internalWrite("INFO:", text, *args)

    @classmethod
    def error(self, text, *args):
        self.internalWrite("ERROR:", text, *args)

    @classmethod
    def internalWrite(self, level, text, *args):
        timestamp = '{date:%H:%M:%S.%f}'.format(date=datetime.datetime.now())
        data = f"{level:5} {timestamp}: {text.format(*args)}"
        print(data)
        self._file.write(data + '\n')
        self._file.flush()

class Config:

    def __init__(self):
        self.update()

    def update(self):
        vars = {}
        fullPath = fileAtScript("fan.cfg")
        if (os.path.exists(fullPath)):
            with open(fullPath, "r") as file:
                for line in file:
                    name, var = line.partition("=")[::2]
                    vars[name.strip()] = var.strip()

        self._dict = vars

    @property
    def limit(self):
        return self.readInteger('limit', 50)

    @property
    def cooldown(self):
        return self.readInteger('cooldown', 10)

    @property
    def interval(self):
        return self.readInteger('interval', 5)

    @property
    def controlPin(self):
        return self.readInteger('controlPin', 14)

    def readInteger(self, key, default):
        
        try:
            valueFromDict = self._dict.get(key)
            if valueFromDict is None:
                return default
            else:
                return int(valueFromDict)
        except:
            Log.error("Exception in reading " + key + ":\n" + traceback.format_exc())
            return default

class Fan:

    def __init__(self, controlPin):
        self.init(controlPin)

    def init(self, controlPin):
        self._controlPin = controlPin
        self._isOn = False
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(controlPin, GPIO.OUT, initial=0)

    def isOn(self):
        return self._isOn

    def setState(self, enable):
        Log.info(f"Enabling fan: {enable}")
        self._isOn = enable
        GPIO.output(self._controlPin, enable)

def writePidFile():
    currentPid = os.getpid()
    with open(fileAtScript('fan.pid'), 'w', encoding='utf-8') as f:
        f.write(str(currentPid))

    
    Log.info(f'PID created: {currentPid}')

def get_temp():
    f = open("/sys/class/thermal/thermal_zone0/temp")
    temp = int(f.read())
    f.close()
    
    return (temp/1000)

def cleanup():
    Log.info('Performing cleanup')
    GPIO.cleanup()

def initiateShutdown():
    None

def doMeasuring(fan, config):
    
    Log.info('Start measuring')
    
    coolingStarted = None # Moment when cooling started

    errorLimit = 10
    measuresCount = 0
    updateEvery = 30

    while errorLimit != 0:
        try:
            measuresCount = measuresCount + 1

            if (measuresCount == updateEvery):
                config.update()
                Log.info('Config updated')
                measuresCount = 0

            limit = config.limit
            cooldownDelta = config.cooldown
            currentTemp = get_temp()

            if currentTemp >= limit and not fan.isOn():
                Log.info(f'Limit reached. Current temp is {currentTemp}')
                fan.setState(True)
                coolingStarted = datetime.datetime.now()
            elif currentTemp >= limit and fan.isOn():
                alreadyOn = datetime.datetime.now() - coolingStarted
                minutesLimit = datetime.timedelta(minutes=5)
                if (alreadyOn > minutesLimit):
                    Log.error(f'Overheated! Temp is {currentTemp} for {minutesLimit} while fan is on')
                    initiateShutdown()
            elif currentTemp <= limit - cooldownDelta and fan.isOn():
                Log.info(f'We\'re cool now. Current temp is {currentTemp}')
                fan.setState(False)

            sleep(config.interval)

        except KeyboardInterrupt:
            raise
        except:
            lines = traceback.format_exc() 
            Log.error("Exception in measure:\n" + lines)
            cleanup()
            config.update()
            fan.init(config.controlPin)
            errorLimit = errorLimit - 1

# Main program
def start():

    writePidFile()

    Log.info('Initializing')
    config = Config()
    fan = Fan(config.controlPin)
    
    try:

        doMeasuring(fan, config)

    except KeyboardInterrupt:
        Log.info("Exit pressed Ctrl+C")
    except:
        # ...
        Log.error("Other Exception")
        lines = traceback.format_exc()
        Log.error(lines)
    finally:
        cleanup()

    Log.info('Exited')

start()