ajoute l'option de mensualisation dans la pipeline

leximpact_prepare_data/run_pipeline.py

@@ -37,7 +37,15 @@
     show_default=True,
     type=int,
 )
-def run_pipeline(annee_erfs, annee_pote, annee_de_calcul):
+@click.option(
+    "-mensualisation",
+    "--mensualisation",
+    default=False,
+    help="Mensualisation des revenus",
+    show_default=True,
+    type=bool,
+)
+def run_pipeline(annee_erfs, annee_pote, annee_de_calcul, mensualisation):
     log.debug("Create FranceTaxBenefitSystem")
 
     # Instanciation du scenario
@@ -84,10 +92,16 @@ def run_pipeline(annee_erfs, annee_pote, annee_de_calcul):
         ].create_data_frame_by_entity(variables=variables, period=year, index=True)
 
         collection = "leximpact"
+        if mensualisation:
+            survey_name = f"leximpact_mensualise_{annee_de_calcul}"
+        else:
             survey_name = f"leximpact_{annee_de_calcul}"
+
         for entity, input_dataframe in data_frame_by_entity.items():
             assert isinstance(input_dataframe, pd.DataFrame)
 
+            if mensualisation:
+
                 if entity == "individu":
 
                     # Mensualiser les données
@@ -119,14 +133,15 @@ def run_pipeline(annee_erfs, annee_pote, annee_de_calcul):
 
             # Enregistrer la table annuelle dans le survey
             set_table_in_survey(
-                input_dataframe,
-                entity,
+                input_dataframe=input_dataframe,
+                entity=entity,
                 period=year,
                 collection=collection,
                 survey_name=survey_name,
                 config_files_directory=survey_scenario.data["config_files_directory"],
                 source_format="parquet",
             )
+
             print(
                 f"set_table_in_survey of {entity} for {year} in {collection}.{survey_name}"
             )

leximpact_prepare_data/scenario_tools/mensualisation.py

@@ -409,13 +409,14 @@ def mensualiser_revenus(
         nom_fichier_export = f"individu_{annee_table}_{str(i).zfill(2)}"
 
         set_table_in_survey(
-            df_mois,
+            input_dataframe=df_mois,
             entity="individu",
             parquet_file=os.path.join(dossier_export, f"{nom_fichier_export}.parquet"),
             period=f"{annee_table}-{str(i).zfill(2)}",
             collection=collection,
             survey_name=survey_name,
             config_files_directory=config_files_directory,
+            source_format="parquet",
         )
 
         # incrémenter i
@@ -522,13 +523,14 @@ def mensualiser(
     df_indiv = df_indiv[["noindiv", "TRIM", "MOIS"] + spr_columns]
     df_indiv[spr_columns] = df_indiv[spr_columns].replace(np.nan, 9)
 
-    # --- Initialiser le survey ---
+    # --- Initialiser le survey pour récupérer le dossier d'export ---
 
-    survey_collection = SurveyCollection.load(
-        config_files_directory=config_files_directory, collection=collection
+    survey_collection = SurveyCollection(
+        config_files_directory=config_files_directory, name=collection
+    )
+    dossier_export = os.path.join(
+        survey_collection.config.get("data", "output_directory"), survey_name
     )
-    survey = survey_collection.get_survey(survey_name)
-    dossier_export = survey.parquet_file_path
 
     # --- Merger la table ERFS importée avec la table LexImpact ---