プログラム - 栄商会

2022年8月14日2022年8月14日

WordPressにbokehのグラフをのせる（備忘録）

bokeh、pandas-bokehはインタラクティブなグラフを作成することができるが、これをWordPressに添付するのも割と簡単であったので、のせておきます。

１．bokehグラフのhtmlコードを取得する

bokehのプログムをrunさせるとrunさせたディレクトリにhtmlファイルが生成されます。これをIDE（私はVisualStudioCode)で開いて、クリップボードにコピーする。

２．WordPressにコピーする

ブロック挿入ツールのウィジェット中に「カスタムHTML」があるので、これを選択し、ダイアログ中にコピーしたhtmlをペーストして、出来上がり

プレビューすると、しっかりbokehのグラフを表示することができます。マウスでドラッグやピンチもできます。

Bokeh Plot

2022年8月11日2022年8月14日

MySQLデータをpandas-bokehでグラフ化（備忘録）

毎日暑いので、エアコンの効いた部屋でノンビリＰＣいじってます。今日は、先日サーバーを入れ替え、自宅温度などをESP32からMySQLに取り込み、これをグラフ化するところまで備忘録にしておきます。

必要モジュールのインストール

今回はpandas-bokehとsqlalchemyをインストールしました。それぞれpipを用いてインストールします。

pip install pandas-bokeh
pip install sqlalchemy

全体のコードは以下のとおりです。

import pandas as pd
import sqlalchemy as sa
import pandas_bokeh

user='esp32'
password='*******'
host='192.168.1.**'
db='esp_data'
port=3306

url = f'mysql+pymysql://{user}:{password}@{host}:{port}/{db}?charset=utf8'
engine = sa.create_engine(url,echo=False)

query = "select reading_time,temp from SensorData order by reading_time desc limit 1000"
#query="select * from SensorData order by reading_time desc limit 10"
df = pd.read_sql(query,con=engine,index_col='reading_time')
#print(df)

df.plot_bokeh()

たったこれだけのコードです。16行目のindex_col=’reading_time’でX座標が時間軸になり、ソートはdescですが、プロットするときは経時的に描画されます。

Bokeh Plot

2022年7月30日2022年7月30日

MySQLクライアント接続設定（備忘録）

サーバーにMySQLを入れた際にターミナル接続は可能なのに、クライアントソフトからの接続ができなかったため、cnfファイルの変更備忘録。

cnfファイル　/etc/mysql/mysql.conf.d/mysqld.conf

bind-address = 192.168.1.51 #デフォルトは127.0.0.1
mysqlx-bind-address = 192.168.1.51　＃デフォルトは127.0.0.1

sudo service mysql restart　でMySQLをリスタートし、アプリで確認。無事アクセスできました。

2021年3月4日2021年3月3日

ESP32　BME680で測定したデータをMariaDBにとばす　その２

先日BME680のデータをMariaDBに飛ばしましたが、スタンドアロンで自動測定できるようにプログラムを更新しました。また、Wi-Fiに接続している際にＬＥＤを点灯させることとしました。ピン４番にプルアップ抵抗を介してＬＥＤを接続します。

以下がプログラムです。ブートした際に自動的に走らせるために、プログラム名はmain.pyにしてあります。

from machine import Pin,SoftI2C
from bme680 import *
import ntptime
import utime
import ujson
import urequests

UTC_OFFSET=9
p21 = Pin(21, Pin.IN, Pin.PULL_UP)
p23 = Pin(23, Pin.IN, Pin.PULL_UP)
led = Pin(4, Pin.OUT)
i2c = SoftI2C(scl=Pin(21), sda=Pin(23))

def do_connect():
    import network
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print('connecting to network...')
        wlan.connect('essid', 'password')
        while not wlan.isconnected():
            pass
    print('network config:', wlan.ifconfig())
    led.value(1)

     
def get_jst():
    sleep_ms(1000)
    ntptime.settime()
    tm=utime.localtime(utime.mktime(utime.localtime())+UTC_OFFSET*3600)
    jst = str(tm[0])+'/'+str(tm[1])+'/'+str(tm[2])+' '+str(tm[3])+':'+str(tm[4])+':'+str(tm[5])
    return(jst)
     
do_connect()
print(get_jst())

while True:
  try:
    bme = BME680_I2C(i2c=i2c) 
    temp = str(round(bme.temperature, 2))
    hum = str(round(bme.humidity, 2))
    pres = str(round(bme.pressure, 2))
    gas = str(round(bme.gas/1000, 2))

    dict = {
            "temp_value":temp,
            "hum_value":hum,
            "pres_value":pres,
            "gas_value":gas
            }
    response = urequests.post(
            'https://***url***/datarcv.php',
            data=ujson.dumps(dict).encode("utf-8"),
                headers={'Content-Type': 'application/json'}
            )
    print  (response.text)
  except OSError as e:
    print('Failed to read sensor.')
 
  sleep(10)

from machine import Pin,SoftI2C

from bme680 import *

import ntptime

import utime

import ujson

import urequests

UTC_OFFSET=9

p21 = Pin(21, Pin.IN, Pin.PULL_UP)

p23 = Pin(23, Pin.IN, Pin.PULL_UP)

led = Pin(4, Pin.OUT)

i2c = SoftI2C(scl=Pin(21), sda=Pin(23))

def do_connect():

import network

wlan = network.WLAN(network.STA_IF)

wlan.active(True)

if not wlan.isconnected():

print('connecting to network...')

wlan.connect('essid', 'password')

while not wlan.isconnected():

pass

print('network config:', wlan.ifconfig())

led.value(1)

def get_jst():

sleep_ms(1000)

ntptime.settime()

tm=utime.localtime(utime.mktime(utime.localtime())+UTC_OFFSET*3600)

jst = str(tm[0])+'/'+str(tm[1])+'/'+str(tm[2])+' '+str(tm[3])+':'+str(tm[4])+':'+str(tm[5])

return(jst)

do_connect()

print(get_jst())

while True:

try:

bme = BME680_I2C(i2c=i2c)

temp = str(round(bme.temperature, 2))

hum = str(round(bme.humidity, 2))

pres = str(round(bme.pressure, 2))

gas = str(round(bme.gas/1000, 2))

dict = {

"temp_value":temp,

"hum_value":hum,

"pres_value":pres,

"gas_value":gas

}

response = urequests.post(

'https://***url***/datarcv.php',

data=ujson.dumps(dict).encode("utf-8"),

headers={'Content-Type': 'application/json'}

)

print (response.text)

except OSError as e:

print('Failed to read sensor.')

sleep(10)

この通りWi-Fiに接続するとＬＥＤがともります。phpMyAdminで確認すると10秒おきにデータが蓄積していきます。

2021年3月3日

ESP32 複合センサBME680で測定したデータをMariaDBにとばす

先日仕入れたBME680を用いて温度、湿度、気圧、ガス濃度を測定してMariaDBに飛ばして記録したいと思います。今回のプログラムです。先にWi-Fiにつないでからrunさせます。

import ujson
import urequests

from machine import Pin,I2C
from time import sleep
from bme680 import *
p21 = Pin(21, Pin.IN, Pin.PULL_UP)
p23 = Pin(23, Pin.IN, Pin.PULL_UP)

i2c = I2C(scl=Pin(21), sda=Pin(23))

while True:
  try:
    bme = BME680_I2C(i2c=i2c) 
    temp = str(round(bme.temperature, 2))
    hum = str(round(bme.humidity, 2))
    pres = str(round(bme.pressure, 2))
    gas = str(round(bme.gas/1000, 2))

    print('Temperature:', temp)
    print('Humidity:', hum)
    print('Pressure:', pres)
    print('Gas:', gas)

    dict = {
            "temp_value":temp,
            "hum_value":hum,
            "pres_value":pres,
            "gas_value":gas
            }
    response = urequests.post(
            'https://***url***/postrcv.php',
            data=ujson.dumps(dict).encode("utf-8"),
                headers={'Content-Type': 'application/json'}
            )
    print  (response.text)
  except OSError as e:
    print('Failed to read sensor.')
 
  sleep(10)

import ujson

import urequests

from machine import Pin,I2C

from time import sleep

from bme680 import *

p21 = Pin(21, Pin.IN, Pin.PULL_UP)

p23 = Pin(23, Pin.IN, Pin.PULL_UP)

i2c = I2C(scl=Pin(21), sda=Pin(23))

while True:

try:

bme = BME680_I2C(i2c=i2c)

temp = str(round(bme.temperature, 2))

hum = str(round(bme.humidity, 2))

pres = str(round(bme.pressure, 2))

gas = str(round(bme.gas/1000, 2))

print('Temperature:', temp)

print('Humidity:', hum)

print('Pressure:', pres)

print('Gas:', gas)

dict = {

"temp_value":temp,

"hum_value":hum,

"pres_value":pres,

"gas_value":gas

}

response = urequests.post(

'https://***url***/postrcv.php',

data=ujson.dumps(dict).encode("utf-8"),

headers={'Content-Type': 'application/json'}

)

print (response.text)

except OSError as e:

print('Failed to read sensor.')

sleep(10)

実行結果です。

C:\Users\>ampy -p com3 run c:\micropython\posttest8.py
Warning: I2C(-1, ...) is deprecated, use SoftI2C(...) instead
Temperature: 17.36
Humidity: 54.01
Pressure: 1019.21
Gas: 56.24
"temp=17.36 , hum=54.01 , pres=1019.21, gas=56.24"New record created successfully
Temperature: 17.35
Humidity: 54.14
Pressure: 1019.22
Gas: 54.43
"temp=17.35 , hum=54.14 , pres=1019.22, gas=54.43"New record created successfully
Temperature: 17.41
Humidity: 54.15
Pressure: 1019.24
Gas: 53.23
"temp=17.41 , hum=54.15 , pres=1019.24, gas=53.23"New record created successfully
Temperature: 17.48
Humidity: 54.03
Pressure: 1019.2
Gas: 52.34
"temp=17.48 , hum=54.03 , pres=1019.2, gas=52.34"New record created successfully

Aborted!

しっかりと10秒おきにDBに記録されています。

次回はプログラムを見直し、単独で自動運転できるようにします。

2021年3月1日2021年3月1日

ESP32 複合センサBME680

先日お知らせしたBME680による温湿度気圧、ガス濃度の分析です。

今回は秋月電子のセンサを用いていますが、同じようにAdafruitでも同様のセンサキットを用意しており、GithubにてMicroPython用に用意されているAdafruithttps://github.com/robert-hh/BME680-Micropythonのモジュールを用います。

まずはモジュールです。

# The MIT License (MIT)
#
# Copyright (c) 2017 ladyada for Adafruit Industries
#
# 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.

# We have a lot of attributes for this complex sensor.
# pylint: disable=too-many-instance-attributes

"""
`bme680` - BME680 - Temperature, Humidity, Pressure & Gas Sensor
================================================================

MicroPython driver from BME680 air quality sensor, based on Adafruit_bme680

* Author(s): Limor 'Ladyada' Fried of Adafruit
             Jeff Raber (SPI support)
             and many more contributors
"""

import time
import math
from micropython import const
from ubinascii import hexlify as hex
try:
    import struct
except ImportError:
    import ustruct as struct

#    I2C ADDRESS/BITS/SETTINGS
#    -----------------------------------------------------------------------
_BME680_CHIPID = const(0x61)

_BME680_REG_CHIPID = const(0xD0)
_BME680_BME680_COEFF_ADDR1 = const(0x89)
_BME680_BME680_COEFF_ADDR2 = const(0xE1)
_BME680_BME680_RES_HEAT_0 = const(0x5A)
_BME680_BME680_GAS_WAIT_0 = const(0x64)

_BME680_REG_SOFTRESET = const(0xE0)
_BME680_REG_CTRL_GAS = const(0x71)
_BME680_REG_CTRL_HUM = const(0x72)
_BME280_REG_STATUS = const(0xF3)
_BME680_REG_CTRL_MEAS = const(0x74)
_BME680_REG_CONFIG = const(0x75)

_BME680_REG_PAGE_SELECT = const(0x73)
_BME680_REG_MEAS_STATUS = const(0x1D)
_BME680_REG_PDATA = const(0x1F)
_BME680_REG_TDATA = const(0x22)
_BME680_REG_HDATA = const(0x25)

_BME680_SAMPLERATES = (0, 1, 2, 4, 8, 16)
_BME680_FILTERSIZES = (0, 1, 3, 7, 15, 31, 63, 127)

_BME680_RUNGAS = const(0x10)

_LOOKUP_TABLE_1 = (2147483647.0, 2147483647.0, 2147483647.0, 2147483647.0, 2147483647.0,
                   2126008810.0, 2147483647.0, 2130303777.0, 2147483647.0, 2147483647.0,
                   2143188679.0, 2136746228.0, 2147483647.0, 2126008810.0, 2147483647.0,
                   2147483647.0)

_LOOKUP_TABLE_2 = (4096000000.0, 2048000000.0, 1024000000.0, 512000000.0, 255744255.0, 127110228.0,
                   64000000.0, 32258064.0, 16016016.0, 8000000.0, 4000000.0, 2000000.0, 1000000.0,
                   500000.0, 250000.0, 125000.0)


def _read24(arr):
    """Parse an unsigned 24-bit value as a floating point and return it."""
    ret = 0.0
    #print([hex(i) for i in arr])
    for b in arr:
        ret *= 256.0
        ret += float(b & 0xFF)
    return ret


class Adafruit_BME680:
    """Driver from BME680 air quality sensor

       :param int refresh_rate: Maximum number of readings per second. Faster property reads
         will be from the previous reading."""
    def __init__(self, *, refresh_rate=10):
        """Check the BME680 was found, read the coefficients and enable the sensor for continuous
           reads."""
        self._write(_BME680_REG_SOFTRESET, [0xB6])
        time.sleep(0.005)

        # Check device ID.
        chip_id = self._read_byte(_BME680_REG_CHIPID)
        if chip_id != _BME680_CHIPID:
            raise RuntimeError('Failed to find BME680! Chip ID 0x%x' % chip_id)

        self._read_calibration()

        # set up heater
        self._write(_BME680_BME680_RES_HEAT_0, [0x73])
        self._write(_BME680_BME680_GAS_WAIT_0, [0x65])

        self.sea_level_pressure = 1013.25
        """Pressure in hectoPascals at sea level. Used to calibrate ``altitude``."""

        # Default oversampling and filter register values.
        self._pressure_oversample = 0b011
        self._temp_oversample = 0b100
        self._humidity_oversample = 0b010
        self._filter = 0b010

        self._adc_pres = None
        self._adc_temp = None
        self._adc_hum = None
        self._adc_gas = None
        self._gas_range = None
        self._t_fine = None

        self._last_reading = 0
        self._min_refresh_time = 1000 / refresh_rate

    @property
    def pressure_oversample(self):
        """The oversampling for pressure sensor"""
        return _BME680_SAMPLERATES[self._pressure_oversample]

    @pressure_oversample.setter
    def pressure_oversample(self, sample_rate):
        if sample_rate in _BME680_SAMPLERATES:
            self._pressure_oversample = _BME680_SAMPLERATES.index(sample_rate)
        else:
            raise RuntimeError("Invalid oversample")

    @property
    def humidity_oversample(self):
        """The oversampling for humidity sensor"""
        return _BME680_SAMPLERATES[self._humidity_oversample]

    @humidity_oversample.setter
    def humidity_oversample(self, sample_rate):
        if sample_rate in _BME680_SAMPLERATES:
            self._humidity_oversample = _BME680_SAMPLERATES.index(sample_rate)
        else:
            raise RuntimeError("Invalid oversample")

    @property
    def temperature_oversample(self):
        """The oversampling for temperature sensor"""
        return _BME680_SAMPLERATES[self._temp_oversample]

    @temperature_oversample.setter
    def temperature_oversample(self, sample_rate):
        if sample_rate in _BME680_SAMPLERATES:
            self._temp_oversample = _BME680_SAMPLERATES.index(sample_rate)
        else:
            raise RuntimeError("Invalid oversample")

    @property
    def filter_size(self):
        """The filter size for the built in IIR filter"""
        return _BME680_FILTERSIZES[self._filter]

    @filter_size.setter
    def filter_size(self, size):
        if size in _BME680_FILTERSIZES:
            self._filter = _BME680_FILTERSIZES[size]
        else:
            raise RuntimeError("Invalid size")

    @property
    def temperature(self):
        """The compensated temperature in degrees celsius."""
        self._perform_reading()
        calc_temp = (((self._t_fine * 5) + 128) / 256)
        return calc_temp / 100

    @property
    def pressure(self):
        """The barometric pressure in hectoPascals"""
        self._perform_reading()
        var1 = (self._t_fine / 2) - 64000
        var2 = ((var1 / 4) * (var1 / 4)) / 2048
        var2 = (var2 * self._pressure_calibration[5]) / 4
        var2 = var2 + (var1 * self._pressure_calibration[4] * 2)
        var2 = (var2 / 4) + (self._pressure_calibration[3] * 65536)
        var1 = (((((var1 / 4) * (var1 / 4)) / 8192) *
                (self._pressure_calibration[2] * 32) / 8) +
                ((self._pressure_calibration[1] * var1) / 2))
        var1 = var1 / 262144
        var1 = ((32768 + var1) * self._pressure_calibration[0]) / 32768
        calc_pres = 1048576 - self._adc_pres
        calc_pres = (calc_pres - (var2 / 4096)) * 3125
        calc_pres = (calc_pres / var1) * 2
        var1 = (self._pressure_calibration[8] * (((calc_pres / 8) * (calc_pres / 8)) / 8192)) / 4096
        var2 = ((calc_pres / 4) * self._pressure_calibration[7]) / 8192
        var3 = (((calc_pres / 256) ** 3) * self._pressure_calibration[9]) / 131072
        calc_pres += ((var1 + var2 + var3 + (self._pressure_calibration[6] * 128)) / 16)
        return calc_pres/100

    @property
    def humidity(self):
        """The relative humidity in RH %"""
        self._perform_reading()
        temp_scaled = ((self._t_fine * 5) + 128) / 256
        var1 = ((self._adc_hum - (self._humidity_calibration[0] * 16)) -
                ((temp_scaled * self._humidity_calibration[2]) / 200))
        var2 = (self._humidity_calibration[1] *
                (((temp_scaled * self._humidity_calibration[3]) / 100) +
                 (((temp_scaled * ((temp_scaled * self._humidity_calibration[4]) / 100)) /
                   64) / 100) + 16384)) / 1024
        var3 = var1 * var2
        var4 = self._humidity_calibration[5] * 128
        var4 = (var4 + ((temp_scaled * self._humidity_calibration[6]) / 100)) / 16
        var5 = ((var3 / 16384) * (var3 / 16384)) / 1024
        var6 = (var4 * var5) / 2
        calc_hum = (((var3 + var6) / 1024) * 1000) / 4096
        calc_hum /= 1000  # get back to RH

        if calc_hum > 100:
            calc_hum = 100
        if calc_hum < 0:
            calc_hum = 0
        return calc_hum

    @property
    def altitude(self):
        """The altitude based on current ``pressure`` vs the sea level pressure
           (``sea_level_pressure``) - which you must enter ahead of time)"""
        pressure = self.pressure # in Si units for hPascal
        return 44330 * (1.0 - math.pow(pressure / self.sea_level_pressure, 0.1903))

    @property
    def gas(self):
        """The gas resistance in ohms"""
        self._perform_reading()
        var1 = ((1340 + (5 * self._sw_err)) * (_LOOKUP_TABLE_1[self._gas_range])) / 65536
        var2 = ((self._adc_gas * 32768) - 16777216) + var1
        var3 = (_LOOKUP_TABLE_2[self._gas_range] * var1) / 512
        calc_gas_res = (var3 + (var2 / 2)) / var2
        return int(calc_gas_res)

    def _perform_reading(self):
        """Perform a single-shot reading from the sensor and fill internal data structure for
           calculations"""
        if (time.ticks_diff(self._last_reading, time.ticks_ms()) * time.ticks_diff(0, 1)
                < self._min_refresh_time):
            return

        # set filter
        self._write(_BME680_REG_CONFIG, [self._filter << 2])
        # turn on temp oversample & pressure oversample
        self._write(_BME680_REG_CTRL_MEAS,
                    [(self._temp_oversample << 5)|(self._pressure_oversample << 2)])
        # turn on humidity oversample
        self._write(_BME680_REG_CTRL_HUM, [self._humidity_oversample])
        # gas measurements enabled
        self._write(_BME680_REG_CTRL_GAS, [_BME680_RUNGAS])

        ctrl = self._read_byte(_BME680_REG_CTRL_MEAS)
        ctrl = (ctrl & 0xFC) | 0x01  # enable single shot!
        self._write(_BME680_REG_CTRL_MEAS, [ctrl])
        new_data = False
        while not new_data:
            data = self._read(_BME680_REG_MEAS_STATUS, 15)
            new_data = data[0] & 0x80 != 0
            time.sleep(0.005)
        self._last_reading = time.ticks_ms()

        self._adc_pres = _read24(data[2:5]) / 16
        self._adc_temp = _read24(data[5:8]) / 16
        self._adc_hum = struct.unpack('>H', bytes(data[8:10]))[0]
        self._adc_gas = int(struct.unpack('>H', bytes(data[13:15]))[0] / 64)
        self._gas_range = data[14] & 0x0F

        var1 = (self._adc_temp / 8) - (self._temp_calibration[0] * 2)
        var2 = (var1 * self._temp_calibration[1]) / 2048
        var3 = ((var1 / 2) * (var1 / 2)) / 4096
        var3 = (var3 * self._temp_calibration[2] * 16) / 16384

        self._t_fine = int(var2 + var3)

    def _read_calibration(self):
        """Read & save the calibration coefficients"""
        coeff = self._read(_BME680_BME680_COEFF_ADDR1, 25)
        coeff += self._read(_BME680_BME680_COEFF_ADDR2, 16)

        coeff = list(struct.unpack('<hbBHhbBhhbbHhhBBBHbbbBbHhbb', bytes(coeff[1:39])))
        # print("\n\n",coeff)
        coeff = [float(i) for i in coeff]
        self._temp_calibration = [coeff[x] for x in [23, 0, 1]]
        self._pressure_calibration = [coeff[x] for x in [3, 4, 5, 7, 8, 10, 9, 12, 13, 14]]
        self._humidity_calibration = [coeff[x] for x in [17, 16, 18, 19, 20, 21, 22]]
        self._gas_calibration = [coeff[x] for x in [25, 24, 26]]

        # flip around H1 & H2
        self._humidity_calibration[1] *= 16
        self._humidity_calibration[1] += self._humidity_calibration[0] % 16
        self._humidity_calibration[0] /= 16

        self._heat_range = (self._read_byte(0x02) & 0x30) / 16
        self._heat_val = self._read_byte(0x00)
        self._sw_err = (self._read_byte(0x04) & 0xF0) / 16

    def _read_byte(self, register):
        """Read a byte register value and return it"""
        return self._read(register, 1)[0]

    def _read(self, register, length):
        raise NotImplementedError()

    def _write(self, register, values):
        raise NotImplementedError()

class BME680_I2C(Adafruit_BME680):
    """Driver for I2C connected BME680.

        :param i2c: I2C device object
        :param int address: I2C device address
        :param bool debug: Print debug statements when True.
        :param int refresh_rate: Maximum number of readings per second. Faster property reads
          will be from the previous reading."""
    def __init__(self, i2c, address=0x77, debug=False, *, refresh_rate=10):
        """Initialize the I2C device at the 'address' given"""
        self._i2c = i2c
        self._address = address
        self._debug = debug
        super().__init__(refresh_rate=refresh_rate)

    def _read(self, register, length):
        """Returns an array of 'length' bytes from the 'register'"""
        result = bytearray(length)
        self._i2c.readfrom_mem_into(self._address, register & 0xff, result)
        if self._debug:
            print("\t${:x} read ".format(register), " ".join(["{:02x}".format(i) for i in result]))
        return result

    def _write(self, register, values):
        """Writes an array of 'length' bytes to the 'register'"""
        if self._debug:
            print("\t${:x} write".format(register), " ".join(["{:02x}".format(i) for i in values]))
        for value in values:
            self._i2c.writeto_mem(self._address, register, bytearray([value & 0xFF]))
            register += 1


class BME680_SPI(Adafruit_BME680):
    """Driver for SPI connected BME680.

        :param spi: SPI device object, configured
        :param cs: Chip Select Pin object, configured to OUT mode
        :param bool debug: Print debug statements when True.
        :param int refresh_rate: Maximum number of readings per second. Faster property reads
          will be from the previous reading.
      """

    def __init__(self, spi, cs, debug=False, *, refresh_rate=10):
        self._spi = spi
        self._cs = cs
        self._debug = debug
        self._cs(1)
        super().__init__(refresh_rate=refresh_rate)

    def _read(self, register, length):
        if register != _BME680_REG_PAGE_SELECT:
            # _BME680_REG_PAGE_SELECT exists in both SPI memory pages
            # For all other registers, we must set the correct memory page
            self._set_spi_mem_page(register)
        register = (register | 0x80) & 0xFF  # Read single, bit 7 high.

        try:
            self._cs(0)
            self._spi.write(bytearray([register]))  # pylint: disable=no-member
            result = bytearray(length)
            self._spi.readinto(result)  # pylint: disable=no-member
            if self._debug:
                print("\t${:x} read ".format(register), " ".join(["{:02x}".format(i) for i in result]))
        except Exception as e:
            print (e)
            result = None
        finally:
            self._cs(1)
        return result

    def _write(self, register, values):
        if register != _BME680_REG_PAGE_SELECT:
            # _BME680_REG_PAGE_SELECT exists in both SPI memory pages
            # For all other registers, we must set the correct memory page
            self._set_spi_mem_page(register)
        register &= 0x7F  # Write, bit 7 low.
        try:
            self._cs(0)
            buffer = bytearray(2 * len(values))
            for i, value in enumerate(values):
                buffer[2 * i] = register + i
                buffer[2 * i + 1] = value & 0xFF
            self._spi.write(buffer)  # pylint: disable=no-member
            if self._debug:
                print("\t${:x} write".format(register), " ".join(["{:02x}".format(i) for i in values]))
        except Exception as e:
            print (e)
        finally:
            self._cs(1)

    def _set_spi_mem_page(self, register):
        spi_mem_page = 0x00
        if register < 0x80:
            spi_mem_page = 0x10
        self._write(_BME680_REG_PAGE_SELECT, [spi_mem_page])

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

# The MIT License (MIT)

# 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.

# We have a lot of attributes for this complex sensor.

# pylint: disable=too-many-instance-attributes

"""

`bme680` - BME680 - Temperature, Humidity, Pressure & Gas Sensor

================================================================

MicroPython driver from BME680 air quality sensor, based on Adafruit_bme680

* Author(s): Limor 'Ladyada' Fried of Adafruit

Jeff Raber (SPI support)

and many more contributors

"""

import time

import math

from micropython import const

from ubinascii import hexlify as hex

try:

import struct

except ImportError:

import ustruct as struct

# I2C ADDRESS/BITS/SETTINGS

# -----------------------------------------------------------------------

_BME680_CHIPID = const(0x61)

_BME680_REG_CHIPID = const(0xD0)

_BME680_BME680_COEFF_ADDR1 = const(0x89)

_BME680_BME680_COEFF_ADDR2 = const(0xE1)

_BME680_BME680_RES_HEAT_0 = const(0x5A)

_BME680_BME680_GAS_WAIT_0 = const(0x64)

_BME680_REG_SOFTRESET = const(0xE0)

_BME680_REG_CTRL_GAS = const(0x71)

_BME680_REG_CTRL_HUM = const(0x72)

_BME280_REG_STATUS = const(0xF3)

_BME680_REG_CTRL_MEAS = const(0x74)

_BME680_REG_CONFIG = const(0x75)

_BME680_REG_PAGE_SELECT = const(0x73)

_BME680_REG_MEAS_STATUS = const(0x1D)

_BME680_REG_PDATA = const(0x1F)

_BME680_REG_TDATA = const(0x22)

_BME680_REG_HDATA = const(0x25)

_BME680_SAMPLERATES = (0, 1, 2, 4, 8, 16)

_BME680_FILTERSIZES = (0, 1, 3, 7, 15, 31, 63, 127)

_BME680_RUNGAS = const(0x10)

_LOOKUP_TABLE_1 = (2147483647.0, 2147483647.0, 2147483647.0, 2147483647.0, 2147483647.0,

2126008810.0, 2147483647.0, 2130303777.0, 2147483647.0, 2147483647.0,

2143188679.0, 2136746228.0, 2147483647.0, 2126008810.0, 2147483647.0,

2147483647.0)

_LOOKUP_TABLE_2 = (4096000000.0, 2048000000.0, 1024000000.0, 512000000.0, 255744255.0, 127110228.0,

64000000.0, 32258064.0, 16016016.0, 8000000.0, 4000000.0, 2000000.0, 1000000.0,

500000.0, 250000.0, 125000.0)

def _read24(arr):

"""Parse an unsigned 24-bit value as a floating point and return it."""

ret = 0.0

#print([hex(i) for i in arr])

for b in arr:

ret *= 256.0

ret += float(b & 0xFF)

return ret

class Adafruit_BME680:

"""Driver from BME680 air quality sensor

:param int refresh_rate: Maximum number of readings per second. Faster property reads

will be from the previous reading."""

def __init__(self, *, refresh_rate=10):

"""Check the BME680 was found, read the coefficients and enable the sensor for continuous

reads."""

self._write(_BME680_REG_SOFTRESET, [0xB6])

time.sleep(0.005)

# Check device ID.

chip_id = self._read_byte(_BME680_REG_CHIPID)

if chip_id != _BME680_CHIPID:

raise RuntimeError('Failed to find BME680! Chip ID 0x%x' % chip_id)

self._read_calibration()

# set up heater

self._write(_BME680_BME680_RES_HEAT_0, [0x73])

self._write(_BME680_BME680_GAS_WAIT_0, [0x65])

self.sea_level_pressure = 1013.25

"""Pressure in hectoPascals at sea level. Used to calibrate ``altitude``."""

# Default oversampling and filter register values.

self._pressure_oversample = 0b011

self._temp_oversample = 0b100

self._humidity_oversample = 0b010

self._filter = 0b010

self._adc_pres = None

self._adc_temp = None

self._adc_hum = None

self._adc_gas = None

self._gas_range = None

self._t_fine = None

self._last_reading = 0

self._min_refresh_time = 1000 / refresh_rate

@property

def pressure_oversample(self):

"""The oversampling for pressure sensor"""

return _BME680_SAMPLERATES[self._pressure_oversample]

@pressure_oversample.setter

def pressure_oversample(self, sample_rate):

if sample_rate in _BME680_SAMPLERATES:

self._pressure_oversample = _BME680_SAMPLERATES.index(sample_rate)

else:

raise RuntimeError("Invalid oversample")

@property

def humidity_oversample(self):

"""The oversampling for humidity sensor"""

return _BME680_SAMPLERATES[self._humidity_oversample]

@humidity_oversample.setter

def humidity_oversample(self, sample_rate):

if sample_rate in _BME680_SAMPLERATES:

self._humidity_oversample = _BME680_SAMPLERATES.index(sample_rate)

else:

raise RuntimeError("Invalid oversample")

@property

def temperature_oversample(self):

"""The oversampling for temperature sensor"""

return _BME680_SAMPLERATES[self._temp_oversample]

@temperature_oversample.setter

def temperature_oversample(self, sample_rate):

if sample_rate in _BME680_SAMPLERATES:

self._temp_oversample = _BME680_SAMPLERATES.index(sample_rate)

else:

raise RuntimeError("Invalid oversample")

@property

def filter_size(self):

"""The filter size for the built in IIR filter"""

return _BME680_FILTERSIZES[self._filter]

@filter_size.setter

def filter_size(self, size):

if size in _BME680_FILTERSIZES:

self._filter = _BME680_FILTERSIZES[size]

else:

raise RuntimeError("Invalid size")

@property

def temperature(self):

"""The compensated temperature in degrees celsius."""

self._perform_reading()

calc_temp = (((self._t_fine * 5) + 128) / 256)

return calc_temp / 100

@property

def pressure(self):

"""The barometric pressure in hectoPascals"""

self._perform_reading()

var1 = (self._t_fine / 2) - 64000

var2 = ((var1 / 4) * (var1 / 4)) / 2048

var2 = (var2 * self._pressure_calibration[5]) / 4

var2 = var2 + (var1 * self._pressure_calibration[4] * 2)

var2 = (var2 / 4) + (self._pressure_calibration[3] * 65536)

var1 = (((((var1 / 4) * (var1 / 4)) / 8192) *

(self._pressure_calibration[2] * 32) / 8) +

((self._pressure_calibration[1] * var1) / 2))

var1 = var1 / 262144

var1 = ((32768 + var1) * self._pressure_calibration[0]) / 32768

calc_pres = 1048576 - self._adc_pres

calc_pres = (calc_pres - (var2 / 4096)) * 3125

calc_pres = (calc_pres / var1) * 2

var1 = (self._pressure_calibration[8] * (((calc_pres / 8) * (calc_pres / 8)) / 8192)) / 4096

var2 = ((calc_pres / 4) * self._pressure_calibration[7]) / 8192

var3 = (((calc_pres / 256) ** 3) * self._pressure_calibration[9]) / 131072

calc_pres += ((var1 + var2 + var3 + (self._pressure_calibration[6] * 128)) / 16)

return calc_pres/100

@property

def humidity(self):

"""The relative humidity in RH %"""

self._perform_reading()

temp_scaled = ((self._t_fine * 5) + 128) / 256

var1 = ((self._adc_hum - (self._humidity_calibration[0] * 16)) -

((temp_scaled * self._humidity_calibration[2]) / 200))

var2 = (self._humidity_calibration[1] *

(((temp_scaled * self._humidity_calibration[3]) / 100) +

(((temp_scaled * ((temp_scaled * self._humidity_calibration[4]) / 100)) /

64) / 100) + 16384)) / 1024

var3 = var1 * var2

var4 = self._humidity_calibration[5] * 128

var4 = (var4 + ((temp_scaled * self._humidity_calibration[6]) / 100)) / 16

var5 = ((var3 / 16384) * (var3 / 16384)) / 1024

var6 = (var4 * var5) / 2

calc_hum = (((var3 + var6) / 1024) * 1000) / 4096

calc_hum /= 1000 # get back to RH

if calc_hum > 100:

calc_hum = 100

if calc_hum < 0:

calc_hum = 0

return calc_hum

@property

def altitude(self):

"""The altitude based on current ``pressure`` vs the sea level pressure

(``sea_level_pressure``) - which you must enter ahead of time)"""

pressure = self.pressure # in Si units for hPascal

return 44330 * (1.0 - math.pow(pressure / self.sea_level_pressure, 0.1903))

@property

def gas(self):

"""The gas resistance in ohms"""

self._perform_reading()

var1 = ((1340 + (5 * self._sw_err)) * (_LOOKUP_TABLE_1[self._gas_range])) / 65536

var2 = ((self._adc_gas * 32768) - 16777216) + var1

var3 = (_LOOKUP_TABLE_2[self._gas_range] * var1) / 512

calc_gas_res = (var3 + (var2 / 2)) / var2

return int(calc_gas_res)

def _perform_reading(self):

"""Perform a single-shot reading from the sensor and fill internal data structure for

calculations"""

if (time.ticks_diff(self._last_reading, time.ticks_ms()) * time.ticks_diff(0, 1)

< self._min_refresh_time):

return

# set filter

self._write(_BME680_REG_CONFIG, [self._filter << 2])

# turn on temp oversample & pressure oversample

self._write(_BME680_REG_CTRL_MEAS,

[(self._temp_oversample << 5)|(self._pressure_oversample << 2)])

# turn on humidity oversample

self._write(_BME680_REG_CTRL_HUM, [self._humidity_oversample])

# gas measurements enabled

self._write(_BME680_REG_CTRL_GAS, [_BME680_RUNGAS])

ctrl = self._read_byte(_BME680_REG_CTRL_MEAS)

ctrl = (ctrl & 0xFC) | 0x01 # enable single shot!

self._write(_BME680_REG_CTRL_MEAS, [ctrl])

new_data = False

while not new_data:

data = self._read(_BME680_REG_MEAS_STATUS, 15)

new_data = data[0] & 0x80 != 0

time.sleep(0.005)

self._last_reading = time.ticks_ms()

self._adc_pres = _read24(data[2:5]) / 16

self._adc_temp = _read24(data[5:8]) / 16

self._adc_hum = struct.unpack('>H', bytes(data[8:10]))[0]

self._adc_gas = int(struct.unpack('>H', bytes(data[13:15]))[0] / 64)

self._gas_range = data[14] & 0x0F

var1 = (self._adc_temp / 8) - (self._temp_calibration[0] * 2)

var2 = (var1 * self._temp_calibration[1]) / 2048

var3 = ((var1 / 2) * (var1 / 2)) / 4096

var3 = (var3 * self._temp_calibration[2] * 16) / 16384

self._t_fine = int(var2 + var3)

def _read_calibration(self):

"""Read & save the calibration coefficients"""

coeff = self._read(_BME680_BME680_COEFF_ADDR1, 25)

coeff += self._read(_BME680_BME680_COEFF_ADDR2, 16)

coeff = list(struct.unpack('<hbBHhbBhhbbHhhBBBHbbbBbHhbb', bytes(coeff[1:39])))

# print("\n\n",coeff)

coeff = [float(i) for i in coeff]

self._temp_calibration = [coeff[x] for x in [23, 0, 1]]

self._pressure_calibration = [coeff[x] for x in [3, 4, 5, 7, 8, 10, 9, 12, 13, 14]]

self._humidity_calibration = [coeff[x] for x in [17, 16, 18, 19, 20, 21, 22]]

self._gas_calibration = [coeff[x] for x in [25, 24, 26]]

# flip around H1 & H2

self._humidity_calibration[1] *= 16

self._humidity_calibration[1] += self._humidity_calibration[0] % 16

self._humidity_calibration[0] /= 16

self._heat_range = (self._read_byte(0x02) & 0x30) / 16

self._heat_val = self._read_byte(0x00)

self._sw_err = (self._read_byte(0x04) & 0xF0) / 16

def _read_byte(self, register):

"""Read a byte register value and return it"""

return self._read(register, 1)[0]

def _read(self, register, length):

raise NotImplementedError()

def _write(self, register, values):

raise NotImplementedError()

class BME680_I2C(Adafruit_BME680):

"""Driver for I2C connected BME680.

:param i2c: I2C device object

:param int address: I2C device address

:param bool debug: Print debug statements when True.

:param int refresh_rate: Maximum number of readings per second. Faster property reads

will be from the previous reading."""

def __init__(self, i2c, address=0x77, debug=False, *, refresh_rate=10):

"""Initialize the I2C device at the 'address' given"""

self._i2c = i2c

self._address = address

self._debug = debug

super().__init__(refresh_rate=refresh_rate)

def _read(self, register, length):

"""Returns an array of 'length' bytes from the 'register'"""

result = bytearray(length)

self._i2c.readfrom_mem_into(self._address, register & 0xff, result)

if self._debug:

print("\t${:x} read ".format(register), " ".join(["{:02x}".format(i) for i in result]))

return result

def _write(self, register, values):

"""Writes an array of 'length' bytes to the 'register'"""

if self._debug:

print("\t${:x} write".format(register), " ".join(["{:02x}".format(i) for i in values]))

for value in values:

self._i2c.writeto_mem(self._address, register, bytearray([value & 0xFF]))

class BME680_SPI(Adafruit_BME680):

"""Driver for SPI connected BME680.

:param spi: SPI device object, configured

:param cs: Chip Select Pin object, configured to OUT mode

:param bool debug: Print debug statements when True.

:param int refresh_rate: Maximum number of readings per second. Faster property reads

will be from the previous reading.

"""

def __init__(self, spi, cs, debug=False, *, refresh_rate=10):

self._spi = spi

self._cs = cs

self._debug = debug

self._cs(1)

super().__init__(refresh_rate=refresh_rate)

def _read(self, register, length):

if register != _BME680_REG_PAGE_SELECT:

# _BME680_REG_PAGE_SELECT exists in both SPI memory pages

# For all other registers, we must set the correct memory page

self._set_spi_mem_page(register)

try:

self._cs(0)

self._spi.write(bytearray([register])) # pylint: disable=no-member

result = bytearray(length)

self._spi.readinto(result) # pylint: disable=no-member

if self._debug:

print("\t${:x} read ".format(register), " ".join(["{:02x}".format(i) for i in result]))

except Exception as e:

print (e)

result = None

finally:

self._cs(1)

return result

def _write(self, register, values):

if register != _BME680_REG_PAGE_SELECT:

# _BME680_REG_PAGE_SELECT exists in both SPI memory pages

# For all other registers, we must set the correct memory page

self._set_spi_mem_page(register)

try:

self._cs(0)

buffer = bytearray(2 * len(values))

for i, value in enumerate(values):

buffer[2 * i] = register + i

buffer[2 * i + 1] = value & 0xFF

self._spi.write(buffer) # pylint: disable=no-member

if self._debug:

print("\t${:x} write".format(register), " ".join(["{:02x}".format(i) for i in values]))

except Exception as e:

print (e)

finally:

self._cs(1)

def _set_spi_mem_page(self, register):

spi_mem_page = 0x00

if register < 0x80:

spi_mem_page = 0x10

self._write(_BME680_REG_PAGE_SELECT, [spi_mem_page])

このモジュールを用いますと簡単に測定できます。Adafruitで用意しているテスト用プログラムを使用し、毎度のプルアップをソフト的に行います。

from bme680 import *
from machine import I2C, Pin
import time
p21 = Pin(21, Pin.IN, Pin.PULL_UP)
p23 = Pin(23, Pin.IN, Pin.PULL_UP)
bme = BME680_I2C(I2C(-1, Pin(21), Pin(23)))

for _ in range(3):
    print(bme.temperature, bme.humidity, bme.pressure, bme.gas)
    time.sleep(1)

from bme680 import *

from machine import I2C, Pin

import time

p21 = Pin(21, Pin.IN, Pin.PULL_UP)

p23 = Pin(23, Pin.IN, Pin.PULL_UP)

bme = BME680_I2C(I2C(-1, Pin(21), Pin(23)))

for _ in range(3):

print(bme.temperature, bme.humidity, bme.pressure, bme.gas)

time.sleep(1)

あっという間に出来上がりです。

C:\Users\>ampy -p com3 run c:\micropython\bmetest.py
Warning: I2C(-1, ...) is deprecated, use SoftI2C(...) instead
15.30285 45.3333 1031.108 269211
15.31457 45.40735 1031.111 116460
15.33859 45.47064 1031.105 124723

次回はこれを用いて、DBにつなぎます。

2021年2月25日2021年2月25日

ESP32からMariaDBにデータをとばす

センサからのデータもとれるようになり、Wi-Fiにつながり、ＮＴＰで時計の補正もできるようになりました。

IoTを進める中で必要なのはデータベースにひたすらデータを放り込み、その中で、異常のあるデータをいかに早く見つけ出すかが重要になります。このため、今回はデータベースにMariaDBを用い、ESP32からデータをPHPにPOSTしデータベースに取り込む一連の流れを紹介します。

まずはデータベースを作成します。MariaDBのインストールその他は省略しまして、既にesp_dataというデータベースを作成してあるところから始めます。phpMyAdminを使用してデータベース中にSensorDataというテーブルを作ります。次のSQL文をphpMyAdminのSQLクエリを実行し、テーブルを作ります。今回テストの都合でMACadd列は省略してあります。

CREATE TABLE SensorData (
    id INT(6) UNSIGNED AUTO_INCREMENT PRIMARY KEY,
//    MACadd VARCHAR(30) NOT NULL,
    value1 VARCHAR(10),
    value2 VARCHAR(10),
    value3 VARCHAR(10),
    reading_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
)

テーブルが出来上がったら、適当にデータを入れてみます。

INSERT INTO SensorData(value1, value2, value3) VALUES (12.3,23.4,35.3)

実行を押すとSQL文が実行され、行が挿入されます。ここまで確認したら、次はMicroPython上でjsonを用いたPOST文を作成します。

import ujson
import urequests
 

dict = {
        "data_value":"12.4",
        "temp_value":"23.4",
        "humi_value":"35.2"
        }
response = urequests.post(
        'http://**url**/postrcv.php',
        data=ujson.dumps(dict).encode("utf-8"),
        headers={'Content-Type': 'application/json'}
        )
print  (response.text)

import ujson

import urequests

dict = {

"data_value":"12.4",

"temp_value":"23.4",

"humi_value":"35.2"

}

response = urequests.post(

'http://**url**/postrcv.php',

data=ujson.dumps(dict).encode("utf-8"),

headers={'Content-Type': 'application/json'}

)

print (response.text)

MicroPythonではpythonで使用するjsonとrequestsの頭にuをつける形になります。この形でpostできます。

次にデータを受けるphp側を作成します。

<?php

$in = json_decode(stripslashes(file_get_contents('php://input')),true);
$data_value = $in["data_value"];
$temp_value = $in["temp_value"]; 
$humi_value = $in["humi_value"];

$servername = "localhost";
$dbname = "esp_data";
$username = "username";
$password = "password";

$conn = new mysqli($servername, $username, $password, $dbname);
if ($conn->connect_error) {
     die("Connection failed: " . $conn->connect_error);
} 

$sql="INSERT INTO `SensorData`(`value1`, `value2`, `value3`) VALUES ($data_value,$temp_value,$humi_value)";
if ($conn->query($sql) === TRUE) {
    echo "New record created successfully";
} 
else {
    echo "Error: " . $sql . "<br>" . $conn->error;
}
    
$conn->close();

?>

<?php

$in = json_decode(stripslashes(file_get_contents('php://input')),true);

$data_value = $in["data_value"];

$temp_value = $in["temp_value"];

$humi_value = $in["humi_value"];

$servername = "localhost";

$dbname = "esp_data";

$username = "username";

$password = "password";

$conn = new mysqli($servername, $username, $password, $dbname);

if ($conn->connect_error) {

die("Connection failed: " . $conn->connect_error);

}

$sql="INSERT INTO `SensorData`(`value1`, `value2`, `value3`) VALUES ($data_value,$temp_value,$humi_value)";

if ($conn->query($sql) === TRUE) {

echo "New record created successfully";

}

else {

echo "Error: " . $sql . "<br>" . $conn->error;

}

$conn->close();

コマンドラインからampyを用いて上記MicroPythonのプログラムを走らせた結果が次です。

C:\Users\>ampy -p com4 run c:\micropython\posttest.py
New record created successfully

phpMyAdminで確認しても新しいレコードが挿入されています。

編集コピー削除1712.323.435.32021-02-25 19:35:51

2021年2月23日

ESP32 時計まわり

ESP32をWi-Fiにてネットワークにつなぎましたので、次は時計まわりを整備します。

ESP32にはリアルタイムクロック（ＲＴＣ）が実装されていますので、まずはどのような反応が出るか見ます。

>>> from machine import RTC
>>> rtc=RTC()
>>> rtc.datetime()
(2000, 1, 1, 5, 0, 2, 41, 237515)

2000年1月1日ですね。これでは時計として使えませんので、NTPサーバーに接続して現在時刻（UTC)に合わせます。

>>> import ntptime
>>> ntptime.settime()
>>> rtc.datetime()
(2021, 2, 22, 0, 22, 6, 52, 654317)

2021年2月22日22時6分52秒ですか、4個目の０は多分曜日を表しており月曜日、654317は・・・

>>> import utime
>>> utime.localtime()
(2021, 2, 22, 22, 16, 34, 0, 53)

utimeでローカルタイム表示で見てみましょう。

2021年2月22日22時16分34秒月曜日、年内の通算日数53日目（３１＋２２＝５３）これはＵＴＣですので、ＪＳＴに合わせます。

>>> utime.mktime(utime.localtime())
667355194
>>> utime.mktime(utime.localtime())+(9*3600)
667387595
>>> utime.localtime(utime.mktime(utime.localtime())+(9*3600))
(2021, 2, 23, 9, 26, 38, 1, 54)

9時間進めますutime.mktime()関数でローカルタイムを秒数の整数に戻します。この整数にJSTは9時間進んでいますので9時間×3600秒を足してあげます。この整数をutime.localtime()関数で時間に戻します。すると2021年2月23日9時26分38秒火曜日54日経過と日本時間に変更できました。

全体のプログラムです

from time import sleep_ms
import ntptime
import utime

UTC_OFFSET=9


def do_connect():
    import network
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print('connecting to network...')
        wlan.connect('essid', 'password')
        while not wlan.isconnected():
            pass
    print('network config:', wlan.ifconfig())

     
def get_jst():
    sleep_ms(1000)
    ntptime.settime()
    tm=utime.localtime(utime.mktime(utime.localtime())+UTC_OFFSET*3600)
    jst = str(tm[0])+'/'+str(tm[1])+'/'+str(tm[2])+' '+str(tm[3])+':'+str(tm[4])+':'+str(tm[5])
    return(jst)
     
  
do_connect()
print(get_jst())

from time import sleep_ms

import ntptime

import utime

UTC_OFFSET=9

def do_connect():

import network

wlan = network.WLAN(network.STA_IF)

wlan.active(True)

if not wlan.isconnected():

print('connecting to network...')

wlan.connect('essid', 'password')

while not wlan.isconnected():

pass

print('network config:', wlan.ifconfig())

def get_jst():

sleep_ms(1000)

ntptime.settime()

tm=utime.localtime(utime.mktime(utime.localtime())+UTC_OFFSET*3600)

jst = str(tm[0])+'/'+str(tm[1])+'/'+str(tm[2])+' '+str(tm[3])+':'+str(tm[4])+':'+str(tm[5])

return(jst)

do_connect()

print(get_jst())

実行経過です。

C:\Users\>ampy -p com4 run c:\micropython\timetest1.py
network config: ('192.168.11.25', '255.255.255.0', '192.168.11.1', '192.168.11.1')
2021/2/23 9:39:41

2021年2月21日2021年2月21日

ESP32 ネットワーク接続

今回はESP32をWi-Fi接続してみました。

Wi-Fi接続は結構簡単で、MicroPython DocのESP32用クイックリファレンスhttps://micropython-docs-ja.readthedocs.io/ja/latest/esp32/quickref.htmlに詳しく書かれています。今回はこれをそっくり踏襲します。

まずはTera Term上でコマンドでの反応を見ます。

>>> wlan = network.WLAN(network.STA_IF)　　　#Wi-Fiをステーションモードにします
>>> wlan.active(True)　　　　　　　　　　　　　#アクティブ化確認します
True
>>> wlan.connect('essid','password')　　　　　#essidにはつなぐWi-Fiのessidを、passwordにはパスワードを入れます
>>> print('network config:',wlan.ifconfig())　#Wi-Fi接続のコンフィギュレーションをプリントします、ipアドレス、サブネット、ゲートウェイアドレス、DNSアドレス
network config: ('192.168.11.25', '255.255.255.0', '192.168.11.1', '192.168.11.1')
>>>
>>> wlan.scan()　　　　　　　　　　　　　　　　　#Wi-Fiの電波のスキャンをします
[(b'Buffalo-G-****', b'4=\xc4\xc3\x17\x98', 11, -41, 3, False), (b'rs500m-e17ec9-1', b'\x10KF\xe1~\xcc', 6, -67, 3, False), (b'rs500m-e17ec9-2', b'\x12KF\xe1~\xcc', 6, -69, 3, False), (b'TOBU_Free_Wi-Fi', b'\x00\x14\x06(\x191', 1, -84, 0, False), (b'aterm-36fa28-g', b'\x10f\x82M\xd2\x08', 8, -85, 4, False), (b'SPWH_H33_129E8B', b'0\xa1\xfa\x12\x9e\x8b', 2, -86, 4, False)]
>>>

これで、Wi-Fiへの接続ができることを確認しました。実際のプログラムではクイックリファレンスにあるように、関数で定義して、つなげばオッケーです。

def do_connect():
    import network
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print('connecting to network...')
        wlan.connect('essid', 'password')
        while not wlan.isconnected():
            pass
    print('network config:', wlan.ifconfig())

do_connect()

def do_connect():

import network

wlan = network.WLAN(network.STA_IF)

wlan.active(True)

if not wlan.isconnected():

print('connecting to network...')

wlan.connect('essid', 'password')

while not wlan.isconnected():

pass

print('network config:', wlan.ifconfig())

do_connect()

ampyを使用してプログラムを走らせると、以下のような結果になります。

C:\Users\>ampy -p com4 run c:\micropython\wifitest.py
connecting to network...
network config: ('192.168.11.25', '255.255.255.0', '192.168.11.1', '192.168.11.1')

2021年2月19日2021年2月19日

温度センサADT7410を用いた温度測定

前回のDHT11から、時間がたちましたが、ブレッドボードに乗っていたADT7410を用いて、温度測定をしてみたいと思います。DHT11はワンワイヤーで、ESP32自体でライブラリをもっていたので楽でしたが、ADT7410ではしっかりとメーカーの仕様を読んでおかないといけません。

メーカーのアナログデバイシスのホームページから、資料をダウンロードします。https://www.analog.com/media/en/technical-documentation/data-sheets/ADT7410.pdfの12~13ページに書いてありますが、13ビットでの測定では0x00から2バイト拾って、くっつけて、うしろの3ビット分を切ると言う作業が必要になります。

今回はsclを18ピンにsdaを19ピンに設定しました。ADT7410もプルアップが必要なので、まずはプルアップします。

>>> from machine import SoftI2C, Pin
>>> p18 = Pin(18, Pin.IN, Pin.PULL_UP)
>>> p19 = Pin(19, Pin.IN, Pin.PULL_UP)

その後softi2cでデータを読み込みます。

>>> data = SoftI2C(scl = Pin(18), sda = Pin(19), freq = 10000).readfrom_mem(0x48, 0x00, 2)

ここで、2バイト分のデータを見てみます。

>>> print(data[0])
10
>>> print(data[1])
224

10進数でプリントされるので、2進数でプリントさせます。

>>> print(bin(data[0]))
0b1010
>>> print(bin(data[1]))
0b11100000

data[0]は00001010、data[1]は11100000となってます。data[0]を左に8ビットずらしてdata[1]をつなげます。

>>> print(bin(data[0]<<8 |data[1]))
0b101011100000

0000101011100000となりました。右側3ビットは不要ですから、右に3ビットずらします。

>>> print(bin((data[0]<<8 |data[1]) >>3))
0b101011100

0000101011100になりました。10進数に治すと、

>>> print(((data[0]<<8 |data[1]) >>3))  # temp=print(((data[0]<<8 |data[1]) >>3))
348

これを13ビットですから、資料にあるとおり16で割ります。

>>> print(((data[0]<<8 |data[1]) >>3)/16)
21.75

これで温度21.75度が出てきました。

このセンサではマイナスになるとファーストビットに1が立つので、13ビットで1000000000000は10進数で4096ですから、tempがこの数より大きい場合はマイナスの温度になるのでtemp=temp-8192とすればOKです。例フルビット1111111111111は10進数で8191ですから、8191-8192=-1これを16で割って-0.0625度となります。