"""Pydantic models for the Indicateurs Territoriaux API responses.""" from typing import Any from pydantic import BaseModel, Field class IndicatorMetadata(BaseModel): """Metadata for a territorial indicator.""" id: int = Field(..., description="Unique identifier of the indicator") libelle: str = Field(..., description="Human-readable name of the indicator") unite: str | None = Field(None, description="Unit of measurement") description: str | None = Field(None, description="Detailed description") methode_calcul: str | None = Field(None, description="Calculation method") fonction_calcul: str | None = Field(None, description="Calculation function") date_debut: int | None = Field(None, description="First available year") date_fin: int | None = Field(None, description="Last available year") annees_disponibles: str | None = Field( None, description="Available years (comma-separated)" ) annees_manquantes: str | None = Field( None, description="Missing years (comma-separated)" ) mailles_disponibles: str | None = Field( None, description="Available geographic levels (e.g., 'region,departement,epci')" ) maille_mini_disponible: str | None = Field( None, description="Finest available geographic level" ) couverture_geographique: str | None = Field( None, description="Geographic coverage (France métro, DOM, etc.)" ) liste_drom: str | None = Field(None, description="Covered DROM territories") completion_region: float | None = Field( None, description="Completion percentage at region level" ) completion_departement: float | None = Field( None, description="Completion percentage at department level" ) completion_epci: float | None = Field( None, description="Completion percentage at EPCI level" ) completion_commune: float | None = Field( None, description="Completion percentage at commune level" ) compte_region: int | None = Field( None, description="Number of regions with data" ) compte_departement: int | None = Field( None, description="Number of departments with data" ) compte_epci: int | None = Field(None, description="Number of EPCIs with data") compte_commune: int | None = Field( None, description="Number of communes with data" ) thematique_fnv: str | None = Field( None, description="France Nation Verte thematic" ) secteur_fnv: str | None = Field(None, description="FNV sector") enjeux_fnv: str | None = Field(None, description="FNV challenges") levier_fnv: str | None = Field(None, description="FNV lever") projets_associes: str | None = Field(None, description="Associated projects") valeur_axes: str | None = Field( None, description="Breakdown axes (JSON stringified)" ) @classmethod def from_api_response(cls, data: dict[str, Any]) -> "IndicatorMetadata": """Create an IndicatorMetadata from a Cube.js API response row. The API returns dimension names prefixed with the cube name. This method strips the prefix. """ # Strip the cube name prefix from keys prefix = "indicateur_metadata." cleaned = {} for key, value in data.items(): clean_key = key.replace(prefix, "") cleaned[clean_key] = value return cls(**cleaned) def has_geographic_level(self, level: str) -> bool: """Check if the indicator has data at the specified geographic level.""" if not self.mailles_disponibles: return False return level.lower() in self.mailles_disponibles.lower() def get_completion_for_level(self, level: str) -> float | None: """Get the completion percentage for a geographic level.""" level_map = { "region": self.completion_region, "departement": self.completion_departement, "epci": self.completion_epci, "commune": self.completion_commune, } return level_map.get(level.lower()) class SourceMetadata(BaseModel): """Metadata for a data source associated with an indicator.""" id_indicateur: int = Field(..., description="ID of the related indicator") nom_source: str | None = Field(None, description="Source identifier") libelle: str | None = Field(None, description="Human-readable source name") description: str | None = Field(None, description="Source description") producteur_source: str | None = Field(None, description="Data producer") distributeur_source: str | None = Field(None, description="Data distributor") license_source: str | None = Field(None, description="Data license") lien_page: str | None = Field(None, description="Source URL") annees_disponibles_source: str | None = Field( None, description="Available years from this source" ) annees_manquantes_source: str | None = Field( None, description="Missing years from this source" ) maille_mini_disponible: str | None = Field( None, description="Finest geographic level" ) couverture_geographique: str | None = Field( None, description="Geographic coverage" ) date_derniere_extraction: str | None = Field( None, description="Last extraction date" ) @classmethod def from_api_response(cls, data: dict[str, Any]) -> "SourceMetadata": """Create a SourceMetadata from a Cube.js API response row.""" prefix = "indicateur_x_source_metadata." cleaned = {} for key, value in data.items(): clean_key = key.replace(prefix, "") cleaned[clean_key] = value return cls(**cleaned) class IndicatorListItem(BaseModel): """Simplified indicator info for list responses.""" id: int libelle: str unite: str | None = None mailles_disponibles: str | None = None thematique_fnv: str | None = None class IndicatorDetails(BaseModel): """Complete indicator details with sources.""" metadata: IndicatorMetadata sources: list[SourceMetadata] = Field(default_factory=list) class GeographicDataPoint(BaseModel): """A single data point with geographic information.""" geocode: str = Field(..., description="INSEE code of the territory") libelle: str | None = Field(None, description="Name of the territory") valeur: float | str | None = Field(None, description="Indicator value") annee: str | None = Field(None, description="Year of the data") unite: str | None = Field(None, description="Unit of measurement") class QueryResult(BaseModel): """Result of a data query.""" indicator_id: int indicator_name: str geographic_level: str data: list[GeographicDataPoint] total_count: int = 0 query_info: dict[str, Any] = Field(default_factory=dict) class SearchResult(BaseModel): """Result of an indicator search.""" indicators: list[IndicatorListItem] query: str total_count: int class CubeInfo(BaseModel): """Information about a data cube.""" name: str = Field(..., description="Cube name (e.g., 'conso_enaf_com')") maille: str = Field(..., description="Geographic level (commune, epci, departement, region)") indicator_ids: list[int] = Field(default_factory=list, description="Indicator IDs in this cube") # Geographic level constants GEOGRAPHIC_LEVELS = ["region", "departement", "epci", "commune"] # Maille suffix mapping for cube names MAILLE_SUFFIX_MAP = { "commune": "_com", "epci": "_epci", "departement": "_dpt", "region": "_reg", } # Dimension patterns for each geographic level (validated by API tests) # Format: geocode_{maille} and libelle_{maille} GEO_DIMENSION_PATTERNS = { "region": { "geocode": "geocode_region", "libelle": "libelle_region", }, "departement": { "geocode": "geocode_departement", "libelle": "libelle_departement", }, "epci": { "geocode": "geocode_epci", "libelle": "libelle_epci", }, "commune": { "geocode": "geocode_commune", "libelle": "libelle_commune", }, } # Region code reference REGION_CODES = { "11": "Île-de-France", "24": "Centre-Val de Loire", "27": "Bourgogne-Franche-Comté", "28": "Normandie", "32": "Hauts-de-France", "44": "Grand Est", "52": "Pays de la Loire", "53": "Bretagne", "75": "Nouvelle-Aquitaine", "76": "Occitanie", "84": "Auvergne-Rhône-Alpes", "93": "Provence-Alpes-Côte d'Azur", "94": "Corse", }