OMGeeky/homecontrol.esp-sensors
1
0
Fork 0
mirror of https://github.com/OMGeeky/homecontrol.esp-sensors.git synced 2026-02-23 15:49:52 +01:00
Code Issues Packages Projects Releases Wiki Activity

a bunch of changes

Browse Source
This commit is contained in:
OMGeeky
2025-05-17 15:22:06 +02:00
parent 8bc58e4056
commit baccbfbfdd
15 changed files with 331 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
src/esp_sensors/config.py
View File

@@ -7,8 +7,6 @@ sensor names, and intervals without modifying the code.
"""
import json
import os
from typing import Dict, Any, Optional, Union, List
# Default configuration file path
DEFAULT_CONFIG_PATH = "config.json"
@@ -48,10 +46,14 @@ DEFAULT_CONFIG = {
"ssl": False,
"keepalive": 60,
},
"network": {
"ssid": "<your ssid>",
"password": "<your password>",
}
}
def load_config(config_path: str = DEFAULT_CONFIG_PATH) -> Dict[str, Any]:
def load_config(config_path: str = DEFAULT_CONFIG_PATH) :
"""
Load configuration from a JSON file.
@@ -64,43 +66,18 @@ def load_config(config_path: str = DEFAULT_CONFIG_PATH) -> Dict[str, Any]:
If the file doesn't exist or can't be read, returns the default configuration.
"""
try:
if os.path.exists(config_path):
with open(config_path, "r") as f:
config = json.load(f)
return config
else:
print(
f"Configuration file {config_path} not found. Using default configuration."
)
return DEFAULT_CONFIG
with open(config_path, "r") as f:
print(f"Loading configuration from '{config_path}'")
config = json.load(f)
return config
except Exception as e:
print(f"Error loading configuration: {e}. Using default configuration.")
return DEFAULT_CONFIG
def save_config(config: Dict[str, Any], config_path: str = DEFAULT_CONFIG_PATH) -> bool:
"""
Save configuration to a JSON file.
Args:
config: Configuration dictionary to save
config_path: Path to the configuration file (default: config.json)
Returns:
True if the configuration was saved successfully, False otherwise
"""
try:
with open(config_path, "w") as f:
json.dump(config, f, indent=4)
return True
except Exception as e:
print(f"Error saving configuration: {e}")
return False
def get_sensor_config(
sensor_type: str, config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
sensor_type: str, config: dict | None = None
) -> dict:
"""
Get configuration for a specific sensor type.
@@ -123,8 +100,8 @@ def get_sensor_config(
def get_display_config(
display_type: str, config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
display_type: str, config: dict | None = None
) -> dict:
"""
Get configuration for a specific display type.
@@ -147,8 +124,8 @@ def get_display_config(
def get_button_config(
button_name: str = "main_button", config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
button_name: str = "main_button", config: dict | None = None
) -> dict:
"""
Get configuration for a specific button.
@@ -170,7 +147,7 @@ def get_button_config(
return button_config
def get_mqtt_config(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
def get_mqtt_config(config: dict | None = None) -> dict:
"""
Get MQTT configuration.
@@ -189,16 +166,3 @@ def get_mqtt_config(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
mqtt_config = DEFAULT_CONFIG.get("mqtt", {})
return mqtt_config
def create_default_config(config_path: str = DEFAULT_CONFIG_PATH) -> bool:
"""
Create a default configuration file.
Args:
config_path: Path to the configuration file (default: config.json)
Returns:
True if the configuration was created successfully, False otherwise
"""
return save_config(DEFAULT_CONFIG, config_path)

15
src/esp_sensors/dht22.py
View File

@@ -2,9 +2,6 @@
DHT22 temperature and humidity sensor module for ESP32.
"""
import time
from typing import Dict, Any, Optional
try:
import dht
from machine import Pin
@@ -29,7 +26,7 @@ class DHT22Sensor(TemperatureSensor, HumiditySensor):
pin: int = None,
interval: int = None,
temperature_unit: str = None,
sensor_config: Dict[str, Any] = None,
sensor_config: dict | None = None,
):
"""
Initialize a new DHT22 sensor.
@@ -67,7 +64,10 @@ class DHT22Sensor(TemperatureSensor, HumiditySensor):
# Initialize the sensor if not in simulation mode
if not SIMULATION:
self._sensor = dht.DHT22(Pin(pin))
print("Initializing DHT22 sensor...")
pin1 = Pin(self.pin)
self._sensor = dht.DHT22(pin1)
print(f"DHT22 sensor initialized on pin {pin1}")
def apply_parameters(self, interval, name, pin, sensor_config):
# Get main parameters from config if not provided
@@ -160,7 +160,10 @@ class DHT22Sensor(TemperatureSensor, HumiditySensor):
humidity_metadata = HumiditySensor.get_metadata(self)
# Combine metadata from both parent classes
metadata = {**temp_metadata, **humidity_metadata}
metadata = {}
metadata.update(temp_metadata)
metadata.update(humidity_metadata)
# metadata = {**temp_metadata, **humidity_metadata}
# Ensure the name is the main sensor name, not the humidity sensor name
metadata["name"] = self.name
metadata["type"] = "DHT22"

3
src/esp_sensors/humidity.py
View File

@@ -3,7 +3,6 @@ Humidity sensor module for ESP-based sensors.
"""
import random
from typing import Dict, Any, Optional
from .sensor import Sensor
from .config import get_sensor_config
@@ -16,7 +15,7 @@ class HumiditySensor(Sensor):
name: str = None,
pin: int = None,
interval: int = None,
sensor_config: Dict[str, Any] = None,
sensor_config: dict | None = None,
**kwargs,
):
"""

9
src/esp_sensors/mqtt.py
View File

@@ -7,7 +7,6 @@ It supports both real hardware and simulation mode.
import time
import json
from typing import Dict, Any, Optional, Union
# Import hardware-specific modules if available (for ESP32/ESP8266)
try:
@@ -56,7 +55,7 @@ except ImportError:
return
def setup_mqtt(mqtt_config: Dict[str, Any]) -> Optional[MQTTClient]:
def setup_mqtt(mqtt_config: dict) -> MQTTClient | None:
"""
Set up and connect to the MQTT broker.
@@ -92,9 +91,9 @@ def setup_mqtt(mqtt_config: Dict[str, Any]) -> Optional[MQTTClient]:
def publish_sensor_data(
client: Optional[MQTTClient],
mqtt_config: Dict[str, Any],
sensor: Any,
client: MQTTClient | None,
mqtt_config: dict,
sensor,
temperature: float,
humidity: float,
) -> bool:

5
src/esp_sensors/oled_display.py
View File

@@ -2,9 +2,6 @@
OLED display module for ESP32 using SSD1306 controller.
"""
import time
from typing import Dict, Any, Optional
try:
from machine import Pin, I2C
import ssd1306
@@ -30,7 +27,7 @@ class OLEDDisplay(Sensor):
address: int | str = None,
interval: int = None,
on_time: int = None,
display_config: Dict[str, Any] = None,
display_config = None,
):
"""
Initialize a new OLED display.

8
src/esp_sensors/sensor.py
View File

@@ -2,8 +2,6 @@
Base sensor module for ESP-based sensors.
"""
from typing import Dict, Any, Optional
from .config import get_sensor_config
class Sensor:
@@ -14,7 +12,7 @@ class Sensor:
name: str = None,
pin: int = None,
interval: int = None,
sensor_config: Dict[str, Any] = None,
sensor_config = None,
):
"""
Initialize a new sensor.
@@ -37,7 +35,7 @@ class Sensor:
interval if interval is not None else sensor_config.get("interval", 60)
)
self._last_reading: Optional[float] = None
self._last_reading= None
def read(self) -> float:
"""
@@ -51,7 +49,7 @@ class Sensor:
self._last_reading = 0.0
return self._last_reading
def get_metadata(self) -> Dict[str, Any]:
def get_metadata(self) :
"""
Get sensor metadata.

3
src/esp_sensors/temperature.py
View File

@@ -3,7 +3,6 @@ Temperature sensor module for ESP-based sensors.
"""
import random
from typing import Dict, Any, Optional
from .sensor import Sensor
from .config import get_sensor_config
@@ -17,7 +16,7 @@ class TemperatureSensor(Sensor):
pin: int = None,
interval: int = None,
unit: str = None,
sensor_config: Dict[str, Any] = None,
sensor_config: dict | None = None,
):
"""
Initialize a new temperature sensor.

202
src/main.py
View File

@@ -10,7 +10,6 @@ This program:
"""
import time
import sys
from esp_sensors.oled_display import OLEDDisplay
from esp_sensors.dht22 import DHT22Sensor
@@ -81,151 +80,92 @@ def main():
"""
Main function to demonstrate button-triggered sensor display with MQTT publishing.
"""
# record start time
last_read_time = time.time() # this is to make sure, the sleep time is correct
# Load configuration
config = load_config()
button_config = get_button_config("main_button", config)
# print('config: ', config)
# button_config = get_button_config("main_button", config)
mqtt_config = get_mqtt_config(config)
dht_config = get_sensor_config("dht22", config)
display_config = get_display_config("oled", config)
network_config = config.get("network", {})
# Initialize a DHT22 sensor using configuration
dht_sensor = DHT22Sensor(
sensor_config=get_sensor_config("dht22", config) # Pass the loaded config
)
dht_sensor = DHT22Sensor(sensor_config=dht_config)
if network_config.get("ssid"):
ssid = network_config.get("ssid")
password = network_config.get("password")
connect_wifi(ssid, password)
# Initialize an OLED display using configuration
display = OLEDDisplay(
display_config=get_display_config("oled", config) # Pass the loaded config
)
display = OLEDDisplay(display_config=display_config)
# Set up MQTT client if enabled
mqtt_client = setup_mqtt(mqtt_config)
mqtt_publish_interval = mqtt_config.get("publish_interval", 60)
last_publish_time = 0
last_read_time = 0 # Track the last time sensors were read
# Set up button using configuration
button_pin = button_config.get("pin", 0)
if not SIMULATION:
pull_up = button_config.get("pull_up", True)
button = Pin(button_pin, Pin.IN, Pin.PULL_UP if pull_up else None)
# Display initialization message
# # Set up button using configuration
# button_pin = button_config.get("pin", 0)
# if not SIMULATION:
# pull_up = button_config.get("pull_up", True)
# button = Pin(button_pin, Pin.IN, Pin.PULL_UP if pull_up else None)
#
# # Display initialization message
display.clear()
display.display_text("Ready - Auto & Button", 0, 0)
print(
f"System initialized. Will run every {mqtt_publish_interval} seconds or on button press..."
)
print(f"System initialized. Will run every {mqtt_publish_interval} seconds or on button press...")
# Main loop - sleep until button press, then read and display sensor data
try:
while True:
# Calculate time until next scheduled reading
current_time = time.time()
time_since_last_read = current_time - last_read_time
time_until_next_read = max(
0, mqtt_publish_interval - int(time_since_last_read)
)
# while True:
print('sleeping for 5 seconds for debugging')
time.sleep(5)
# Wait for button press or until next scheduled reading
if SIMULATION:
# In simulation mode, wait for Enter key with timeout
# Read sensor values
temperature = dht_sensor.read_temperature()
humidity = dht_sensor.read_humidity()
#
# # Format values for display
temp_str = f"Temp: {temperature:.1f} C"
hum_str = f"Humidity: {humidity:.1f}%"
time_str = f"Time: {time.time():.0f}"
name_str = f"Sensor: {dht_sensor.name}"
if not simulate_button_press(timeout=time_until_next_read):
break # Exit if 'q' was pressed or Ctrl+C
# Display values
# TODO: only display values, if the button has been clicked
display.display_values(
[name_str, temp_str, hum_str, time_str, "Press button again"]
)
time.sleep(3)
else:
# In hardware mode, check if button is pressed (active low)
button_pressed_value = 0 if not pull_up else 1 # TODO: check if 1 is correct
if button.value() == button_pressed_value: # Button is pressed
pass
elif time_since_last_read >= mqtt_publish_interval:
# Time for scheduled reading
pass
else:
# Go to light sleep mode to save power
# Wake up on pin change (button press) or timer
print(
f"Entering light sleep mode for {time_until_next_read:.1f} seconds or until button press..."
)
# Print to console
print('='*20)
print(f"Sensor data: {temp_str}, {hum_str}")
print('='*20)
# Set up wake on button press
esp32.wake_on_ext0(
pin=button, level=button_pressed_value
) # Wake on button press (low)
# Publish to MQTT
publish_sensor_data(mqtt_client, mqtt_config, dht_sensor, temperature, humidity)
# Set up wake on timer
if last_read_time > 0: # Skip timer on first run
# Convert seconds to milliseconds for sleep
esp32.wake_on_timer(time_until_next_read * 1000)
if mqtt_client:
try:
mqtt_client.disconnect()
print("MQTT client disconnected")
except Exception as e:
print(f"Error disconnecting MQTT client: {e}")
# Enter light sleep
esp32.light_sleep() # Light sleep preserves RAM but saves power
time_until_next_read = mqtt_publish_interval - (time.time() - last_read_time)
print('sleeping for', time_until_next_read, 'seconds')
if not SIMULATION:
deepsleep(time_until_next_read * 1000)
else:
# Simulate deep sleep
print(f"Simulated deep sleep for {time_until_next_read:.1f} seconds")
time.sleep(time_until_next_read)
# When we get here, either the button was pressed or the timer expired
print(f"Awake from light sleep")
# Determine if this was triggered by a button press or scheduled interval
if SIMULATION:
trigger_source = "user input or scheduled interval"
else:
trigger_source = (
"button press" if button.value() == 0 else "scheduled interval"
)
print(f"Reading sensor data (triggered by {trigger_source})...")
# Read sensor values
temperature = dht_sensor.read_temperature()
humidity = dht_sensor.read_humidity()
# Update last read time
last_read_time = time.time()
# Adjust for actual sleep duration (so if we sleep longer or shorter (because of
# the button) than expected, we don't miss the next scheduled read)
actual_sleep_duration = last_read_time - current_time
last_read_time -= (
(actual_sleep_duration - time_until_next_read)
% mqtt_publish_interval
)
# Format values for display
temp_str = f"Temp: {temperature:.1f} C"
hum_str = f"Humidity: {humidity:.1f}%"
time_str = f"Time: {time.time():.0f}"
name_str = f"Sensor: {dht_sensor.name}"
# Display values
# TODO: only display values, if the button has been clicked
display.display_values(
[name_str, temp_str, hum_str, time_str, "Press button again"]
)
# Print to console
print(f"Updated display with: {temp_str}, {hum_str}")
# Publish to MQTT if enabled
current_time = time.time()
publish_sensor_data(
mqtt_client, mqtt_config, dht_sensor, temperature, humidity
)
if mqtt_client and (
current_time - last_publish_time >= mqtt_publish_interval
):
last_publish_time = current_time
print(f"Next normal MQTT publish in {mqtt_publish_interval} seconds")
# Keep display on for a few seconds before going back to sleep
time.sleep(display.on_time)
# Clear display to save power
display.clear()
display.display_text("Ready - Auto & Button", 0, 0)
print("last_read_time", last_read_time)
if SIMULATION:
print(
f"Display cleared. Will run again in {mqtt_publish_interval - (time.time() - last_read_time):.1f} seconds or on button press."
)
except KeyboardInterrupt:
# Clean up on exit
@@ -245,5 +185,23 @@ def main():
print("Program terminated by user")
def connect_wifi(ssid, password):
import network
print(f'Connecting to WIFI: "{ssid}"')
# Connect to your network
station = network.WLAN(network.STA_IF)
station.active(True)
station.connect(ssid, password)
connection_start_time = time.time()
while not station.isconnected():
# Check if connection attempt has timed out
if time.time() - connection_start_time > 10:
print("Connection timed out")
return False
pass
print('Connection successful')
print(station.ifconfig())
if __name__ == "__main__":
main()