Source code for bqss.bqss.models.finess

import pandas as pd
import pandera as pa
from pandera.engines.pandas_engine import FLOAT64, INT64, Category, DateTime
from pandera.typing import Series

# pylint: disable-next=unexpected-keyword-arg,no-value-for-parameter
Date: DateTime = DateTime(unit="D", to_datetime_kwargs={"format": "%Y-%m-%d"})  # type: ignore

# Ajout suite à une discussion sur data-gouv concernant ces cas d'erreur
# https://www.data.gouv.fr/fr/datasets/finess-extraction-du-fichier-des-etablissements/#/discussions/65cfe1d80cfffe68d42c2720
TEMPORARY_GEOOCODING_EXCLUSION = [
    "060790185",
    "930031398",
    "060019957",
    "120001888",
    "120006937",
    "120009659",
    "120009667",
    "120009675",
    "140018284",
    "210003059",
    "250022290",
    "270000342",
    "270029416",
    "270030935",
    "300021748",
    "310793583",
    "310796891",
    "380802728",
    "420018921",
    "420780678",
    "450000252",
    "490536109",
    "560031742",
    "600011795",
    "620000349",
    "680017829",
    "710011560",
    "760000166",
    "760041368",
    "940000409",
    "970404406",
]


[docs] class FinessSchema(pa.DataFrameModel): """ Modèle de données de la table Finess. Ce référentiel historisé FINESS est construit à partir des exports FINESS mis à disposition en Open Data. Le détail des sources de données est accessible dans le dossier `resources` du repository. """ # pylint: disable=too-few-public-methods,no-self-argument
[docs] class Config: strict = True coerce = True
date_export: Series[Date] = pa.Field( # type: ignore title="Date de l'export source", description=( "Date de l'export de données source ayant servi" " à reconstituer cette ligne (ex: 2021-12-31)" ), coerce=True, ) num_finess_et: Series[str] = pa.Field( title="Numéro FINESS ET", description="Numéro FINESS du site géographique (ex: 920000650)", ) num_finess_ej: Series[str] = pa.Field( title="Numéro FINESS EJ", description="Numéro FINESS de l'entité juridique (ex: 920150059)", ) raison_sociale_et: Series[str] = pa.Field( title="Raison sociale ET", description="(ex: HOPITAL FOCH)" ) raison_sociale_longue_et: Series[str] = pa.Field( title="Raison sociale longue ET", description="(ex: HOPITAL FOCH)", nullable=True, ) complement_raison_sociale: Series[str] = pa.Field( title="Complément de raison sociale", nullable=True ) complement_distribution: Series[str] = pa.Field( title="Complément de distribution", nullable=True ) num_voie: Series[str] = pa.Field( title="Numéro de voie", description="(ex: 40)", nullable=True ) type_voie: Series[str] = pa.Field( title="Type de voie", description="(ex: Rue)", nullable=True ) libelle_voie: Series[str] = pa.Field( title="Libellé de voie", description="(ex: DE LA REPUBLIQUE)", nullable=True, ) complement_voie: Series[str] = pa.Field( title="Complément de voie", description="(ex: B)", nullable=True ) lieu_dit_bp: Series[str] = pa.Field( title="Lieu-dit / BP", description="(ex: BP 69)", nullable=True ) commune: Series[INT64] = pa.Field( title="Code Commune", description="(ex: 73)", nullable=True ) departement: Series[str] = pa.Field( title="Département", description=( "96 départements en France métropolitaine (1-95) " "dont la Corse du Sud (2A) et la Haute Corse (2B) " "et les 5 départements d'outre mer (9A, 9B, 9C, etc)" ), ) libelle_departement: Series[str] = pa.Field( title="Libellé département", description="(ex: HAUTS-DE-SEINE)" ) ligne_acheminement: Series[str] = pa.Field( title="Ligne d'acheminement (CodePostal+Lib commune)", description="(ex: 92151 SURESNES CEDEX)", ) telephone: Series[str] = pa.Field( title="Téléphone", description="Numéro de téléphone de l'établissement (ex: 08 26 20 72 20)", nullable=True, ) telecopie: Series[str] = pa.Field( title="Télécopie", description="(ex: 01 46 25 20 94)", nullable=True ) categorie_et: Series[INT64] = pa.Field( title="Catégorie d'établissement", description="(ex: 365)" ) libelle_categorie_et: Series[str] = pa.Field( title="Libellé catégorie d'établissement", description="(ex: Établissement de Soins Pluridisciplinaire)", ) categorie_agregat_et: Series[INT64] = pa.Field( title="Catégorie d'agrégat d'établissement", description="(ex: 1110)" ) libelle_categorie_agregat_et: Series[str] = pa.Field( title="Libellé catégorie d'agrégat d'établissement", description="(ex: Soins Suite & Réadap)", ) siret: Series[str] = pa.Field( title="Numéro de SIRET", description="(ex: 40845729900019)", nullable=True, ) code_ape: Series[str] = pa.Field( title="Code APE", description="Code Activité Principale Exercée (ex: 8899B)", nullable=True, ) code_mft: Series[str] = pa.Field( title="Code MFT", description="Code Mode de Fixation des Tarifs (ex: 07)", ) libelle_mft: Series[str] = pa.Field( title="Libellé MFT", description=( "Libellé Mode de Fixation des Tarifs" "(ex: ARS établissements Publics de santé dotation globale)" ), ) code_sph: Series[INT64] = pa.Field( title="Code SPH", description="(ex: 6)", nullable=True ) libelle_sph: Series[str] = pa.Field( title="Libellé SPH", description="(ex: Etablissement de santé privé d'intérêt collectif)", nullable=True, ) # Certaines dates sont trop anciennes (1536) pour être stockées sous forme de date # dans un dataframe pandas date_ouverture: Series[str] = pa.Field( title="Date d'ouverture", description="(ex: 1949-04-12)", str_matches=r"[12]\d{3}-[01]\d-[0123]\d", ) date_autorisation: Series[str] = pa.Field( title="Date d'autorisation", description="(ex: 1949-04-12)", str_matches=r"[12]\d{3}-[01]\d-[0123]\d", ) date_maj: Series[Date] = pa.Field( # type: ignore title="Date de mise à jour sur la structure", description="(ex: 2017-12-07)", ) num_uai: Series[str] = pa.Field( title="Numéro éducation nationale", description=( "Chaque établissement reconnu par l'éducation nationale " "possède un numéro UAI (Unité Administrative Immatriculée) " "(ex: 0195047H)" ), nullable=True, ) coord_x_et: Series[FLOAT64] = pa.Field( title="Coordonnées X", description="(ex: 642877.5)", nullable=True ) coord_y_et: Series[FLOAT64] = pa.Field( title="Coordonnées Y", description="(ex: 6863746.5)", nullable=True ) source_coord_et: Series[str] = pa.Field( title="Source des coordonnées", description="(ex: 1;ATLASANTE;100;IGN;BD_ADRESSE;V2.1;LAMBERT_93)", nullable=True, ) date_geocodage: Series[Date] = pa.Field( # type: ignore title="Date de calcul des coordonnées", description="(ex: 2021-05-10)", nullable=True, ) region: Series[INT64] = pa.Field( title="Région", description="(ex: 11)", nullable=True ) libelle_region: Series[str] = pa.Field( title="Libellé région", description="(ex: ILE DE FRANCE)", nullable=True, ) code_officiel_geo: Series[str] = pa.Field( title="Code officiel géographique", description="(ex: 92073 (autre exemple: 2B077))", nullable=True, ) code_postal: Series[str] = pa.Field( title="Code postal", description="(ex: 92151)" ) libelle_routage: Series[str] = pa.Field( title="Libellé routage", description="(ex: SURESNES CEDEX)", nullable=True, ) libelle_code_ape: Series[str] = pa.Field( title="Libellé code APE", description="(ex: Activités hospitalières)", nullable=True, ) ferme_cette_annee: Series[bool] = pa.Field( title="Fermé cette année", description=( "Champ indiquant si le FINESS historisé a fermé pour l'année correspondante " "(champ dateexport)" ), ) latitude: Series[float] = pa.Field( title="Latitude", description=( "Coordonnée latitudinale de l'établissement " "(système géodésique : WGS 84) (ex: 48.84512493935407)" ), nullable=True, le=90, ge=-90, ) longitude: Series[float] = pa.Field( title="Longitude", description=( "Coordonnée longitudinale de l'établissement " "(système géodésique : WGS 84) (ex: 2.319538289041818)" ), nullable=True, le=180, ge=-180, ) libelle_commune: Series[str] = pa.Field( title="Libelle de la commune", description="Nom de la commune de l'établissement (ex: CHATILLON)", ) adresse_postale_ligne_1: Series[str] = pa.Field( title="Adresse postale ligne 1", description=( "Première ligne de l'adresse postale de l'établissement " "(ex: 17 RUE DES FAUVETTES)" ), nullable=True, ) adresse_postale_ligne_2: Series[str] = pa.Field( title="Adresse postale ligne 2", description=( "Seconde ligne de l'adresse postale de l'établissement " "(ex: 92321 CHATILLON)" ), ) raison_sociale_ej: Series[str] = pa.Field( title="Raison sociale EJ", description="(ex: HOPITAL FOCH)" ) raison_sociale_longue_ej: Series[str] = pa.Field( title="Raison sociale longue EJ", description="(ex: HOPITAL FOCH)", nullable=True, ) statut_juridique_ej: Series[INT64] = pa.Field( title="Statut juridique de l'EJ", description=( "Statut juridique de l'entité juridique liée à cet établissement " "(ex: 13)" ), ) libelle_statut_juridique_ej: Series[str] = pa.Field( title="Libellé du statut juridique de l'EJ", description=( "Libellé du statut juridique de l'entité juridique liée à cet établissement " "(ex: Etablissement Public Communal d'Hospitalisation)" ), ) statut_juridique: Series[Category] = pa.Field( title="Statut juridique", description=( "Statut juridique simplifié sur 3 modalités : " "Public, Privé, Privé à but non lucratif " "(ex: Public)" ), dtype_kwargs={ "categories": ["Public", "Privé", "Privé à but non lucratif"] }, ) type_etablissement: Series[Category] = pa.Field( title="Type d'établissement", description=( "Type d'établissement : catégorie simplifié sur 6 modalités " "(Public, Privé, Privé à but non lucratif, CH, CHU, CLCC)" "(ex: CLCC)" ), dtype_kwargs={ "categories": [ "Public", "Privé", "Privé à but non lucratif", "CH", "CHU", "CLCC", ] }, ) actif_qualiscope: Series[bool] = pa.Field( title="Actif sur Qualiscope", description="Établissement visible sur Qualiscope", ) dernier_enregistrement: Series[bool] = pa.Field( title="Dernier enregistrement connu de l'établissement" )
[docs] @pa.check("num_finess_e.", regex=True, name="check_num_finess_format") def check_num_finess_format( cls, num_finess_et: Series[str] ) -> Series[bool]: """ Les numéros finess sont une suite de 9 chiffres """ return num_finess_et.str.match("^\\w{9}$")
[docs] @pa.dataframe_check def check_coords_non_null(cls, data_df: pd.DataFrame) -> Series[bool]: """ Les coordonnées géographiques ne sont disponibles qu'à partir de 2018 """ return ( ( (data_df["date_export"].dt.year >= 2018) & data_df["coord_x_et"].notna() & data_df["coord_y_et"].notna() & data_df["source_coord_et"].notna() & data_df["date_geocodage"].notna() & data_df["latitude"].notna() & data_df["longitude"].notna() ) | (data_df["date_export"].dt.year < 2018) | (data_df["num_finess_et"].isin(TEMPORARY_GEOOCODING_EXCLUSION)) )
[docs] @pa.dataframe_check def check_region_non_null(cls, data_df: pd.DataFrame) -> bool: """ Les exports trimestrielles sur l'année en cours ne contiennent pas la région. """ current_year_export_date = data_df["date_export"].max() return ( ( data_df[data_df["region"].isna()]["date_export"] == current_year_export_date ).all() and ( data_df[data_df["libelle_region"].isna()]["date_export"] == current_year_export_date ).all() and ( data_df[data_df["code_officiel_geo"].isna()]["date_export"] == current_year_export_date ).all() and ( data_df[data_df["libelle_routage"].isna()]["date_export"] == current_year_export_date ).all() )