hakkoso/sensors.py

221 lines
6.5 KiB
Python
Raw Normal View History

from time import perf_counter
from datetime import datetime, timedelta
from collections import deque
2023-02-21 22:47:49 +01:00
from statistics import mean
from itertools import islice
import re
2022-11-19 19:25:46 +01:00
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
2023-02-19 20:06:15 +01:00
# TODO client: add a way to disable live plotting? server: stop
# sending data periodically to everybody!
from random import random
from time import sleep
2023-02-21 22:47:49 +01:00
# How many seconds to use
WINDOW_SIZE_S = 5
class Sensor():
def __init__(self, autocorr=0.7):
2023-02-21 22:53:22 +01:00
# FIXME: DEPENDS ON SAMPLING RATE
self.history = deque([], int(WINDOW_SIZE_S * 2))
2023-02-21 22:47:49 +01:00
## This is just for debugging
self.prevval = random() * 100
self.auto = autocorr
self.measure = None
def read(self):
sleep(0.5)
2023-02-21 22:47:49 +01:00
return self.store(
perf_counter(),
int((random() * (1-self.auto) + self.prevval * self.auto) * 100) / 100)
def smooth(self):
# TODO: Implement *correctly* this (take into account the time)
2023-02-22 00:38:07 +01:00
# m = mean([el[1] for el in self.history])
# Non mi piace proprio in realtà
2023-02-21 22:47:49 +01:00
return (self.history[-1][0], m)
def store(self, value, time):
self.history.append((value, time))
2023-02-22 00:38:07 +01:00
# self.smooth()
return (value, time)
2023-02-21 22:47:49 +01:00
2023-01-05 15:44:41 +01:00
class GPIOState(Sensor):
def __init__(self, pin, transform=lambda x: 1-x):
2023-02-21 22:47:49 +01:00
super().__init__()
2023-01-05 15:51:31 +01:00
self.measure = 'Switch'
2023-01-05 15:44:41 +01:00
self.pin = pin
self.transform = transform
def read(self):
2023-01-05 15:49:33 +01:00
try:
self.value = GPIO.input(self.pin)
except:
print(f"Could not read pin {self.pin}")
self.value = 0
2023-01-05 15:49:33 +01:00
self.time = perf_counter()
2023-02-21 22:53:22 +01:00
# Even if we don't need to return the smoothed value, we still smooth it
# to get the percentage of use over the period (if it will be needed)
value = self.transform(self.value)
self.store(self.time, value)
return (self.time, value)
2023-01-05 15:44:41 +01:00
class Temperature1W(Sensor):
def __init__(self, address):
2023-02-21 22:47:49 +01:00
super().__init__()
self.measure = 'Temperature'
self.address = address
def path(self):
return f'/sys/bus/w1/devices/{self.address}/w1_slave'
def read(self):
2022-11-19 19:25:46 +01:00
path = self.path()
if not exists(path):
# WIP
return (perf_counter(), 0.0)
with open(path, "r") as f:
content = f.readlines()
2022-11-19 19:04:56 +01:00
time = perf_counter()
if len(content) < 2:
return None
if content[0].strip()[-3:] != "YES":
print("INVALID CHECKSUM")
2023-01-05 13:27:31 +01:00
return (time, None)
2023-02-21 22:47:49 +01:00
return self.store(
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
2023-02-19 20:06:15 +01:00
class SHT40(Sensor):
def __init__(self, what, every=600):
2023-02-21 22:47:49 +01:00
super().__init__()
if what not in ("Temperature", "Humidity"):
print("ERROR: invalid sensor value: ", what)
return
2023-02-19 20:06:15 +01:00
self.measure = what
self.heat_every = every
if self.is_humidity():
self.last_heat = perf_counter()
if not enable_sht:
return
2023-02-21 14:19:12 +01:00
# alternative: HIGHHEAT_1S
self.heatmode = adafruit_sht4x.Mode.LOWHEAT_100MS
self.standardmode = SHT40_DEFAULT
2023-02-21 14:19:12 +01:00
self.serial = hex(sht.serial_number)
# mode: adafruit_sht4x.Mode.string[sht.mode]
2023-02-19 20:06:15 +01:00
# Can also set the mode to enable heater
# sht.mode = adafruit_sht4x.Mode.LOWHEAT_100MS
2023-02-21 14:19:12 +01:00
def is_temperature(self):
return self.measure == "Temperature"
def is_humidity(self):
return self.measure == "Humidity"
def reset_mode(self):
sht.mode = self.standardmode
2023-02-19 20:06:15 +01:00
def read(self):
time = perf_counter()
if not enable_sht:
return (time, None)
reset = False
if self.is_humidity():
timediff = time - self.last_heat
if timediff > self.heat_every:
reset = True
sht.mode = self.heatmode
2023-02-19 20:06:15 +01:00
temperature, relative_humidity = sht.measurements
if reset:
self.reset_mode()
2023-02-21 14:19:12 +01:00
self.last_heat = perf_counter()
2023-02-21 22:47:49 +01:00
return self.store(
time, temperature if self.measure == 'Temperature' else relative_humidity)
2023-02-19 20:06:15 +01:00
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 = {
2023-02-19 20:06:15 +01:00
'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, n=None):
l = len(self.history)
return tuple(islice(self.history, max(0, l - n), l))
sensors = Sensors()
def get_sensor_value(sensor_name):
return sensors.get_sensor_value(sensor_name)[1]