OMGeeky/homecontrol.esp-sensors
1
0
Fork 0
mirror of https://github.com/OMGeeky/homecontrol.esp-sensors.git synced 2025-12-26 17:02:29 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat: add display power management and improve WiFi connection handling

Browse Source
This commit is contained in:
OMGeeky
2025-06-07 15:00:29 +02:00
parent 21ad532cf4
commit 171336b6fd
2 changed files with 150 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/esp_sensors/oled_display.py
View File

@@ -116,6 +116,26 @@ class OLEDDisplay(Sensor):
self._display = None
# region basic display methods
def power_off(self):
"""
Turn off the display to save power.
"""
if SIMULATION:
print("Simulated OLED display powered off")
else:
if self._display:
self._display.poweroff()
def power_on(self):
"""
Turn on the display.
"""
if SIMULATION:
print("Simulated OLED display powered on")
else:
if self._display:
self._display.poweron()
def clear(self):
"""
Clear the display.

218
src/main.py
View File

@@ -80,9 +80,18 @@ def main():
# Initialize an OLED display using configuration
display = OLEDDisplay(display_config=config.display_config)
display.clear()
display.set_header(f"Device: {config.device_name}")
display.set_status("Initializing...")
# Check if display is enabled in config
display_enabled = config.display_config.get("enabled", True)
display_always_on = config.display_config.get("always_on", False)
if display_enabled:
display.power_on() # Explicitly power on the display
display.clear()
display.set_header(f"Device: {config.device_name}")
display.set_status("Initializing...")
else:
print("Display disabled in config, not initializing")
# Initialize a DHT22 sensor using configuration
dht_sensor = DHT22Sensor(sensor_config=config.dht_config)
@@ -129,86 +138,88 @@ def main():
if mqtt_enabled:
# Initialize Wi-Fi connection
display.set_status("Connecting WiFi...")
connect_wifi(config.network_config, config.network_fallback_config)
wifi_connected, station = connect_wifi(config.network_config, config.network_fallback_config)
# Set up MQTT client if enabled
display.set_status("Setting up MQTT...")
print(
f"MQTT enabled: {mqtt_enabled}, broker: {config.mqtt_config.get('broker')}"
)
mqtt_client = setup_mqtt(config.mqtt_config)
if mqtt_client:
if not mqtt_client.connected:
try:
mqtt_client.connect()
display.set_status("MQTT connected")
print("MQTT client connected")
except Exception as e:
print(f"Error connecting MQTT client: {e}")
mqtt_client = None
if mqtt_client and mqtt_client.connected and load_config_from_mqtt:
display.set_status("Checking MQTT config...")
print("Checking for configuration updates before publishing...")
updated_config = check_config_update(
mqtt_client, config.mqtt_config, config.config
)
# If we got an updated configuration with a newer version, save it
if (
updated_config != config.config
and updated_config.get("version", 0) > config.current_version
):
display.set_status("Updating config...")
print(
f"Found newer configuration (version {updated_config.get('version')}), updating..."
)
config.save_config(updated_config)
publish_success = mqtt_client.publish(
get_data_topic(config.mqtt_config) + "/config_status",
"Configuration updated",
)
if not publish_success:
print("Failed to publish configuration update status")
# Note: We continue with the current config for this cycle
# The updated config will be used after the next reboot
else:
print(
f"No configuration updates found or no newer version available (local version: {config.current_version})"
)
if not wifi_connected:
display.set_status("WiFi connection failed")
print("Failed to connect to WiFi, skipping MQTT operations")
else:
# Set up MQTT client if enabled
display.set_status("Setting up MQTT...")
print(
"MQTT client not connected or not configured to load config from broker, skipping config check"
f"MQTT enabled: {mqtt_enabled}, broker: {config.mqtt_config.get('broker')}"
)
display.set_status("MQTT not loading config")
mqtt_client = setup_mqtt(config.mqtt_config)
if mqtt_client and mqtt_client.connected:
# Now publish sensor data using the same MQTT client
display.set_status("Publishing to MQTT...")
print(
f"Publishing sensor data to MQTT at {config.mqtt_config.get('broker')}:{config.mqtt_config.get('port')}"
)
publish_success = publish_sensor_data(
mqtt_client, config.mqtt_config, dht_sensor, temperature, humidity
)
if publish_success:
print("Sensor data published to MQTT")
display.set_status("Published to MQTT")
else:
print("Failed to publish sensor data to MQTT")
display.set_status("MQTT publish failed")
if mqtt_client:
if not mqtt_client.connected:
try:
mqtt_client.connect()
display.set_status("MQTT connected")
print("MQTT client connected")
except Exception as e:
print(f"Error connecting MQTT client: {e}")
mqtt_client = None
# Disconnect MQTT client after both operations
try:
mqtt_client.disconnect()
display.set_status("MQTT disconnected")
print("MQTT client disconnected")
except Exception as e:
print(f"Error disconnecting MQTT client: {e}")
if mqtt_client:
# Save the updated reconnection state to the configuration
config.save_config()
if mqtt_client and mqtt_client.connected:
# First publish sensor data (most important task)
display.set_status("Publishing to MQTT...")
print(
f"Publishing sensor data to MQTT at {config.mqtt_config.get('broker')}:{config.mqtt_config.get('port')}"
)
publish_success = publish_sensor_data(
mqtt_client, config.mqtt_config, dht_sensor, temperature, humidity
)
if publish_success:
print("Sensor data published to MQTT")
display.set_status("Published to MQTT")
else:
print("Failed to publish sensor data to MQTT")
display.set_status("MQTT publish failed")
# Only check for config updates if publishing was successful
if publish_success and load_config_from_mqtt:
display.set_status("Checking MQTT config...")
print("Checking for configuration updates...")
updated_config = check_config_update(
mqtt_client, config.mqtt_config, config.config
)
# If we got an updated configuration with a newer version, save it
if (
updated_config != config.config
and updated_config.get("version", 0) > config.current_version
):
display.set_status("Updating config...")
print(
f"Found newer configuration (version {updated_config.get('version')}), updating..."
)
config.save_config(updated_config)
mqtt_client.publish(
get_data_topic(config.mqtt_config) + "/config_status",
"Configuration updated",
)
# Note: We continue with the current config for this cycle
# The updated config will be used after the next reboot
else:
print(
f"No configuration updates found or no newer version available (local version: {config.current_version})"
)
# Disconnect MQTT client after operations
try:
mqtt_client.disconnect()
display.set_status("MQTT disconnected")
print("MQTT client disconnected")
except Exception as e:
print(f"Error disconnecting MQTT client: {e}")
if mqtt_client:
# Save the updated reconnection state to the configuration
config.save_config()
# Disconnect WiFi to save power
disconnect_wifi(station)
else:
print("MQTT is disabled, not publishing data")
@@ -221,23 +232,42 @@ def main():
display.set_status(f"Sleeping {time_until_next_read}s")
print("sleeping for", time_until_next_read, "seconds")
# Power off display to save battery before going to sleep
# Only if display is enabled and not set to always_on
if display_enabled and not display_always_on:
display.power_off()
print("Display powered off to save battery")
elif display_enabled and display_always_on:
print("Display kept on as per always_on setting")
deepsleep(time_until_next_read * 1000)
except KeyboardInterrupt:
# Clean up on exit
display.clear()
display.display_text("Shutting down...", 0, 0)
if display_enabled:
display.clear()
display.display_text("Shutting down...", 0, 0)
# Disconnect MQTT if connected
if mqtt_client:
if 'mqtt_client' in locals() and mqtt_client:
try:
mqtt_client.disconnect()
print("MQTT client disconnected")
except Exception as e:
print(f"Error disconnecting MQTT client: {e}")
time.sleep(1)
display.clear()
# Disconnect WiFi if connected
if 'station' in locals() and station:
disconnect_wifi(station)
# Power off display to save battery if it's enabled and not set to always_on
if display_enabled and not display_always_on:
display.power_off()
print("Display powered off during shutdown")
elif display_enabled and display_always_on:
print("Display kept on during shutdown as per always_on setting")
print("Program terminated by user")
@@ -246,10 +276,10 @@ def connect_wifi(network_config: dict, fallback_config: dict = None):
ssid = network_config.get("ssid")
password = network_config.get("password")
timeout = network_config.get("timeout", 10)
timeout = network_config.get("timeout", 5) # Reduced timeout for faster failure
if not ssid or not password:
print("SSID and password are required for WiFi connection")
return False
return False, None
print(f'Connecting to WIFI: "{ssid}"')
# Connect to your network
@@ -266,19 +296,31 @@ def connect_wifi(network_config: dict, fallback_config: dict = None):
if fallback_config:
print("Trying fallback network")
return connect_wifi(fallback_config)
return False
return False, None
time.sleep(1)
time.sleep(0.5) # Reduced sleep time for faster checking
print("Connection successful")
print(station.ifconfig())
return True
return True, station
def disconnect_wifi(station):
"""
Disconnect from WiFi to save power.
Args:
station: The WiFi station interface
"""
if station:
station.active(False)
print("WiFi disconnected to save power")
if __name__ == "__main__":
try:
main()
except Exception as e:
# this should never really be reachable. But if it is, just reset the system
print(f"An error occurred: {e}")
time.sleep(5) # give time to read the error message and respond
deepsleep(1) # dummy deepsleep to basically reset the system
deepsleep(60) # dummy deepsleep to basically reset the system