Trafasi: Laundry monitor

trafasi.md

Trafasi

Simple laundry monitor

Hardware:

• Raspberry Pi Zero
• lechacal RPIZ CT3T1 power monitor hat for Raspberry Pi Zero
• Two sct-013-000 clamps to measure the Washer and Dryer power consumption

Software:

• Using Raspberry OS lite
• To update and install dependencies:

sudo apt update
sudo apt upgrade
sudo apt install python3-serial git
git clone https://github.com/SqyD/hapy

• Then make sure the serial port of the GPIO works following these instructions

trafasi.py

#!/usr/bin/python3

# Simple laundry monitor for HA
# Found at https://gist.github.com/SqyD/98e00b8f709e0e87f3ec9be08e72568c

# Settings. Change these during setup:
config = {
  'tty': '/dev/ttyAMA0',
  'update_freq': 60,
  'devices': {
    'washer': {
      'sensor': 3,
      'read_threshold': 25,
      'state_threshold': 40,
      'repeat_threshold': 5,
      'power_default_state': {
        'state': False,
        'attributes': {'state_class': 'measurement', 'device_class': 'power', 'unit_of_measurement': 'W'}
      },
    },
    'dryer': {
      'sensor': 2,
      'read_threshold': 50,
      'state_threshold': 40,
      'repeat_threshold': 15,
      'power_default_state': {
        'state': False,
        'attributes': {'state_class': 'measurement', 'device_class': 'power', 'unit_of_measurement': 'W'}
      },
    },
  },
}
secrets = {
  'url': 'http://ha_host',
  'access_token': 'changeme',
}

import serial, time
from hapy import hapy

class Device():
  def __init__(self, id, config):
    self.id = id
    self.config = config
    self.state = False
    self.power_hist = list()

  def read(self, raw_data):
    power = int(float(raw_data[self.config['sensor']]))
    if power < self.config['read_threshold']:
      power = 0
    self.update(power)

  def update(self, power):
    self.power_hist.insert(0, power)
    if len(self.power_hist) > self.config['repeat_threshold']:
      del self.power_hist[self.config['repeat_threshold']]
    state_data = self.config['power_default_state']
    state_data['state'] = power
    ha.entity_set('sensor.' + self.id + '_power', state_data)
    count = 0
    for reading in self.power_hist:
      if (reading > self.config['state_threshold']) != self.state:
         count = count + 1
    if count >= self.config['repeat_threshold']:
      self.setstate(not(self.state))

  def setstate(self, state):
    self.state = state
    if state:
      ha.entity_set_state('binary_sensor.' + self.id, 'on')
    else:
      ha.entity_set_state('binary_sensor.' + self.id, 'off')

def read_data():
  # Read one line from the serial buffer
  line = ser.readline().decode().strip()

  # Create an array of the data
  raw_data = line.split(' ')
  return raw_data

ser = serial.Serial(config['tty'], 38400)
ha = hapy.HaPyRest(secrets)

devices = {}

for id, dev_config in config['devices'].items():
  devices[id] = Device(id, dev_config)

while True:
  last_update = time.time()
  raw_data = read_data()
  for device in devices:
    devices[device].read(raw_data)
  wait_time = config['update_freq'] - (time.time() - last_update)
  time.sleep(wait_time)