101 lines
2.7 KiB
Python
101 lines
2.7 KiB
Python
# main.py
|
|
|
|
import logging
|
|
import time
|
|
import traceback
|
|
from datetime import datetime, timedelta
|
|
|
|
import schedule
|
|
from icmplib import ping as _ping
|
|
|
|
from tools.weather_logger import set_up_logger
|
|
|
|
set_up_logger()
|
|
|
|
from constants import MONTH_NAMES, WEEK_DAYS
|
|
from display import Display
|
|
from environment import Weather
|
|
from tools.mqtt import MQTT
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.info("Setting up components...")
|
|
|
|
weather = Weather()
|
|
display = Display()
|
|
mqtt = MQTT()
|
|
|
|
|
|
def is_network_up(retries: int = 2, timeout: int = 2) -> bool:
|
|
"""Checks if the internet is accessible by pinging Cloudflare."""
|
|
try:
|
|
host = _ping("1.1.1.1", count=retries, timeout=timeout, privileged=False)
|
|
return host.is_alive
|
|
except Exception as e:
|
|
logger.warning(f"Ping failed: {e}")
|
|
return False
|
|
|
|
|
|
def date_string(date: datetime) -> tuple[str, str]:
|
|
"""Formats the date and time strings for the display."""
|
|
date_line = (
|
|
f"{WEEK_DAYS[date.weekday()]}, {date.day} {MONTH_NAMES[date.month]} {date.year}"
|
|
)
|
|
time_line = f"{date.hour:02d}:{date.minute:02d}"
|
|
return date_line, time_line
|
|
|
|
|
|
def update(first: bool = False) -> None:
|
|
start_timestamp = time.time()
|
|
logger.debug("Running scheduled update")
|
|
date = datetime.now() if first else datetime.now() + timedelta(seconds=20)
|
|
logger.debug(f"Target display time: {date}")
|
|
|
|
weather_data = weather.get(date)
|
|
mqtt.send(weather_data)
|
|
|
|
str_date = date_string(date)
|
|
|
|
logger.debug("Constructing the Image...")
|
|
image = display.image.make_image(str_date, weather_data, is_network_up(timeout=1))
|
|
|
|
logger.debug("Refreshing the E-ink display...")
|
|
display.refresh(image, date, first)
|
|
|
|
logger.debug(
|
|
f"Update completed in {round(time.time() - start_timestamp, 2)} seconds"
|
|
)
|
|
|
|
|
|
def main():
|
|
logger.info("Checking network connection...")
|
|
is_network_up()
|
|
|
|
try:
|
|
logger.info("Starting weather station")
|
|
update(first=True)
|
|
schedule.every().minute.at(":40").do(update)
|
|
logger.info("Schedule started")
|
|
|
|
while True:
|
|
schedule.run_pending()
|
|
time.sleep(1)
|
|
|
|
except KeyboardInterrupt:
|
|
logger.info("Manual exit requested (Ctrl+C). Shutting down...")
|
|
except Exception as e:
|
|
logger.error(f"Schedule failed - ERROR: {e}")
|
|
logger.error(traceback.format_exc())
|
|
finally:
|
|
try:
|
|
Display.exit()
|
|
logger.info("E-ink display safely shut down.")
|
|
except Exception as display_error:
|
|
logger.error(
|
|
f"Failed to shutdown the E-ink Display cleanly: {display_error}"
|
|
)
|
|
exit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|