## Exemple CircuitPython pour l'ES32D26
## Certaines bibliothèques peuvent ne pas être utilisées, j'ai nettoyé cela à partir d'un autre projet.
# boot_out.txt
## Adafruit CircuitPython 10.0.0 le 2025-10-01 ; ESP32 Devkit V1 avec ESP32
## ID de la carte : doit_esp32_devkit_v1
## UID : 8875121BD145
## MAC : 88:57:21:B1:1D:54
# config.json
#{
# "wifi_info":{
# "hostname":"ES32D26",
# "ssid": "myNetwork",
# "psk": "lookattheceiling"
# },
# "mqtt_info":{
# "broker": "mosquitto.local",
# "user": "myuser",
# "pass": "dumbpassword",
# "topic": "ES32D26/"
# }
#}
import wifi
import adafruit_74hc595
import wws_74hc165 ## https://github.com/WoolseyWorkshop/WoolseyWorkshop_CircuitPython_74HC165
import board
import microcontroller
import busio
import digitalio
import time
import json
import os
import adafruit_simplemath
import adafruit_connection_manager
import socketpool
import adafruit_minimqtt.adafruit_minimqtt as MQTT
import analogio
# mètres en pieds
def mtf(meters):
return meters * 3.28084
# pieds en mètres
def ftm(feet):
return feet * 0.3048
# je fais ces deux-là en premier pour pouvoir utiliser la config dans d'autres fonctions.
def load_config_json():
try:
# Ouvrir le fichier JSON en mode lecture ('r')
f = open("/config.json", 'r')
config = json.loads(f.read())
f.close
return config
except:
pass
def save_config_json(config):
try:
f = open("/config.json", 'w')
f.write(config)
f.close
except:
pass
# Mon logger paresseux TODO : rendre ça un peu plus intelligent. Enregistrer un niveau de log dans la config, ne rien loguer au-dessus du niveau de log
def log(level,log_msg):
# niveaux de log
# debug
# verbose
# par défaut
# aucun
mqtt_client.publish(config["mqtt_info"]["topic"]+"log/"+level, log_msg)
## callbacks mqtt
def connected(client, userdata, flags, rc):
# Cette fonction sera appelée lorsque le client sera connecté
# avec succès au broker.
#print(f"Connecté à "+config["mqtt_info"]["broker"]+" ! Écoute les changements de topic sur "+config["mqtt_info"]["topic"])
# S'abonner à tous les changements sur le topic suivant. Cela devrait peut-être avoir sa propre variable de config.
client.subscribe(config["mqtt_info"]["topic"]+"listenHere/#")
def disconnected(client, userdata, rc):
# Cette méthode est appelée lorsque le client est déconnecté
#print("Déconnecté de "+config["mqtt_info"]["broker"]+" !")
pass
def message(client, topic, message):
# Cette méthode est appelée lorsqu'un topic auquel le client est abonné
# a un nouveau message.
# Ajoutez votre code ici pour faire quelque chose lorsqu'il reçoit un message.
# Assurez-vous de ne pas vous abonner au même topic auquel vous publiez
# , il n'y a pas de protection de boucle locale.
print("Topic:"+topic+" Message:"+message)
def map_range(value,input_upper,input_lower,output_upper,output_lower):
# Cela va mettre à l'échelle vos entrées vers la sortie souhaitée
# input_lower est 0 et input_upper est 8195 pour l'esp32
return adafruit_simplemath.map_range(value, input_lower, input_upper, output_lower, output_upper)
# charger le fichier de config
config = load_config_json()
# configurer le réseau
wifi.radio.hostname = config["wifi_info"]["hostname"]
wifi.radio.connect(config["wifi_info"]["ssid"], config["wifi_info"]["psk"])
pool = socketpool.SocketPool(wifi.radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)
# configurer les E/S
## configurer les relais 74hc595
relay_pin_data = board.D12
relay_pin_clk = board.D22
relay_pin_latch = digitalio.DigitalInOut(board.D23)
relay_pin_oe = digitalio.DigitalInOut(board.D13) ## active_low
relay_spi = busio.SPI(relay_pin_clk, MOSI=relay_pin_data)
sr = adafruit_74hc595.ShiftRegister74HC595(relay_spi, relay_pin_latch)
relay_pin_oe.direction = digitalio.Direction.OUTPUT
## maintenant définir la broche dans une variable
relay1 = sr.get_pin(0)
relay2 = sr.get_pin(1)
relay3 = sr.get_pin(2)
relay4 = sr.get_pin(3)
relay5 = sr.get_pin(4)
relay6 = sr.get_pin(5)
relay7 = sr.get_pin(6)
relay8 = sr.get_pin(7)
## faire un peu de nettoyage
relay1.value = False # mettre un relais sur off avant d'activer les sorties pour avoir un état connu lors de l'activation de la sortie
relay_pin_oe.value = False # Cette sortie est active low
## configurer les entrées discrètes 74hc165
## rappel, celles-ci retournent false lorsqu'elles sont reliées à la terre.
inputs_pin_data = board.D15
inputs_pin_clk = board.D2
inputs_pin_load = digitalio.DigitalInOut(microcontroller.pin.GPIO0)
inputs_spi = busio.SPI(inputs_pin_clk, MISO=inputs_pin_data)
inputs_sr = wws_74hc165.ShiftRegister74HC165(inputs_spi, inputs_pin_load)
input1 = inputs_sr.get_pin(0)
input2 = inputs_sr.get_pin(1)
input3 = inputs_sr.get_pin(2)
input4 = inputs_sr.get_pin(3)
input5 = inputs_sr.get_pin(4)
input6 = inputs_sr.get_pin(5)
input7 = inputs_sr.get_pin(6)
input8 = inputs_sr.get_pin(7)
# entrées analogiques
## entrées 4-20ma
analog_i_1 = analogio.AnalogIn(microcontroller.pin.GPIO34)
analog_i_2 = analogio.AnalogIn(microcontroller.pin.GPIO39)
analog_i_3 = analogio.AnalogIn(microcontroller.pin.GPIO35)
analog_i_4 = analogio.AnalogIn(microcontroller.pin.GPIO36)
## entrées 0-10v
analog_v_1 = analogio.AnalogIn(microcontroller.pin.GPIO14)
analog_v_2 = analogio.AnalogIn(microcontroller.pin.GPIO33)
analog_v_3 = analogio.AnalogIn(microcontroller.pin.GPIO27)
analog_v_4 = analogio.AnalogIn(microcontroller.pin.GPIO32)
# sorties analogiques
## sorties analogiques, configurables par dip pour 0-10v ou 4-20ma
analog_out_1 = analogio.AnalogOut(microcontroller.pin.GPIO25)
analog_out_2 = analogio.AnalogOut(microcontroller.pin.GPIO26)
# configurer le client mqtt
mqtt_client = MQTT.MQTT(
broker=config["mqtt_info"]["broker"],
username=config["mqtt_info"]["user"],
password=config["mqtt_info"]["pass"],
is_ssl=False,
socket_pool=pool,
ssl_context=ssl_context,
)
# définir notre message LWT
mqtt_client.will_set(config["mqtt_info"]["topic"]+'status','offline')
# définir nos callbacks
mqtt_client.on_connect = connected
mqtt_client.on_disconnect = disconnected
mqtt_client.on_message = message
# Connecter le client au broker MQTT.
print("Connexion à "+config["mqtt_info"]["broker"])
mqtt_client.connect()
log("default","Démarrage du système")
fiveSecondTimer = thirtySecondTimer = sixtySecondTimer = ninetySecondTimer = startTime = time.time()
while True:
# Interroger la file de messages
mqtt_client.loop()
if fiveSecondTimer <= time.time():
# Construire nos listes
discrete_input_status = [
input1.value,
input2.value,
input3.value,
input4.value,
input4.value,
input5.value,
input6.value,
input7.value,
input8.value
]
analog_input_status = [
analog_i_1.value,
analog_i_2.value,
analog_i_3.value,
analog_i_4.value,
#analog_v_1.value,
#analog_v_2.value,
#analog_v_3.value,
#analog_v_4.value
]
relay_output_status = [
relay1.value,
relay2.value,
relay3.value,
relay4.value,
relay5.value,
relay6.value,
relay7.value,
relay8.value
]
# Publier nos listes
mqtt_client.publish(config["mqtt_info"]["topic"]+"config", json.dumps(config))
mqtt_client.publish(config["mqtt_info"]["topic"]+"discrete_input_status", str(discrete_input_status))
mqtt_client.publish(config["mqtt_info"]["topic"]+"analog_input_status", str(analog_input_status))
mqtt_client.publish(config["mqtt_info"]["topic"]+"relay_output_status", str(relay_output_status))
fiveSecondTimer = time.time()+5
if thirtySecondTimer <= time.time():
thirtySecondTimer = time.time()+30
if sixtySecondTimer <= time.time():
sixtySecondTimer = time.time()+30
if ninetySecondTimer <= time.time():
ninetySecondTimer = time.time()+90
# fin de l'exécution
log("debug","fin de l'exécution")
