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 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]
if days == 0: days = 1
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': 'meterReadingsBackup2020022',
'raise_on_warnings': True
}
zeropi = {
'user': 'andrew',
'password': 'password123',
'host': 'zeropi',
'port': '3306',
'database': 'meterReadings',
'raise_on_warnings': True
}
weekdays = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday") # Weekdays represented as a tuple dataset (round brackets)
today = datetime.date.today() # using the date.today call in datetime to get today's date in yyyy-mm-dd format
dayOfWeek = today.weekday() # using the date.weekday call in datetime to get an integer representation of the day of the week (where 0=Monday, 6=Sunday)
print("")
print("")
if dayOfWeek == 6: # or !=
print("Today is: " + str(today) + " which is a Sunday") # n.b. had to use the str() function to turn the date into a string so that it could be concatenated into a single string to be printed
readingDate = today # set readingDate to today's date
go = input("Do you want to enter readings for today? (y/n)") # Record the readings against today's date if user so wishes
else:
lastSunday = today - datetime.timedelta(days=today.weekday()) + datetime.timedelta(days=6, weeks=-1) # find last Sunday by taking today's number away from today (i.e. go back to Monday) and then adding a delta of Sunday (6) of last week (-1)
print("Today is " + str(today) + ", which is a " + weekdays[dayOfWeek] + ". Last Sunday was " + str(lastSunday) + ".")
readingDate = lastSunday
go = input("Do you want to enter readings for last Sunday? (y/n)")
if go == 'n':
print("Then you'll have to enter the readings yourself!") # if user's input was 'n' then program stops here! Need to consider error conditions, or maybe other choices
exit()
else:
print("The date recorded for these readings will be " + str(readingDate)) # if user's input was 'y' then confirm date (either today or last Sunday depending on choices above) and proceed to input readings
print("")
print("")
dB = DatabaseManager(localhost)
sqlQuery = "SELECT * FROM Readings ORDER BY readingID DESC LIMIT 1"
result = dB.query(sqlQuery)
gas = input("Gas reading: ")
electric = input("Electricity reading: ")
index1 = result[0][0] + 1
index2 = index1 + 1
records = [
(index1, readingDate, 0, electric),
(index2, readingDate, 1, gas)
]
sqlQuery = "INSERT INTO Readings (readingID, readingDate, readingType, readingValue) VALUES (%s, %s, %s, %s)"
dB.update(sqlQuery, records)
elecReadings = []
gasReadings = []
sqlQuery = "SELECT * FROM Readings ORDER BY readingID DESC LIMIT 12"
result = dB.query(sqlQuery)
elecReadings = formatReading(result, 0)
gasReadings = formatReading(result, 1)
printReadings(gasReadings, "New gas")
printReadings(elecReadings, "New electricity")
print(result)
print(index1, index2)
dB.close()
# webbrowser.open("https://octopus.energy/dashboard/accounts/A-A4570FF5/properties/1327603/meter-readings/meter/")
if __name__ == "__main__":
main()