Stage 2: package logic as a wheel with unit tests

.gitignore

@@ -22,3 +22,5 @@ build/
 .DS_Store
 .env
 .env.local
+dist/
+.venv/

README.md

@@ -125,7 +125,16 @@ Batch aggregation and anomaly detection over valid readings only.
 
 ```
 windmill_classcell/
-├── databricks.yml          # Bundle config: variables, includes, targets
+├── databricks.yml          # Bundle config: variables, artifacts, includes, targets
+├── pyproject.toml          # Wheel build + dev dependencies
+├── src/windmill/           # Transformation logic, shipped as a wheel
+│   ├── schema.py           # Schema and domain constants
+│   ├── validation.py       # Silver: dedup and row-level validity
+│   └── aggregation.py      # Gold: daily statistics and anomaly detection
+├── tests/                  # Unit tests, local Spark, no cluster needed
+│   ├── conftest.py
+│   ├── test_validation.py
+│   └── test_aggregation.py
 ├── resources/              # One file per resource, <name>.<resource_type>.yml
 │   ├── bronze.schema.yml
 │   ├── silver.schema.yml
@@ -159,7 +168,37 @@ windmill_classcell/
 
 ## Assumptions
 
-1. Data is hourly per turbine.
+1. **The feed delivers exactly one reading per turbine per hour, on the hour.**
+   True of the provided data — all 11,160 timestamps land on `:00:00`, there are
+   no duplicate `(timestamp, turbine_id)` pairs, and every one of the 465
+   turbine-days has exactly 24 readings.
+
+   This is load-bearing in two places. Deduplication treats
+   `(timestamp, turbine_id)` as the natural key, so it collapses re-delivered
+   rows but correctly leaves genuinely distinct sub-hour readings alone. And
+   `avg_power_output` is an unweighted mean over readings, which is only right
+   while each hour contributes one sample.
+
+   If the feed ever sent finer-grained timestamps — say two readings at `00:15`
+   and `00:47` — nothing would break, but the daily mean would quietly reweight
+   toward the busier hours. Gold therefore emits `distinct_hours` alongside
+   `measurement_count` and sets **`has_sub_hourly_readings`** when they disagree,
+   so the condition is visible instead of silent.
+
+   The right correction depends on what the sensor actually emits, which cannot
+   be determined from the data:
+
+   | If each reading is… | Correct treatment |
+   |---|---|
+   | An instantaneous sample | Average them (what the pipeline does today) |
+   | A cumulative counter | Take the last value in the hour, and difference it |
+   | A pre-averaged period value | Weight by the period each covers |
+
+   The spec states output is measured in **megawatts** — a rate, not an energy
+   total — which points at the first reading and makes averaging defensible. But
+   that is inference, not confirmation. Rather than guess at a resampling rule,
+   the pipeline holds the assumption explicitly and flags violations; the rule
+   can be implemented once someone who knows the sensor confirms the semantics.
 2. 5 turbines per CSV file (3 files = 15 turbines), and a turbine always appears
    in the same file — so the filename is a usable source of group identity.
 3. **Invalid readings are flagged, not dropped or imputed.** Nulls and negatives
@@ -200,10 +239,27 @@ and boundary values that must stay valid. Writes to `test_bronze` / `test_silver
 
 See `docs/testing.md` for the defect table, expected counts, and known gaps.
 
-Validation runs as a Databricks job rather than local pytest, so it needs no
-local Spark install — the recruiter runs one `bundle run` and the job fails on any
-violated assertion. A unit-testable transformation layer is introduced on the
-`medallion_whl` branch.
+### Unit tests
+
+The transformation logic lives in `src/windmill/` as pure DataFrame transforms —
+no I/O, no `dbutils`, no session handling — so it runs against small local
+DataFrames without a cluster. 41 tests, ~11 seconds.
+
+```bash
+python3 -m venv .venv
+.venv/bin/pip install -e ".[dev]"
+.venv/bin/pytest
+```
+
+Requires a JDK (17 or later) for local Spark. The suite unsets `SPARK_HOME`
+before starting a session, so a Spark distribution already installed on the
+machine cannot cause a version clash.
+
+The most important test is `test_baseline_is_built_from_daily_means_not_raw_readings`.
+It constructs data whose daily means are tightly clustered while the individual
+readings are spread wide, so the anomalous day is only detectable against the
+daily-mean baseline. It fails if the √24 defect is ever reintroduced — verified
+by injecting that defect and watching it fail, rather than assuming.
 
 ## Cleanup
 

databricks.yml

@@ -10,6 +10,25 @@ bundle:
 include:
   - resources/*.yml
 
+# The transformation logic ships as a wheel rather than being inlined in the
+# notebooks, so it can be unit tested locally without a cluster. `bundle deploy`
+# runs this build and uploads the artifact; jobs attach it via `libraries`.
+artifacts:
+  windmill_wheel:
+    type: whl
+    build: python3 -m build --wheel
+    path: .
+
+# Applied to every target; each target adds its own environment tag and prefix.
+# Deployments are otherwise anonymous -- the branch and commit tags are what let
+# you answer "which version of the code produced these tables?" from the Jobs UI
+# without guessing.
+presets:
+  tags:
+    project: windmill
+    git_branch: ${bundle.git.branch}
+    git_commit: ${bundle.git.commit}
+
 variables:
   catalog_name:
     description: Catalog name
@@ -41,7 +60,6 @@ targets:
       name_prefix: "[dev ${workspace.current_user.short_name}] "
       tags:
         environment: dev
-        project: windmill
     variables:
       data_dir: data
 
@@ -59,6 +77,5 @@ targets:
       name_prefix: "[test ${workspace.current_user.short_name}] "
       tags:
         environment: test
-        project: windmill
     variables:
       data_dir: data_test

docs/testing.md

@@ -163,8 +163,38 @@ Known gaps, roughly in priority order:
 - **Late-arriving data beyond the watermark** — a row older than the 2-day
   watermark is dropped by design. Worth an explicit test so the behaviour is
   recorded as intentional rather than discovered later.
-- **Unit tests for `src/`** — `tests/` currently only asserts that functions are
-  callable. The transformation helpers should be tested directly against small
-  local DataFrames, which is much faster than a full pipeline run.
 - **Idempotency of gold** — gold is `overwrite`, so re-running should be a no-op.
   Untested.
+
+## Unit tests
+
+The integration harness above proves the pipeline end to end, but a full run
+costs minutes and needs a workspace. The transformation logic is packaged in
+`src/windmill/` as pure DataFrame transforms, so it is also tested directly
+against small local DataFrames: 41 tests in ~11 seconds, no cluster.
+
+```bash
+.venv/bin/pytest
+```
+
+The two layers cover different things and neither replaces the other. Unit tests
+cover statistical behaviour and edge cases that are awkward to express as
+fixtures — null spread, single-day turbines, boundary values, per-turbine versus
+fleet-wide baselines. The integration harness covers everything the package
+cannot see: Auto Loader, checkpoints, streaming dedup across real re-runs, and
+the wiring between jobs.
+
+### Mutation-checked
+
+`test_baseline_is_built_from_daily_means_not_raw_readings` guards the √24 defect.
+Rather than assume it works, the defect was reinjected into `build_gold` and the
+suite re-run.
+
+The first attempt **passed with the defect present** — the test called
+`turbine_baseline()` directly, so it guarded the unit while the defect lived in
+how `build_gold` wired it together. Every unit test passed; the composed path was
+wrong. The test now asserts through `build_gold` as well, and fails with the
+defect injected.
+
+Worth recording because it is the failure mode unit tests are most prone to:
+correct components, wrong composition.

notebooks/02_bronze_ingestion.py

@@ -2,23 +2,15 @@
 # Bronze layer: Stream raw turbine data with Auto Loader (cloudFiles)
 
 from pyspark.sql import SparkSession
-from pyspark.sql.types import StructType, StructField, DoubleType, TimestampType, IntegerType
 from pyspark.sql.functions import current_timestamp, col
 
+from windmill.schema import TURBINE_SCHEMA
+
 catalog_name = dbutils.widgets.get("catalog_name")
 schema_name = dbutils.widgets.get("schema_name")
 
 spark = SparkSession.builder.appName("bronze_ingestion").getOrCreate()
 
-# Explicit schema (no inference for streaming reliability)
-turbine_schema = StructType([
-    StructField("timestamp", TimestampType(), True),
-    StructField("turbine_id", IntegerType(), True),
-    StructField("wind_speed", DoubleType(), True),
-    StructField("wind_direction", DoubleType(), True),
-    StructField("power_output", DoubleType(), True),
-])
-
 volume_path = f"/Volumes/{catalog_name}/{schema_name}/dropzone/inbox"
 archive_path = f"/Volumes/{catalog_name}/{schema_name}/dropzone/archive"
 
@@ -36,7 +28,7 @@
     .option("cloudFiles.cleanSource", "MOVE")
     .option("cloudFiles.cleanSource.moveDestination", archive_path)
     .option("cloudFiles.cleanSource.retentionDuration", "7 days")
-    .schema(turbine_schema)
+    .schema(TURBINE_SCHEMA)
     .load(volume_path)
     .withColumn("ingestion_timestamp", current_timestamp())
     .withColumn("source_file", col("_metadata.file_path"))

notebooks/03_silver_transform.py

@@ -1,8 +1,13 @@
 # Databricks notebook source
-# Silver layer: Clean & enrich turbine data incrementally from bronze stream
+# Silver layer: clean and enrich turbine data incrementally from bronze.
+#
+# Orchestration only -- read, transform, write. The transform itself lives in
+# the windmill package (installed as a wheel by the bundle) so it can be unit
+# tested against small local DataFrames without a cluster.
 
 from pyspark.sql import SparkSession
-from pyspark.sql.functions import col, when, regexp_extract, date_trunc, lit
+
+from windmill.validation import transform_silver
 
 dbutils.widgets.text("catalog_name", "windmill_classcell")
 dbutils.widgets.text("bronze_schema", "bronze")
@@ -16,80 +21,22 @@
 
 bronze_table = f"{catalog}.{bronze_schema}.turbine_raw"
 silver_table = f"{catalog}.{silver_schema}.turbine_clean"
-# Checkpoint lives in bronze dropzone volume (only volume available)
-checkpoint_path = f"/Volumes/{catalog}/{bronze_schema}/dropzone/_checkpoints/silver_turbine_clean"
-
-# Incremental read from bronze Delta table
-df = spark.readStream.table(bronze_table)
-
-# 1. Deduplicate on (timestamp, turbine_id) — bounded state via watermark.
-#    Duplicates share identical event timestamp, so always within any window.
-df_dedup = (
-    df.withWatermark("timestamp", "2 days")
-      .dropDuplicatesWithinWatermark(["timestamp", "turbine_id"])
-)
-
-# 2. Extract turbine group from source_file (UNKNOWN if pattern misses)
-df_with_group = df_dedup.withColumn(
-    "turbine_group",
-    when(col("source_file").rlike(r"data_group_\d+"),
-         regexp_extract(col("source_file"), r"data_group_(\d+)", 1))
-    .otherwise("UNKNOWN")
+# Checkpoint lives in the bronze dropzone volume -- the only volume the bundle
+# provisions.
+checkpoint_path = (
+    f"/Volumes/{catalog}/{bronze_schema}/dropzone/_checkpoints/silver_turbine_clean"
 )
 
-# 3. Validate turbine IDs per group. Group mapping: 1→[1-5], 2→[6-10], 3→[11-15]
-df_validated = df_with_group.withColumn(
-    "is_turbine_id_valid",
-    when(col("turbine_group") == "UNKNOWN", False)
-    .when((col("turbine_group") == "1") & col("turbine_id").between(1, 5), True)
-    .when((col("turbine_group") == "2") & col("turbine_id").between(6, 10), True)
-    .when((col("turbine_group") == "3") & col("turbine_id").between(11, 15), True)
-    .otherwise(False)
-)
-
-# 4. Row-level validity checks.
-#    These are context-free — a single row is enough to judge them, so they belong
-#    here. Statistical outlier detection needs a population and lives in gold.
-#    Zero is a legitimate reading (a becalmed or idling turbine), so the bound is
-#    >= 0, not > 0. Null means the sensor reported nothing.
-df_checked = (
-    df_validated
-    .withColumn(
-        "is_power_output_valid",
-        col("power_output").isNotNull() & (col("power_output") >= 0),
-    )
-    .withColumn(
-        "is_wind_speed_valid",
-        col("wind_speed").isNotNull() & (col("wind_speed") >= 0),
-    )
-    .withColumn(
-        "is_wind_direction_valid",
-        col("wind_direction").isNotNull() & col("wind_direction").between(0, 360),
-    )
-)
-
-# 5. Single rollup flag — gold filters on this before computing statistics so a
-#    sensor-error reading cannot skew a turbine's baseline mean and stddev.
-df_checked = df_checked.withColumn(
-    "is_reading_valid",
-    col("is_turbine_id_valid")
-    & col("is_power_output_valid")
-    & col("is_wind_speed_valid")
-    & col("is_wind_direction_valid"),
-)
+df = spark.readStream.table(bronze_table)
 
-# 6. Add date column from timestamp
-df_silver = df_checked.withColumn("date", date_trunc("day", col("timestamp")))
+silver = transform_silver(df)
 
-# Invalid rows are flagged, not dropped — quarantining in place keeps them
-# available for quality reporting and root-cause work on the sensors.
-# Incremental append to silver (statistical outlier detection deferred to gold)
 (
-    df_silver.writeStream
-    .option("checkpointLocation", checkpoint_path)
+    silver.writeStream.option("checkpointLocation", checkpoint_path)
     .outputMode("append")
     .trigger(availableNow=True)
     .toTable(silver_table)
+    .awaitTermination()
 )
 
 print(f"✓ Streamed to {silver_table}")