155 lines
6.0 KiB
Python
155 lines
6.0 KiB
Python
# file to create a database via python script
|
|
import sqlite3
|
|
from typing import Optional
|
|
import logging
|
|
|
|
|
|
class PlantDataBase:
|
|
"""
|
|
Class to create Makeathon database
|
|
"""
|
|
|
|
def __init__(self, database_name: str):
|
|
self.db_file = database_name # 'backend_database.db'
|
|
self.conn = None
|
|
try:
|
|
self.conn = sqlite3.connect(self.db_file, check_same_thread=False)
|
|
except sqlite3.Error as e:
|
|
logging.error("Database init error: " + str(e))
|
|
self.cur = self.conn.cursor()
|
|
|
|
def create_tables(self):
|
|
try:
|
|
table_config = "CREATE TABLE IF NOT EXISTS plants " \
|
|
"(PlantID INTEGER PRIMARY KEY," \
|
|
"PlantName TEXT)"
|
|
self.cur.execute(table_config)
|
|
|
|
table_config = "CREATE TABLE IF NOT EXISTS measurement_values " \
|
|
"(measurementID INTEGER PRIMARY KEY," \
|
|
"Timestamp DATETIME DEFAULT (datetime('now', 'localtime'))," \
|
|
"PlantID INTEGER, " \
|
|
"AirTemperature REAL," \
|
|
"AirHumidity REAL," \
|
|
"SoilMoisture REAL," \
|
|
"Brightness REAL," \
|
|
"FOREIGN KEY (PlantID)" \
|
|
"REFERENCES plants (PlantID) )"
|
|
self.cur.execute(table_config)
|
|
return True
|
|
except sqlite3.Warning as e:
|
|
logging.error("Could not create tables: " + str(e))
|
|
return False
|
|
|
|
def insert_plant(self, plantname: str, plant_id: int) -> bool:
|
|
try:
|
|
self.cur.execute("INSERT INTO plants (PlantName, PlantID) VALUES (?,?)", (plantname, plant_id))
|
|
self.conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Could not insert plant: " + str(e))
|
|
return False
|
|
|
|
def configure_plant(self, plant_id: int, plantname: str) -> bool:
|
|
try:
|
|
self.cur.execute("UPDATE plants SET PlantID = ?, PlantName = ? WHERE PlantID= ?",
|
|
(plant_id, plantname, plant_id))
|
|
self.conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Could not configure plant: " + str(e))
|
|
return False
|
|
|
|
def delete_plant(self, plant_id):
|
|
try:
|
|
self.cur.execute('DELETE FROM plants WHERE PlantID = ?', (plant_id,))
|
|
self.conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Could not delete plant: " + str(e))
|
|
return False
|
|
|
|
def insert_measurement_data(self, plant_id,
|
|
sensordata_temp,
|
|
sensordata_humidity,
|
|
sensordata_soil_moisture,
|
|
sensordata_brightness) -> bool:
|
|
try:
|
|
self.cur.execute(f"INSERT INTO measurement_values (PlantID, AirTemperature, AirHumidity,"
|
|
f"SoilMoisture, Brightness) VALUES "
|
|
f"({plant_id}, {sensordata_temp}, {sensordata_humidity}, {sensordata_soil_moisture}"
|
|
f", {sensordata_brightness})")
|
|
self.conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Could not insert measurement data: " + str(e))
|
|
return False
|
|
|
|
def get_latest_data(self, plant_name: Optional[str] = None, plant_id: Optional[int] = None):
|
|
"""
|
|
Gets the newest parameter of specific plant and returns all parameters in json format
|
|
:param plant_id:
|
|
:param plant_name:
|
|
:return:
|
|
"""
|
|
try:
|
|
if plant_name is not None and plant_id is None:
|
|
self.cur.execute("SELECT PlantID FROM plants where PlantName = ?", (plant_name,))
|
|
plant_id = self.cur.fetchone()[0]
|
|
elif (plant_id is not None and plant_name is not None) or (plant_id is None and plant_name is None):
|
|
raise TypeError("Can't pass plant_id and plant_name to the function. Just one allowed !")
|
|
|
|
self.cur.execute("SELECT * FROM measurement_values where PlantID = ? ORDER BY Timestamp DESC LIMIT 1",
|
|
(plant_id,))
|
|
data = self.cur.fetchone()
|
|
json_file = {
|
|
"MeasurementID": data[0],
|
|
"PlantID": data[2],
|
|
"Timestamp": data[1],
|
|
"AirTemperature": data[3],
|
|
"AirHumidity": data[4],
|
|
"SoilMoisture": data[5],
|
|
"Brightness": data[6],
|
|
"PlantName": plant_name
|
|
}
|
|
return json_file
|
|
|
|
except Exception as e:
|
|
logging.error("Could not get measurement values: " + str(e))
|
|
|
|
def delete_data(self, table_name):
|
|
try:
|
|
self.cur.execute(f'DELETE FROM {table_name}')
|
|
self.conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Could not delete data: " + str(e))
|
|
|
|
def plant_count(self) -> int:
|
|
"""
|
|
returns the number of plants registered in the database
|
|
:return:
|
|
"""
|
|
try:
|
|
self.cur.execute("SELECT COUNT(*) FROM plants")
|
|
return self.cur.fetchone()[0]
|
|
except Exception as e:
|
|
logging.error("Could not count plants: " + str(e))
|
|
|
|
def get_plant_names(self) -> list:
|
|
try:
|
|
self.cur.execute("SELECT PlantName FROM plants")
|
|
return self.cur.fetchall()
|
|
except Exception as e:
|
|
logging.error("Could not get plant names: " + str(e))
|
|
|
|
def get_plant_id(self, plant_name: str) -> int:
|
|
try:
|
|
self.cur.execute("SELECT PlantID FROM plants WHERE PlantName=?", (plant_name,))
|
|
return self.cur.fetchone()[0]
|
|
except Exception as e:
|
|
logging.error("Could not get plant id: " + str(e))
|
|
|
|
def __del__(self):
|
|
self.conn.close()
|