CO2 (二酸化炭素) 濃度に反応する「自動換気システム」を2通り自作してみた

#!/bin/bash

# co2_fan.sh (ver.20211002)
# usage: $0 [CO2_NOW(ppm)] [CO2_THRESHOLD(ppm)] [{ON_only | OFF_only}]

if [[ $# -lt 2 ]]; then
	exit 1
fi
CO2_NOW=$1
if [[ ${CO2_NOW} -lt 1 ]]; then
	exit 2
fi
CO2_THRESHOLD=$2 # 閾値(ppm)
CO2_MIN=400 # 測定上の最低値(ppm)
CO2_TMP=`echo "${CO2_MIN} + (( ${CO2_THRESHOLD} - ${CO2_MIN} ) * 0.93)" | bc` # 下げたい値(ppm)
CO2_SAFE=`echo "${CO2_TMP}" | awk '{printf("%d", $1 + 0.5)}'` # 下げたい値の整数値(ppm)
FILE_STATUS='/tmp/fan_status.dat'
MODE_SWITCH=$3 # ON_only or OFF_only
FAN_KEEPSEC=300 # FILE_STATUSの有効時間(sec)

if [[ ${HOSTNAME} == 'raspi4' ]]; then
	CMD_FAN='switch_tuya.py'
elif [[ ${HOSTNAME} == 'raspi0' ]]; then
	CMD_FAN='switch_meross.py'
fi

func_switchfan() {
	if [[ "$1" == 'on' ]]; then
		${CMD_FAN} "ON"
	elif [[ "$1" == 'off' ]]; then
		${CMD_FAN} "OFF"
	fi
}
func_savestatus() {
	echo "$1" > ${FILE_STATUS}
}
func_checkstatus() {
	local FAN_STATUS="$1"
	if [[ -e ${FILE_STATUS} ]]; then
		FAN_SWITCHING=0
		local FAN_ELAPSEDSEC=$(( `date "+%s"` - `date "+%s" -r ${FILE_STATUS}` ))
		local FAN_LASTSTATUS=`cat ${FILE_STATUS}`
		if [[ ${FAN_ELAPSEDSEC} -gt ${FAN_KEEPSEC} ]]; then
			FAN_SWITCHING=1
		elif [[ ${FAN_STATUS} != ${FAN_LASTSTATUS} ]]; then
			FAN_SWITCHING=1
		fi
	else
		FAN_SWITCHING=1
	fi
}

if [[ ${CO2_NOW} -ge ${CO2_SAFE} ]]; then # 閾値以上
	if [[ ${MODE_SWITCH} != 'OFF_only' ]]; then
		func_checkstatus 'on'
		if [[ ${FAN_SWITCHING} -eq 1 ]]; then
			func_switchfan 'on'
			func_savestatus 'on'
		fi
	fi
elif [[ ${CO2_NOW} -ge ${CO2_MIN} ]]; then # 最低値以上
	if [[ ${MODE_SWITCH} != 'ON_only' ]]; then
		func_checkstatus 'off'
		if [[ ${FAN_SWITCHING} -eq 1 ]]; then
			func_switchfan 'off'
			func_savestatus 'off'
		fi
	fi
else # 何らかの異常値
	:
fi

#!/usr/bin/env python3

# switch_tuya.py (ver.20210903)

import tinytuya # https://github.com/jasonacox/tinytuya
import sys

d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', 'LOCAL_KEY_HERE')
d.set_version(3.1)
data = d.status()

print('Dictionary %r' % data)
switch_state = data['dps']['1']

args = sys.argv
if len(args) == 2:
	if args[1] == 'ON':
		data = d.set_status(True)
	elif args[1] == 'OFF':
		data = d.set_status(False)
	elif args[1] == 'TOGGLE':
		data = d.set_status(not switch_state)
	else:
		pass
	if data:
		print('set_status() result %r' % data)
else:
	print('Usage: %s [ON|OFF|TOGGLE]' % args[0])

#!/usr/bin/env python3

# switch_meross.py (ver.20211109)
# ref: albertogeniola/MerossIot: Simple Python library for Meross devices https://github.com/albertogeniola/MerossIot

import asyncio
import os
import sys
from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or 'YOUR_MEROSS_CLOUD_EMAIL'
PASSWORD = os.environ.get('MEROSS_PASSWORD') or 'YOUR_MEROSS_CLOUD_PASSWORD'
NAME_PLUG = 'HOGEHOGE' # 制御したいスマートプラグの名称

async def main(status):
	http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD)
	manager = MerossManager(http_client=http_api_client)
	await manager.async_init()
	await manager.async_device_discovery()
	plugs = manager.find_devices(device_type="mss110")
	if len(plugs) < 1:
		print("No MSS110 plugs found...")
	else:
		for dev in plugs:
			print(f"Device = {dev.name}")
			await dev.async_update()
			if dev.name == NAME_PLUG:
				if status == 'on':
					print(f"Turning on {dev.name}...")
					await dev.async_turn_on(channel=0)
				elif status == 'off':
					print(f"Turing off {dev.name}")
					await dev.async_turn_off(channel=0)
	manager.close()
	await http_api_client.async_logout()

if __name__ == '__main__':
	args = sys.argv
	status = ''
	if len(args) == 2:
		if args[1] == 'ON':
			status = 'on'
		elif args[1] == 'OFF':
			status = 'off'
		else:
			print('Usage: %s [ON|OFF]' % args[0])
	else:
		print('Usage: %s [ON|OFF]' % args[0])
	if status:
		loop = asyncio.get_event_loop()
		loop.run_until_complete(main(status))
		loop.close()