doctor/mini-IOT
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add

Browse Source
This commit is contained in:
doctor
2023-06-10 20:23:16 +02:00
parent 81cbb45920
commit c3a93c6365
14 changed files with 1385 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
ESP-codes/C/main.ino Normal file
View File

@@ -0,0 +1,74 @@
#include <WiFi.h>
#include <PubSubClient.h>
#include <LiquidCrystal.h>
const char* ssid = "PT4ObjConnect";
const char* password = "Admin2022";
const char* mqttServer = "192.168.1.64";
const int mqttPort = 1883;
const char* mqttUser = "patate";
const char* mqttPassword = "patate";
LiquidCrystal lcd(19, 23, 18, 17, 16, 15);
WiFiClient espClient;
PubSubClient client(espClient);
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived in topic: ");
Serial.println(topic);
Serial.print("Message:");
lcd.clear();
for (int i = 0; i < length; i++) {
Serial.print((char)payload[i]);
lcd.print((char)payload[i]);
}
Serial.println();
Serial.println("-----------------------");
}
void setup() {
Serial.begin(115200);
lcd.begin(16, 2);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.println("Connecting to WiFi..");
}
Serial.println("Connected to the WiFi network");
client.setServer(mqttServer, mqttPort);
client.setCallback(callback);
while (!client.connected()) {
Serial.println("Connecting to MQTT...");
if (client.connect("ESP32Client", mqttUser, mqttPassword )) {
Serial.println("connected");
} else {
Serial.print("failed with state ");
Serial.print(client.state());
delay(2000);
}
}
client.subscribe("esp32/lcd");
}
void loop() {
client.loop();
}

229
ESP-codes/MicroPython/bme680.py Normal file
View File

@@ -0,0 +1,229 @@
# Spaces, comments and some functions have been removed from the original file to save memory
# Original source: https://github.com/adafruit/Adafruit_CircuitPython_BME680/blob/master/adafruit_bme680.py
import time
import math
from micropython import const
from ubinascii import hexlify as hex
try:
import struct
except ImportError:
import ustruct as struct
_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):
ret = 0.0
for b in arr:
ret *= 256.0
ret += float(b & 0xFF)
return ret
class Adafruit_BME680:
def __init__(self, *, refresh_rate=10):
self._write(_BME680_REG_SOFTRESET, [0xB6])
time.sleep(0.005)
chip_id = self._read_byte(_BME680_REG_CHIPID)
if chip_id != _BME680_CHIPID:
raise RuntimeError('Failed 0x%x' % chip_id)
self._read_calibration()
self._write(_BME680_BME680_RES_HEAT_0, [0x73])
self._write(_BME680_BME680_GAS_WAIT_0, [0x65])
self.sea_level_pressure = 1013.25
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):
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")
@property
def humidity_oversample(self):
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")
@property
def temperature_oversample(self):
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")
@property
def filter_size(self):
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")
@property
def temperature(self):
self._perform_reading()
calc_temp = (((self._t_fine * 5) + 128) / 256)
return calc_temp / 100
@property
def pressure(self):
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):
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
if calc_hum > 100:
calc_hum = 100
if calc_hum < 0:
calc_hum = 0
return calc_hum
@property
def altitude(self):
pressure = self.pressure
return 44330 * (1.0 - math.pow(pressure / self.sea_level_pressure, 0.1903))
@property
def gas(self):
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):
if (time.ticks_diff(self._last_reading, time.ticks_ms()) * time.ticks_diff(0, 1)
< self._min_refresh_time):
return
self._write(_BME680_REG_CONFIG, [self._filter << 2])
self._write(_BME680_REG_CTRL_MEAS,
[(self._temp_oversample << 5)|(self._pressure_oversample << 2)])
self._write(_BME680_REG_CTRL_HUM, [self._humidity_oversample])
self._write(_BME680_REG_CTRL_GAS, [_BME680_RUNGAS])
ctrl = self._read_byte(_BME680_REG_CTRL_MEAS)
ctrl = (ctrl & 0xFC) | 0x01
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):
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])))
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]]
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):
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):
def __init__(self, i2c, address=0x77, debug=False, *, refresh_rate=10):
self._i2c = i2c
self._address = address
self._debug = debug
super().__init__(refresh_rate=refresh_rate)
def _read(self, register, length):
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):
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

1
ESP-codes/MicroPython/boot.py Normal file
View File

@@ -0,0 +1 @@
import ssid_connect

129
ESP-codes/MicroPython/main.py Normal file
View File

@@ -0,0 +1,129 @@
from machine import Pin, SoftI2C
from time import sleep
from bme680 import *
import ssd1306
import machine
import network
from umqtt.robust import MQTTClient
import sys
pwm = machine.PWM(machine.Pin(16), freq=50)
wifi = network.WLAN(network.STA_IF)
client = MQTTClient(client_id=b'esp32',
server=b'192.168.1.64',
port=1883,
user=b'patate',
password=b'patate',
keepalive=30,
ssl=False)
try:
client.connect()
print('Connection ok');
except Exception as e:
print('Could not connect to MQTT server {}{}'.format(type(e).__name__, e))
sys.exit()
# Fonction pour déplacer le servo moteur vers une position donnée
def set_position(position):
# Convertion de la position en valeur de duty
duty = int(position / 180 * 1023)
# Envoi de la valeur de duty au servo moteur
pwm.duty(duty)
i2c = SoftI2C(scl=Pin(22), sda=Pin(21))
bme = BME680_I2C(i2c=i2c)
# Définition de la tailler de l'écran oled
oled_width = 128
oled_height = 64
oled = ssd1306.SSD1306_I2C(oled_width, oled_height, i2c)
if wifi.isconnected():
print ("connecté au réseau wifi")
print(wifi.ifconfig())
else:
print ("NON connecté au réseau wifi")
pin_led = Pin(2, mode=Pin.OUT) # Led sur l'ESP
ledR=Pin(14,Pin.OUT) # LED externe
def control(topic,msg):
print(msg)
if 'ledE' in topic:
if 'true' in msg :
pin_led.on()
else :
pin_led.off()
elif 'ledR' in topic:
if 'true' in msg :
ledR.value(1)
else :
ledR.value(0)
elif 'motor' in topic:
if 'Droite' in msg :
set_position(10)
time.sleep(0.4)
set_position(0)
else:
set_position(20)
time.sleep(0.4)
set_position(0)
client.set_callback(control)
client.subscribe('esp32/ledE', 1)
client.subscribe('esp32/ledR', 1)
client.subscribe('esp32/motor', 1)
count=120
while True:
try:
client.check_msg()
count+=1
if (count>120) :
# Données du BME680 :
temperature = str(round(bme.temperature, 2))
humidite = str(round(bme.humidity, 2))
pression = str(round(bme.pressure, 2))
gas = str(round(bme.gas/1000, 2))
#print('temperature:', temperature,'|','humidite:', humidite,'|','pression:', pression, '|', 'Gas:', gas)
oled.text(temperature + ' C', 0, 5) # Affichage de la temperatureerature
oled.text(humidite + ' %', 0, 20) # Affichage de l'humiditeidité
oled.text(pression + ' hPa', 0, 35) # Affichage de la pressionsion atmosphérique
oled.text(gas + ' ppm', 0, 50) # Affichage du gaz
oled.show() # Affiche les valeurs sur l'écran
oled.fill(0) # Renitialise l'affichage
bme680='{"temperature": '+ temperature + ',' + '"humidite": '+ humidite + ',' + '"pression": ' + pression + ',' + '"gas": ' + gas + '}'
count=0
client.publish('esp32/bme', bme680)
print("Publish : " + str(bme680))
time.sleep(0.1)
except KeyboardInterrupt:
print('Ctrl-C pressionsed...exiting')
client.disconnect()
sys.exit()

167
ESP-codes/MicroPython/ssd1306.py Normal file
View File

@@ -0,0 +1,167 @@
#MicroPython SSD1306 OLED driver, I2C and SPI interfaces created by Adafruit
import time
import framebuf
# register definitions
SET_CONTRAST = const(0x81)
SET_ENTIRE_ON = const(0xa4)
SET_NORM_INV = const(0xa6)
SET_DISP = const(0xae)
SET_MEM_ADDR = const(0x20)
SET_COL_ADDR = const(0x21)
SET_PAGE_ADDR = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP = const(0xa0)
SET_MUX_RATIO = const(0xa8)
SET_COM_OUT_DIR = const(0xc0)
SET_DISP_OFFSET = const(0xd3)
SET_COM_PIN_CFG = const(0xda)
SET_DISP_CLK_DIV = const(0xd5)
SET_PRECHARGE = const(0xd9)
SET_VCOM_DESEL = const(0xdb)
SET_CHARGE_PUMP = const(0x8d)
class SSD1306:
def __init__(self, width, height, external_vcc):
self.width = width
self.height = height
self.external_vcc = external_vcc
self.pages = self.height // 8
# Note the subclass must initialize self.framebuf to a framebuffer.
# This is necessary because the underlying data buffer is different
# between I2C and SPI implementations (I2C needs an extra byte).
self.poweron()
self.init_display()
def init_display(self):
for cmd in (
SET_DISP | 0x00, # off
# address setting
SET_MEM_ADDR, 0x00, # horizontal
# resolution and layout
SET_DISP_START_LINE | 0x00,
SET_SEG_REMAP | 0x01, # column addr 127 mapped to SEG0
SET_MUX_RATIO, self.height - 1,
SET_COM_OUT_DIR | 0x08, # scan from COM[N] to COM0
SET_DISP_OFFSET, 0x00,
SET_COM_PIN_CFG, 0x02 if self.height == 32 else 0x12,
# timing and driving scheme
SET_DISP_CLK_DIV, 0x80,
SET_PRECHARGE, 0x22 if self.external_vcc else 0xf1,
SET_VCOM_DESEL, 0x30, # 0.83*Vcc
# display
SET_CONTRAST, 0xff, # maximum
SET_ENTIRE_ON, # output follows RAM contents
SET_NORM_INV, # not inverted
# charge pump
SET_CHARGE_PUMP, 0x10 if self.external_vcc else 0x14,
SET_DISP | 0x01): # on
self.write_cmd(cmd)
self.fill(0)
self.show()
def poweroff(self):
self.write_cmd(SET_DISP | 0x00)
def contrast(self, contrast):
self.write_cmd(SET_CONTRAST)
self.write_cmd(contrast)
def invert(self, invert):
self.write_cmd(SET_NORM_INV | (invert & 1))
def show(self):
x0 = 0
x1 = self.width - 1
if self.width == 64:
# displays with width of 64 pixels are shifted by 32
x0 += 32
x1 += 32
self.write_cmd(SET_COL_ADDR)
self.write_cmd(x0)
self.write_cmd(x1)
self.write_cmd(SET_PAGE_ADDR)
self.write_cmd(0)
self.write_cmd(self.pages - 1)
self.write_framebuf()
def fill(self, col):
self.framebuf.fill(col)
def pixel(self, x, y, col):
self.framebuf.pixel(x, y, col)
def scroll(self, dx, dy):
self.framebuf.scroll(dx, dy)
def text(self, string, x, y, col=1):
self.framebuf.text(string, x, y, col)
class SSD1306_I2C(SSD1306):
def __init__(self, width, height, i2c, addr=0x3c, external_vcc=False):
self.i2c = i2c
self.addr = addr
self.temp = bytearray(2)
# Add an extra byte to the data buffer to hold an I2C data/command byte
# to use hardware-compatible I2C transactions. A memoryview of the
# buffer is used to mask this byte from the framebuffer operations
# (without a major memory hit as memoryview doesn't copy to a separate
# buffer).
self.buffer = bytearray(((height // 8) * width) + 1)
self.buffer[0] = 0x40 # Set first byte of data buffer to Co=0, D/C=1
self.framebuf = framebuf.FrameBuffer1(memoryview(self.buffer)[1:], width, height)
super().__init__(width, height, external_vcc)
def write_cmd(self, cmd):
self.temp[0] = 0x80 # Co=1, D/C#=0
self.temp[1] = cmd
self.i2c.writeto(self.addr, self.temp)
def write_framebuf(self):
# Blast out the frame buffer using a single I2C transaction to support
# hardware I2C interfaces.
self.i2c.writeto(self.addr, self.buffer)
def poweron(self):
pass
class SSD1306_SPI(SSD1306):
def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
self.rate = 10 * 1024 * 1024
dc.init(dc.OUT, value=0)
res.init(res.OUT, value=0)
cs.init(cs.OUT, value=1)
self.spi = spi
self.dc = dc
self.res = res
self.cs = cs
self.buffer = bytearray((height // 8) * width)
self.framebuf = framebuf.FrameBuffer1(self.buffer, width, height)
super().__init__(width, height, external_vcc)
def write_cmd(self, cmd):
self.spi.init(baudrate=self.rate, polarity=0, phase=0)
self.cs.high()
self.dc.low()
self.cs.low()
self.spi.write(bytearray([cmd]))
self.cs.high()
def write_framebuf(self):
self.spi.init(baudrate=self.rate, polarity=0, phase=0)
self.cs.high()
self.dc.high()
self.cs.low()
self.spi.write(self.buffer)
self.cs.high()
def poweron(self):
self.res.high()
time.sleep_ms(1)
self.res.low()
time.sleep_ms(10)
self.res.high()

22
ESP-codes/MicroPython/ssid_connect.py Normal file
View File

@@ -0,0 +1,22 @@
ssid = "PT4ObjConnect" # wifi router name
pw = "Admin2022" # wifi router password
def do_connect():
import network
sta_if = network.WLAN(network.STA_IF)
if not sta_if.isconnected():
print('connecting to network...')
sta_if.active(True)
sta_if.connect(ssid, pw)
while not sta_if.isconnected():
pass
print('network config:', sta_if.ifconfig())
def configure_AP():
import network
ap = network.WLAN(network.AP_IF)
ap.active(True)
ap.config(essid='MP115', password='MP115')
do_connect()
#configure_AP()

25
ESP-codes/MicroPython/wifi.py Normal file
View File

@@ -0,0 +1,25 @@
import esp
import network
def disable_dhcp():
esp.osdebug(None)
nic = network.WLAN(network.STA_IF)
nic.active(True)
nic.ifconfig((nic.ifconfig()[0], '255.255.255.0', '0.0.0.0', '0.0.0.0'))
def connect(ssid, password):
esp.sleep_type(esp.SLEEP_LIGHT)
nic = network.WLAN(network.STA_IF)
nic.active(True)
if not nic.isconnected():
nic.connect(ssid, password)
def config(ip, subnet, gateway):
esp.sleep_type(esp.SLEEP_LIGHT)
nic = network.WLAN(network.STA_IF)
nic.ifconfig((ip, subnet, gateway, '0.0.0.0'))
def isconnected():
esp.sleep_type(esp.SLEEP_LIGHT)
nic = network.WLAN(network.STA_IF)
return nic.isconnected()