185 lines
6.7 KiB
Python
185 lines
6.7 KiB
Python
import pandas as pd
|
|
|
|
import folium
|
|
import plotly.graph_objects as go
|
|
import io
|
|
|
|
# Chargement du fichier JSON
|
|
with open('/home/poule/encrypted/stockage-syncable/www/development/html/mapping-geojson-osm/etalab_data/irve_bornes_recharge/latest.csv', 'r') as file:
|
|
data = pd.read_csv(file)
|
|
|
|
|
|
def format_puissance_nominale(value):
|
|
"""Formate la valeur de la puissance nominale"""
|
|
if value >= 1000:
|
|
return "{:,.1f} kW".format(value / 1000)
|
|
else:
|
|
return "{:,.0f} W".format(value)
|
|
|
|
|
|
|
|
|
|
# Création du DataFrame pandas
|
|
dataframe_irve = pd.DataFrame(data)
|
|
print("Nombre total de lignes avant suppression de duplicatas: ", len(dataframe_irve))
|
|
|
|
|
|
# enlever les duplicatas selon l'id du point de charge
|
|
dataframe_irve = dataframe_irve.drop_duplicates(subset=["id_pdc_itinerance"])
|
|
# enlever les duplicatas de placement
|
|
dataframe_irve = dataframe_irve.drop_duplicates(subset=["coordonneesXY"])
|
|
# enlever les lignes où on ne connait pas l'opérateur
|
|
dataframe_irve = dataframe_irve[(dataframe_irve["nom_operateur"].notnull()) & (dataframe_irve["nom_amenageur"].notnull())]
|
|
dataframe_irve = dataframe_irve[(dataframe_irve["coordonneesXY"].notnull())]
|
|
|
|
# Regroupement par opérateur
|
|
groups = dataframe_irve.groupby("nom_operateur")
|
|
|
|
dataframe_irve_sorted = dataframe_irve.sort_values(by="nom_operateur")
|
|
|
|
# Affichage du résultat
|
|
# print(dataframe_irve_sorted)
|
|
|
|
|
|
# Filtres pour identifier les lignes incorrectes
|
|
filtre_non_numerique = dataframe_irve["puissance_nominale"] % 1 > 0
|
|
filtre_incorrect = filtre_non_numerique
|
|
|
|
# Lignes incorrectes
|
|
erreurs_puissance_nominale = dataframe_irve[filtre_incorrect][[ "puissance_nominale", "id_station_itinerance"]]
|
|
|
|
groups_erreur = erreurs_puissance_nominale
|
|
print('groups en erreur: \n',)
|
|
groups_erreur.head()
|
|
|
|
# Minimum de puissance_nominale
|
|
puissance_nominale_min = dataframe_irve["puissance_nominale"].min()
|
|
|
|
# Maximum de puissance_nominale
|
|
puissance_nominale_max = dataframe_irve["puissance_nominale"].max()
|
|
|
|
print("\nMinimum de puissance_nominale : {}".format(puissance_nominale_min))
|
|
print("Maximum de puissance_nominale : {}\n".format(puissance_nominale_max))
|
|
|
|
# Impression du nombre de stations restantes
|
|
print("Nombre total de lignes : ", len(dataframe_irve))
|
|
|
|
|
|
# Application de la fonction de formatage à la colonne "puissance_nominale"
|
|
dataframe_irve["puissance_nominale_formatted"] = dataframe_irve["puissance_nominale"].apply(format_puissance_nominale)
|
|
|
|
# Remplacement de la colonne originale par celle nouvellement formatée
|
|
dataframe_irve["puissance_nominale"] = dataframe_irve["puissance_nominale_formatted"]
|
|
del dataframe_irve["puissance_nominale_formatted"]
|
|
|
|
|
|
|
|
# Compte le nombre d'anomalies pour chaque opérateur
|
|
def count_errors_for_each_operator():
|
|
# Initialisation du dictionnaire vide servant au stockage temporaire
|
|
temp_dict = dict()
|
|
|
|
for _, row in dataframe_irve.iterrows():
|
|
operator = str(row["nom_operateur"])
|
|
|
|
# Test uniquement les entrées textuelles présentes dans la colonne "puissance_nominale"
|
|
power_entry = str(row["puissance_nominale"]).replace('W', '').replace(' ', '')
|
|
if 'k' in power_entry or power_entry[-2:] == ". " or int(power_entry) > 6000 or int(power_entry) < 0:
|
|
if operator not in temp_dict:
|
|
temp_dict[operator] = {"error_count": 1}
|
|
else:
|
|
current_error_count = temp_dict[operator]["error_count"] + 1
|
|
temp_dict[operator] = {"error_count": current_error_count}
|
|
|
|
# Transformation du dictionnaire local en liste de tuples
|
|
return sorted([(key, val["error_count"]) for key, val in temp_dict.items()], key=lambda x: -x[1])
|
|
|
|
# Passage du DataFrame à la fonction
|
|
countland = count_errors_for_each_operator()
|
|
|
|
print("\nListe des opérateurs classés par nombre d'anomalies :'" , countland)
|
|
|
|
# Affichage des résultats
|
|
# print("\nListe des opérateurs classés par nombre d'anomalies : ")
|
|
for i, item in enumerate(countland):
|
|
print("{} - Opérateur : {}, Anomalies : {}".format(i+1, item[0], item[1]))
|
|
|
|
# ============== graphique ============================
|
|
import plotly.express as px
|
|
import plotly.io as pio
|
|
pio.renderers.default = "svg"
|
|
# Spécification du nom et du path du fichier de sortie
|
|
output_filename = "puissances_repartition.svg"
|
|
|
|
# Génération du graphe
|
|
# fig = px.histogram(
|
|
# dataframe_irve,
|
|
# x="puissance_nominale",
|
|
# title="Répartition des Puissances",
|
|
# labels={'puissance_nominale': 'Puissance'},
|
|
# color_discrete_sequence=px.colors.sequential.Plasma_r,
|
|
# opacity=0.7,
|
|
# hover_name="nom_operateur",
|
|
# category_orders={'puissance_nominale': ['<1 kW', '1 kW', '2 kW', '3 kW', '5 kW', '6 kW', '8 kW', '11 kW', '12 kW', '15 kW', '22 kW', '30 kW', '43 kW', '50 kW', '150 kW']},
|
|
# )
|
|
|
|
# Visualisation du graphe
|
|
# fig.show()
|
|
|
|
|
|
# Sauvegarde du graphe Plotly sous forme de fichier SVG
|
|
# fig.write_image(output_filename, engine='kaleido', scale=1.0)
|
|
#
|
|
# # Display the generated filename
|
|
# print("\nGraph saved to:", output_filename)
|
|
|
|
dataframe_irve.to_csv("output/cleaned_irve_from_cipherbliss.csv", index=False)
|
|
|
|
|
|
# ============ carte svg
|
|
|
|
|
|
# Colonne "coordonneesXY" est un string, transformation en tuple
|
|
dataframe_irve[['latitude','longitude']] = dataframe_irve['coordonneesXY'].str.findall(r'(-?[\d]+(\.[0-9]*)?)').apply(pd.Series)
|
|
|
|
dataframe_irve['latitude'].head()
|
|
|
|
# Centrer la map sur la france
|
|
france_center = [47.5, 2.5]
|
|
# map_folium = folium.Map(location=france_center, zoom_start=5)
|
|
#
|
|
# # Itération sur chaque point géographique pour l'ajouter à la map Folium
|
|
# for index, row in dataframe_irve.iterrows():
|
|
# location = [row['latitude'], row['longitude']]
|
|
# icon = folium.Icon(color='blue', icon='info-sign')
|
|
# tooltip = "<strong>{}</strong>{}".format(row['id_station_itinerance'], row['adresse_station'])
|
|
# popup = '<div style="width:10rem;">{}</div>'.format(row['nom_station'])
|
|
# marker = folium.Marker(location, popup=popup, tooltip=tooltip, icon=icon)
|
|
# marker.add_to(map_folium)
|
|
#
|
|
# # Objet graphique Plotly contenant la carte Folium encodée
|
|
# plotly_chart = go.Figure(go.Scattergeo(
|
|
# lon=[row['longitude'] for index, row in dataframe_irve.iterrows()],
|
|
# lat=[row['latitude'] for index, row in dataframe_irve.iterrows()]
|
|
# ))
|
|
#
|
|
# # Configuration de l'apparence visuelle
|
|
# plotly_chart.update_geos(fitbounds="locations", visible=True)
|
|
# plotly_chart.update_layout(
|
|
# geo_scope="world",
|
|
# margin={"r":0,"t":0,"l":0,"b":0},
|
|
# paper_bgcolor="#FFFFFF",
|
|
# plot_bgcolor="#FFFFFF",
|
|
# showlegend=False
|
|
# )
|
|
|
|
|
|
|
|
# Encapsulation de la carte Plotly dans un objet BytesIO
|
|
# buffer = io.BytesIO()
|
|
# plotly_chart.write_html(buffer, auto_open=False)
|
|
# buffer.seek(0)
|
|
|
|
# Affichage de la carte Plotly
|
|
# plotly.io.write_image(plotly_chart, "carte_stations.svg", format="svg", engine="kaleido")
|