Some code refactored

Author: Daniel Todor

.gitignore

@@ -1,2 +1,4 @@
 dist
 rpi_gpio_devices.egg-info
+__pycache__
+.coverage

src/__init__.py

@@ -1,6 +1,6 @@
 from .base import BaseDevice, SwitchableDevice, PWMDevice
 from .button import Button
-from .fan import Fan
+from .fan import Fan, SM
 from .led import LED
 from .pwmled import PWMLED
 from .switch import Switch

src/base.py

@@ -1,21 +1,22 @@
 from datetime import datetime
+
 import RPi.GPIO as gpio
 
 
-class BaseDevice(object):
+class BaseDevice:
     """
     Base class for other devices or classes
 
     -:param gpiomode: Gpio pin numbering mode. "board" or "bcm"
     -:param gpiowarnings: Gpio configuration warnings
     -:param verbose: Verbose mode
     """
-    def __init__(self, gpiomode="board", gpiowarnings=False, verbose=False):
+    def __init__(self, gpiomode='board', gpiowarnings=False, verbose=False):
         self.verbose = verbose
         # GPIO basic settings
-        if gpiomode == "board":
+        if gpiomode == 'board':
             gpio.setmode(gpio.BOARD)
-        elif gpiomode == "bcm":
+        elif gpiomode == 'bcm':
             gpio.setmode(gpio.BCM)
         gpio.setwarnings(gpiowarnings)
 
@@ -26,7 +27,7 @@ def message(self, val):
 
     def cleanup(self):
         """ Cleanup GPIOs """
-        self.message("Cleanup GPIOs")
+        self.message('Cleanup GPIOs...')
         gpio.cleanup()
 
 
@@ -55,15 +56,17 @@ def turn_on(self):
         """ Turn on the device """
         if self.is_on():
             return
-        self.message("Device turned on")
+        self.message('Device turned on.')
+        self.turned_on_at = datetime.now()
         self.on = True
         gpio.output(self.power, 1)
 
     def turn_off(self):
         """ Turn off the device """
         if self.is_off():
             return
-        self.message("Device turned off")
+        self.message('Device turned off.')
+        self.turned_on_at = False
         self.on = False
         gpio.output(self.power, 0)
 
@@ -79,19 +82,19 @@ def ontime(self):
         if self.is_off():
             return 0
         difference = datetime.now() - self.turned_on_at
-        ontime = int(difference.total_seconds() / 60)
-        self.message("Ontime {} minute(s)".format(ontime))
+        ontime = int(difference.total_seconds())
+        self.message(f'Ontime {ontime} seconds.')
         return ontime
 
 
 class PWMDevice(SwitchableDevice):
     """ Device that can be controlled with PWM signals
 
-    -:param power: Pin used for giving power supply to the fan
     -:param pwm: Pin used for pwm control
+    -:param power: Pin used for giving power supply (optional)
     -:param frequency: Frequency used for the pwm signal
     """
-    def __init__(self, power, pwm, frequency=100, **kwargs):
+    def __init__(self, pwm, power=False, frequency=100, **kwargs):
         super().__init__(power, **kwargs)
         self.frequency = frequency
         self.duty_cycle = 0 # Duty cycle between 0.0 and 100.0
@@ -104,7 +107,7 @@ def turn_on(self):
         """ Turn on the power and pwm on the lowest setting """
         if self.is_on():
             return
-        self.message("Device turned on")
+        self.message('Device turned on.')
         self.duty_cycle = 0
         self.turned_on_at = datetime.now()
         if self.pwm:
@@ -117,7 +120,7 @@ def turn_off(self):
         """ Turn off the power and pwm """
         if self.is_off():
             return
-        self.message("Device turned off")
+        self.message('Device turned off.')
         self.duty_cycle = 0
         self.turned_on_at = False
         if self.pwm:
@@ -133,14 +136,14 @@ def set_duty_cycle(self, percent, z_off=True):
         -:param z_off: Turn off at zero
         """
         if not self.pwm:
-            raise ValueError("No pin provided for outputting PWM signal")
+            raise ValueError('No pin provided for outputting PWM signal!')
         if z_off and percent == 0:
             self.turn_off()
             return
         if self.is_off():
             self.turn_on()
         elif self.duty_cycle == percent:
             return
-        self.message("Duty cycle set to {}%".format(percent))
+        self.message(f'Duty cycle set to {percent}%.')
         self.duty_cycle = percent
         self.pwm.ChangeDutyCycle(percent)

src/button.py

@@ -1,5 +1,7 @@
 import time
+
 import RPi.GPIO as gpio
+
 from .base import BaseDevice
 
 
@@ -10,24 +12,20 @@ class Button(BaseDevice):
     -:param debounce_time: Debounce time when checking button state (ms)
     -:param polarity: When you press the button, will the pin be connected to LOW or HIGH?
     """
-    def __init__(self, pin, debounce_time=200, polarity='LOW', **kwargs):
+    def __init__(self, pin, debounce_time=200, polarity=gpio.LOW, **kwargs):
         super().__init__(**kwargs)
         self.debounce_time = debounce_time / 1000
         self.polarity = polarity
         self.pin = pin
-        gpio.setup(pin, gpio.IN, pull_up_down={'HIGH': gpio.PUD_DOWN, 'LOW': gpio.PUD_UP}[polarity])
+        gpio.setup(pin, gpio.IN, pull_up_down={'0': gpio.PUD_UP, '1': gpio.PUD_DOWN}[str(polarity)])
 
     def is_pressed(self):
         """ True if the button is pressed """
-        def HIGH(pin):
-            if gpio.input(pin):
-                return True
-            return False
         def LOW(pin):
-            if not gpio.input(pin):
-                return True
-            return False
-        test = {'HIGH': HIGH, 'LOW': LOW}[self.polarity]
+            return bool(not gpio.input(pin))
+        def HIGH(pin):
+            return bool(gpio.input(pin))
+        test = {'0': LOW, '1': HIGH}[str(self.polarity)]
 
         if test(self.pin):
             time.sleep(self.debounce_time)

src/fan.py

@@ -1,10 +1,19 @@
 from time import sleep
 from datetime import datetime
 import subprocess
+from enum import IntEnum
+
 import RPi.GPIO as gpio
+
 from .base import PWMDevice
 
 
+class SM(IntEnum):
+    """ Variable positions inside speed_mapping """
+    TEMP = 0
+    SPEED = 1
+
+
 class Fan(PWMDevice):
     """ Class for controlling a fan. You can use any fan that has at least 2 wires.
     If your fan has 3-4 wires (sense and PWM) you can control it with PWM signal,
@@ -16,23 +25,23 @@ class Fan(PWMDevice):
     -:param pwm: Pin used for PWM control
     -:param cycletime: Cycle time of auto fan control (s)
     -:param frequency: Frequency used for the pwm signal (hz)
-    -:param idle_limit: If fan is on automode, and the temp is below min, only turn off after this limit is reached (m)
-    -:param speed_map: This is used to match temperature readings to fan speed
+    -:param idle_limit: If fan is on automode, and the temp is below min, only turn off after this limit is reached (s)
+    -:param speed_mapping: This is used to match temperature readings to fan speed
         Values are series of tuples in form of: (from_temp, duty_cycle)
     -:param rpm_measurement_timeout: Timeout for the edge detection (ms)
     -:param rpm_measurement_edges: How many edges to record for the calculation
     -:param rpm_measurement_bouncetime: After an edge detection, the next edge will be ignored for this duration (ms)
         This value is in miliseconds, and should be less than the time for half revolution on maximum speed.
         Calculation: rev_per_sec = max_rpm / 60 --> 1000 / rev_per_sec / 2 = max_bouncetime
     """
-    def __init__(self, power=False, sense=False, pwm=False, cycletime=5, frequency=25000, idle_limit=5,
-                 speed_map=False, rpm_measurement_timeout=2000, rpm_measurement_edges=40,
+    def __init__(self, power=False, sense=False, pwm=False, cycletime=5, frequency=25000, idle_limit=300,
+                 speed_mapping=False, rpm_measurement_timeout=2000, rpm_measurement_edges=40,
                  rpm_measurement_bouncetime=10, **kwargs):
-        super().__init__(power, pwm, frequency, **kwargs)
+        super().__init__(pwm, power, frequency, **kwargs)
         self.idle_limit = idle_limit
-        self.speed_map = speed_map or [
-            (60, 1),
-            (65, 30),
+        self.speed_mapping = speed_mapping or [
+            (20, 1),
+            (40, 30),
             (70, 50),
             (75, 70),
             (80, 100)
@@ -46,7 +55,7 @@ def __init__(self, power=False, sense=False, pwm=False, cycletime=5, frequency=2
         if sense:
             gpio.setup(sense, gpio.IN, pull_up_down=gpio.PUD_UP)
         if not power and not pwm:
-            raise ValueError("No pins provided for controlling the fan!")
+            raise ValueError('No pins provided for controlling the fan!')
 
     def set_speed(self, percent):
         """ Set fan speed.
@@ -63,38 +72,39 @@ def set_speed(self, percent):
 
     def smart_set_speed(self, percent):
         """ Set fan speed with respect to other parameters """
-        # If the desired speed is 0, but the fan not reached the idle limit, do nothing
+        # If the desired speed is 0, but the fan not reached the idle limit, set idle speed
         if percent == 0 and self.is_on() and self.ontime() < self.idle_limit:
-            return
+            percent = 1
         self.set_speed(percent)
 
     def read_hw_temperature(self):
         """ Read hardware temperatures """
-        cpu_temp = round(int(subprocess.check_output(["cat", "/sys/class/thermal/thermal_zone0/temp"]).strip()) / 1000)
-        self.message("Temperature reading {}c".format(cpu_temp))
+        cpu_temp = round(int(subprocess.check_output(['cat', '/sys/class/thermal/thermal_zone0/temp']).strip()) / 1000)
+        self.message(f'Temperature reading {cpu_temp}c.')
         return cpu_temp
 
     def measure_rpm(self):
         """ Measure fan rpm.
 
         Note: Measuring RPM from a script running under an OS (with many other things) can be inaccurate.
+        You can improve it by fine tune the "rpm_measurement_..." variables for a given fan.
         """
         if not self.sense:
-            raise ValueError("No pin was provided for sensing rpm.")
+            raise ValueError('No pin was provided for sensing rpm.')
         edges = self.rpm_measurement_edges
         start_time = datetime.now()
         for _ in range(edges):
             channel = gpio.wait_for_edge(self.sense, gpio.FALLING,
                                          timeout=self.rpm_measurement_timeout,
                                          bouncetime=self.rpm_measurement_bouncetime)
             if channel is None:
-                self.message("RPM measurement timeout")
+                self.message('RPM measurement timeout.')
                 edges = 0
                 break
         end_time = datetime.now()
         difference = (end_time - start_time).total_seconds()
         rpm = int((edges * (60 / difference)) / 2)
-        self.message("Current rpm: {}".format(rpm))
+        self.message(f'Current rpm: {rpm}')
         return rpm
 
     def temp_to_speed(self, temp):
@@ -103,19 +113,15 @@ def temp_to_speed(self, temp):
         -:param temp: Temperature in celsius
         """
         # If the temp is below the minimum level, we dont need further processing
-        if temp < self.speed_map[0][0]:
-            self.message("Temp below minimum")
+        if temp < self.speed_mapping[0][SM.TEMP]:
+            self.message('Temperature below minimum.')
             return 0
 
-        for i in range(len(self.speed_map)):
-            current_map = self.speed_map[i]
-            next_map = self.speed_map[i+1] if len(self.speed_map) > i+1 else (500, 100)
-            if current_map[0] <= temp < next_map[0]:
-                return current_map[1]
-
-        # If nothing was matched raise error
-        self.cleanup()
-        raise ValueError("Matching speed to temperature failed!")
+        for i in range(len(self.speed_mapping)):
+            current_mapping = self.speed_mapping[i]
+            next_mapping = self.speed_mapping[i+1] if len(self.speed_mapping) > i+1 else (500, 100)
+            if current_mapping[SM.TEMP] <= temp < next_mapping[SM.TEMP]:
+                return current_mapping[SM.SPEED]
 
     def auto_set(self, temp=False):
         """ Set fan speed automatically based on temperature and the speedmap