Make task dependencies explicit

.gitignore

@@ -45,3 +45,6 @@ sandbox/
 
 # ignore pickle files
 *.pickle
+
+# ignore yaml files
+*.yaml

pyproject.toml

@@ -67,7 +67,7 @@ allow-direct-references = true
 # Pixi configuration
 # ======================================================================================
 
-[tool.pixi.project]
+[tool.pixi.workspace]
 channels = ["conda-forge"]
 platforms = ["linux-64", "osx-64", "osx-arm64", "win-64"]
 
@@ -78,6 +78,7 @@ pre-commit = "*"
 pygraphviz = "*"
 pyarrow = ">=21.0.0,<22"
 pandas = ">=2.3.1,<3"
+pytask-parallel = ">=0.5.1,<0.6"
 
 [tool.pixi.pypi-dependencies]
 soep_preparation = {path = ".", editable = true}

src/soep_preparation/clean_variables/hwealth.py → src/soep_preparation/clean_modules/hwealth.py

@@ -60,42 +60,51 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     out["hh_net_overall_wealth_d"] = apply_smallest_float_dtype(raw_data["w011hd"])
     out["hh_net_overall_wealth_e"] = apply_smallest_float_dtype(raw_data["w011he"])
 
-    out["hh_vehicles_value_a"] = apply_smallest_float_dtype(raw_data["v010ha"])
-    out["hh_vehicles_value_b"] = apply_smallest_float_dtype(raw_data["v010hb"])
-    out["hh_vehicles_value_c"] = apply_smallest_float_dtype(raw_data["v010hc"])
-    out["hh_vehicles_value_d"] = apply_smallest_float_dtype(raw_data["v010hd"])
-    out["hh_vehicles_value_e"] = apply_smallest_float_dtype(raw_data["v010he"])
+    out["hh_vehicles_value_a"] = apply_smallest_float_dtype(
+        raw_data["v010ha"].replace({-8: pd.NA})
+    )
+    out["hh_vehicles_value_b"] = apply_smallest_float_dtype(
+        raw_data["v010hb"].replace({-8: pd.NA})
+    )
+    out["hh_vehicles_value_c"] = apply_smallest_float_dtype(
+        raw_data["v010hc"].replace({-8: pd.NA})
+    )
+    out["hh_vehicles_value_d"] = apply_smallest_float_dtype(
+        raw_data["v010hd"].replace({-8: pd.NA})
+    )
+    out["hh_vehicles_value_e"] = apply_smallest_float_dtype(
+        raw_data["v010he"].replace({-8: pd.NA})
+    )
 
     out["hh_gross_overall_wealth_including_vehicles_a"] = apply_smallest_float_dtype(
-        raw_data["n010ha"]
+        raw_data["n010ha"].replace({-8: pd.NA})
     )
     out["hh_gross_overall_wealth_including_vehicles_b"] = apply_smallest_float_dtype(
-        raw_data["n010hb"]
+        raw_data["n010hb"].replace({-8: pd.NA})
     )
     out["hh_gross_overall_wealth_including_vehicles_c"] = apply_smallest_float_dtype(
-        raw_data["n010hc"]
+        raw_data["n010hc"].replace({-8: pd.NA})
     )
     out["hh_gross_overall_wealth_including_vehicles_d"] = apply_smallest_float_dtype(
-        raw_data["n010hd"]
+        raw_data["n010hd"].replace({-8: pd.NA})
     )
     out["hh_gross_overall_wealth_including_vehicles_e"] = apply_smallest_float_dtype(
-        raw_data["n010he"]
+        raw_data["n010he"].replace({-8: pd.NA})
     )
-
     out["hh_net_overall_wealth_including_vehicles_and_student_loans_a"] = (
-        apply_smallest_float_dtype(raw_data["n011ha"])
+        apply_smallest_float_dtype(raw_data["n011ha"].replace({-8: pd.NA}))
     )
     out["hh_net_overall_wealth_including_vehicles_and_student_loans_b"] = (
-        apply_smallest_float_dtype(raw_data["n011hb"])
+        apply_smallest_float_dtype(raw_data["n011hb"].replace({-8: pd.NA}))
     )
     out["hh_net_overall_wealth_including_vehicles_and_student_loans_c"] = (
-        apply_smallest_float_dtype(raw_data["n011hc"])
+        apply_smallest_float_dtype(raw_data["n011hc"].replace({-8: pd.NA}))
     )
     out["hh_net_overall_wealth_including_vehicles_and_student_loans_d"] = (
-        apply_smallest_float_dtype(raw_data["n011hd"])
+        apply_smallest_float_dtype(raw_data["n011hd"].replace({-8: pd.NA}))
     )
     out["hh_net_overall_wealth_including_vehicles_and_student_loans_e"] = (
-        apply_smallest_float_dtype(raw_data["n011he"])
+        apply_smallest_float_dtype(raw_data["n011he"].replace({-8: pd.NA}))
     )
 
     out[

src/soep_preparation/clean_variables/pkal.py → src/soep_preparation/clean_modules/pkal.py

@@ -78,6 +78,7 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
         The processed pkal data.
     """
     out = pd.DataFrame()
+    tmp = pd.DataFrame()
 
     out["p_id"] = apply_smallest_int_dtype(raw_data["pid"])
     out["hh_id"] = apply_smallest_int_dtype(raw_data["hid"])
@@ -102,20 +103,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     # the second the timeframe 1998 until 2022
     # individual employment status by month
     # Month 1 - Jan
-    out["tmp_ft_employed_m_v1_1"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_1"] = object_to_str_categorical(
         series=raw_data["kal1a001_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_1"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_1"] = object_to_str_categorical(
         series=raw_data["kal1a001_v2"],
         renaming={
             "[1] Jan Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Jan Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_1"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_1"],
-        series_2=out["tmp_ft_employed_m_v2_1"],
+        series_1=tmp["tmp_ft_employed_m_v1_1"],
+        series_2=tmp["tmp_ft_employed_m_v2_1"],
         ordered=False,
     )
     out["pt_employed_m_1"] = object_to_str_categorical(
@@ -131,20 +132,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 2 - Feb
-    out["tmp_ft_employed_m_v1_2"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_2"] = object_to_str_categorical(
         series=raw_data["kal1a002_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_2"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_2"] = object_to_str_categorical(
         series=raw_data["kal1a002_v2"],
         renaming={
             "[1] Feb Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Feb Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_2"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_2"],
-        series_2=out["tmp_ft_employed_m_v2_2"],
+        series_1=tmp["tmp_ft_employed_m_v1_2"],
+        series_2=tmp["tmp_ft_employed_m_v2_2"],
         ordered=False,
     )
     out["pt_employed_m_2"] = object_to_str_categorical(
@@ -160,20 +161,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 3 - Mrz
-    out["tmp_ft_employed_m_v1_3"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_3"] = object_to_str_categorical(
         series=raw_data["kal1a003_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_3"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_3"] = object_to_str_categorical(
         series=raw_data["kal1a003_v2"],
         renaming={
             "[1] Mrz Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Mrz Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_3"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_3"],
-        series_2=out["tmp_ft_employed_m_v2_3"],
+        series_1=tmp["tmp_ft_employed_m_v1_3"],
+        series_2=tmp["tmp_ft_employed_m_v2_3"],
         ordered=False,
     )
     out["pt_employed_m_3"] = object_to_str_categorical(
@@ -189,20 +190,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 4 - Apr
-    out["tmp_ft_employed_m_v1_4"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_4"] = object_to_str_categorical(
         series=raw_data["kal1a004_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_4"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_4"] = object_to_str_categorical(
         series=raw_data["kal1a004_v2"],
         renaming={
             "[1] Apr Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Apr Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_4"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_4"],
-        series_2=out["tmp_ft_employed_m_v2_4"],
+        series_1=tmp["tmp_ft_employed_m_v1_4"],
+        series_2=tmp["tmp_ft_employed_m_v2_4"],
         ordered=False,
     )
     out["pt_employed_m_4"] = object_to_str_categorical(
@@ -218,20 +219,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 5 - Mai
-    out["tmp_ft_employed_m_v1_5"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_5"] = object_to_str_categorical(
         series=raw_data["kal1a005_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_5"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_5"] = object_to_str_categorical(
         series=raw_data["kal1a005_v2"],
         renaming={
             "[1] Mai Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Mai Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_5"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_5"],
-        series_2=out["tmp_ft_employed_m_v2_5"],
+        series_1=tmp["tmp_ft_employed_m_v1_5"],
+        series_2=tmp["tmp_ft_employed_m_v2_5"],
         ordered=False,
     )
     out["pt_employed_m_5"] = object_to_str_categorical(
@@ -247,20 +248,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 6 - Jun
-    out["tmp_ft_employed_m_v1_6"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_6"] = object_to_str_categorical(
         series=raw_data["kal1a006_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_6"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_6"] = object_to_str_categorical(
         series=raw_data["kal1a006_v2"],
         renaming={
             "[1] Jun Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Jun Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_6"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_6"],
-        series_2=out["tmp_ft_employed_m_v2_6"],
+        series_1=tmp["tmp_ft_employed_m_v1_6"],
+        series_2=tmp["tmp_ft_employed_m_v2_6"],
         ordered=False,
     )
     out["pt_employed_m_6"] = object_to_str_categorical(
@@ -276,20 +277,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 7 - Jul
-    out["tmp_ft_employed_m_v1_7"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_7"] = object_to_str_categorical(
         series=raw_data["kal1a007_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_7"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_7"] = object_to_str_categorical(
         series=raw_data["kal1a007_v2"],
         renaming={
             "[1] Jul Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Jul Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_7"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_7"],
-        series_2=out["tmp_ft_employed_m_v2_7"],
+        series_1=tmp["tmp_ft_employed_m_v1_7"],
+        series_2=tmp["tmp_ft_employed_m_v2_7"],
         ordered=False,
     )
     out["pt_employed_m_7"] = object_to_str_categorical(
@@ -305,20 +306,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 8 - Aug
-    out["tmp_ft_employed_m_v1_8"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_8"] = object_to_str_categorical(
         series=raw_data["kal1a008_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_8"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_8"] = object_to_str_categorical(
         series=raw_data["kal1a008_v2"],
         renaming={
             "[1] Aug Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Aug Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_8"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_8"],
-        series_2=out["tmp_ft_employed_m_v2_8"],
+        series_1=tmp["tmp_ft_employed_m_v1_8"],
+        series_2=tmp["tmp_ft_employed_m_v2_8"],
         ordered=False,
     )
     out["pt_employed_m_8"] = object_to_str_categorical(
@@ -334,20 +335,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 9 - Sep
-    out["tmp_ft_employed_m_v1_9"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_9"] = object_to_str_categorical(
         series=raw_data["kal1a009_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_9"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_9"] = object_to_str_categorical(
         series=raw_data["kal1a009_v2"],
         renaming={
             "[1] Sep Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Sep Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_9"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_9"],
-        series_2=out["tmp_ft_employed_m_v2_9"],
+        series_1=tmp["tmp_ft_employed_m_v1_9"],
+        series_2=tmp["tmp_ft_employed_m_v2_9"],
         ordered=False,
     )
     out["pt_employed_m_9"] = object_to_str_categorical(
@@ -363,20 +364,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 10 - Okt
-    out["tmp_ft_employed_m_v1_10"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_10"] = object_to_str_categorical(
         series=raw_data["kal1a010_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_10"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_10"] = object_to_str_categorical(
         series=raw_data["kal1a010_v2"],
         renaming={
             "[1] Okt Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Okt Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_10"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_10"],
-        series_2=out["tmp_ft_employed_m_v2_10"],
+        series_1=tmp["tmp_ft_employed_m_v1_10"],
+        series_2=tmp["tmp_ft_employed_m_v2_10"],
         ordered=False,
     )
     out["pt_employed_m_10"] = object_to_str_categorical(
@@ -392,20 +393,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 11 - Nov
-    out["tmp_ft_employed_m_v1_11"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_11"] = object_to_str_categorical(
         series=raw_data["kal1a011_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_11"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_11"] = object_to_str_categorical(
         series=raw_data["kal1a011_v2"],
         renaming={
             "[1] Nov Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Nov Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_11"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_11"],
-        series_2=out["tmp_ft_employed_m_v2_11"],
+        series_1=tmp["tmp_ft_employed_m_v1_11"],
+        series_2=tmp["tmp_ft_employed_m_v2_11"],
         ordered=False,
     )
     out["pt_employed_m_11"] = object_to_str_categorical(
@@ -421,20 +422,20 @@ def clean(raw_data: pd.DataFrame) -> pd.DataFrame:
     )
 
     # Month 12 - Dez
-    out["tmp_ft_employed_m_v1_12"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v1_12"] = object_to_str_categorical(
         series=raw_data["kal1a012_v1"],
         renaming={"[1] Ja": "Vollzeit erwerbstätig"},
     )
-    out["tmp_ft_employed_m_v2_12"] = object_to_str_categorical(
+    tmp["tmp_ft_employed_m_v2_12"] = object_to_str_categorical(
         series=raw_data["kal1a012_v2"],
         renaming={
             "[1] Dez Vollzeit erwerbst.": "Vollzeit erwerbstätig",
             "[8] Dez Werkstatt fuer behinderte Menschen": "Werkstatt für behinderte Menschen",  # noqa: E501
         },
     )
     out["ft_employed_m_12"] = combine_first_and_make_categorical(
-        series_1=out["tmp_ft_employed_m_v1_12"],
-        series_2=out["tmp_ft_employed_m_v2_12"],
+        series_1=tmp["tmp_ft_employed_m_v1_12"],
+        series_2=tmp["tmp_ft_employed_m_v2_12"],
         ordered=False,
     )
     out["pt_employed_m_12"] = object_to_str_categorical(