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
def get_loi_hospitaliere_mask(data_df):
"""
Retourne un mask indiquant si l'et relève de la loi hospitalière ou assimilé
(ancien périmètre finess de la BQSS avant l'élargissement ESSMS)
"""
return (
(
(data_df["categorie_agregat_et"] >= 1000)
& (data_df["categorie_agregat_et"] < 2000)
)
| (data_df["categorie_agregat_et"] == 2204)
| (data_df["categorie_agregat_et"] == 2205)
)
[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)",
nullable=True,
)
libelle_mft: Series[str] = pa.Field(
title="Libellé MFT",
description=(
"Libellé Mode de Fixation des Tarifs"
"(ex: ARS établissements Publics de santé dotation globale)"
),
nullable=True,
)
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)",
)
date_autorisation: Series[str] = pa.Field(
title="Date d'autorisation",
description="(ex: 1949-04-12)",
nullable=True,
)
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)",
nullable=True,
)
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)"
),
nullable=True,
)
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)"
),
nullable=True,
)
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",
"Inconnu",
]
},
)
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",
"Inconnu",
]
},
)
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.dataframe_check
def check_date_ouverture_autorisation(
cls, data_df: pd.DataFrame
) -> Series[bool]:
"""
Les dates d'ouverture et d'autorisations sont parfois erronées
"""
is_loi_hospitaliere = get_loi_hospitaliere_mask(data_df)
really_clean_ouv_date = data_df["date_ouverture"].str.match(
r"[12]\d{3}-[01]\d-[0123]\d"
)
really_clean_autor_date = (
data_df["date_autorisation"]
.str.match(r"[12]\d{3}-[01]\d-[0123]\d")
.fillna(False)
)
# certaines dates sont erronées, par exe:
# 0 41147 0202-02-24
# 1 79712 0426-03-01
# 2 331035 0202-02-24
# 3 367690 0426-03-01
# 4 382658 0201-12-01
# 5 402836 0426-03-01
# 6 439628 0201-12-01
lax_clean_ouv_date = data_df["date_ouverture"].str.match(
r"[012]\d{3}-[01]\d-[0123]\d"
)
lax_clean_autor_date = data_df["date_autorisation"].str.match(
r"[012]\d{3}-[01]\d-[0123]\d"
)
return (
(is_loi_hospitaliere & really_clean_ouv_date)
| (
~is_loi_hospitaliere
& (lax_clean_ouv_date | data_df["date_ouverture"].isna())
)
) & (
(is_loi_hospitaliere & really_clean_autor_date)
| (
~is_loi_hospitaliere
& (lax_clean_autor_date | data_df["date_autorisation"].isna())
)
)
[docs]
@pa.dataframe_check
def check_nullable_ej(cls, data_df: pd.DataFrame) -> Series[bool]:
"""
Les raisons sociale ej sont non nulles pour les établissements relevant
de la loi hospitalière
"""
is_loi_hospitaliere = get_loi_hospitaliere_mask(data_df)
return (
is_loi_hospitaliere
& data_df["raison_sociale_ej"].notna()
& data_df["statut_juridique_ej"].notna()
& data_df["libelle_statut_juridique_ej"].notna()
) | (~is_loi_hospitaliere)
[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,
On ne vérifie que sur le dernier enregistrement
"""
return (
(
(data_df["date_export"].dt.year >= 2018)
& data_df["dernier_enregistrement"]
& 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["dernier_enregistrement"]
)
| (data_df["date_export"].dt.year < 2018)
)
[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()
)
[docs]
@pa.dataframe_check
def check_qualiscope_non_null(cls, data_df: pd.DataFrame) -> Series[bool]:
"""
Vérifie que le code_mft/libelle_mft pour les finess actif_qualiscope
"""
is_loi_hospitaliere = get_loi_hospitaliere_mask(data_df)
return (
(is_loi_hospitaliere & data_df["code_mft"].notna())
& (is_loi_hospitaliere & data_df["libelle_mft"].notna())
& (is_loi_hospitaliere & data_df["date_autorisation"].notna())
) | (~is_loi_hospitaliere)