d/pyptqs1005
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

initial

Browse Source
This commit is contained in:
quackerd 2022-01-22 05:03:29 +08:00
parent 1b6066d258
commit 39191643b5
2 changed files with 242 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
libptqs1005/__init__.py Normal file
View File

@ -0,0 +1,109 @@
import serial
import time
import periphery
class PTQS1005AC:
RESP_LEN = 42
def __make_16bit_int(self, high, low) -> int:
return int(high) << 8 | int(low)
def __init__(self):
self.pm10 = 0
self.pm25 = 0
self.pm100 = 0
self.pm10_atm = 0
self.pm25_atm = 0
self.pm100_atm = 0
self.part3 = 0
self.part5 = 0
self.part10 = 0
self.part25 = 0
self.part50 = 0
self.part100 = 0
self.tvoc = 0
self.tvoc_quan = 0
self.hcho = 0
self.tvoc_quan = 0
self.co2 = 0
self.temp = 0
self.hum = 0
def __init__(self, raw_resp : bytes):
if len(raw_resp) != 42:
raise Exception("Invalid response length: " + str(len(raw_resp)))
if (raw_resp[0] != 0x42 or raw_resp[1] != 0x4d or raw_resp[2] != 0x00 or raw_resp[3] != 0x26):
raise Exception("Invalid magic header")
checksum = self.__make_16bit_int(raw_resp[40], raw_resp[41])
sum = 0
for i in range(0, 40):
sum = sum + int(raw_resp[i])
if sum != checksum:
raise Exception("Invalid checksum")
self.pm10 = self.__make_16bit_int(raw_resp[4], raw_resp[5])
self.pm25 = self.__make_16bit_int(raw_resp[6], raw_resp[7])
self.pm100 = self.__make_16bit_int(raw_resp[8], raw_resp[9])
self.pm10_atm = self.__make_16bit_int(raw_resp[10], raw_resp[11])
self.pm25_atm = self.__make_16bit_int(raw_resp[12], raw_resp[13])
self.pm100_atm = self.__make_16bit_int(raw_resp[14], raw_resp[15])
self.part3 = self.__make_16bit_int(raw_resp[16], raw_resp[17])
self.part5 = self.__make_16bit_int(raw_resp[18], raw_resp[19])
self.part10 = self.__make_16bit_int(raw_resp[20], raw_resp[21])
self.part25 = self.__make_16bit_int(raw_resp[22], raw_resp[23])
self.part50 = self.__make_16bit_int(raw_resp[24], raw_resp[25])
self.part100 = self.__make_16bit_int(raw_resp[26], raw_resp[27])
self.tvoc = self.__make_16bit_int(raw_resp[28], raw_resp[29]) / 100.0
self.tvoc_quan = int(raw_resp[30])
self.hcho = self.__make_16bit_int(raw_resp[31], raw_resp[32]) / 100.0
self.tvoc_quan = int(raw_resp[33])
self.co2 = self.__make_16bit_int(raw_resp[34], raw_resp[35])
self.temp = self.__make_16bit_int(raw_resp[36], raw_resp[37]) / 10.0
self.hum = self.__make_16bit_int(raw_resp[38], raw_resp[39]) / 10.0
class PTQS1005Driver:
def __init__(self, tty : str, reset_pin : int = None):
# init serial port
self.ser = serial.Serial(tty, 9600, timeout = 1)
self.ser.flushInput()
self.ser.flushOutput()
# time.sleep(10)
if reset_pin != None:
self.reset_gpio = periphery.GPIO(reset_pin)
else:
self.reset_gpio = None
def __make_cmd(self, cmd : int, data : int) -> bytes :
arr = bytearray(7)
arr[0] = 0x42
arr[1] = 0x4d
arr[2] = cmd & 0xFF
arr[3] = (data & 0xFF00) >> 8
arr[4] = data & 0xFF
checksum = 0
for i in range(0,5):
checksum = checksum + int(arr[i])
arr[5] = (checksum & 0xFF00) >> 8
arr[6] = (checksum & 0xFF)
return bytes(arr)
def read(self) -> PTQS1005AC:
cmd = self.__make_cmd(0xac, 0)
self.ser.write(cmd)
self.ser.flush()
time.sleep(1e-3)
resp = self.ser.read(PTQS1005AC.RESP_LEN, timeout = 3)
if(len(resp) != PTQS1005AC.RESP_LEN):
raise Exception("invalid response length")
return PTQS1005AC(resp)
def reset(self):
raise Exception("unimplemented yet!")

133
monitor.py Normal file
View File

@ -0,0 +1,133 @@
from warnings import catch_warnings
import libptqs1005
import time
import logging
import influxdb
import singal
NUM_READS = 10
DELAY_PER_READ = 1 * 1000
DELAY_PER_MEASURE = 300 * 1000
DB_USER = 'sensor'
DB_PSWD = 'sensor'
DB_NAME = 'sensor'
g_exit = False
COLUMN_MAP = {
"temp" : "Temperature",
"hum" : "Humidity",
"pm10" : "PM1 (std)",
"pm25" : "PM2.5 (std)",
"pm100" : "PM10 (std)",
"pm10_atm" : "PM10 (atmosphere)",
"pm25_atm" : "PM2.5 (atmosphere)",
"pm100_atm" : "PM10 (atmosphere)",
"tvoc" : "TVOC",
"tvoc_quan" : "TVOC quantity",
"hcho" : "HCHO",
"hcho_quan" : "HCHO quantity",
"co2" : "CO2",
"part3" : "0.3um particles",
"part5" : "0.5um particles",
"part10" : "1um particles",
"part25" : "2.5um particles",
"part50" : "5um particles",
"part100" : "10um particles"
}
def signal_handler():
global g_exit
logging.warn("SIGINT detected, exiting gracefully...")
g_exit = True
def current_ms():
return round(time.time() * 1000)
def measure(drv : libptqs1005.PTQS1005Driver) -> libptqs1005.PTQS1005AC:
all_data = []
for i in range(0, NUM_READS):
all_data.append(drv.read())
logging.debug("Measurement " + str(i) + " out of " + str(NUM_READS) + ".")
ret = libptqs1005.PTQS1005AC()
for attr in COLUMN_MAP:
# average all measurements
val = 0
for dat in all_data:
val += getattr(dat, attr)
val = val / NUM_READS
setattr(ret, attr, val)
return ret
def setup_db() -> influxdb.InfluxDBClient:
logging.info("Setting up InfluxDB connection...")
conn = influxdb.InfluxDBClient(host="nas", port=8086, ssl=True, username=DB_USER, password=DB_PSWD)
ret = conn.get_list_database()
skip_create = False
logging.info("Checking whether database exists...")
for dicti in ret:
if dicti['name'] == DB_NAME:
skip_create = True
if (not skip_create):
logging.info("Creating database...")
conn.create_database(DB_NAME)
conn.switch_database(DB_NAME)
return conn
def data_to_line(data : libptqs1005.PTQS1005AC) -> str:
ret = "air "
for col in COLUMN_MAP:
desc = COLUMN_MAP[col].replace(" ", "\\ ")
attr = col
ret = ret + desc + "=" + str(getattr(data, attr)) + ","
ret = ret[:-1]
return ret
def main():
logging.basicConfig(filename="monitor.log", encoding='utf-8', level=logging.DEBUG)
logging.info("Setting up signal handler...")
signal.signal(signal.SIGINT, signal_handler)
logging.info("Setting up sensor driver...")
drv = libptqs1005.PTQS1005Driver(tty="/dev/ttyAMA0", reset_pin=23)
dbconn = setup_db()
logging.info("Begin monitoring...")
next_ms = current_ms()
while(True):
if g_exit:
break
cur_ms = current_ms()
if cur_ms >= next_ms:
logging.info("Reading data from sensor. Current milliseconds: " + str(cur_ms))
try:
data = measure(drv)
except Exception as ex:
logging.error("Failed to read data from sensor. Error: " + ex.args)
logging.info("Uploading data to database...")
dbline = data_to_line(data)
logging.debug("DB line: " + dbline)
if not dbconn.write(data = dbline, protocol='line'):
logging.error("Failed to update database.")
#print("PM2.5: " + str(data.pm25) + " PM10: " + str(data.pm100) + " PM1: " + str(data.pm10) + " Temp: " + str(data.temp) + " Humidity: " + str(data.hum) + " CO2: " + str(data.co2) + " HCHO: " + str(data.hcho) + " TVOC: " + str(data.tvoc) )
next_ms = cur_ms + DELAY_PER_MEASURE
logging.debug("Next measurement point: " + str(next_ms))
else:
logging.debug("Next measurement point hasn't arrived. Sleeping for 1s...")
time.sleep(1)
logging.info("Stopped monitoring...")
main()