285 lines
9.4 KiB
Python
Executable file
285 lines
9.4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
"""Connecteur MySQL python, qui implémente toutes les requêtes nécessaires"""
|
|
|
|
from configparser import ConfigParser
|
|
from contextlib import ContextDecorator
|
|
from datetime import datetime
|
|
import sys
|
|
|
|
import mysql.connector # type: ignore
|
|
from mysql.connector.cursor import MySQLCursor # type: ignore
|
|
import matplotlib.pyplot as plt # type: ignore
|
|
|
|
|
|
VERBOSE = False
|
|
|
|
|
|
class my_connect(ContextDecorator):
|
|
"""Context manager for database connection"""
|
|
|
|
def __init__(self) -> None:
|
|
self.data_base_connection = ''
|
|
|
|
def __enter__(self) -> MySQLCursor:
|
|
sql_config = ConfigParser()
|
|
sql_config.read("sql_config.ini")
|
|
self.data_base_connection = mysql.connector.connect(
|
|
user=sql_config["global"]["user"],
|
|
password=sql_config["global"]["password"],
|
|
host=sql_config["global"]["host"],
|
|
database=sql_config["global"]["database"],
|
|
)
|
|
cursor = self.data_base_connection.cursor()
|
|
return cursor
|
|
|
|
def __exit__(self, *exc):
|
|
self.data_base_connection.close()
|
|
return False
|
|
|
|
|
|
def main() -> None:
|
|
"""Make database connection and queries"""
|
|
|
|
if "-v" in sys.argv:
|
|
global VERBOSE
|
|
VERBOSE = True
|
|
|
|
sql_config = ConfigParser()
|
|
sql_config.read("sql_config.ini")
|
|
|
|
with my_connect() as cursor:
|
|
cursor.execute("USE bdd_project")
|
|
|
|
choice, choice_data = ask_what_to_do()
|
|
|
|
if choice.lower() == "1":
|
|
print("Calcul en cours...")
|
|
for avg in get_average_by_city(cursor, city=choice_data):
|
|
print(avg)
|
|
elif choice.lower() == "2":
|
|
print("Calcul en cours...")
|
|
for avg in get_average_by_zip_code(cursor, zip_code=choice_data):
|
|
print(avg)
|
|
elif choice.lower() == "3":
|
|
print("Calcul en cours...")
|
|
for avg in get_average_surface_by_city(cursor, city=choice_data):
|
|
print(avg)
|
|
elif choice.lower() == "4":
|
|
print("Calcul en cours...")
|
|
for avg in get_average_surface_by_zip_code(
|
|
cursor, zip_code=choice_data
|
|
):
|
|
print(avg)
|
|
elif choice.lower() == "5":
|
|
print("Calcul en cours...")
|
|
for avg in get_nb_nature_mutation_by_departement(
|
|
cursor, departement=choice_data
|
|
):
|
|
print(avg)
|
|
elif choice.lower() == "6":
|
|
print("Calcul en cours...")
|
|
_, avg, dates = zip(*get_average_by_city_graph(cursor, city=choice_data))
|
|
generate_average_graph(choice_data, dates, avg)
|
|
|
|
|
|
def ask_what_to_do() -> tuple[str, str]:
|
|
"""Prompt the user with a menu displaying the possible actions"""
|
|
|
|
while True:
|
|
print(
|
|
"Bienvenue, veuillez choisir une action à effectuer :\n"
|
|
" 1 - Calcul de la moyenne des valeurs foncières pour une commune\n"
|
|
" 2 - Calcul de la moyenne des valeurs foncières pour un code postal\n"
|
|
" 3 - Calcul de la moyenne du m2 pour une commune\n"
|
|
" 4 - Calcul de la moyenne du m2 pour un code postal\n"
|
|
" 5 - Calcul du nombre de Nature de mutation pour un département\n"
|
|
" 6 - Calcul de la moyenne des valeurs foncières pour une commune au cours du temps\n"
|
|
)
|
|
choice = input()
|
|
if choice == "1":
|
|
city = input("Commune : ")
|
|
return choice, city
|
|
if choice == "2":
|
|
zip_code = input("Code postal : ")
|
|
return choice, zip_code
|
|
if choice == "3":
|
|
city = input("Commune : ")
|
|
return choice, city
|
|
if choice == "4":
|
|
zip_code = input("Code postal : ")
|
|
return choice, zip_code
|
|
if choice == "5":
|
|
departement = input("N° Département : ")
|
|
return choice, departement
|
|
if choice == "6":
|
|
city = input("Commune : ")
|
|
return choice, city
|
|
print("Mauvais choix !")
|
|
|
|
|
|
def get_average_by_zip_code(
|
|
cursor: MySQLCursor, zip_code: str
|
|
) -> list[tuple[str, ...]]:
|
|
"""Calculate the average of value for a zip code"""
|
|
|
|
query = (
|
|
"SELECT Code_postal.Code_postal, ROUND(AVG(Mutation.Valeur_fonciere)) FROM "
|
|
"((((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id) "
|
|
"RIGHT JOIN Code_postal ON Commune.Code_postal_id = Code_postal.id "
|
|
f"WHERE Code_postal.Code_postal='{zip_code}' "
|
|
"AND Mutation.Valeur_fonciere IS NOT NULL "
|
|
"GROUP BY Code_postal.Code_postal;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def get_average_by_city(cursor: MySQLCursor, city: str) -> list[tuple[str, ...]]:
|
|
"""Calculate the average of value for a city"""
|
|
|
|
query = (
|
|
"SELECT Commune.Commune, ROUND(AVG(Mutation.Valeur_fonciere)) FROM "
|
|
"(((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id "
|
|
f"WHERE Commune.Commune='{city}' AND Mutation.Valeur_fonciere IS NOT NULL "
|
|
"GROUP BY Commune.Commune;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def get_average_surface_by_zip_code(
|
|
cursor: MySQLCursor, zip_code: str
|
|
) -> list[tuple[str, ...]]:
|
|
"""Calculate the average surface for a zip code"""
|
|
|
|
query = (
|
|
"SELECT Code_postal.Code_postal, "
|
|
"SUM(Mutation.Valeur_fonciere) / SUM(Bien.Surface_terrain) "
|
|
"FROM ((((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id) "
|
|
"RIGHT JOIN Code_postal ON Commune.Code_postal_id = Code_postal.id "
|
|
f"WHERE Code_postal.Code_postal='{zip_code}' "
|
|
"AND Mutation.Valeur_fonciere IS NOT NULL "
|
|
"GROUP BY Code_postal.Code_postal;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def get_average_surface_by_city(
|
|
cursor: MySQLCursor, city: str
|
|
) -> list[tuple[str, ...]]:
|
|
"""Calculate the average surface for a city"""
|
|
|
|
query = (
|
|
"SELECT Commune.Commune, "
|
|
"SUM(Mutation.Valeur_fonciere) / SUM(Bien.Surface_terrain) "
|
|
"FROM (((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id "
|
|
f"WHERE Commune.Commune='{city}' "
|
|
"AND Mutation.Valeur_fonciere IS NOT NULL "
|
|
"GROUP BY Commune.Commune;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def get_nb_nature_mutation_by_departement(
|
|
cursor: MySQLCursor, departement: str
|
|
) -> list[tuple[str, ...]]:
|
|
"""Calculate the number of nature_mutation"""
|
|
|
|
query = (
|
|
"SELECT Nature_mutation.Nature_mutation, "
|
|
"COUNT(Nature_mutation.Nature_mutation) "
|
|
"FROM (((((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id) "
|
|
"RIGHT JOIN Departement ON Commune.Departement_id = Departement.id) "
|
|
"RIGHT JOIN Nature_mutation "
|
|
"ON Mutation.Nature_mutation_id = Nature_mutation.id "
|
|
f"WHERE Departement.Code_departement = {departement} "
|
|
"GROUP BY Nature_mutation.Nature_mutation "
|
|
"ORDER BY COUNT(Nature_mutation.Nature_mutation) DESC;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def get_average_by_city_graph(cursor: MySQLCursor, city: str) -> list[tuple[str, ...]]:
|
|
"""Calculate the average of value for a city"""
|
|
|
|
query = (
|
|
"SELECT Commune.Commune, AVG(Mutation.Valeur_fonciere), Mutation.Date_mutation "
|
|
"FROM (((Mutation "
|
|
"RIGHT JOIN Bien ON Mutation.Bien_id = Bien.id) "
|
|
"RIGHT JOIN Adresse ON Bien.Adresse_id = Adresse.id) "
|
|
"RIGHT JOIN Voie ON Adresse.Voie_id = Voie.id) "
|
|
"RIGHT JOIN Commune ON Voie.Commune_id = Commune.id "
|
|
f"WHERE Commune.Commune='{city}' AND Mutation.Valeur_fonciere IS NOT NULL "
|
|
"GROUP BY Mutation.Date_mutation "
|
|
"ORDER BY Mutation.Date_mutation;"
|
|
)
|
|
cursor.execute(query)
|
|
if VERBOSE:
|
|
print(cursor.statement)
|
|
|
|
result: list[tuple[str, ...]] = cursor.fetchall()
|
|
|
|
return result
|
|
|
|
|
|
def generate_average_graph(title: str, dates: tuple[datetime], avg: tuple[float]) -> None:
|
|
"""Plot average versus datetime"""
|
|
|
|
plt.plot(dates, avg)
|
|
plt.title(title)
|
|
plt.xlabel("Dates")
|
|
plt.ylabel("Moyennes")
|
|
plt.show()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|