From c559150d40a9a7d61e3c099dce7fd81736a0283e Mon Sep 17 00:00:00 2001
From: Marton Vago <marci.vago@gmail.com>
Date: Thu, 28 May 2026 09:19:22 +0100
Subject: [PATCH 1/5] feat: :sparkles: stage raw data

---
 justfile              |   4 ++
 scripts/stage_data.py | 133 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 137 insertions(+)
 create mode 100644 scripts/stage_data.py

diff --git a/justfile b/justfile
index dfe00d4..f85dfc8 100644
--- a/justfile
+++ b/justfile
@@ -86,3 +86,7 @@ download-data-dict:
 # Download data from REDCap
 download-data:
   uv run python scripts/redcap_data.py
+
+# Stage raw data
+stage-data:
+  uv run python scripts/stage_data.py
diff --git a/scripts/stage_data.py b/scripts/stage_data.py
new file mode 100644
index 0000000..dbfe35a
--- /dev/null
+++ b/scripts/stage_data.py
@@ -0,0 +1,133 @@
+import re
+from datetime import datetime
+from operator import itemgetter
+from pathlib import Path
+from typing import cast
+
+import polars as pl
+import seedcase_soil as so
+
+VAS_TIME_FIELD_PATTERN = re.compile(
+    r"^vas_(?P<field_name>.+?)(_fasted)?_(?P<time>minus10|30|60|90|120|180|240)min$"
+)
+
+
+def load_raw_data() -> pl.DataFrame:
+    """Loads the latest raw data from `raw/redcap/<timestamp>.csv.gz`."""
+    file_path = Path("raw") / "redcap"
+    latest_file = max(file_path.glob("*.csv.gz"), key=lambda file: file.name)
+    so.pretty_print(f"Loading data from '{latest_file}'.")
+    return pl.read_csv(latest_file)
+
+
+def get_fields_by_resource() -> dict[str, list[str]]:
+    """Gets a mapping from resource names to a list of field names in that resource."""
+    properties = so.read_properties(so.parse_source("datapackage.json"))
+    return dict(
+        so.fmap(
+            properties["resources"],
+            lambda resource: (
+                resource["name"],
+                so.fmap(resource["schema"]["fields"], itemgetter("name")),
+            ),
+        )
+    )
+
+
+def select_with_base_cols(
+    raw_df: pl.DataFrame, resource_name: str, cols: list[str]
+) -> pl.DataFrame:
+    """Selects columns and adds base columns common to all dataframes."""
+    return (
+        raw_df.select(["redcap_event_name"] + cols)
+        .rename({"redcap_event_name": "event"})
+        .with_columns(
+            pl.lit("Copenhagen").alias("center"),
+            pl.lit(resource_name).alias("resource_name"),
+        )
+    )
+
+
+def raw_to_staged(raw_df: pl.DataFrame) -> list[pl.DataFrame]:
+    """Transforms the raw data into staged dataframes."""
+    resources = get_fields_by_resource()
+
+    # Resources with special handling
+    resources.pop("vas", None)
+    dfs = [create_vas_df(raw_df)]
+
+    # Resources without special handling
+    return dfs + so.pairwise_fmap(
+        list(resources.items()), [raw_df], create_df_for_resource
+    )
+
+
+def create_df_for_resource(
+    resource_entry: tuple[str, list[str]], raw_df: pl.DataFrame
+) -> pl.DataFrame:
+    """Creates a dataframe for a resource."""
+    resource_name, field_names = resource_entry
+
+    # Remove fields not in the raw data
+    field_names.remove("event")
+    field_names.remove("center")
+
+    return select_with_base_cols(raw_df, resource_name, field_names)
+
+
+def create_vas_df(raw_df: pl.DataFrame) -> pl.DataFrame:
+    """Creates a dataframe for the VAS resource."""
+    vas_cols = so.keep(
+        raw_df.columns,
+        lambda column: VAS_TIME_FIELD_PATTERN.match(column) is not None,
+    )
+
+    cols_grouped_by_time: dict[int, list[str]] = {}
+    for col in vas_cols:
+        match = cast(re.Match[str], VAS_TIME_FIELD_PATTERN.match(col))
+
+        time = match.group("time")
+        if time == "minus10":
+            time = "-10"
+
+        cols_grouped_by_time.setdefault(int(time), []).append(col)
+
+    vas_dfs = so.pairwise_fmap(
+        list(cols_grouped_by_time.items()), [raw_df], create_df_for_time_group
+    )
+    return pl.concat(vas_dfs, how="vertical")
+
+
+def create_df_for_time_group(
+    time_group: tuple[int, list[str]], raw_df: pl.DataFrame
+) -> pl.DataFrame:
+    """Creates a dataframe for a group of VAS columns with the same time."""
+    time, cols = time_group
+
+    renamed_cols = {
+        col: cast(re.Match[str], VAS_TIME_FIELD_PATTERN.match(col)).group("field_name")
+        for col in cols
+    }
+
+    return (
+        select_with_base_cols(raw_df, "vas", cols)
+        .rename(renamed_cols)
+        .with_columns(pl.lit(time).alias("minutes_from_meal"))
+    )
+
+
+def write_staged_df(df: pl.DataFrame) -> None:
+    """Writes the dataframe to `staging/redcap/<resource-name>/<timestamp>.parquet`."""
+    resource_name = df["resource_name"][0]
+    timestamp = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
+    file_path = Path("staging") / "redcap" / resource_name / f"{timestamp}.parquet"
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+
+    df.drop("resource_name").write_parquet(file_path)
+    so.pretty_print(f"Wrote staged data to '{file_path}'.")
+
+
+if __name__ == "__main__":
+    raw = load_raw_data()
+    for df in raw_to_staged(raw):
+        write_staged_df(df)

From b4608bbc64e96450b9d9a260c634812236b50007 Mon Sep 17 00:00:00 2001
From: Marton Vago <marci.vago@gmail.com>
Date: Thu, 28 May 2026 09:42:03 +0100
Subject: [PATCH 2/5] feat: :sparkles: track staging with git lfs

---
 .gitattributes | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.gitattributes b/.gitattributes
index 51b0677..743a5c6 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -1 +1,2 @@
 raw/** filter=lfs diff=lfs merge=lfs -text
+staging/** filter=lfs diff=lfs merge=lfs -text

From 30214ad33d13ee48fa829110713a9fc517774987 Mon Sep 17 00:00:00 2001
From: Marton Vago <marci.vago@gmail.com>
Date: Thu, 28 May 2026 10:16:09 +0100
Subject: [PATCH 3/5] refactor: :recycle: end early if no raw data

---
 scripts/stage_data.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/scripts/stage_data.py b/scripts/stage_data.py
index dbfe35a..f2dcfe4 100644
--- a/scripts/stage_data.py
+++ b/scripts/stage_data.py
@@ -15,7 +15,14 @@
 def load_raw_data() -> pl.DataFrame:
     """Loads the latest raw data from `raw/redcap/<timestamp>.csv.gz`."""
     file_path = Path("raw") / "redcap"
-    latest_file = max(file_path.glob("*.csv.gz"), key=lambda file: file.name)
+    files = list(file_path.glob("*.csv.gz"))
+    if not files:
+        raise FileNotFoundError(
+            f"No raw data files found in '{file_path}'. "
+            "Have you run `just download-data`?"
+        )
+
+    latest_file = max(files, key=lambda file: file.name)
     so.pretty_print(f"Loading data from '{latest_file}'.")
     return pl.read_csv(latest_file)
 

From 627ee1629382b50d1fc528ab0dff966344ff965f Mon Sep 17 00:00:00 2001
From: Marton Vago <marci.vago@gmail.com>
Date: Thu, 28 May 2026 10:36:46 +0100
Subject: [PATCH 4/5] chore: :wrench: add vas to known words

---
 .typos.toml | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/.typos.toml b/.typos.toml
index a8ad854..7699225 100644
--- a/.typos.toml
+++ b/.typos.toml
@@ -5,3 +5,8 @@ extend-exclude = [
   # Auto-generated by Flower
   "resources"
 ]
+
+[default]
+
+[default.extend-words]
+vas = "vas"

From 455a4acacdcf78185f39d5b3f2e0a532827cb820 Mon Sep 17 00:00:00 2001
From: Marton Vago <marci.vago@gmail.com>
Date: Thu, 28 May 2026 10:43:03 +0100
Subject: [PATCH 5/5] refactor: :recycle: add underscore prefix to functions

---
 scripts/stage_data.py | 22 +++++++++++-----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/scripts/stage_data.py b/scripts/stage_data.py
index f2dcfe4..a646a27 100644
--- a/scripts/stage_data.py
+++ b/scripts/stage_data.py
@@ -27,7 +27,7 @@ def load_raw_data() -> pl.DataFrame:
     return pl.read_csv(latest_file)
 
 
-def get_fields_by_resource() -> dict[str, list[str]]:
+def _get_fields_by_resource() -> dict[str, list[str]]:
     """Gets a mapping from resource names to a list of field names in that resource."""
     properties = so.read_properties(so.parse_source("datapackage.json"))
     return dict(
@@ -41,7 +41,7 @@ def get_fields_by_resource() -> dict[str, list[str]]:
     )
 
 
-def select_with_base_cols(
+def _select_with_base_cols(
     raw_df: pl.DataFrame, resource_name: str, cols: list[str]
 ) -> pl.DataFrame:
     """Selects columns and adds base columns common to all dataframes."""
@@ -57,19 +57,19 @@ def select_with_base_cols(
 
 def raw_to_staged(raw_df: pl.DataFrame) -> list[pl.DataFrame]:
     """Transforms the raw data into staged dataframes."""
-    resources = get_fields_by_resource()
+    resources = _get_fields_by_resource()
 
     # Resources with special handling
     resources.pop("vas", None)
-    dfs = [create_vas_df(raw_df)]
+    dfs = [_create_vas_df(raw_df)]
 
     # Resources without special handling
     return dfs + so.pairwise_fmap(
-        list(resources.items()), [raw_df], create_df_for_resource
+        list(resources.items()), [raw_df], _create_df_for_resource
     )
 
 
-def create_df_for_resource(
+def _create_df_for_resource(
     resource_entry: tuple[str, list[str]], raw_df: pl.DataFrame
 ) -> pl.DataFrame:
     """Creates a dataframe for a resource."""
@@ -79,10 +79,10 @@ def create_df_for_resource(
     field_names.remove("event")
     field_names.remove("center")
 
-    return select_with_base_cols(raw_df, resource_name, field_names)
+    return _select_with_base_cols(raw_df, resource_name, field_names)
 
 
-def create_vas_df(raw_df: pl.DataFrame) -> pl.DataFrame:
+def _create_vas_df(raw_df: pl.DataFrame) -> pl.DataFrame:
     """Creates a dataframe for the VAS resource."""
     vas_cols = so.keep(
         raw_df.columns,
@@ -100,12 +100,12 @@ def create_vas_df(raw_df: pl.DataFrame) -> pl.DataFrame:
         cols_grouped_by_time.setdefault(int(time), []).append(col)
 
     vas_dfs = so.pairwise_fmap(
-        list(cols_grouped_by_time.items()), [raw_df], create_df_for_time_group
+        list(cols_grouped_by_time.items()), [raw_df], _create_df_for_time_group
     )
     return pl.concat(vas_dfs, how="vertical")
 
 
-def create_df_for_time_group(
+def _create_df_for_time_group(
     time_group: tuple[int, list[str]], raw_df: pl.DataFrame
 ) -> pl.DataFrame:
     """Creates a dataframe for a group of VAS columns with the same time."""
@@ -117,7 +117,7 @@ def create_df_for_time_group(
     }
 
     return (
-        select_with_base_cols(raw_df, "vas", cols)
+        _select_with_base_cols(raw_df, "vas", cols)
         .rename(renamed_cols)
         .with_columns(pl.lit(time).alias("minutes_from_meal"))
     )