sensing_07.py の履歴(No.1)

履歴一覧
差分を表示
現在との差分を表示
ソースを表示
sensing_07.py へ行く。
- 1 (2025-03-16 (日) 20:20:13)

# -*- coding: utf-8 -*-
from machine import UART,Pin,ADC, Timer
import micropython,time,gc
# for analog co2 sensor, gravity:analog infrared co2 sensor
# 20250114 
 
# Updated 2018 and 2020
# This module is based on the below cited resources, which are all
# based on the documentation as provided in the Bosch Data Sheet and
# the sample implementation provided therein.
#
# Final Document: BST-BME280-DS002-15
#
# Authors: Paul Cunnane 2016, Peter Dahlebrg 2016
#
# This module borrows from the Adafruit BME280 Python library. Original
# Copyright notices are reproduced below.
#
# Those libraries were written for the Raspberry Pi. This modification is
# intended for the MicroPython and esp8266 boards.
#
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
#
# Based on the BMP280 driver with BME280 changes provided by
# David J Taylor, Edinburgh (www.satsignal.eu)
#
# Based on Adafruit_I2C.py created by Kevin Townsend.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
 
from ustruct import unpack, unpack_from
from array import array
 
# BME280 default address.
BME280_I2CADDR = 0x76
 
# Operating Modes
BME280_OSAMPLE_1 = 1
BME280_OSAMPLE_2 = 2
BME280_OSAMPLE_4 = 3
BME280_OSAMPLE_8 = 4
BME280_OSAMPLE_16 = 5
 
BME280_REGISTER_CONTROL_HUM = 0xF2
BME280_REGISTER_STATUS = 0xF3
BME280_REGISTER_CONTROL = 0xF4
 
MODE_SLEEP = const(0)
MODE_FORCED = const(1)
MODE_NORMAL = const(3)
 
BME280_TIMEOUT = const(100)  # about 1 second timeout
 
 
class BME280:
 
    def __init__(self,
                 mode=BME280_OSAMPLE_8,
                 address=BME280_I2CADDR,
                 i2c=None,
                 **kwargs):
        # Check that mode is valid.
        if type(mode) is tuple and len(mode) == 3:
            self._mode_hum, self._mode_temp, self._mode_press = mode
        elif type(mode) == int:
            self._mode_hum, self._mode_temp, self._mode_press = mode, mode, mode
        else:
            raise ValueError("Wrong type for the mode parameter, must be int or a 3 element tuple")
 
        for mode in (self._mode_hum, self._mode_temp, self._mode_press):
            if mode not in [BME280_OSAMPLE_1, BME280_OSAMPLE_2, BME280_OSAMPLE_4,
                            BME280_OSAMPLE_8, BME280_OSAMPLE_16]:
                raise ValueError(
                    'Unexpected mode value {0}. Set mode to one of '
                    'BME280_ULTRALOWPOWER, BME280_STANDARD, BME280_HIGHRES, or '
                    'BME280_ULTRAHIGHRES'.format(mode))
 
        self.address = address
        if i2c is None:
            raise ValueError('An I2C object is required.')
        self.i2c = i2c
        self.__sealevel = 101325
 
        # load calibration data
        dig_88_a1 = self.i2c.readfrom_mem(self.address, 0x88, 26)
        dig_e1_e7 = self.i2c.readfrom_mem(self.address, 0xE1, 7)
        self.dig_T1, self.dig_T2, self.dig_T3, self.dig_P1, \
            self.dig_P2, self.dig_P3, self.dig_P4, self.dig_P5, \
            self.dig_P6, self.dig_P7, self.dig_P8, self.dig_P9, \
            _, self.dig_H1 = unpack("<HhhHhhhhhhhhBB", dig_88_a1)
 
        self.dig_H2, self.dig_H3, self.dig_H4,\
            self.dig_H5, self.dig_H6 = unpack("<hBbhb", dig_e1_e7)
        # unfold H4, H5, keeping care of a potential sign
        self.dig_H4 = (self.dig_H4 * 16) + (self.dig_H5 & 0xF)
        self.dig_H5 //= 16
 
        self.t_fine = 0
 
        # temporary data holders which stay allocated
        self._l1_barray = bytearray(1)
        self._l8_barray = bytearray(8)
        self._l3_resultarray = array("i", [0, 0, 0])
 
        self._l1_barray[0] = self._mode_temp << 5 | self._mode_press << 2 | MODE_SLEEP
        self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL,
                             bytearray([0x3c | MODE_SLEEP]))
 
    def read_raw_data(self, result):
        """ Reads the raw (uncompensated) data from the sensor.
 
            Args:
                result: array of length 3 or alike where the result will be
                stored, in temperature, pressure, humidity order
            Returns:
                None
        """
 
        self._l1_barray[0] = self._mode_hum
        self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL_HUM,
                             self._l1_barray)
        self._l1_barray[0] = self._mode_temp << 5 | self._mode_press << 2 | MODE_FORCED
        self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL,
                             self._l1_barray)
 
        # Wait for conversion to complete
        for _ in range(BME280_TIMEOUT):
            if self.i2c.readfrom_mem(self.address, BME280_REGISTER_STATUS, 1)[0] & 0x08:
                time.sleep_ms(10)  # still busy
            else:
                break  # Sensor ready
        else:
            raise RuntimeError("Sensor BME280 not ready")
 
        # burst readout from 0xF7 to 0xFE, recommended by datasheet
        self.i2c.readfrom_mem_into(self.address, 0xF7, self._l8_barray)
        readout = self._l8_barray
        # pressure(0xF7): ((msb << 16) | (lsb << 8) | xlsb) >> 4
        raw_press = ((readout[0] << 16) | (readout[1] << 8) | readout[2]) >> 4
        # temperature(0xFA): ((msb << 16) | (lsb << 8) | xlsb) >> 4
        raw_temp = ((readout[3] << 16) | (readout[4] << 8) | readout[5]) >> 4
        # humidity(0xFD): (msb << 8) | lsb
        raw_hum = (readout[6] << 8) | readout[7]
 
        result[0] = raw_temp
        result[1] = raw_press
        result[2] = raw_hum
 
    def read_compensated_data(self, result=None):
        """ Reads the data from the sensor and returns the compensated data.
 
            Args:
                result: array of length 3 or alike where the result will be
                stored, in temperature, pressure, humidity order. You may use
                this to read out the sensor without allocating heap memory
 
            Returns:
                array with temperature, pressure, humidity. Will be the one
                from the result parameter if not None
        """
        self.read_raw_data(self._l3_resultarray)
        raw_temp, raw_press, raw_hum = self._l3_resultarray
        # temperature
        var1 = (((raw_temp // 8) - (self.dig_T1 * 2)) * self.dig_T2) // 2048
        var2 = (raw_temp // 16) - self.dig_T1
        var2 = (((var2 * var2) // 4096) * self.dig_T3) // 16384
        self.t_fine = var1 + var2
        temp = (self.t_fine * 5 + 128) // 256
 
        # pressure
        var1 = self.t_fine - 128000
        var2 = var1 * var1 * self.dig_P6
        var2 = var2 + ((var1 * self.dig_P5) << 17)
        var2 = var2 + (self.dig_P4 << 35)
        var1 = (((var1 * var1 * self.dig_P3) >> 8) +
                ((var1 * self.dig_P2) << 12))
        var1 = (((1 << 47) + var1) * self.dig_P1) >> 33
        if var1 == 0:
            pressure = 0
        else:
            p = ((((1048576 - raw_press) << 31) - var2) * 3125) // var1
            var1 = (self.dig_P9 * (p >> 13) * (p >> 13)) >> 25
            var2 = (self.dig_P8 * p) >> 19
            pressure = ((p + var1 + var2) >> 8) + (self.dig_P7 << 4)
 
        # humidity
        h = self.t_fine - 76800
        h = (((((raw_hum << 14) - (self.dig_H4 << 20) -
                (self.dig_H5 * h)) + 16384) >> 15) *
             (((((((h * self.dig_H6) >> 10) *
                (((h * self.dig_H3) >> 11) + 32768)) >> 10) + 2097152) *
              self.dig_H2 + 8192) >> 14))
        h = h - (((((h >> 15) * (h >> 15)) >> 7) * self.dig_H1) >> 4)
        h = 0 if h < 0 else h
        h = 419430400 if h > 419430400 else h
        humidity = h >> 12
        if humidity < 0:
            humidity = 0
        if humidity > 100 * 1024:
            humidity = 100 * 1024
 
        if result:
            result[0] = temp
            result[1] = pressure
            result[2] = humidity
            return result
 
        return array("i", (temp, pressure, humidity))
 
    @property
    def sealevel(self):
        return self.__sealevel
 
    @sealevel.setter
    def sealevel(self, value):
        if 30000 < value < 120000:  # just ensure some reasonable value
            self.__sealevel = value
 
    @property
    def altitude(self):
        '''
        Altitude in m.
        '''
        from math import pow
        try:
            p = 44330 * (1.0 - pow((self.read_compensated_data()[1] / 256) /
                                   self.__sealevel, 0.1903))
        except:
            p = 0.0
        return p
 
    @property
    def dew_point(self):
        """
        Compute the dew point temperature for the current Temperature
        and Humidity measured pair
        """
        from math import log
        t, p, h = self.read_compensated_data()
        t /= 100
        h /= 1024
        h = (log(h, 10) - 2) / 0.4343 + (17.62 * t) / (243.12 + t)
        return (243.12 * h / (17.62 - h)) * 100
 
    @property
    def values(self):
        """ human readable values """
 
        t, p, h = self.read_compensated_data()
 
        p = p / 256
 
        h = h / 1024
        return ("{}C".format(t / 100), "{:.02f}hPa".format(p/100),
                "{:.02f}%".format(h))
 
class Sensing:
 
    def get_co2(self):
        return self.co2_value
    def get_motion(self):
        return self.motion_value
    def get_lx(self):
        return self.lx_value
    def get_pressure(self):
        return self.pressure_value
    def get_temperature(self):
        return self.temperature_value
    def get_humidity(self):
        return self.humidity_value
    def co2_read(self):
        #print("start read co2")
        #global o_b, r_b, co2_value
        self.o_b = bytearray([0xFF, 0x01, 0x86, 0x00, 0x00, 0x00, 0x00, 0x00, 0x79])
        self.r_b=bytearray([0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00 ])
        try:
            self.u.write(self.o_b)
            time.sleep(10)
            result = self.u.readinto(self.r_b,9)
            #print(result)
            v=int(0x00000 | (self.r_b[2]<<8 | self.r_b[3]))
            #print("co2:"+str(v))
            self.co2_value = v
        except Exception as e:
            #print(e)
            return
        #adcVal = self.co2.read_u16()
        #vx = float(adcVal)*(3.3/65536.0)
        #if vx == 0:
        #    #print("A problem has occurred with the sensor.")
        #    self.co2_value=0
        #elif vx < 0.4:
        #    #print("Pre-heating the sensor...")
        #    self.co2_value=0
        #else:
        #    vd=vx-0.4
        #    concentration=(vd*5000.0)/1.6
        #    self.co2_value = int(concentration)
    def get_smell(self):
        return self.smell_value
    def get_smell2(self):
        return self.smell2_value
    def get_particle(self):
        self.particle_read()
        return self.particle_concentration
    motion_counter = 0
    motion_accum = 0
    def motion_read(self):
        #print("start read motion")
        #global motion_counter, motion_accum, motion_value
        if self.motion_counter < 100:
            self.motion_counter = self.motion_counter + 1
            self.motion_accum = self.motion_accum + self.pir.value()
        else:
            self.motion_value = self.motion_accum
            self.motion_counter = 0
            self.motion_accum = 0
            #print("motion:"+str(self.motion_value))    
 
    def lx_read(self):
        #global lx_value
        self.lx_value = self.lx.read_u16()
        #print("lx:"+str(self.lx_value))
    def smell_read(self):
        #global lx_value
        self.smell_value = self.smell.read_u16()
        #print("lx:"+str(self.lx_value))
 
    def pressure_read(self):
        #global pressure_value
        self.pressure_value = self.bme.values[1]
        #print("pressure:"+self.pressure_value)
 
    def temperature_read(self):
        #global temperature_value
        self.temperature_value = self.bme.values[0]
        #print("temperature:"+self.temperature_value)
 
    def humidity_read(self):
        #global humidity_value
        self.humidity_value = self.bme.values[2]
        #print("humidity:"+self.humidity_value)
 
    def smell_capture(self):
        self.smell_a0.value(1)
        time.sleep(0.001)
        self.smell_read()
        time.sleep(0.001)
        self.smell_a0.value(0)
        #print("smell="+str(self.smell_value))
 
    def smell2_capture(self): #called by each 1sec.
        if self.smell2_step==0:
            val=0
            self.smell2_out.value(1)
        if self.smell2_step==3:
            #time.sleep(3)
            self.smell2_value=self.smell2_in.read_u16()
        if self.smell2_step==5:
            #time.sleep(2)
            self.smell2_out.value(0)
            self.smell2_heater.value(0)
        if self.smell2_step==12:
            #time.sleep(8)
            self.smell2_heater.value(1)
        if self.smell2_step==240:
            self.smell2_step=0
        self.smell2_step=self.smell2_step+1
        #print("smell2="+str(self.sm
 
    def particle_read(self):
        data=[]
        data=self.i2c.readfrom_mem(0x15,0x00,2)
        self.particle_concentration=(data[0] & 0xff)<<8|(data[1]&0xff)
 
    count_minutes=0 
    def sensor_read_one_minute(self,timer):
        self.co2_read()
        #self.motion_read()
        self.lx_read()
        self.temperature_read()
        self.pressure_read()
        self.humidity_read()
        if self.count_minutes>=4:
            #self.smell2_capture()
            self.count_minutes=0
        self.count_minutes=self.count_minutes+1
    def sensor_read_five_minute(self,timer):
        self.smell2_capture()
 
    def sensor_read_100ms(self,timer):
        if self.c100ms>=60:            
            self.c100ms=0
        if self.c100ms%6==0:
            self.motion_read()
        if self.c100ms%10==0:
            self.smell_capture()
            self.smell2_capture()
        self.c100ms=self.c100ms+1
        #print("start sensor read one second")
        
    #def sensor_read_1s(self,timer):
    #
 
    def __init__(self):
        #print ("start setup")
        self.u = UART(1, baudrate=9600, tx=Pin(4), rx=Pin(5)) 
        #print('exception buf')
        #interrupt exception
        micropython.alloc_emergency_exception_buf(100)
        #print('otimer')
        self.one_minute_timer = Timer()
        #self.timer_1s = Timer()
        self.timer_100ms = Timer()
        #print('adc')
        self.pir=Pin(14, Pin.IN)
        self.lx = ADC(Pin(27))
        #self.co2 = ADC(Pin(28))
        self.smell = ADC(Pin(26))
        self.smell_a0=Pin(15,Pin.OUT)
        self.smell2_heater=Pin(12,Pin.OUT)
        self.smell2_out=Pin(13,Pin.OUT)
        self.smell2_heater.value(1)
        self.smell2_out.value(0)
        self.smell2_in=ADC(Pin(28))
        self.smell2_step=0
        #print('i2c')
        self.sda=Pin(0)
        self.scl=Pin(1)
        #print('i2c2')
        self.i2c = machine.I2C(0, sda = self.sda, scl = self.scl, freq = 100000)
        #print('bme280')
        self.bme = BME280(i2c = self.i2c)
        #print('x')
        self.co2_value = 0
        self.motion_value = 0
        self.smell_value = 0
        self.smell2_value = 0
        self.lx_value = 0
        self.pressure_value = ""
        self.temperature_value = ""
        self.humidity_value = ""
        #self.one_hour_timer.init(period=600000,mode=Timer.PERIODIC, callback=self.reset_pico)
        self.one_minute_timer.init(period=60000, mode=Timer.PERIODIC, callback=self.sensor_read_one_minute)
        self.timer_100ms.init(period=100, mode=Timer.PERIODIC, callback=self.sensor_read_100ms)
        #self.timer_1s.init(period=1000,mode=Timer.PERIODIC, callback=self.sensor_read_1s)
        ##time.sleep(60)
        #print ("end setup")
        self.c100ms=0
 
class Parser:
    def __init__(self,line,sensing):
        self.line=line
        self.sensing=sensing
    def p_key(self,key,p):
        #print("p_key key="+key+" p="+str(p)+" "+self.html[p:p+10])
        keylen=len(key)
        if p+keylen>=len(self.line):
            return p
        q=p
        if self.line[q:q+keylen]==key:  
            #print("key="+key+" have found, q="+str(q+keylen))
            return q+keylen
        #print("key="+key+" is not found, p="+str(p))
        return p 
      
    def p_String_Constant(self,p,strc):
        #print("p_String_Constant p="+str(p)+' '+self.html[p:p+10])
        rx=[""]
        xstr=""
        pw=p
        q=self.p_key('"',p)
        if p!=q:
            #print("p_String_Constant p="+str(p)+' '+self.html[p:p+10])
            p=q
            fx=self.line[p]
            xlen=len(self.line)-p
            while fx!='"':
                if xlen==0:
                    return pw
                if fx in ['\n','\t','\r']:
                    return pw
                if fx=='\\':
                   p=p+1
                p=p+1
                fx=self.line[p]
                xlen=len(self.line)-p
            q=self.p_key('"',p)
            if p!=q:
                strc[0]=self.line[pw+1:q-1]
                #print("p_string_constant, strc="+str(strc[0]))
                return q
        #print("p_String_Constant, fail to find \"")
        return pw
    def p_Date(self,p,date_str):
        #print("p_Date p="+str(p)+' '+self.html[p:p+10])
        rx=[""]
        xstr=""
        pw=p
        year=[0]
        q=self.p_number(p,year)
        if p==q:
            return pw
        p=q
        q=self.p_key('/',p)
        if p==q:
            return pw
        p=q
        month=[0]
        q=self.p_number(p,month)
        if p==q:
            return pw
        p=q
        q=self.p_key('/',p)
        if p==q:
            return pw
        p=q
        day=[0]
        q=self.p_number(p,day)
        if p==q:
            return False
        p=q
        q=self.p_b(p)
        if p==q:
            return pw
        p=q
        hour=[0]
        q=self.p_number(p,hour)
        if p==q:
            return pw
        p=q
        q=self.p_key(':',p)
        if p==q:
            return pw
        p=q
        minute=[0]
        q=self.p_number(p,minute)
        if p==q:
            return pw
        p=q
        q=self.p_key(':',p)
        if p==q:
            return pw
        p=q
        second=[0]
        q=self.p_number(p,second)
        if p==q:
            return pw
        p=q
        date_str[0]=str(year[0])+'/'+str(month[0])+'/'+str(day[0])+' '+str(hour[0])+':'+str(minute[0])+':'+str(second[0])
        return p
    
    def reset(self):
        machine.reset()
    
    def parse_the_line(self):
      p=0
      p=self.p_b(p)
      q=self.p_key('ex',p)
      if q>p:
            p=q
            p=self.p_b(p)
            exp=self.line[p:]
            #print(exp)
            exec_locals={'self':self}
            try:
                rtn=eval(exp,globals(),exec_locals)
                print(rtn)
            except Exception as e:
                print("error in "+exp)
                print(e)
    def p_b(self,p):
        #print("parse_b")
        #print("p="+str(p))
        xlen=len(self.line)
        if p>=xlen:
            return p
        while self.line[p]==' ':
            p=p+1
            if p>=xlen:
                p=xlen
                break
        return p
 
    def p_name(self,p,name):
        #print("p_name, self.html="+ self.html[p:p+10])
        #print("p="+str(p)+"..."+self.html[p:p+10])
        q=p
        last_p=len(self.line)
        while self.line[q] in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_':
            #print("self.html["+str(q)+"]="+self.html[q])
            q=q+1
            if q>=last_p:
                break
        name[0]=self.line[p:q]
        #print("p_name p="+str(p)+" q="+str(q)+" name="+str(name[0]))
        #print("p="+str(p)+" q="+str(q)+" name="+name[0])
        #print("name[0]= "+name[0])
        return q
    def p_number(self,p,num):
        #print("p_number")
        #print("p="+str(p))
        n=""
        last_p=len(self.line)
        q=p
        while self.line[p] in '0123456789':
            q=q+1
            if q>=last_p:
                break
        num[0]=int(self.line[p:q])
        #print("p_number p="+str(p)+" q="+str(q)+" num="+str(num[0]))
        return q
    def p_an_equation(self,p,eqs):
        #print("p_an_equation p="+str(p)+' '+self.html[p:p+10])
        pw=p
        p=self.p_b(p)
        nx=[""]
        strc=[""]
        q=self.p_name(p,nx)
        #print("p_an_equation name="+nx[0])
        if p==q:
            #print("p_an_equation fail to find name")
            return pw
        p=q
        p=self.p_b(p)
        q=self.p_key('=',p)
        if p==q:
            #print("p_an_equation fail to find =")
            return pw
        p=q
        p=self.p_b(p)
        q=self.p_String_Constant(p,strc)
        if p==q:
            num=[0]
            q=self.p_number(p,num)
            if p==q:
                return pw
            else:
                eqs[nx[0]]=num[0]
                #print("p_qn_equation get "+nx[0]+"="+str(num[0]))
                return q
        else:
            eqs[nx[0]]=strc[0]
            #print("p_an_equation get "+nx[0]+"="+str(strc[0]))
            return q
    def p_equations(self,p,eqs):
        #print("parse_equations")
        #print("p="+str(p))
        pw=p
        while True:
            q=self.p_an_equation(p,eqs)
            if p==q:
                return p
            p=q
    def p_get_equations_of_the_tag(self,p,tag_name,eqs):
        #print("p_get_equations_of_the_tag <"+tag_name)
        #print("p="+str(p)+' '+self.html[p:p+10])
        pw=p
        try:
            q=self.line.index('<'+tag_name,pw)
        except:
            #print("p_get_equation_of_the_tag fail to find <"+tag_name)
            return p
        #print("q="+str(q)+' '+self.html[q:q+10])
        q=q+len('<'+tag_name)
        #print("q="+str(q)+' '+self.html[q:q+10])
        if pw==q:
            #print("p_get_equation_of_the_tag fail to find <"+tag_name)
            return p
        pw=q
        q=self.p_equations(pw,eqs)
        #print("after equations q="+str(q) +' '+self.html[q:q+10])
        if pw==q:
            return p
        pw=q
        pw=self.p_b(pw)
        #print("pw="+str(pw))
        q=self.p_key('>',pw)
        #print("after > q="+str(q)+' '+self.html[q:q+10])
        if pw!=q:
            return q
        q=self.p_key('/>',pw)
        if pw!=q:
            return q
        return p
 
if __name__ == '__main__':
    #
    # ex self.sensing.get_co2()
    # ex self.sensing.get_motion()
    # ex self.sensing.get_lx()
    # ex self.sensing.get_pressure()
    # ex self.sensing.get_temperature()
    # ex self.sensing.get_humidity()
    #
    sensing=Sensing()
    while True:
        line=input()
        parser=Parser(line,sensing)
        parser.parse_the_line()
        gc.collect()
        time.sleep(0.1)