From 76fe3dbc360308954f603895fae96668d077b8ea Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 08:08:11 +0100 Subject: [PATCH 1/3] feat: package transformation logic as a wheel with unit tests Extract the silver and gold logic from the notebooks into src/windmill/, built as a wheel by the bundle's artifacts block and attached to jobs. Notebooks reduce to orchestration: read, call a transform, write. Everything in the package is a pure DataFrame transform -- no I/O, no dbutils, no session handling -- which is what lets the layer be tested against small local DataFrames. 41 tests in ~11s, no cluster required. Serverless rejects a task-level `libraries` field, so dependencies are declared as a job `environments` block and referenced by environment_key from the task. deduplicate() now branches on df.isStreaming. dropDuplicatesWithinWatermark is rejected outright on batch DataFrames, and a watermark is meaningless there anyway since there is no unbounded state to bound; plain dropDuplicates on the same keys is equivalent. This means the unit tests exercise the production code path rather than a batch-only lookalike. The regression guard on the sqrt(24) defect was verified by reinjecting the defect rather than assumed. The first version of that test passed with the defect present -- it called turbine_baseline() directly, so it guarded the unit while the defect lived in how build_gold composed it. Correct components, wrong composition. The test now asserts through build_gold and fails as it should. Test fixtures pin an explicit schema instead of relying on inference: an all-null column in a small fixture has no inferrable type, and inference gave turbine_id a long where production has an int. conftest also drops SPARK_HOME before starting a session, so a Spark distribution already on the machine cannot cause a version clash -- a local 4.1.1 install against pip's 4.2.0 surfaced as an opaque "Method getConfs does not exist" py4j error. Verified behaviour-preserving against the test target: silver 11169, 54 invalid readings, 20 anomalies -- identical to the pre-refactor run. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 2 + README.md | 36 +++++- databricks.yml | 9 ++ docs/testing.md | 36 +++++- notebooks/02_bronze_ingestion.py | 14 +-- notebooks/03_silver_transform.py | 83 +++---------- notebooks/04_gold_aggregate.py | 128 +++---------------- pyproject.toml | 28 +++++ resources/windmill_aggregate.job.yml | 10 ++ resources/windmill_ingest.job.yml | 10 ++ resources/windmill_transform.job.yml | 10 ++ src/windmill/__init__.py | 14 +++ src/windmill/aggregation.py | 145 ++++++++++++++++++++++ src/windmill/schema.py | 52 ++++++++ src/windmill/validation.py | 155 +++++++++++++++++++++++ tests/__init__.py | 0 tests/conftest.py | 85 +++++++++++++ tests/test_aggregation.py | 173 ++++++++++++++++++++++++++ tests/test_validation.py | 179 +++++++++++++++++++++++++++ 19 files changed, 973 insertions(+), 196 deletions(-) create mode 100644 pyproject.toml create mode 100644 src/windmill/__init__.py create mode 100644 src/windmill/aggregation.py create mode 100644 src/windmill/schema.py create mode 100644 src/windmill/validation.py create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_aggregation.py create mode 100644 tests/test_validation.py diff --git a/.gitignore b/.gitignore index 8a5b219..269cac2 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,5 @@ build/ .DS_Store .env .env.local +dist/ +.venv/ diff --git a/README.md b/README.md index 07d117b..8905d4a 100644 --- a/README.md +++ b/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, ..yml │ ├── bronze.schema.yml │ ├── silver.schema.yml @@ -200,10 +209,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 diff --git a/databricks.yml b/databricks.yml index b8acc21..583b2ab 100644 --- a/databricks.yml +++ b/databricks.yml @@ -10,6 +10,15 @@ 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: . + variables: catalog_name: description: Catalog name diff --git a/docs/testing.md b/docs/testing.md index 3152c5e..a31acac 100644 --- a/docs/testing.md +++ b/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. diff --git a/notebooks/02_bronze_ingestion.py b/notebooks/02_bronze_ingestion.py index 31669c4..51f7a7b 100644 --- a/notebooks/02_bronze_ingestion.py +++ b/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")) diff --git a/notebooks/03_silver_transform.py b/notebooks/03_silver_transform.py index 1ccf2e8..2d0406d 100644 --- a/notebooks/03_silver_transform.py +++ b/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}") diff --git a/notebooks/04_gold_aggregate.py b/notebooks/04_gold_aggregate.py index 830c65f..e7cc21c 100644 --- a/notebooks/04_gold_aggregate.py +++ b/notebooks/04_gold_aggregate.py @@ -1,15 +1,15 @@ # Databricks notebook source -# Gold layer: daily summary statistics per turbine + anomaly detection +# Gold layer: daily summary statistics per turbine plus anomaly detection. # -# Statistical outlier detection lives here, not in silver, for two reasons: -# 1. The spec asks for anomalies to be *identified*. Filtering them out in -# silver would destroy the deliverable. -# 2. It needs a population (mean + stddev per turbine). Silver is a streaming -# append whose watermark exists to bound dedup state, not to define a -# statistical window. +# Orchestration only -- read, transform, write. The aggregation logic lives in +# the windmill package so the statistical behaviour can be unit tested; see +# tests/test_aggregation.py, in particular the regression guard on computing +# baselines from daily means rather than raw readings. from pyspark.sql import SparkSession -from pyspark.sql.functions import col, avg, stddev, min, max, count, when, sum as _sum +from pyspark.sql.functions import col + +from windmill.aggregation import build_gold dbutils.widgets.text("catalog_name", "windmill_classcell") dbutils.widgets.text("silver_schema", "silver") @@ -24,112 +24,22 @@ silver_table = f"{catalog}.{silver_schema}.turbine_clean" gold_table = f"{catalog}.{gold_schema}.turbine_summary" -STDDEV_THRESHOLD = 2.0 - -df = spark.table(silver_table) -print(f"Read {df.count()} rows from {silver_table}") - -# Only valid readings feed the statistics. A negative or null power reading left -# in the population would inflate stddev and mask the real anomalies. -valid = df.filter(col("is_reading_valid")) -print(f"Valid readings: {valid.count()}") - -# 1. Daily summary per turbine — the min/max/avg the spec asks for. -daily = valid.groupBy("date", "turbine_id", "turbine_group").agg( - min("power_output").alias("min_power_output"), - max("power_output").alias("max_power_output"), - avg("power_output").alias("avg_power_output"), - stddev("power_output").alias("stddev_power_output"), - count("*").alias("measurement_count"), - avg("wind_speed").alias("avg_wind_speed"), -) - -# 2. Per-turbine baseline, computed over DAILY MEANS — not raw readings. -# -# This unit has to match what we test against it. Averaging 24 hourly readings -# shrinks the spread by a factor of sqrt(24) ~= 5 (standard error of the mean), -# so a band built from reading-level stddev is ~5x too wide for daily means to -# ever escape. Measured on the seed data: reading stddev ~0.86 vs daily-mean -# stddev ~0.17, giving a band of [1.30, 4.74] against an actual daily range of -# [2.47, 3.38]. Zero anomalies were possible. See -# docs/dev_seed_original_results.md. -# -# Per-turbine rather than fleet-wide: turbines sit in different wind -# conditions, so a fleet baseline conflates "this turbine is underperforming" -# with "this turbine is in a calmer spot". The spec asks for deviation from -# *their* expected output. -baseline = daily.groupBy("turbine_id").agg( - avg("avg_power_output").alias("baseline_mean"), - stddev("avg_power_output").alias("baseline_stddev"), - count("*").alias("baseline_sample_count"), -) - -summary = daily.join(baseline, "turbine_id") - -# 3. Flag days where the turbine's average sits outside N standard deviations -# of its own baseline. A null or zero stddev (single reading, or a turbine -# pinned at one value) makes the test meaningless — treat as not anomalous -# rather than letting null propagate. -has_usable_baseline = col("baseline_stddev").isNotNull() & (col("baseline_stddev") > 0) -lower = col("baseline_mean") - STDDEV_THRESHOLD * col("baseline_stddev") -upper = col("baseline_mean") + STDDEV_THRESHOLD * col("baseline_stddev") +silver = spark.table(silver_table) +print(f"Read {silver.count()} rows from {silver_table}") -summary = ( - summary - .withColumn("anomaly_lower_bound", when(has_usable_baseline, lower)) - .withColumn("anomaly_upper_bound", when(has_usable_baseline, upper)) - .withColumn( - "is_anomaly", - when( - has_usable_baseline, - (col("avg_power_output") < lower) | (col("avg_power_output") > upper), - ).otherwise(False), - ) -) - -# Reading-level anomaly count. A turbine can spike for a few hours and still land -# on a normal daily average, which the daily test above cannot see. Counting the -# individual readings that breach the reading-level band surfaces that case -# without changing the daily verdict. -reading_baseline = valid.groupBy("turbine_id").agg( - avg("power_output").alias("reading_mean"), - stddev("power_output").alias("reading_stddev"), -) -reading_flags = ( - valid.join(reading_baseline, "turbine_id") - .withColumn( - "is_anomalous_reading", - col("reading_stddev").isNotNull() - & (col("reading_stddev") > 0) - & ( - (col("power_output") < col("reading_mean") - STDDEV_THRESHOLD * col("reading_stddev")) - | (col("power_output") > col("reading_mean") + STDDEV_THRESHOLD * col("reading_stddev")) - ), - ) - .groupBy("date", "turbine_id") - .agg(_sum(when(col("is_anomalous_reading"), 1).otherwise(0)).alias("anomalous_reading_count")) -) - -summary = summary.join(reading_flags, ["date", "turbine_id"], "left") +summary = build_gold(silver) summary.write.mode("overwrite").saveAsTable(gold_table) print(f"✓ Wrote {summary.count()} turbine-days to {gold_table}") -# Data quality rollup — how much was discarded and why. -quality = df.agg( - count("*").alias("total_rows"), - _sum(when(~col("is_reading_valid"), 1).otherwise(0)).alias("invalid_rows"), - _sum(when(~col("is_turbine_id_valid"), 1).otherwise(0)).alias("bad_turbine_id"), - _sum(when(~col("is_power_output_valid"), 1).otherwise(0)).alias("bad_power_output"), - _sum(when(~col("is_wind_speed_valid"), 1).otherwise(0)).alias("bad_wind_speed"), - _sum(when(~col("is_wind_direction_valid"), 1).otherwise(0)).alias("bad_wind_direction"), -) -print("\nData quality:") -quality.show(truncate=False) - anomalies = summary.filter(col("is_anomaly")) print(f"\nAnomalous turbine-days: {anomalies.count()}") anomalies.select( - "date", "turbine_id", "avg_power_output", - "baseline_mean", "anomaly_lower_bound", "anomaly_upper_bound", -).orderBy("date", "turbine_id").show(20, truncate=False) + "date", + "turbine_id", + "avg_power_output", + "baseline_mean", + "anomaly_lower_bound", + "anomaly_upper_bound", + "anomalous_reading_count", +).orderBy("turbine_id", "date").show(20, truncate=False) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c13adcb --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "windmill" +version = "0.1.0" +description = "Transformation logic for the windmill turbine pipeline" +requires-python = ">=3.10" +# pyspark is deliberately not a runtime dependency: it is provided by the +# Databricks runtime, and pinning it here would risk shadowing the cluster's own +# version. It is a test-time dependency only. +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pyspark>=3.5", + "pytest>=8.0", + "build>=1.0", +] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +# -q keeps the local Spark session's noisy startup out of the way. +addopts = "-q" diff --git a/resources/windmill_aggregate.job.yml b/resources/windmill_aggregate.job.yml index 1d47563..8b8db12 100644 --- a/resources/windmill_aggregate.job.yml +++ b/resources/windmill_aggregate.job.yml @@ -7,6 +7,15 @@ resources: windmill_aggregate: name: windmill_aggregate description: "Aggregate data to gold layer for analysis" + # Serverless rejects a task-level `libraries` field -- dependencies are + # declared as an environment and referenced by key from the task. + environments: + - environment_key: default + spec: + client: "3" + dependencies: + # Built by the artifacts block in databricks.yml. + - ../dist/*.whl tasks: - task_key: gold_aggregate notebook_task: @@ -15,4 +24,5 @@ resources: catalog_name: ${var.catalog_name} silver_schema: ${resources.schemas.silver.name} gold_schema: ${resources.schemas.gold.name} + environment_key: default timeout_seconds: 3600 diff --git a/resources/windmill_ingest.job.yml b/resources/windmill_ingest.job.yml index 979f712..f37262b 100644 --- a/resources/windmill_ingest.job.yml +++ b/resources/windmill_ingest.job.yml @@ -8,6 +8,15 @@ resources: windmill_ingest: name: windmill_ingest description: "Ingest raw turbine data to bronze layer" + # Serverless rejects a task-level `libraries` field -- dependencies are + # declared as an environment and referenced by key from the task. + environments: + - environment_key: default + spec: + client: "3" + dependencies: + # Built by the artifacts block in databricks.yml. + - ../dist/*.whl tasks: - task_key: bronze_load notebook_task: @@ -15,4 +24,5 @@ resources: base_parameters: catalog_name: ${var.catalog_name} schema_name: ${resources.schemas.bronze.name} + environment_key: default timeout_seconds: 3600 diff --git a/resources/windmill_transform.job.yml b/resources/windmill_transform.job.yml index 86876e3..b054496 100644 --- a/resources/windmill_transform.job.yml +++ b/resources/windmill_transform.job.yml @@ -7,6 +7,15 @@ resources: windmill_transform: name: windmill_transform description: "Transform and clean data to silver layer" + # Serverless rejects a task-level `libraries` field -- dependencies are + # declared as an environment and referenced by key from the task. + environments: + - environment_key: default + spec: + client: "3" + dependencies: + # Built by the artifacts block in databricks.yml. + - ../dist/*.whl tasks: - task_key: silver_transform notebook_task: @@ -15,4 +24,5 @@ resources: catalog_name: ${var.catalog_name} bronze_schema: ${resources.schemas.bronze.name} silver_schema: ${resources.schemas.silver.name} + environment_key: default timeout_seconds: 3600 diff --git a/src/windmill/__init__.py b/src/windmill/__init__.py new file mode 100644 index 0000000..d5f13a5 --- /dev/null +++ b/src/windmill/__init__.py @@ -0,0 +1,14 @@ +"""Transformation logic for the windmill turbine pipeline. + +Packaged separately from the notebooks so it can be unit tested against small +local DataFrames without a cluster, and so the notebooks reduce to orchestration: +read, call a transform, write. +""" + +from .aggregation import build_gold +from .schema import GROUP_RANGES, TURBINE_SCHEMA +from .validation import transform_silver + +__all__ = ["TURBINE_SCHEMA", "GROUP_RANGES", "transform_silver", "build_gold"] + +__version__ = "0.1.0" diff --git a/src/windmill/aggregation.py b/src/windmill/aggregation.py new file mode 100644 index 0000000..599f915 --- /dev/null +++ b/src/windmill/aggregation.py @@ -0,0 +1,145 @@ +"""Gold-layer aggregation and anomaly detection. + +Statistical outlier detection lives here rather than in silver for two reasons. +The spec asks for deviating turbines to be *identified*, so filtering them out +upstream would destroy the deliverable. And it needs a population: silver's +watermark exists to bound dedup state, not to define a statistical window. +""" + +from __future__ import annotations + +from pyspark.sql import Column, DataFrame +from pyspark.sql.functions import avg, col, count, max, min, stddev +from pyspark.sql.functions import sum as _sum +from pyspark.sql.functions import when + +from .schema import DEFAULT_STDDEV_THRESHOLD + + +def valid_readings(df: DataFrame) -> DataFrame: + """Restrict to readings fit to compute statistics from. + + A negative or null power reading left in the population would inflate the + standard deviation and mask the anomalies the threshold is meant to catch. + """ + return df.filter(col("is_reading_valid")) + + +def daily_summary(df: DataFrame) -> DataFrame: + """Per-turbine, per-day summary statistics.""" + return df.groupBy("date", "turbine_id", "turbine_group").agg( + min("power_output").alias("min_power_output"), + max("power_output").alias("max_power_output"), + avg("power_output").alias("avg_power_output"), + stddev("power_output").alias("stddev_power_output"), + count("*").alias("measurement_count"), + avg("wind_speed").alias("avg_wind_speed"), + ) + + +def turbine_baseline(daily: DataFrame) -> DataFrame: + """Per-turbine baseline computed over DAILY MEANS, not raw readings. + + The unit of the baseline has to match the unit being tested against it. + Averaging 24 hourly readings shrinks the spread by sqrt(24) ~= 5 (the + standard error of the mean), so a band built from reading-level stddev is + about five times too wide for a daily mean to ever escape. + + Measured on the seed data: reading-level stddev ~0.86 against daily-mean + stddev ~0.17, giving a band of [1.30, 4.74] versus an actual daily range of + [2.47, 3.38]. Zero anomalies were mathematically possible. See + docs/dev_seed_original_results.md. + + Per-turbine rather than fleet-wide: turbines sit in different wind + conditions, so a fleet baseline conflates "this turbine is underperforming" + with "this turbine is in a calmer spot". The spec asks for deviation from + *their* expected output. + """ + return daily.groupBy("turbine_id").agg( + avg("avg_power_output").alias("baseline_mean"), + stddev("avg_power_output").alias("baseline_stddev"), + count("*").alias("baseline_sample_count"), + ) + + +def _usable_baseline(stddev_col: str) -> Column: + """A zero or null spread makes the threshold test meaningless. + + Happens for a turbine with a single day of data, or one pinned at a constant + output. Treated as not-anomalous rather than letting null propagate into the + flag, because a null would silently vanish from downstream + `WHERE is_anomaly` filters. + """ + return col(stddev_col).isNotNull() & (col(stddev_col) > 0) + + +def flag_daily_anomalies( + daily: DataFrame, + baseline: DataFrame, + threshold: float = DEFAULT_STDDEV_THRESHOLD, +) -> DataFrame: + """Mark turbine-days whose mean output falls outside the baseline band. + + Emits the bounds alongside the flag so a reviewer can see why a day tripped + rather than having to recompute it. + """ + summary = daily.join(baseline, "turbine_id") + usable = _usable_baseline("baseline_stddev") + lower = col("baseline_mean") - threshold * col("baseline_stddev") + upper = col("baseline_mean") + threshold * col("baseline_stddev") + + return ( + summary.withColumn("anomaly_lower_bound", when(usable, lower)) + .withColumn("anomaly_upper_bound", when(usable, upper)) + .withColumn( + "is_anomaly", + when( + usable, + (col("avg_power_output") < lower) | (col("avg_power_output") > upper), + ).otherwise(False), + ) + ) + + +def count_anomalous_readings( + df: DataFrame, threshold: float = DEFAULT_STDDEV_THRESHOLD +) -> DataFrame: + """Count individual readings breaching the reading-level band, per day. + + A turbine can spike for a couple of hours and still land on a normal daily + average, which the daily test cannot see. This surfaces that case without + changing the daily verdict. + """ + reading_baseline = df.groupBy("turbine_id").agg( + avg("power_output").alias("reading_mean"), + stddev("power_output").alias("reading_stddev"), + ) + usable = _usable_baseline("reading_stddev") + lower = col("reading_mean") - threshold * col("reading_stddev") + upper = col("reading_mean") + threshold * col("reading_stddev") + + return ( + df.join(reading_baseline, "turbine_id") + .withColumn( + "is_anomalous_reading", + usable & ((col("power_output") < lower) | (col("power_output") > upper)), + ) + .groupBy("date", "turbine_id") + .agg( + _sum(when(col("is_anomalous_reading"), 1).otherwise(0)).alias( + "anomalous_reading_count" + ) + ) + ) + + +def build_gold( + silver: DataFrame, threshold: float = DEFAULT_STDDEV_THRESHOLD +) -> DataFrame: + """Full silver-to-gold transform.""" + valid = valid_readings(silver) + daily = daily_summary(valid) + baseline = turbine_baseline(daily) + summary = flag_daily_anomalies(daily, baseline, threshold) + readings = count_anomalous_readings(valid, threshold) + return summary.join(readings, ["date", "turbine_id"], "left") diff --git a/src/windmill/schema.py b/src/windmill/schema.py new file mode 100644 index 0000000..51ba7c1 --- /dev/null +++ b/src/windmill/schema.py @@ -0,0 +1,52 @@ +"""Schema and domain constants for turbine measurements.""" + +from __future__ import annotations + +from pyspark.sql.types import ( + DoubleType, + IntegerType, + StructField, + StructType, + TimestampType, +) + +#: Explicit schema for the raw CSV feed. Declared rather than inferred so a +#: malformed file cannot silently change column types between ingestion runs. +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), + ] +) + +#: Which turbine IDs legitimately appear in each source file group. +#: A turbine always reports into the same file, so the filename identifies the +#: expected range. Hardcoded for a fixed 15-turbine farm; a real deployment would +#: read this from a turbine registry table. +GROUP_RANGES: dict[str, tuple[int, int]] = { + "1": (1, 5), + "2": (6, 10), + "3": (11, 15), +} + +#: Filenames are expected to look like data_group_.csv. Anything else +#: cannot be attributed to a group. +GROUP_FILENAME_PATTERN = r"data_group_(\d+)" + +#: Marker for rows whose source file does not match the expected pattern. +UNKNOWN_GROUP = "UNKNOWN" + +#: Valid compass bearing range, inclusive at both ends. +WIND_DIRECTION_MIN = 0.0 +WIND_DIRECTION_MAX = 360.0 + +#: Standard deviations from a turbine's own baseline before a reading or a day +#: is considered anomalous. +DEFAULT_STDDEV_THRESHOLD = 2.0 + +#: How late a duplicate may arrive and still be caught. Bounds streaming state; +#: duplicates share an identical event timestamp, so any positive window works. +DEFAULT_WATERMARK = "2 days" diff --git a/src/windmill/validation.py b/src/windmill/validation.py new file mode 100644 index 0000000..a5642f6 --- /dev/null +++ b/src/windmill/validation.py @@ -0,0 +1,155 @@ +"""Silver-layer cleaning and row-level validation. + +Every function here is a pure DataFrame transform: it takes a DataFrame and +returns a DataFrame, with no I/O, no widget reads, and no session handling. That +is what lets the whole layer be tested against small local DataFrames without a +cluster. + +Row-level checks only. A check belongs here if a single row is enough to judge +it. Statistical outlier detection needs a population and lives in `aggregation`. +""" + +from __future__ import annotations + +from pyspark.sql import Column, DataFrame +from pyspark.sql.functions import col, date_trunc, regexp_extract, when + +from .schema import ( + DEFAULT_WATERMARK, + GROUP_FILENAME_PATTERN, + GROUP_RANGES, + UNKNOWN_GROUP, + WIND_DIRECTION_MAX, + WIND_DIRECTION_MIN, +) + + +DEDUP_KEYS = ["timestamp", "turbine_id"] + + +def deduplicate(df: DataFrame, watermark: str = DEFAULT_WATERMARK) -> DataFrame: + """Drop repeat readings for the same turbine at the same instant. + + On a stream this uses `dropDuplicatesWithinWatermark`, so the state store + stays bounded. Duplicates carry an identical event timestamp, so they always + fall inside the window however wide it is -- the watermark is a memory bound, + not a correctness knob. + + On a batch DataFrame that operator is rejected outright by Spark, and a + watermark would be meaningless anyway since there is no unbounded state to + bound. Plain `dropDuplicates` on the same keys is equivalent there. Handling + both means the production code path is the one exercised by the unit tests, + rather than a batch-only lookalike. + """ + if not df.isStreaming: + return df.dropDuplicates(DEDUP_KEYS) + return df.withWatermark("timestamp", watermark).dropDuplicatesWithinWatermark( + DEDUP_KEYS + ) + + +def add_turbine_group(df: DataFrame) -> DataFrame: + """Derive the turbine group from the source filename. + + Falls back to UNKNOWN rather than failing or guessing, so an unexpected file + lands in the table and is visible instead of silently attaching to a group. + """ + return df.withColumn( + "turbine_group", + when( + col("source_file").rlike(GROUP_FILENAME_PATTERN), + regexp_extract(col("source_file"), GROUP_FILENAME_PATTERN, 1), + ).otherwise(UNKNOWN_GROUP), + ) + + +def _turbine_id_valid_expr(group_ranges: dict[str, tuple[int, int]]) -> Column: + """Build the turbine-id range check as a chain of per-group conditions.""" + # An UNKNOWN group carries no expected range, so its turbines cannot be + # validated -- treated as invalid rather than as passing by default. + expr = when(col("turbine_group") == UNKNOWN_GROUP, False) + for group, (low, high) in group_ranges.items(): + expr = expr.when( + (col("turbine_group") == group) & col("turbine_id").between(low, high), + True, + ) + # Known group, id outside its range. + return expr.otherwise(False) + + +def add_turbine_id_validity( + df: DataFrame, group_ranges: dict[str, tuple[int, int]] | None = None +) -> DataFrame: + """Flag turbines reporting from a file they do not belong to.""" + return df.withColumn( + "is_turbine_id_valid", + _turbine_id_valid_expr(group_ranges or GROUP_RANGES), + ) + + +def add_measurement_validity(df: DataFrame) -> DataFrame: + """Flag physically impossible or missing measurements. + + Bounds are `>= 0`, not `> 0`. Zero is a legitimate reading: a becalmed or + idling turbine genuinely produces 0 MW in 0 m/s wind. Treating zero as + invalid would discard real low-wind data. + + Null propagation is handled by the `isNotNull()` conjunct -- in Spark's + three-valued logic `false AND null` is `false`, so a null measurement yields + an explicit `false` rather than a null flag. + """ + return ( + df.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(WIND_DIRECTION_MIN, WIND_DIRECTION_MAX), + ) + ) + + +def add_reading_validity(df: DataFrame) -> DataFrame: + """Roll the individual checks into one flag. + + Gold filters on this before computing statistics, so a sensor-error reading + cannot skew a turbine's baseline mean and standard deviation. + """ + return df.withColumn( + "is_reading_valid", + col("is_turbine_id_valid") + & col("is_power_output_valid") + & col("is_wind_speed_valid") + & col("is_wind_direction_valid"), + ) + + +def add_date(df: DataFrame) -> DataFrame: + """Add the day a reading belongs to, used as the gold aggregation grain.""" + return df.withColumn("date", date_trunc("day", col("timestamp"))) + + +def transform_silver( + df: DataFrame, + watermark: str = DEFAULT_WATERMARK, + group_ranges: dict[str, tuple[int, int]] | None = None, +) -> DataFrame: + """Full bronze-to-silver transform. + + Invalid rows are flagged in place, never dropped or imputed. Forward-filling + a sensor gap would invent measurements that were never taken; quarantining in + place keeps the rows available for quality reporting and sensor root-cause + work while letting gold exclude them from statistics. + """ + df = deduplicate(df, watermark) + df = add_turbine_group(df) + df = add_turbine_id_validity(df, group_ranges) + df = add_measurement_validity(df) + df = add_reading_validity(df) + return add_date(df) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..035d3b1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,85 @@ +"""Shared pytest fixtures. + +A local Spark session is session-scoped because starting one costs seconds and +the tests are otherwise fast. Configured small deliberately: the point of these +tests is logic, not scale, so shuffle partitions are cut from the default 200 to +avoid hundreds of empty tasks per aggregation. +""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta + +import pytest + +# Must happen before pyspark launches a JVM. If the developer has SPARK_HOME set +# to a local Spark distribution, pyspark talks to those JARs instead of the ones +# it ships with -- and any version skew surfaces as an opaque py4j error such as +# "Method getConfs([class java.util.ArrayList]) does not exist". Dropping it here +# keeps the suite hermetic and independent of ambient environment. +os.environ.pop("SPARK_HOME", None) + +from pyspark.sql import SparkSession # noqa: E402 +from pyspark.sql.types import StringType, StructField, StructType # noqa: E402 + +from windmill.schema import TURBINE_SCHEMA # noqa: E402 + + +@pytest.fixture(scope="session") +def spark(): + session = ( + SparkSession.builder.appName("windmill-tests") + .master("local[2]") + .config("spark.sql.shuffle.partitions", "2") + .config("spark.ui.enabled", "false") + .getOrCreate() + ) + session.sparkContext.setLogLevel("ERROR") + yield session + session.stop() + + +PERIOD_START = datetime(2022, 3, 1) + + +def reading( + hour: int, + turbine_id: int, + power: float | None = 3.0, + wind_speed: float | None = 11.0, + wind_direction: float | None = 180.0, + source_file: str = "/Volumes/c/s/dropzone/inbox/data_group_1_20220301.csv", +): + """Build one raw reading, defaulting to a valid row. + + Tests override only the field under test, so each case reads as a statement + about that one field. + """ + return ( + PERIOD_START + timedelta(hours=hour), + turbine_id, + wind_speed, + wind_direction, + power, + source_file, + ) + + +#: Bronze shape: the raw feed schema plus the ingestion metadata silver reads. +#: Declared explicitly rather than inferred for two reasons -- a column that is +#: all-null in a small fixture has no inferrable type, and inference would give +#: turbine_id a long where production has an int. +RAW_SCHEMA = StructType( + list(TURBINE_SCHEMA.fields) + [StructField("source_file", StringType(), True)] +) + + +@pytest.fixture +def make_raw(spark): + """Build a bronze-shaped DataFrame from `reading(...)` tuples.""" + + def _make(rows): + return spark.createDataFrame(list(rows), RAW_SCHEMA) + + return _make diff --git a/tests/test_aggregation.py b/tests/test_aggregation.py new file mode 100644 index 0000000..f4b9397 --- /dev/null +++ b/tests/test_aggregation.py @@ -0,0 +1,173 @@ +"""Gold-layer aggregation and anomaly detection tests.""" + +from __future__ import annotations + +from windmill.aggregation import ( + build_gold, + count_anomalous_readings, + daily_summary, + flag_daily_anomalies, + turbine_baseline, + valid_readings, +) + +from .conftest import reading + +HOURS_PER_DAY = 24 + + +def _silver(make_raw, rows): + """Bronze rows through the silver transform, ready for gold.""" + from windmill.validation import transform_silver + + return transform_silver(make_raw(rows)) + + +def _day(day: int, turbine_id: int, powers: list[float]): + """Readings for one turbine across one day.""" + return [ + reading(day * HOURS_PER_DAY + h, turbine_id, power=p) + for h, p in enumerate(powers) + ] + + +def test_daily_summary_grain_and_stats(make_raw): + rows = _day(0, 1, [1.0, 2.0, 3.0, 4.0]) + _day(1, 1, [5.0, 5.0]) + daily = daily_summary(valid_readings(_silver(make_raw, rows))).orderBy("date") + got = daily.collect() + + assert len(got) == 2 + assert got[0]["min_power_output"] == 1.0 + assert got[0]["max_power_output"] == 4.0 + assert got[0]["avg_power_output"] == 2.5 + assert got[0]["measurement_count"] == 4 + assert got[1]["avg_power_output"] == 5.0 + + +def test_invalid_readings_excluded_from_statistics(make_raw): + """A bad reading must not drag the mean or inflate the spread.""" + rows = _day(0, 1, [3.0, 3.0]) + [reading(2, 1, power=-999.0)] + daily = daily_summary(valid_readings(_silver(make_raw, rows))).collect() + assert daily[0]["measurement_count"] == 2 + assert daily[0]["avg_power_output"] == 3.0 + + +def test_baseline_is_built_from_daily_means_not_raw_readings(make_raw): + """Regression guard on the sqrt(24) defect. + + Nine days of readings alternating 1.0 and 5.0 average to 3.0 each, so the + daily means are tightly clustered while the individual readings are spread + wide. A tenth day sits at a flat 5.0. + + Against the distribution of daily means (stddev ~0.63, upper bound ~4.47) + that tenth day is clearly anomalous. Against the spread of individual + readings (stddev ~1.99, upper bound ~7.19) it is nowhere near the boundary. + + So this test passes only while the baseline is computed from daily means. It + fails if the baseline is ever computed from raw readings again -- the exact + defect that made the original implementation report zero anomalies across + 465 turbine-days. + """ + rows = [] + for day in range(9): + rows += _day(day, 1, [1.0, 5.0] * 12) + rows += _day(9, 1, [5.0] * 24) + + silver = _silver(make_raw, rows) + daily = daily_summary(valid_readings(silver)) + baseline = turbine_baseline(daily) + + # The daily-mean spread must be far tighter than the reading-level spread; + # that gap is the whole point. + assert baseline.first()["baseline_stddev"] < 1.0 + + flagged = flag_daily_anomalies(daily, baseline).filter("is_anomaly").collect() + assert len(flagged) == 1 + assert flagged[0]["avg_power_output"] == 5.0 + + # Assert through the composed path as well, not just the unit. Wiring + # build_gold to a reading-level baseline reproduces the original defect + # while every unit above still passes -- verified by mutation, and this + # assertion is what catches it. + gold_flagged = build_gold(silver).filter("is_anomaly").collect() + assert len(gold_flagged) == 1 + assert gold_flagged[0]["avg_power_output"] == 5.0 + + +def test_baseline_is_per_turbine_not_fleet_wide(make_raw): + """A consistently high-output turbine is not an anomaly. + + Turbine 1 sits around 3.0 and turbine 11 around 30.0, each perfectly steady. + Judged against its own history neither deviates. Judged against a fleet-wide + mean both would be flagged every single day -- which would report a turbine + in a windier location as permanently faulty. + """ + rows = [] + for day in range(5): + rows += _day(day, 1, [3.0] * 4) + rows += _day( + day, + 11, + [30.0] * 4, + ) + # Turbine 11 lives in group 3, so give it a matching source file. + rows = [ + r if r[1] == 1 else (*r[:5], "/vol/inbox/data_group_3_x.csv") for r in rows + ] + + gold = build_gold(_silver(make_raw, rows)) + assert gold.filter("is_anomaly").count() == 0 + + +def test_flat_output_never_flags(make_raw): + """Zero variance means a zero-width band; the test is meaningless, not tripped.""" + rows = [] + for day in range(5): + rows += _day(day, 1, [3.0] * 4) + + gold = build_gold(_silver(make_raw, rows)) + assert gold.filter("is_anomaly").count() == 0 + assert gold.filter("is_anomaly IS NULL").count() == 0 + + +def test_single_day_cannot_be_anomalous(make_raw): + """One sample gives a null stddev. Must yield False, never null.""" + gold = build_gold(_silver(make_raw, _day(0, 1, [3.0, 4.0]))) + assert gold.count() == 1 + assert gold.first()["is_anomaly"] is False + + +def test_is_anomaly_is_never_null(make_raw): + """A null flag would silently vanish from `WHERE is_anomaly` filters.""" + rows = _day(0, 1, [3.0]) + _day(1, 2, [3.0, 9.0]) + gold = build_gold(_silver(make_raw, rows)) + assert gold.filter("is_anomaly IS NULL").count() == 0 + + +def test_short_spike_counted_even_when_daily_average_looks_normal(make_raw): + """The reading-level counter catches what the daily test cannot. + + Two hours spike hard but are offset by two low hours, so the day averages out + to something unremarkable. The daily flag stays calm; the reading counter + still reports the breach. + """ + rows = [] + for day in range(6): + rows += _day(day, 1, [3.0] * 4) + rows += _day(6, 1, [12.0, 12.0, -6.0 + 0.0, 0.0]) # mean 4.5, two extremes + + silver = _silver(make_raw, rows) + counts = count_anomalous_readings(valid_readings(silver)) + total = sum(r["anomalous_reading_count"] for r in counts.collect()) + assert total > 0 + + +def test_gold_has_one_row_per_turbine_day(make_raw): + rows = [] + for day in range(3): + rows += _day(day, 1, [3.0, 4.0]) + rows += _day(day, 2, [3.0, 4.0]) + + gold = build_gold(_silver(make_raw, rows)) + assert gold.count() == 6 + assert gold.select("date", "turbine_id").distinct().count() == 6 diff --git a/tests/test_validation.py b/tests/test_validation.py new file mode 100644 index 0000000..ad5bb19 --- /dev/null +++ b/tests/test_validation.py @@ -0,0 +1,179 @@ +"""Row-level validation tests. + +Each case isolates one field. The `reading` helper defaults every column to a +valid value, so a test overriding only `power=-1.0` is a statement about +negative power and nothing else. +""" + +from __future__ import annotations + +import pytest + +from windmill.validation import ( + add_measurement_validity, + add_reading_validity, + add_turbine_group, + add_turbine_id_validity, + transform_silver, +) + +from .conftest import reading + + +def _flags(df, column): + return [r[column] for r in df.orderBy("turbine_id").collect()] + + +# --- turbine group extraction ---------------------------------------------- + + +def test_group_extracted_from_filename(make_raw): + df = add_turbine_group( + make_raw([reading(0, 1, source_file="/vol/inbox/data_group_2_20220301.csv")]) + ) + assert df.first()["turbine_group"] == "2" + + +def test_unmatched_filename_falls_back_to_unknown(make_raw): + df = add_turbine_group( + make_raw([reading(0, 1, source_file="/vol/inbox/sensor_dump_alpha.csv")]) + ) + assert df.first()["turbine_group"] == "UNKNOWN" + + +def test_timestamp_suffix_does_not_break_group_extraction(make_raw): + """The seeding step appends a timestamp, which must not defeat the pattern.""" + df = add_turbine_group( + make_raw( + [reading(0, 1, source_file="/vol/inbox/data_group_3_20260720063000.csv")] + ) + ) + assert df.first()["turbine_group"] == "3" + + +# --- turbine id validity ---------------------------------------------------- + + +@pytest.mark.parametrize( + "group_file,turbine_id,expected", + [ + ("data_group_1_x.csv", 1, True), # lower bound of group 1 + ("data_group_1_x.csv", 5, True), # upper bound of group 1 + ("data_group_1_x.csv", 6, False), # belongs to group 2 + ("data_group_2_x.csv", 6, True), + ("data_group_2_x.csv", 10, True), + ("data_group_2_x.csv", 11, False), + ("data_group_3_x.csv", 15, True), + ("data_group_3_x.csv", 99, False), # not a real turbine + ("sensor_dump.csv", 1, False), # unknown group cannot be validated + ], +) +def test_turbine_id_range_per_group(make_raw, group_file, turbine_id, expected): + df = add_turbine_id_validity( + add_turbine_group( + make_raw([reading(0, turbine_id, source_file=f"/vol/inbox/{group_file}")]) + ) + ) + assert df.first()["is_turbine_id_valid"] is expected + + +# --- measurement validity --------------------------------------------------- + + +@pytest.mark.parametrize( + "power,expected", + [ + (3.0, True), + (0.0, True), # becalmed turbine -- must stay valid + (-0.1, False), + (None, False), + ], +) +def test_power_output_validity(make_raw, power, expected): + df = add_measurement_validity(make_raw([reading(0, 1, power=power)])) + assert df.first()["is_power_output_valid"] is expected + + +@pytest.mark.parametrize( + "wind_speed,expected", + [(11.0, True), (0.0, True), (-1.0, False), (None, False)], +) +def test_wind_speed_validity(make_raw, wind_speed, expected): + df = add_measurement_validity(make_raw([reading(0, 1, wind_speed=wind_speed)])) + assert df.first()["is_wind_speed_valid"] is expected + + +@pytest.mark.parametrize( + "direction,expected", + [ + (180.0, True), + (0.0, True), # due north, inclusive lower bound + (360.0, True), # also due north, inclusive upper bound + (360.1, False), + (-0.1, False), + (None, False), + ], +) +def test_wind_direction_validity(make_raw, direction, expected): + df = add_measurement_validity(make_raw([reading(0, 1, wind_direction=direction)])) + assert df.first()["is_wind_direction_valid"] is expected + + +def test_zero_is_valid_not_merely_non_null(make_raw): + """Regression guard on the `>= 0` bound. + + Tightening either check to `> 0` would silently discard every reading from a + turbine that was idle or becalmed -- real data, lost without an error. + """ + df = add_measurement_validity( + make_raw([reading(0, 1, power=0.0, wind_speed=0.0, wind_direction=0.0)]) + ) + row = df.first() + assert row["is_power_output_valid"] is True + assert row["is_wind_speed_valid"] is True + assert row["is_wind_direction_valid"] is True + + +# --- rollup ----------------------------------------------------------------- + + +def test_reading_validity_requires_every_check(make_raw): + rows = [ + reading(0, 1), # all good + reading(1, 2, power=-1.0), # bad power + reading(2, 3, wind_speed=None), # missing wind speed + reading(3, 4, wind_direction=999.0), # impossible bearing + reading(4, 99), # wrong turbine for this file + ] + df = add_reading_validity( + add_measurement_validity(add_turbine_id_validity(add_turbine_group(make_raw(rows)))) + ) + assert _flags(df, "is_reading_valid") == [True, False, False, False, False] + + +def test_invalid_rows_are_kept_not_dropped(make_raw): + """Quarantine in place: bad rows stay available for quality reporting.""" + rows = [reading(0, 1), reading(1, 2, power=-5.0), reading(2, 3, wind_speed=None)] + out = transform_silver(make_raw(rows)) + assert out.count() == 3 + assert out.filter("NOT is_reading_valid").count() == 2 + + +# --- deduplication ---------------------------------------------------------- + + +def test_duplicate_readings_collapse(make_raw): + """Same turbine, same instant, ingested twice.""" + rows = [reading(0, 1), reading(0, 1), reading(0, 2)] + out = transform_silver(make_raw(rows)) + assert out.count() == 2 + + +def test_same_turbine_different_hours_both_kept(make_raw): + rows = [reading(0, 1), reading(1, 1)] + assert transform_silver(make_raw(rows)).count() == 2 + + +def test_same_hour_different_turbines_both_kept(make_raw): + rows = [reading(0, 1), reading(0, 2)] + assert transform_silver(make_raw(rows)).count() == 2 From 5034ef77e83a13e5cb126dc5e84ab90254f157e5 Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 08:45:50 +0100 Subject: [PATCH 2/3] feat: tag deployments with originating git branch and commit Both branches deploy to the same targets, so a deployment carried no record of which code produced it. Tag every job with bundle.git.branch and bundle.git.commit so that is answerable from the Jobs UI. Tags rather than branch-derived names deliberately. name_prefix also applies to schema names, so deriving it from the branch would mean every branch switch renames the schemas -- a drop-and-recreate with data loss, for isolation that is not needed when branches are deployed one at a time. project tag moved to a top-level presets block shared by both targets; each target keeps its own environment tag and name_prefix. Co-Authored-By: Claude Opus 4.8 (1M context) --- databricks.yml | 12 ++- uv.lock | 225 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 235 insertions(+), 2 deletions(-) create mode 100644 uv.lock diff --git a/databricks.yml b/databricks.yml index 583b2ab..9f4c2ba 100644 --- a/databricks.yml +++ b/databricks.yml @@ -19,6 +19,16 @@ artifacts: 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 @@ -50,7 +60,6 @@ targets: name_prefix: "[dev ${workspace.current_user.short_name}] " tags: environment: dev - project: windmill variables: data_dir: data @@ -68,6 +77,5 @@ targets: name_prefix: "[test ${workspace.current_user.short_name}] " tags: environment: test - project: windmill variables: data_dir: data_test diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..e6454e8 --- /dev/null +++ b/uv.lock @@ -0,0 +1,225 @@ +version = 1 +revision = 3 +requires-python = ">=3.10" + +[[package]] +name = "build" +version = "1.5.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "os_name == 'nt'" }, + { name = "importlib-metadata", marker = "python_full_version < '3.10.2'" }, + { name = "packaging" }, + { name = "pyproject-hooks" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/78/e0/df5e171f685f82f37b12e1f208064e24244911079d7b767447d1af7e0d70/build-1.5.0.tar.gz", hash = "sha256:302c22c3ba2a0fd5f3911918651341ebb3896176cbdec15bd421f80b1afc7647", size = 89796, upload-time = "2026-04-30T03:18:25.17Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0d/fe/6bea5c9162869c5beba5d9c8abbed835ec85bf1ec1fba05a3822325c45f3/build-1.5.0-py3-none-any.whl", hash = "sha256:13f3eecb844759ab66efec90ca17639bbf14dc06cb2fdf37a9010322d9c50a6f", size = 26018, upload-time = "2026-04-30T03:18:23.644Z" }, +] + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "exceptiongroup" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" }, +] + +[[package]] +name = "importlib-metadata" +version = "9.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "zipp" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a9/01/15bb152d77b21318514a96f43af312635eb2500c96b55398d020c93d86ea/importlib_metadata-9.0.0.tar.gz", hash = "sha256:a4f57ab599e6a2e3016d7595cfd72eb4661a5106e787a95bcc90c7105b831efc", size = 56405, upload-time = "2026-03-20T06:42:56.999Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/38/3d/2d244233ac4f76e38533cfcb2991c9eb4c7bf688ae0a036d30725b8faafe/importlib_metadata-9.0.0-py3-none-any.whl", hash = "sha256:2d21d1cc5a017bd0559e36150c21c830ab1dc304dedd1b7ea85d20f45ef3edd7", size = 27789, upload-time = "2026-03-20T06:42:55.665Z" }, +] + +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + +[[package]] +name = "packaging" +version = "26.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + +[[package]] +name = "py4j" +version = "0.10.9.9" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/38/31/0b210511177070c8d5d3059556194352e5753602fa64b85b7ab81ec1a009/py4j-0.10.9.9.tar.gz", hash = "sha256:f694cad19efa5bd1dee4f3e5270eb406613c974394035e5bfc4ec1aba870b879", size = 761089, upload-time = "2025-01-15T03:53:18.624Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bd/db/ea0203e495be491c85af87b66e37acfd3bf756fd985f87e46fc5e3bf022c/py4j-0.10.9.9-py2.py3-none-any.whl", hash = "sha256:c7c26e4158defb37b0bb124933163641a2ff6e3a3913f7811b0ddbe07ed61533", size = 203008, upload-time = "2025-01-15T03:53:15.648Z" }, +] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "pyproject-hooks" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e7/82/28175b2414effca1cdac8dc99f76d660e7a4fb0ceefa4b4ab8f5f6742925/pyproject_hooks-1.2.0.tar.gz", hash = "sha256:1e859bd5c40fae9448642dd871adf459e5e2084186e8d2c2a79a824c970da1f8", size = 19228, upload-time = "2024-09-29T09:24:13.293Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bd/24/12818598c362d7f300f18e74db45963dbcb85150324092410c8b49405e42/pyproject_hooks-1.2.0-py3-none-any.whl", hash = "sha256:9e5c6bfa8dcc30091c74b0cf803c81fdd29d94f01992a7707bc97babb1141913", size = 10216, upload-time = "2024-09-29T09:24:11.978Z" }, +] + +[[package]] +name = "pyspark" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "py4j" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c3/33/c987434f5d50aa802779a004ca0fd45ee4350caab50554ad7283d5a22b50/pyspark-4.2.0.tar.gz", hash = "sha256:5ad689d53570ee1674193fd4f9bda065f0db3be9363a27d2a3406cc457b70b61", size = 450129423, upload-time = "2026-07-14T22:16:46Z" } + +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + +[[package]] +name = "tomli" +version = "2.4.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/22/de/48c59722572767841493b26183a0d1cc411d54fd759c5607c4590b6563a6/tomli-2.4.1.tar.gz", hash = "sha256:7c7e1a961a0b2f2472c1ac5b69affa0ae1132c39adcb67aba98568702b9cc23f", size = 17543, upload-time = "2026-03-25T20:22:03.828Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/11/db3d5885d8528263d8adc260bb2d28ebf1270b96e98f0e0268d32b8d9900/tomli-2.4.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:f8f0fc26ec2cc2b965b7a3b87cd19c5c6b8c5e5f436b984e85f486d652285c30", size = 154704, upload-time = "2026-03-25T20:21:10.473Z" }, + { url = "https://files.pythonhosted.org/packages/6d/f7/675db52c7e46064a9aa928885a9b20f4124ecb9bc2e1ce74c9106648d202/tomli-2.4.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4ab97e64ccda8756376892c53a72bd1f964e519c77236368527f758fbc36a53a", size = 149454, upload-time = "2026-03-25T20:21:12.036Z" }, + { url = "https://files.pythonhosted.org/packages/61/71/81c50943cf953efa35bce7646caab3cf457a7d8c030b27cfb40d7235f9ee/tomli-2.4.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:96481a5786729fd470164b47cdb3e0e58062a496f455ee41b4403be77cb5a076", size = 237561, upload-time = "2026-03-25T20:21:13.098Z" }, + { url = "https://files.pythonhosted.org/packages/48/c1/f41d9cb618acccca7df82aaf682f9b49013c9397212cb9f53219e3abac37/tomli-2.4.1-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5a881ab208c0baf688221f8cecc5401bd291d67e38a1ac884d6736cbcd8247e9", size = 243824, upload-time = "2026-03-25T20:21:14.569Z" }, + { url = "https://files.pythonhosted.org/packages/22/e4/5a816ecdd1f8ca51fb756ef684b90f2780afc52fc67f987e3c61d800a46d/tomli-2.4.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:47149d5bd38761ac8be13a84864bf0b7b70bc051806bc3669ab1cbc56216b23c", size = 242227, upload-time = "2026-03-25T20:21:15.712Z" }, + { url = "https://files.pythonhosted.org/packages/6b/49/2b2a0ef529aa6eec245d25f0c703e020a73955ad7edf73e7f54ddc608aa5/tomli-2.4.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:ec9bfaf3ad2df51ace80688143a6a4ebc09a248f6ff781a9945e51937008fcbc", size = 247859, upload-time = "2026-03-25T20:21:17.001Z" }, + { url = "https://files.pythonhosted.org/packages/83/bd/6c1a630eaca337e1e78c5903104f831bda934c426f9231429396ce3c3467/tomli-2.4.1-cp311-cp311-win32.whl", hash = "sha256:ff2983983d34813c1aeb0fa89091e76c3a22889ee83ab27c5eeb45100560c049", size = 97204, upload-time = "2026-03-25T20:21:18.079Z" }, + { url = "https://files.pythonhosted.org/packages/42/59/71461df1a885647e10b6bb7802d0b8e66480c61f3f43079e0dcd315b3954/tomli-2.4.1-cp311-cp311-win_amd64.whl", hash = "sha256:5ee18d9ebdb417e384b58fe414e8d6af9f4e7a0ae761519fb50f721de398dd4e", size = 108084, upload-time = "2026-03-25T20:21:18.978Z" }, + { url = "https://files.pythonhosted.org/packages/b8/83/dceca96142499c069475b790e7913b1044c1a4337e700751f48ed723f883/tomli-2.4.1-cp311-cp311-win_arm64.whl", hash = "sha256:c2541745709bad0264b7d4705ad453b76ccd191e64aa6f0fc66b69a293a45ece", size = 95285, upload-time = "2026-03-25T20:21:20.309Z" }, + { url = "https://files.pythonhosted.org/packages/c1/ba/42f134a3fe2b370f555f44b1d72feebb94debcab01676bf918d0cb70e9aa/tomli-2.4.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:c742f741d58a28940ce01d58f0ab2ea3ced8b12402f162f4d534dfe18ba1cd6a", size = 155924, upload-time = "2026-03-25T20:21:21.626Z" }, + { url = "https://files.pythonhosted.org/packages/dc/c7/62d7a17c26487ade21c5422b646110f2162f1fcc95980ef7f63e73c68f14/tomli-2.4.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7f86fd587c4ed9dd76f318225e7d9b29cfc5a9d43de44e5754db8d1128487085", size = 150018, upload-time = "2026-03-25T20:21:23.002Z" }, + { url = "https://files.pythonhosted.org/packages/5c/05/79d13d7c15f13bdef410bdd49a6485b1c37d28968314eabee452c22a7fda/tomli-2.4.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ff18e6a727ee0ab0388507b89d1bc6a22b138d1e2fa56d1ad494586d61d2eae9", size = 244948, upload-time = "2026-03-25T20:21:24.04Z" }, + { url = "https://files.pythonhosted.org/packages/10/90/d62ce007a1c80d0b2c93e02cab211224756240884751b94ca72df8a875ca/tomli-2.4.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:136443dbd7e1dee43c68ac2694fde36b2849865fa258d39bf822c10e8068eac5", size = 253341, upload-time = "2026-03-25T20:21:25.177Z" }, + { url = "https://files.pythonhosted.org/packages/1a/7e/caf6496d60152ad4ed09282c1885cca4eea150bfd007da84aea07bcc0a3e/tomli-2.4.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:5e262d41726bc187e69af7825504c933b6794dc3fbd5945e41a79bb14c31f585", size = 248159, upload-time = "2026-03-25T20:21:26.364Z" }, + { url = "https://files.pythonhosted.org/packages/99/e7/c6f69c3120de34bbd882c6fba7975f3d7a746e9218e56ab46a1bc4b42552/tomli-2.4.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5cb41aa38891e073ee49d55fbc7839cfdb2bc0e600add13874d048c94aadddd1", size = 253290, upload-time = "2026-03-25T20:21:27.46Z" }, + { url = "https://files.pythonhosted.org/packages/d6/2f/4a3c322f22c5c66c4b836ec58211641a4067364f5dcdd7b974b4c5da300c/tomli-2.4.1-cp312-cp312-win32.whl", hash = "sha256:da25dc3563bff5965356133435b757a795a17b17d01dbc0f42fb32447ddfd917", size = 98141, upload-time = "2026-03-25T20:21:28.492Z" }, + { url = "https://files.pythonhosted.org/packages/24/22/4daacd05391b92c55759d55eaee21e1dfaea86ce5c571f10083360adf534/tomli-2.4.1-cp312-cp312-win_amd64.whl", hash = "sha256:52c8ef851d9a240f11a88c003eacb03c31fc1c9c4ec64a99a0f922b93874fda9", size = 108847, upload-time = "2026-03-25T20:21:29.386Z" }, + { url = "https://files.pythonhosted.org/packages/68/fd/70e768887666ddd9e9f5d85129e84910f2db2796f9096aa02b721a53098d/tomli-2.4.1-cp312-cp312-win_arm64.whl", hash = "sha256:f758f1b9299d059cc3f6546ae2af89670cb1c4d48ea29c3cacc4fe7de3058257", size = 95088, upload-time = "2026-03-25T20:21:30.677Z" }, + { url = "https://files.pythonhosted.org/packages/07/06/b823a7e818c756d9a7123ba2cda7d07bc2dd32835648d1a7b7b7a05d848d/tomli-2.4.1-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:36d2bd2ad5fb9eaddba5226aa02c8ec3fa4f192631e347b3ed28186d43be6b54", size = 155866, upload-time = "2026-03-25T20:21:31.65Z" }, + { url = "https://files.pythonhosted.org/packages/14/6f/12645cf7f08e1a20c7eb8c297c6f11d31c1b50f316a7e7e1e1de6e2e7b7e/tomli-2.4.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:eb0dc4e38e6a1fd579e5d50369aa2e10acfc9cace504579b2faabb478e76941a", size = 149887, upload-time = "2026-03-25T20:21:33.028Z" }, + { url = "https://files.pythonhosted.org/packages/5c/e0/90637574e5e7212c09099c67ad349b04ec4d6020324539297b634a0192b0/tomli-2.4.1-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c7f2c7f2b9ca6bdeef8f0fa897f8e05085923eb091721675170254cbc5b02897", size = 243704, upload-time = "2026-03-25T20:21:34.51Z" }, + { url = "https://files.pythonhosted.org/packages/10/8f/d3ddb16c5a4befdf31a23307f72828686ab2096f068eaf56631e136c1fdd/tomli-2.4.1-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f3c6818a1a86dd6dca7ddcaaf76947d5ba31aecc28cb1b67009a5877c9a64f3f", size = 251628, upload-time = "2026-03-25T20:21:36.012Z" }, + { url = "https://files.pythonhosted.org/packages/e3/f1/dbeeb9116715abee2485bf0a12d07a8f31af94d71608c171c45f64c0469d/tomli-2.4.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:d312ef37c91508b0ab2cee7da26ec0b3ed2f03ce12bd87a588d771ae15dcf82d", size = 247180, upload-time = "2026-03-25T20:21:37.136Z" }, + { url = "https://files.pythonhosted.org/packages/d3/74/16336ffd19ed4da28a70959f92f506233bd7cfc2332b20bdb01591e8b1d1/tomli-2.4.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:51529d40e3ca50046d7606fa99ce3956a617f9b36380da3b7f0dd3dd28e68cb5", size = 251674, upload-time = "2026-03-25T20:21:38.298Z" }, + { url = "https://files.pythonhosted.org/packages/16/f9/229fa3434c590ddf6c0aa9af64d3af4b752540686cace29e6281e3458469/tomli-2.4.1-cp313-cp313-win32.whl", hash = "sha256:2190f2e9dd7508d2a90ded5ed369255980a1bcdd58e52f7fe24b8162bf9fedbd", size = 97976, upload-time = "2026-03-25T20:21:39.316Z" }, + { url = "https://files.pythonhosted.org/packages/6a/1e/71dfd96bcc1c775420cb8befe7a9d35f2e5b1309798f009dca17b7708c1e/tomli-2.4.1-cp313-cp313-win_amd64.whl", hash = "sha256:8d65a2fbf9d2f8352685bc1364177ee3923d6baf5e7f43ea4959d7d8bc326a36", size = 108755, upload-time = "2026-03-25T20:21:40.248Z" }, + { url = "https://files.pythonhosted.org/packages/83/7a/d34f422a021d62420b78f5c538e5b102f62bea616d1d75a13f0a88acb04a/tomli-2.4.1-cp313-cp313-win_arm64.whl", hash = "sha256:4b605484e43cdc43f0954ddae319fb75f04cc10dd80d830540060ee7cd0243cd", size = 95265, upload-time = "2026-03-25T20:21:41.219Z" }, + { url = "https://files.pythonhosted.org/packages/3c/fb/9a5c8d27dbab540869f7c1f8eb0abb3244189ce780ba9cd73f3770662072/tomli-2.4.1-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:fd0409a3653af6c147209d267a0e4243f0ae46b011aa978b1080359fddc9b6cf", size = 155726, upload-time = "2026-03-25T20:21:42.23Z" }, + { url = "https://files.pythonhosted.org/packages/62/05/d2f816630cc771ad836af54f5001f47a6f611d2d39535364f148b6a92d6b/tomli-2.4.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:a120733b01c45e9a0c34aeef92bf0cf1d56cfe81ed9d47d562f9ed591a9828ac", size = 149859, upload-time = "2026-03-25T20:21:43.386Z" }, + { url = "https://files.pythonhosted.org/packages/ce/48/66341bdb858ad9bd0ceab5a86f90eddab127cf8b046418009f2125630ecb/tomli-2.4.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:559db847dc486944896521f68d8190be1c9e719fced785720d2216fe7022b662", size = 244713, upload-time = "2026-03-25T20:21:44.474Z" }, + { url = "https://files.pythonhosted.org/packages/df/6d/c5fad00d82b3c7a3ab6189bd4b10e60466f22cfe8a08a9394185c8a8111c/tomli-2.4.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:01f520d4f53ef97964a240a035ec2a869fe1a37dde002b57ebc4417a27ccd853", size = 252084, upload-time = "2026-03-25T20:21:45.62Z" }, + { url = "https://files.pythonhosted.org/packages/00/71/3a69e86f3eafe8c7a59d008d245888051005bd657760e96d5fbfb0b740c2/tomli-2.4.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7f94b27a62cfad8496c8d2513e1a222dd446f095fca8987fceef261225538a15", size = 247973, upload-time = "2026-03-25T20:21:46.937Z" }, + { url = "https://files.pythonhosted.org/packages/67/50/361e986652847fec4bd5e4a0208752fbe64689c603c7ae5ea7cb16b1c0ca/tomli-2.4.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ede3e6487c5ef5d28634ba3f31f989030ad6af71edfb0055cbbd14189ff240ba", size = 256223, upload-time = "2026-03-25T20:21:48.467Z" }, + { url = "https://files.pythonhosted.org/packages/8c/9a/b4173689a9203472e5467217e0154b00e260621caa227b6fa01feab16998/tomli-2.4.1-cp314-cp314-win32.whl", hash = "sha256:3d48a93ee1c9b79c04bb38772ee1b64dcf18ff43085896ea460ca8dec96f35f6", size = 98973, upload-time = "2026-03-25T20:21:49.526Z" }, + { url = "https://files.pythonhosted.org/packages/14/58/640ac93bf230cd27d002462c9af0d837779f8773bc03dee06b5835208214/tomli-2.4.1-cp314-cp314-win_amd64.whl", hash = "sha256:88dceee75c2c63af144e456745e10101eb67361050196b0b6af5d717254dddf7", size = 109082, upload-time = "2026-03-25T20:21:50.506Z" }, + { url = "https://files.pythonhosted.org/packages/d5/2f/702d5e05b227401c1068f0d386d79a589bb12bf64c3d2c72ce0631e3bc49/tomli-2.4.1-cp314-cp314-win_arm64.whl", hash = "sha256:b8c198f8c1805dc42708689ed6864951fd2494f924149d3e4bce7710f8eb5232", size = 96490, upload-time = "2026-03-25T20:21:51.474Z" }, + { url = "https://files.pythonhosted.org/packages/45/4b/b877b05c8ba62927d9865dd980e34a755de541eb65fffba52b4cc495d4d2/tomli-2.4.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:d4d8fe59808a54658fcc0160ecfb1b30f9089906c50b23bcb4c69eddc19ec2b4", size = 164263, upload-time = "2026-03-25T20:21:52.543Z" }, + { url = "https://files.pythonhosted.org/packages/24/79/6ab420d37a270b89f7195dec5448f79400d9e9c1826df982f3f8e97b24fd/tomli-2.4.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7008df2e7655c495dd12d2a4ad038ff878d4ca4b81fccaf82b714e07eae4402c", size = 160736, upload-time = "2026-03-25T20:21:53.674Z" }, + { url = "https://files.pythonhosted.org/packages/02/e0/3630057d8eb170310785723ed5adcdfb7d50cb7e6455f85ba8a3deed642b/tomli-2.4.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1d8591993e228b0c930c4bb0db464bdad97b3289fb981255d6c9a41aedc84b2d", size = 270717, upload-time = "2026-03-25T20:21:55.129Z" }, + { url = "https://files.pythonhosted.org/packages/7a/b4/1613716072e544d1a7891f548d8f9ec6ce2faf42ca65acae01d76ea06bb0/tomli-2.4.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:734e20b57ba95624ecf1841e72b53f6e186355e216e5412de414e3c51e5e3c41", size = 278461, upload-time = "2026-03-25T20:21:56.228Z" }, + { url = "https://files.pythonhosted.org/packages/05/38/30f541baf6a3f6df77b3df16b01ba319221389e2da59427e221ef417ac0c/tomli-2.4.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:8a650c2dbafa08d42e51ba0b62740dae4ecb9338eefa093aa5c78ceb546fcd5c", size = 274855, upload-time = "2026-03-25T20:21:57.653Z" }, + { url = "https://files.pythonhosted.org/packages/77/a3/ec9dd4fd2c38e98de34223b995a3b34813e6bdadf86c75314c928350ed14/tomli-2.4.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:504aa796fe0569bb43171066009ead363de03675276d2d121ac1a4572397870f", size = 283144, upload-time = "2026-03-25T20:21:59.089Z" }, + { url = "https://files.pythonhosted.org/packages/ef/be/605a6261cac79fba2ec0c9827e986e00323a1945700969b8ee0b30d85453/tomli-2.4.1-cp314-cp314t-win32.whl", hash = "sha256:b1d22e6e9387bf4739fbe23bfa80e93f6b0373a7f1b96c6227c32bef95a4d7a8", size = 108683, upload-time = "2026-03-25T20:22:00.214Z" }, + { url = "https://files.pythonhosted.org/packages/12/64/da524626d3b9cc40c168a13da8335fe1c51be12c0a63685cc6db7308daae/tomli-2.4.1-cp314-cp314t-win_amd64.whl", hash = "sha256:2c1c351919aca02858f740c6d33adea0c5deea37f9ecca1cc1ef9e884a619d26", size = 121196, upload-time = "2026-03-25T20:22:01.169Z" }, + { url = "https://files.pythonhosted.org/packages/5a/cd/e80b62269fc78fc36c9af5a6b89c835baa8af28ff5ad28c7028d60860320/tomli-2.4.1-cp314-cp314t-win_arm64.whl", hash = "sha256:eab21f45c7f66c13f2a9e0e1535309cee140182a9cdae1e041d02e47291e8396", size = 100393, upload-time = "2026-03-25T20:22:02.137Z" }, + { url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" }, +] + +[[package]] +name = "typing-extensions" +version = "4.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" }, +] + +[[package]] +name = "windmill" +version = "0.1.0" +source = { editable = "." } + +[package.optional-dependencies] +dev = [ + { name = "build" }, + { name = "pyspark" }, + { name = "pytest" }, +] + +[package.metadata] +requires-dist = [ + { name = "build", marker = "extra == 'dev'", specifier = ">=1.0" }, + { name = "pyspark", marker = "extra == 'dev'", specifier = ">=3.5" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, +] +provides-extras = ["dev"] + +[[package]] +name = "zipp" +version = "4.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b9/d8/eab98a517c14134c0b2eb4e2387bc5f457334293ec5d2dd3857ec2966802/zipp-4.1.0.tar.gz", hash = "sha256:4cb57381f544315db7688e976e922a2b18cdb513d21cc194eb42232ba2a3e602", size = 26214, upload-time = "2026-05-18T20:08:57.967Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3a/13/547360d81e6d88d58492968ffda9f9542854f11310ee556fef14260cc886/zipp-4.1.0-py3-none-any.whl", hash = "sha256:25ad4e16390cd314347dd8f1de67a2ac538ae658ed4ab9db16029c07c188e97f", size = 10238, upload-time = "2026-05-18T20:08:57.045Z" }, +] From 51a8f9a9e1f65e808d276db327c2c97bb4985e7c Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 08:59:47 +0100 Subject: [PATCH 3/3] feat: make the hourly-grain assumption explicit and detectable Dedup keys on (timestamp, turbine_id) and gold takes an unweighted mean over readings. Both are correct only while the feed sends one reading per turbine per hour -- true of the provided data (all 11160 timestamps on :00:00, no duplicate keys, every one of the 465 turbine-days holding exactly 24 readings), but an observation rather than a contract. Sub-hourly timestamps would not break anything loudly. They are genuinely distinct readings, so dedup leaves them alone as it should, and the daily mean then quietly reweights toward whichever hours sent more samples. Gold now emits distinct_hours alongside measurement_count and sets has_sub_hourly_readings when they disagree, so the condition surfaces instead of silently distorting the average. No resampling rule is applied, because the correct one depends on what the sensor emits: instantaneous samples should be averaged, a cumulative counter needs the last value per hour, a pre-averaged period value needs weighting. The spec says megawatts -- a rate, not an energy total -- which points at the first case and makes averaging defensible, but that is inference rather than confirmation. Documented in the README with the options laid out, to be implemented once the sensor semantics are confirmed. Gold writes with overwriteSchema so adding a summary column no longer requires dropping the table by hand. Safe because gold is fully derived from silver and rebuilt from scratch every run. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 32 +++++++++++++++++++++++++- notebooks/04_gold_aggregate.py | 5 +++- src/windmill/aggregation.py | 34 ++++++++++++++++++++++++--- tests/test_aggregation.py | 42 ++++++++++++++++++++++++++++++++++ 4 files changed, 108 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 8905d4a..8439b31 100644 --- a/README.md +++ b/README.md @@ -168,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 diff --git a/notebooks/04_gold_aggregate.py b/notebooks/04_gold_aggregate.py index e7cc21c..caf9ea2 100644 --- a/notebooks/04_gold_aggregate.py +++ b/notebooks/04_gold_aggregate.py @@ -29,7 +29,10 @@ summary = build_gold(silver) -summary.write.mode("overwrite").saveAsTable(gold_table) +# overwriteSchema so adding or removing a summary column does not require +# dropping the table by hand. Safe here because gold is fully derived from +# silver -- every run rebuilds it from scratch. +summary.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(gold_table) print(f"✓ Wrote {summary.count()} turbine-days to {gold_table}") anomalies = summary.filter(col("is_anomaly")) diff --git a/src/windmill/aggregation.py b/src/windmill/aggregation.py index 599f915..d7c6d79 100644 --- a/src/windmill/aggregation.py +++ b/src/windmill/aggregation.py @@ -9,7 +9,16 @@ from __future__ import annotations from pyspark.sql import Column, DataFrame -from pyspark.sql.functions import avg, col, count, max, min, stddev +from pyspark.sql.functions import ( + avg, + col, + count, + countDistinct, + date_trunc, + max, + min, + stddev, +) from pyspark.sql.functions import sum as _sum from pyspark.sql.functions import when @@ -26,15 +35,34 @@ def valid_readings(df: DataFrame) -> DataFrame: def daily_summary(df: DataFrame) -> DataFrame: - """Per-turbine, per-day summary statistics.""" - return df.groupBy("date", "turbine_id", "turbine_group").agg( + """Per-turbine, per-day summary statistics. + + `avg_power_output` is a mean over readings, which is correct while the feed + delivers one reading per turbine per hour. `has_sub_hourly_readings` exists + to make a violation of that assumption visible rather than silent -- see the + note on `count_distinct_hours` below. + """ + summary = df.groupBy("date", "turbine_id", "turbine_group").agg( min("power_output").alias("min_power_output"), max("power_output").alias("max_power_output"), avg("power_output").alias("avg_power_output"), stddev("power_output").alias("stddev_power_output"), count("*").alias("measurement_count"), + countDistinct(date_trunc("hour", col("timestamp"))).alias("distinct_hours"), avg("wind_speed").alias("avg_wind_speed"), ) + # More readings than distinct hours means the feed sent multiple samples + # within an hour. Deduplication cannot collapse them -- their timestamps + # genuinely differ -- so they silently reweight the daily mean toward the + # busier hours. Whether that is even wrong depends on what the sensor emits: + # an instantaneous sample should be averaged, whereas a cumulative counter + # would need the last value per hour, and a period-average would need + # weighting. That question cannot be answered from the data alone, so this + # flags the condition rather than guessing at a correction. + return summary.withColumn( + "has_sub_hourly_readings", + col("measurement_count") > col("distinct_hours"), + ) def turbine_baseline(daily: DataFrame) -> DataFrame: diff --git a/tests/test_aggregation.py b/tests/test_aggregation.py index f4b9397..c60c692 100644 --- a/tests/test_aggregation.py +++ b/tests/test_aggregation.py @@ -162,6 +162,48 @@ def test_short_spike_counted_even_when_daily_average_looks_normal(make_raw): assert total > 0 +def test_hourly_grain_assumption_holds_for_well_formed_data(make_raw): + """One reading per hour is the expected shape and must not raise the flag.""" + daily = daily_summary(valid_readings(_silver(make_raw, _day(0, 1, [3.0] * 24)))) + row = daily.first() + assert row["measurement_count"] == 24 + assert row["distinct_hours"] == 24 + assert row["has_sub_hourly_readings"] is False + + +def test_sub_hourly_readings_are_detected(make_raw): + """Two readings inside one hour must be visible, not silently averaged. + + Their timestamps genuinely differ, so dedup cannot and should not collapse + them. Left undetected they reweight the daily mean toward that hour. Whether + averaging is even the right treatment depends on what the sensor emits, which + the data cannot tell us -- so the pipeline surfaces the condition instead of + guessing at a correction. + """ + from datetime import timedelta + + from .conftest import PERIOD_START + + rows = [ + (PERIOD_START, 1, 11.0, 180.0, 3.0, "/vol/inbox/data_group_1_x.csv"), + # Same hour, thirty minutes later -- a distinct reading, not a duplicate. + ( + PERIOD_START + timedelta(minutes=30), + 1, + 11.0, + 180.0, + 9.0, + "/vol/inbox/data_group_1_x.csv", + ), + ] + daily = daily_summary(valid_readings(_silver(make_raw, rows))) + row = daily.first() + + assert row["measurement_count"] == 2 + assert row["distinct_hours"] == 1 + assert row["has_sub_hourly_readings"] is True + + def test_gold_has_one_row_per_turbine_day(make_raw): rows = [] for day in range(3):