# Import other functions and classes import heating import errors import sql import weather import time import sqlite3 import dates from flask import Flask, render_template, redirect, url_for from threading import Thread import os app = Flask(__name__) # TODO: # schedule # export as pdf (dictionary/merge sort) # Use of constants # Paramaterised location of database file DBFILE = "data.db" TIME_INCREMENT = 5 # number of increments per day STEPS = (60 * 60 * 24) / TIME_INCREMENT # instantiate classes db = sql.db(DBFILE) date = date.dates() weather = weather.weather_met_no() # Check if running on correct device : defensive coding against running # on the wrong device, the instantiate heating class try: with open("/sys/firmware/devicetree/base/model", "r") as file: device = file.read() except: errors.device_error() if device.startswith("Raspberry Pi 4"): boiler = heating.rpi_heating(db) else: errors.device_error() # Import API Script after Flask app and boiler classexists import api def main_page(): # paramaterised location of template in 'templates' folder return render_template( "main.html", # Rounding value for user interface so number is not too long actual_temp=round(boiler.temperature, 2), target_temp=boiler.target, forecast_temp=weather.temperature(), ) @app.route("/") @app.route("/index") @app.route("/index.html") def index(): return main_page() @app.route("/up") def form(): boiler.up() return redirect(url_for("index")) @app.route("/down") def activity(): boiler.down() return redirect(url_for("index")) @app.route("/export") def export_choice(): return render_template("export_choice.html") @app.route("/export/") def export(opt): # Defensive handling of input if opt in ["temperature", "weather", "history", "schedule"]: con = sqlite3.connect(DBFILE) cur = con.cursor() data = cur.execute("select * from {}".format(opt)) o = "" for i in data: o += str(i) + "\n" con.close() # writing data to a file with open("static/export.txt", "w") as file: file.write(o) return redirect("/static/export.txt") elif opt == "pdf": con = sqlite3.connect(DBFILE) cur = con.cursor() data = cur.execute("select date, on from history") on = 0 total = 0 days = dictionary.dictionary() for i in data: if data[1] == 1: days.inc(data[0], 1 / STEPS * 100) con.close() days.sort_by_value() # writing data to a file with open("static/table.ms", "w") as file: file.write(days.roff_tbl()) os.system("groff -Tpdf -P-pa4 -tk static/table.ms > static/table.pdf") return redirect("/static/table.pdf") else: return render_template( "error.html", error="Invalid export type" ) # Methods that will run continously: the update script for the heating # and the flask webapp def updater(): while True: boiler.update() db.exec( "insert into weather values (?,?,?,?)", (date.day(), date.time(), weather.temperature()), ) time.sleep(TIME_INCREMENT) def webapp(): app.run(host="0.0.0.0") if __name__ == "__main__": # Start the two main methods as threads # Use of multithreaded processing t1 = Thread(target=webapp) t2 = Thread(target=updater) t1.start() t2.start()