py_atmo_data_wrapper/examples/example_data_models.py
2025-07-14 17:56:57 +02:00

367 lines
No EOL
13 KiB
Python

#!/usr/bin/env python3
"""
Exemples d'utilisation des nouveaux modèles de données typés
"""
from atmo_data_wrapper import AtmoDataClient
from atmo_data_wrapper import Coordinates
def example_indices_atmo():
"""Exemple d'utilisation des indices ATMO avec objets typés"""
print("=== Exemple Indices ATMO avec objets typés ===\n")
client = AtmoDataClient()
# Connexion automatique avec credentials.json
# Décommentez les lignes suivantes pour une vraie connexion API :
# try:
# client.auto_login()
# indices = client.get_indices_atmo(aasqa="11") # Données réelles
# except Exception as e:
# print(f"Erreur de connexion: {e}")
# print("Utilisation de données de test...")
# # Fallback sur données de test...
# Récupération des indices ATMO - maintenant retourne AtmoDataCollection
try:
# indices = client.get_indices_atmo(aasqa="11") # Île-de-France
# Pour la démonstration, créons des données de test
test_data = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.3522, 48.8566]
},
"properties": {
"aasqa": "11",
"code_qual": 3,
"lib_qual": "Moyen",
"coul_qual": "#FFFF00",
"lib_zone": "Paris Centre",
"date_dif": "2024-07-07",
"code_no2": 2,
"code_so2": 1,
"code_o3": 3,
"code_pm10": 2,
"code_pm25": 3
}
},
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.4522, 48.9566]
},
"properties": {
"aasqa": "11",
"code_qual": 5,
"lib_qual": "Mauvais",
"coul_qual": "#FF0000",
"lib_zone": "Banlieue Nord",
"date_dif": "2024-07-07",
"code_no2": 4,
"code_so2": 2,
"code_o3": 5,
"code_pm10": 4,
"code_pm25": 5
}
}
]
}
# Créer une collection typée
from atmo_data_wrapper import AtmoDataCollection
indices = AtmoDataCollection(test_data, 'indices')
print(f"Collection: {len(indices)} indices récupérés")
print(f"Résumé: {indices.to_summary()}\n")
# Parcourir les indices avec les objets typés
for i, indice in enumerate(indices):
print(f"Indice {i+1}:")
print(f" Zone: {indice.lib_zone}")
print(f" AASQA: {indice.get_aasqa_name()}")
print(f" Qualité: {indice.get_qualificatif()} (code: {indice.code_qual})")
# Utiliser les méthodes helper
hex_color, rgb_color = indice.get_color()
print(f" Couleur: {hex_color} (RGB: {rgb_color})")
print(f" Bonne qualité: {'Oui' if indice.is_good_quality() else 'Non'}")
print(f" Mauvaise qualité: {'Oui' if indice.is_poor_quality() else 'Non'}")
# Polluant le plus problématique
worst_pol, worst_code = indice.get_worst_pollutant()
print(f" Pire polluant: {worst_pol} (code: {worst_code})")
# Coordonnées
if indice.has_coordinates():
print(f" Coordonnées: {indice.coordinates}")
print()
# Statistiques sur la collection
stats = indices.get_statistics()
print("Statistiques de la collection:")
for key, value in stats.items():
print(f" {key}: {value}")
print()
# Filtrage par qualité
print("=== Filtrage et analyse ===")
mauvaise_qualite = [indice for indice in indices if indice.is_poor_quality()]
print(f"Zones avec mauvaise qualité: {len(mauvaise_qualite)}")
for indice in mauvaise_qualite:
print(f" - {indice.lib_zone}: {indice.get_qualificatif()}")
except Exception as e:
print(f"Erreur: {e}")
def example_episodes_pollution():
"""Exemple d'utilisation des épisodes de pollution avec objets typés"""
print("\n=== Exemple Épisodes de Pollution avec objets typés ===\n")
# Données de test pour épisodes
test_data = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "MultiPolygon",
"coordinates": [[[2.0, 48.0], [3.0, 48.0], [3.0, 49.0], [2.0, 49.0], [2.0, 48.0]]]
},
"properties": {
"aasqa": "11",
"code_pol": "5",
"lib_pol": "PM10",
"lib_zone": "Île-de-France",
"date_dif": "2024-07-07",
"etat": "INFORMATION ET RECOMMANDATIONS"
}
},
{
"type": "Feature",
"geometry": {
"type": "MultiPolygon",
"coordinates": [[[1.0, 47.0], [2.0, 47.0], [2.0, 48.0], [1.0, 48.0], [1.0, 47.0]]]
},
"properties": {
"aasqa": "24",
"code_pol": "3",
"lib_pol": "Ozone",
"lib_zone": "Centre-Val de Loire",
"date_dif": "2024-07-07",
"etat": "ALERTE"
}
}
]
}
from atmo_data_wrapper import AtmoDataCollection
episodes = AtmoDataCollection(test_data, 'episodes')
print(f"Collection: {len(episodes)} épisodes récupérés")
print(f"Résumé: {episodes.to_summary()}\n")
for i, episode in enumerate(episodes):
print(f"Épisode {i+1}:")
print(f" Zone: {episode.lib_zone}")
print(f" Polluant: {episode.lib_pol} (code: {episode.get_polluant_code()})")
print(f" État: {episode.etat}")
print(f" Alerte active: {'Oui' if episode.is_alert_active() else 'Non'}")
print(f" Niveau d'alerte: {episode.get_alert_level()}")
print(f" Géométrie complexe: {'Oui' if episode.is_geometry_complex() else 'Non'}")
print()
def example_emissions_data():
"""Exemple d'utilisation des données d'émissions avec objets typés"""
print("=== Exemple Données d'Émissions avec objets typés ===\n")
# Données de test pour émissions
test_data = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.3522, 48.8566]
},
"properties": {
"aasqa": "11",
"code": "75056",
"name": "Paris",
"population": 2165423,
"superficie": 105.4,
"nox": 15420.5,
"pm10": 890.2,
"pm25": 623.8,
"ges": 12345678.9,
"code_pcaet": "7"
}
}
]
}
from atmo_data_wrapper import AtmoDataCollection
emissions = AtmoDataCollection(test_data, 'emissions')
print(f"Collection: {len(emissions)} données d'émissions récupérées\n")
for emission in emissions:
print(f"Territoire: {emission.name}")
print(f" Population: {emission.population:,} habitants")
print(f" Superficie: {emission.superficie} km²")
print(f" Secteur: {emission.get_secteur_name()}")
print()
# Émissions totales
total_emissions = emission.get_total_emissions()
print(" Émissions totales:")
for polluant, valeur in total_emissions.items():
print(f" {polluant}: {valeur:,.1f} t/an")
print()
# Densités et per capita
print(" Densités d'émission (t/km²):")
for polluant in ['nox', 'pm10', 'pm25']:
density = emission.get_emission_density(polluant)
print(f" {polluant.upper()}: {density:.2f}")
print(" Émissions par habitant (kg/hab):")
for polluant in ['nox', 'pm10', 'pm25']:
per_capita = emission.get_emission_per_capita(polluant) * 1000 # Convert to kg
print(f" {polluant.upper()}: {per_capita:.2f}")
print()
def example_pollen_indices():
"""Exemple d'utilisation des indices pollen avec objets typés"""
print("=== Exemple Indices Pollen avec objets typés ===\n")
# Données de test pour pollens
test_data = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.3522, 48.8566]
},
"properties": {
"aasqa": "11",
"alerte": True,
"code_ambr": 5,
"code_arm": 2,
"code_aul": 1,
"code_boul": 4,
"code_gram": 3,
"code_oliv": 1
}
}
]
}
from atmo_data_wrapper import AtmoDataCollection
pollens = AtmoDataCollection(test_data, 'pollens')
print(f"Collection: {len(pollens)} indices pollen récupérés\n")
for pollen in pollens:
print(f"Station pollen:")
print(f" Alerte active: {'Oui' if pollen.is_alert_active() else 'Non'}")
# Pollen le plus élevé
highest_pollen, highest_code = pollen.get_highest_pollen()
print(f" Plus haut niveau: {highest_pollen} (code: {highest_code})")
# Résumé de tous les pollens
summary = pollen.get_pollens_summary()
print(" Détail par espèce:")
for code, info in summary.items():
print(f" {info['espece']}: {info['qualificatif']} (code: {info['code']})")
# Pollens dangereux
dangerous = pollen.get_dangerous_pollens()
if dangerous:
print(f" Pollens à risque élevé: {', '.join(dangerous)}")
else:
print(" Aucun pollen à risque élevé")
print()
def example_geographic_filtering():
"""Exemple de filtrage géographique"""
print("=== Exemple Filtrage Géographique ===\n")
# Créer des données avec différentes localisations
test_data = {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.3522, 48.8566] # Paris
},
"properties": {
"aasqa": "11",
"code_qual": 3,
"lib_zone": "Paris Centre"
}
},
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [2.2945, 48.8584] # Proche de Paris
},
"properties": {
"aasqa": "11",
"code_qual": 2,
"lib_zone": "Boulogne"
}
},
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [5.3698, 43.2965] # Marseille (loin)
},
"properties": {
"aasqa": "93",
"code_qual": 4,
"lib_zone": "Marseille"
}
}
]
}
from atmo_data_wrapper import AtmoDataCollection, Coordinates
indices = AtmoDataCollection(test_data, 'indices')
print(f"Collection initiale: {len(indices)} éléments")
# Définir un point central (Paris)
paris_center = Coordinates(2.3522, 48.8566)
# Filtrer dans un rayon de 10 km autour de Paris
nearby = indices.filter_by_coordinates(paris_center, 10.0)
print(f"Dans un rayon de 10km de Paris: {len(nearby)} éléments")
for item in nearby:
if item.has_coordinates():
distance = item.coordinates.distance_to(paris_center)
print(f" {item.lib_zone}: {distance:.2f} km de Paris")
print()
if __name__ == "__main__":
example_indices_atmo()
example_episodes_pollution()
example_emissions_data()
example_pollen_indices()
example_geographic_filtering()