Source code for meterReadingBackup_02b

import mysql.connector # for connectivity to MySQL database
import os              # for the 'cls' command to clear the screen
import datetime        # for date.today, timedelta and date.weekday functions
import webbrowser

[docs]class DatabaseManager: """ The Card class represents a single database accessed across a specific mysql connector. It is initialised by passing a set of credentials for opening a connection to a MySQL server; these credentials are passed as a dictionary which defines the values for the 'user', 'password', 'host', 'port' and 'database' variables to be used to access an existing database on a MySQL server and the value of the 'raise_on_warnings' flag to be used over the connection to that database. The class supports query and update functions for a specific (Readings) table in the database, and 'serverInfo' and 'close' functions for the connection. Work ongoing! """ def __init__(self, credentials): self.conn = mysql.connector.connect(**credentials) self.cur = self.conn.cursor()
[docs] def newTable(self, tableName, tableColumns): genericCode = "CREATE TABLE {} ({})" sqlCode = genericCode.format(tableName, tableColumns) self.cur.execute(sqlCode)
[docs] def query(self, sqlCode): self.cur.execute(sqlCode) return self.cur.fetchall()
[docs] def update(self, sqlCode, data): self.cur.executemany(sqlCode, data) self.conn.commit()
[docs] def serverInfo(self): return self.conn.get_server_info()
[docs] def close(self): self.conn.close()
[docs]def cls(): # defining a function to clear the screen os.system('cls' if os.name=='nt' else 'clear')
[docs]def dateSuffix(date): if date == 1 or date == 21 or date == 31: string= "st " elif date == 2 or date == 22: string = "nd " elif date == 3 or date == 23: string = "rd " else: string = "th " return string
[docs]def formatReading(result, fuelFlag): listOfReadings = [] record = {} for x in result: if x[2] == fuelFlag: date = x[1] yearday = int(date.strftime("%j")) year = int(date.strftime("%Y")) month = (date.strftime("%b")) week = int(date.strftime("%W")) day = int(date.strftime("%d")) reading = x[3] record = (day, week, month, year, yearday, reading) listOfReadings.append(record) return listOfReadings
[docs]def printReadings(readings, fuelString): if fuelString == 'Electricity': extra = "========" elif fuelString == 'New electricity': extra = "============" elif fuelString == 'New gas': extra = "====" else: extra = "" print(" ") print(" ") print(fuelString + ' readings:') print('=============' + extra) print(" ") for x in readings: count = readings.index(x) + 1 try: previous_entry = readings[count] string = dateSuffix(x[0]) if x[3] > previous_entry[3]: if (previous_entry[3]%4) == 0: # looking for leap years - fails with 2100, 2200, 2300, 2500, etc days = int(366 + x[4] - previous_entry[4]) else: days = int(365 + x[4] - previous_entry[4]) else: days = x[4] - previous_entry[4] units_used_per_week = int(7 * (x[5] - previous_entry[5])/days) day = str(x[3]) + " Week " + str(x[1]) + ":\t" + str(x[0]) + string + str(x[2] + "\t") print(day, "-\t", x[5], "\t", "Units used per week:\t", units_used_per_week) except IndexError: break
[docs]def main(): cls() localhost = { 'user': 'andrew', 'password': 'password123', 'host': 'localhost', 'port': '3306', 'database': 'testDatabase2020006', 'raise_on_warnings': True } zeropi1 = { 'user': 'root', 'password': 'password123', 'host': 'zeropi', 'port': '3306', 'database': 'Meter_Readings', 'raise_on_warnings': True } zeropi2 = { 'user': 'andrew', 'password': 'password123', 'host': 'zeropi', 'port': '3306', 'database': 'meterReadings', 'raise_on_warnings': True } # to read the contents of the existing Readings database on PiZero zpR1 = DatabaseManager(zeropi1) sqlQuery = "SELECT * FROM Readings WHERE idReadings > 369" result = zpR1.query(sqlQuery) print(result)
if __name__ == "__main__": main()