hakkoso/sensors.py

from time import perf_counter
from datetime import datetime, timedelta
from collections import deque
import re
from os.path import exists

try:
    import RPi.GPIO as GPIO
    GPIO.setmode(GPIO.BCM)
    rpi = True
except:
    rpi = False

try:
    # SHT40
    import board
    import adafruit_sht4x
    enable_sht = True
except:
    print("Adafruit SHT4X not available!")
    enable_sht = False

# TODO client: add a way to disable live plotting? server: stop
# sending data periodically to everybody!

from random import random
from time import sleep

class Sensor():
    def __init__(self, autocorr=0.7):
        self.prevval = random() * 100
        self.auto = autocorr
        self.measure = None

    def read(self):
        sleep(0.5)
        return (perf_counter(),
                int((random() * (1-self.auto) + self.prevval * self.auto) * 100) / 100)

class GPIOState(Sensor):
    def __init__(self, pin, transform=lambda x: 1-x):
        self.measure = 'Switch'

        self.pin = pin
        self.transform = transform

    def read(self):
        try:
            self.value = GPIO.input(self.pin)
        except:
            print(f"Could not read pin {self.pin}")
            self.value = 0
        self.time = perf_counter()
        
        return (self.time, self.transform(self.value))

class Temperature1W(Sensor):
    def __init__(self, address):
        self.measure = 'Temperature'
        self.address = address

    def path(self):
        return f'/sys/bus/w1/devices/{self.address}/w1_slave'

    def read(self):
        path = self.path()
        if not exists(path):
            # WIP
            return (perf_counter(), 0.0)
        with open(path, "r") as f:
            content = f.readlines()
            time = perf_counter()
        if len(content) < 2:
            return None
        if content[0].strip()[-3:] != "YES":
            print("INVALID CHECKSUM")
            return (time, None)
        return (time, int(re.search("t=([0-9]+)", content[1]).group(1)) / 1000.0)

if enable_sht:
    i2c = board.I2C()  # uses board.SCL and board.SDA
    sht = adafruit_sht4x.SHT4x(i2c)
    print('Serial: ', hex(sht.serial_number))
    SHT40_DEFAULT = adafruit_sht4x.Mode.NOHEAT_HIGHPRECISION
    sht.mode = SHT40_DEFAULT

class SHT40(Sensor):
    def __init__(self, what, every=60):
        if what not in ("Temperature", "Humidity"):
            print("ERROR: invalid sensor value: ", what)
            return
        self.measure = what
        if not enable_sht:
            return
        self.heat_every = every
        if self.measure != "Temperature":
            self.last_heat = perf_counter()
        self.heatmode = adafruit_sht4x.Mode.HIGHHEAT_1S
        self.standardmode = SHT40_DEFAULT
        # mode: adafruit_sht4x.Mode.string[sht.mode]
        # Can also set the mode to enable heater
        # sht.mode = adafruit_sht4x.Mode.LOWHEAT_100MS
        # modes = [adafruit_sht4x.Mode.HIGHHEAT_1S,
        #          adafruit_sht4x.Mode.LOWHEAT_100MS,
        #          adafruit_sht4x.Mode.NOHEAT_HIGHPRECISION]
    def is_temperature(self):
        return self.measure == "Temperature"

    def is_humidity(self):
        return self.measure == "Humidity"

    def reset_mode(self):
        sht.mode = self.standardmode

    def read(self):
        time = perf_counter()
        reset = False
        if self.is_humidity():
            timediff = time - self.last_heat
            if timediff > self.heat_every:
                reset = True
                sht.mode = self.heatmode
        temperature, relative_humidity = sht.measurements
        if reset:
            self.reset_mode()
        return (time, temperature if self.measure == 'Temperature' else relative_humidity)

class Sensors():
    def __init__(self, history=2621440):
        self.starttime = (datetime.now(), perf_counter())
        self.scan()
        self.values = {}
        self.history = deque([], maxlen=history)

    def perf_datetime(self, offset):
        time = offset - self.starttime[1]
        return self.starttime[0] + timedelta(seconds=time)

    "Scan for new sensors"
    def scan(self):
        # Read something like this from a stored prefs json
        # sensors_names_map = {}
        # Then, scan, apply existing names and use placholders for new sensors
        # FIXME: should scan, apply stored data and return this
        self.available_sensors = {
            'T_food':     Temperature1W('28-06214252b671'),
            'T_ext':      SHT40('Temperature'),
            'H_ext':      SHT40('Humidity'),
            'heater':     GPIOState(22),
            'humidifier': GPIOState(17),
            'fan':        GPIOState(27),
        }

    def list(self):
        return {
            k: self.available_sensors[k].measure
            for k in self.available_sensors.keys()
        }

    def value_tuple(self):
        return tuple((k, *self.values[k]) for k in self.values.keys())

    def read(self):
        for sensor in self.available_sensors.keys():
            try:
                time, value = self.available_sensors[sensor].read()
                self.values[sensor] = (str(self.perf_datetime(time)), value)
            except Exception as e:
                print("Error reading sensors", sensor, ": ", e)
        self.history.append(self.value_tuple())

    def get_sensor_value(self, sensor_name):
        return self.values.get(sensor_name, None)

    def get(self):
        return self.value_tuple()

    def get_history(self):
        return tuple(self.history)

sensors = Sensors()

def get_sensor_value(sensor_name):
    return sensors.get_sensor_value(sensor_name)[1]