diff --git a/apps/data-management/src/components/Orchestration/data-products/AggregationForm.vue b/apps/data-management/src/components/Orchestration/data-products/AggregationForm.vue index 6891a9f7..a9b58115 100644 --- a/apps/data-management/src/components/Orchestration/data-products/AggregationForm.vue +++ b/apps/data-management/src/components/Orchestration/data-products/AggregationForm.vue @@ -254,7 +254,7 @@ const taskName = ref('') const schedule = ref(null) const inputDatastreamId = ref(null) const outputDatastreamId = ref(null) -const aggregationMethod = ref('mean') +const aggregationMethod = ref('time_weighted_mean') const outputInterval = ref(1) const outputIntervalUnits = ref('hours') const minValues = ref(null) @@ -264,12 +264,13 @@ const timezone = ref(null) const selectedThingId = computed(() => props.initialThingId ?? null) const aggregationMethodOptions = [ - { title: 'Mean', value: 'mean' }, - { title: 'Sum', value: 'sum' }, - { title: 'Min', value: 'min' }, - { title: 'Max', value: 'max' }, + { title: 'Arithmetic Mean', value: 'mean' }, { title: 'First', value: 'first' }, { title: 'Last', value: 'last' }, + { title: 'Max', value: 'max' }, + { title: 'Min', value: 'min' }, + { title: 'Sum', value: 'sum' }, + { title: 'Time-weighted mean', value: 'time_weighted_mean' }, ] const intervalUnitOptions = [ diff --git a/django/contracts/openapi/data.openapi.json b/django/contracts/openapi/data.openapi.json index b7702490..8160e971 100644 --- a/django/contracts/openapi/data.openapi.json +++ b/django/contracts/openapi/data.openapi.json @@ -651,7 +651,8 @@ "min", "max", "first", - "last" + "last", + "time_weighted_mean" ], "title": "Aggregationmethod", "type": "string" @@ -731,7 +732,8 @@ "min", "max", "first", - "last" + "last", + "time_weighted_mean" ], "title": "Aggregationmethod", "type": "string" @@ -823,7 +825,8 @@ "min", "max", "first", - "last" + "last", + "time_weighted_mean" ], "title": "Aggregationmethod", "type": "string" @@ -912,7 +915,8 @@ "min", "max", "first", - "last" + "last", + "time_weighted_mean" ], "title": "Aggregationmethod", "type": "string" diff --git a/django/interfaces/api/schemas/products/transformation.py b/django/interfaces/api/schemas/products/transformation.py index 830d4554..40bae568 100644 --- a/django/interfaces/api/schemas/products/transformation.py +++ b/django/interfaces/api/schemas/products/transformation.py @@ -15,7 +15,7 @@ from interfaces.api.schemas.products.rating_curve import RatingCurveSummaryResponse -AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last"] +AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last", "time_weighted_mean"] Period = Literal["minutes", "hours", "days", "weeks", "months"] TimezoneType = Literal["offset", "iana"] diff --git a/django/processing/products/migrations/0002_alter_dataproducttransformation_aggregation_method.py b/django/processing/products/migrations/0002_alter_dataproducttransformation_aggregation_method.py new file mode 100644 index 00000000..013a80a6 --- /dev/null +++ b/django/processing/products/migrations/0002_alter_dataproducttransformation_aggregation_method.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.2 on 2026-07-15 17:29 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('products', '0001_initial'), + ] + + operations = [ + migrations.AlterField( + model_name='dataproducttransformation', + name='aggregation_method', + field=models.CharField(blank=True, choices=[('mean', 'Mean'), ('sum', 'Sum'), ('min', 'Min'), ('max', 'Max'), ('first', 'First'), ('last', 'Last'), ('time_weighted_mean', 'Time Weighted Mean')], max_length=255, null=True), + ), + ] diff --git a/django/processing/products/models/transformation.py b/django/processing/products/models/transformation.py index 79258b13..28a2203d 100644 --- a/django/processing/products/models/transformation.py +++ b/django/processing/products/models/transformation.py @@ -20,6 +20,7 @@ class AggregationMethod(models.TextChoices): MAX = "max" FIRST = "first" LAST = "last" + TIME_WEIGHTED_MEAN = "time_weighted_mean" class IntervalUnits(models.TextChoices): diff --git a/django/processing/products/services/transformation.py b/django/processing/products/services/transformation.py index 3bee9891..43dc79ae 100644 --- a/django/processing/products/services/transformation.py +++ b/django/processing/products/services/transformation.py @@ -42,7 +42,7 @@ rating_curve_service = RatingCurveService() TransformationType = Literal["rating_curve", "expression", "composite_expression", "aggregation"] -AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last"] +AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last", "time_weighted_mean"] IntervalUnits = Literal["minutes", "hours", "days", "weeks", "months"] diff --git a/django/tests/products/services/test_transformation.py b/django/tests/products/services/test_transformation.py index f836f9c6..5db900b1 100644 --- a/django/tests/products/services/test_transformation.py +++ b/django/tests/products/services/test_transformation.py @@ -320,6 +320,21 @@ def test_create_aggregation_transformation(get_principal, ds_free): assert result.aggregation_method == "mean" +def test_create_aggregation_transformation_time_weighted_mean(get_principal, ds_free): + result = transformation_service.create( + task=uuid.UUID(TASK1), + principal=get_principal("owner"), + transformation_type="aggregation", + output_datastream=ds_free, + aggregation_method="time_weighted_mean", + output_interval_units="hours", + output_interval=1, + input_datastreams=[TransformationInput(datastream=uuid.UUID(DS_IN_RC))], + ) + assert result.transformation_type == "aggregation" + assert result.aggregation_method == "time_weighted_mean" + + def test_create_composite_expression_transformation(get_principal, ds_free): result = transformation_service.create( task=uuid.UUID(TASK1), @@ -906,3 +921,52 @@ def test_run_aggregation(run_context): from core.sta.models.observation import Observation results = sorted(Observation.objects.filter(datastream=ds_out).values_list("result", flat=True)) np.testing.assert_allclose(results, [2.0, 3.0]) + + +def test_run_aggregation_time_weighted_mean(run_context): + """ + time_weighted_mean, 1-day UTC bucket [2025-02-10T00:00Z, 2025-02-11T00:00Z). + + Observations at 06:00=1.0, 08:00=3.0, 10:00=2.0, 12:00=4.0. Boundary values are + flat-extrapolated (00:00 -> 1.0, 24:00 -> 4.0), then trapezoidal integration gives: + [00-06]: (1+1)/2*6 = 6 + [06-08]: (1+3)/2*2 = 4 + [08-10]: (3+2)/2*2 = 5 + [10-12]: (2+4)/2*2 = 6 + [12-24]: (4+4)/2*12= 48 + total = 69 over 24h -> 2.875 + This differs from the simple arithmetic mean of the same observations (2.5), + confirming the time-weighted method is actually wired through the run pipeline. + """ + ctx = run_context + kwargs = {k: ctx[k] for k in ("thing", "sensor", "observed_property", "processing_level", "unit")} + ds_in = _make_datastream(**kwargs) + ds_out = _make_datastream(**kwargs) + + for dt, val in [ + (datetime(2025, 2, 10, 6, 0, tzinfo=dt_timezone.utc), 1.0), + (datetime(2025, 2, 10, 8, 0, tzinfo=dt_timezone.utc), 3.0), + (datetime(2025, 2, 10, 10, 0, tzinfo=dt_timezone.utc), 2.0), + (datetime(2025, 2, 10, 12, 0, tzinfo=dt_timezone.utc), 4.0), + ]: + _add_obs(ds_in, dt, val) + _set_end_time(ds_in, datetime(2025, 2, 11, 0, 0, tzinfo=dt_timezone.utc)) + + t = transformation_service.create( + task=uuid.UUID(TASK1), + principal=ctx["owner"], + transformation_type="aggregation", + output_datastream=ds_out.pk, + aggregation_method="time_weighted_mean", + output_interval_units="days", + output_interval=1, + input_datastreams=[TransformationInput(datastream=ds_in.pk)], + ) + loaded = transformation_service.run(t) + assert loaded == 1 + + from core.sta.models.observation import Observation + out_obs = list(Observation.objects.filter(datastream=ds_out)) + assert len(out_obs) == 1 + assert out_obs[0].phenomenon_time == datetime(2025, 2, 10, 0, 0, tzinfo=dt_timezone.utc) + np.testing.assert_allclose(out_obs[0].result, 2.875) diff --git a/packages/hydroserver-ts/src/generated/data.types.d.ts b/packages/hydroserver-ts/src/generated/data.types.d.ts index ea6d6dce..3adfa962 100644 --- a/packages/hydroserver-ts/src/generated/data.types.d.ts +++ b/packages/hydroserver-ts/src/generated/data.types.d.ts @@ -2258,7 +2258,7 @@ export interface components { * Aggregationmethod * @enum {string} */ - aggregationMethod?: "mean" | "sum" | "min" | "max" | "first" | "last"; + aggregationMethod?: "mean" | "sum" | "min" | "max" | "first" | "last" | "time_weighted_mean"; /** * Inputdatastreamid * Format: uuid @@ -2289,7 +2289,7 @@ export interface components { * Aggregationmethod * @enum {string} */ - aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last"; + aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last" | "time_weighted_mean"; /** * Id * Format: uuid @@ -2325,7 +2325,7 @@ export interface components { * Aggregationmethod * @enum {string} */ - aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last"; + aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last" | "time_weighted_mean"; /** * Id * Format: uuid @@ -2353,7 +2353,7 @@ export interface components { * Aggregationmethod * @enum {string} */ - aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last"; + aggregationMethod: "mean" | "sum" | "min" | "max" | "first" | "last" | "time_weighted_mean"; /** * Id * Format: uuid diff --git a/packages/hydroserverpy/src/hydroserverpy/api/models/products/transformation.py b/packages/hydroserverpy/src/hydroserverpy/api/models/products/transformation.py index ce66f243..e9f32527 100644 --- a/packages/hydroserverpy/src/hydroserverpy/api/models/products/transformation.py +++ b/packages/hydroserverpy/src/hydroserverpy/api/models/products/transformation.py @@ -3,7 +3,7 @@ from pydantic import BaseModel, ConfigDict from pydantic.alias_generators import to_camel -AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last"] +AggregationMethod = Literal["mean", "sum", "min", "max", "first", "last", "time_weighted_mean"] Period = Literal["minutes", "hours", "days", "weeks", "months"] diff --git a/packages/hydroserverpy/src/hydroserverpy/products/aggregation.py b/packages/hydroserverpy/src/hydroserverpy/products/aggregation.py index 8c09f59a..35daf013 100644 --- a/packages/hydroserverpy/src/hydroserverpy/products/aggregation.py +++ b/packages/hydroserverpy/src/hydroserverpy/products/aggregation.py @@ -1,8 +1,9 @@ import logging import pandas as pd +from bisect import bisect_left from datetime import datetime, timezone -from typing import Literal +from typing import Literal, Optional from pydantic import ConfigDict, validate_call @@ -12,13 +13,160 @@ logger = logging.getLogger(__name__) +TIME_WEIGHTED_MEAN = "time_weighted_mean" + + +def _resample(frame: pd.DataFrame, rule, offset: pd.Timedelta): + """ + Resample `frame` on TIMESTAMP_COL, omitting `offset` for calendar-aware + rules (e.g. Day) that don't support it — passing it anyway is harmless + (those rules are only used when offset is the zero default) but pandas + emits a RuntimeWarning on every call otherwise. + """ + + kwargs = dict(rule=rule, on=TIMESTAMP_COL, closed="left", label="left") + if isinstance(rule, pd.Timedelta): + kwargs["offset"] = offset + return frame.resample(**kwargs) + + +def _boundary_value( + target: pd.Timestamp, + timestamps: list, + values: list, + prev_idx: Optional[int], + next_idx: Optional[int], +) -> Optional[float]: + """ + Estimate the value at a window boundary by exact match or linear interpolation. + + If the observation immediately before (prev_idx) or after (next_idx) the boundary + falls exactly on the target timestamp, that value is returned directly. Otherwise, + if observations exist on both sides, the value is linearly interpolated. If only + one side is available, that side's value is used as a flat extrapolation. + """ + + prev = None + if prev_idx is not None and 0 <= prev_idx < len(timestamps): + prev = (timestamps[prev_idx], values[prev_idx]) + + nxt = None + if next_idx is not None and 0 <= next_idx < len(timestamps): + nxt = (timestamps[next_idx], values[next_idx]) + + if prev is not None and prev[0] == target: + return prev[1] + if nxt is not None and nxt[0] == target: + return nxt[1] + + if prev is not None and nxt is not None: + t0, v0 = prev + t1, v1 = nxt + span = (t1 - t0).total_seconds() + if span <= 0: + return v1 + ratio = (target - t0).total_seconds() / span + return v0 + ratio * (v1 - v0) + + if prev is not None: + return prev[1] + if nxt is not None: + return nxt[1] + + return None + + +def _next_bin_start(bin_start: pd.Timestamp, rule, offset: pd.Timedelta) -> pd.Timestamp: + """ + Return the bin boundary immediately after `bin_start`, using pandas' own + resample binning rather than direct date arithmetic. + + A plain `bin_start + rule` (or `pd.date_range`) can drift by the DST offset + across a transition, and that behavior isn't consistent across pandas + versions. Resampling a probe frame anchored at `bin_start` guarantees the + next label is computed by the exact same binning algorithm already used + (and proven correct) for every other bin boundary in the caller. `offset` + only has an effect when `rule` is Tick-based (e.g. a Timedelta); it's + ignored by pandas for calendar-aware offsets like Day, which is fine since + those are only used when no anchor/offset was requested. + """ + + probe = pd.DataFrame({TIMESTAMP_COL: [bin_start, bin_start + 2 * rule]}) + edges = _resample(probe, rule, offset).size().index + return edges[1] + + +def _time_weighted_mean(localized: pd.DataFrame, bin_starts: list, rule, offset: pd.Timedelta) -> pd.Series: + """ + Compute a trapezoidal (linear-interpolation) time-weighted mean for each bin. + + Each bin's value is the area under the piecewise-linear curve through the + observations spanning [bin_start, bin_end), divided by the bin duration. + Boundary values at the bin edges are estimated by linear interpolation between + the nearest observations on each side, or a flat extrapolation if only one side + has data. Bins with no observations are left as NaN (dropped later via _count). + """ + + timestamps = localized[TIMESTAMP_COL].tolist() + values = localized[RESULT_COL].tolist() + n = len(timestamps) + + results = [] + + for i, window_start in enumerate(bin_starts): + if i + 1 < len(bin_starts): + window_end = bin_starts[i + 1] + else: + # No trailing bin label for the last window. + window_end = _next_bin_start(window_start, rule, offset) + + left = bisect_left(timestamps, window_start) + right = bisect_left(timestamps, window_end) + + if left == right: + results.append(float("nan")) + continue + + start_value = _boundary_value( + window_start, timestamps, values, left - 1 if left > 0 else None, left + ) + end_value = _boundary_value( + window_end, timestamps, values, right - 1, right if right < n else None + ) + + points = [(window_start, start_value)] + for idx in range(left, right): + ts, val = timestamps[idx], values[idx] + if ts == window_start: + points[0] = (ts, val) + else: + points.append((ts, val)) + + if points[-1][0] == window_end: + points[-1] = (window_end, end_value) + else: + points.append((window_end, end_value)) + + area = 0.0 + for j in range(1, len(points)): + t0, v0 = points[j - 1] + t1, v1 = points[j] + span = (t1 - t0).total_seconds() + if span > 0: + area += (v0 + v1) * 0.5 * span + + duration = (window_end - window_start).total_seconds() + results.append(area / duration if duration > 0 else float("nan")) + + return pd.Series(results, index=pd.Index(bin_starts, name=TIMESTAMP_COL)) + @validate_call(config=ConfigDict(arbitrary_types_allowed=True)) def apply_aggregation( df: pd.DataFrame, *, interval: Duration, - method: Literal["min", "max", "sum", "mean", "first", "last"], + method: Literal["min", "max", "sum", "mean", "first", "last", "time_weighted_mean"], anchor: datetime | None = None, local_timezone: str | None = None, min_values: int | None = None, @@ -33,6 +181,11 @@ def apply_aggregation( window is handled: 'drop' omits the window, 'raise' raises a ValueError, 'stop' returns windows up to the first window that doesn't meet the threshold, and 'ndv' fills the window result with no_data_value (requires no_data_value to be set). + + 'time_weighted_mean' differs from the other methods in that it weights each + observation by the time it represents rather than treating all observations in + a window equally, via trapezoidal integration between observations (interpolating + across window boundaries as needed). """ df = validate_timeseries(df) @@ -68,22 +221,34 @@ def apply_aggregation( else: offset = pd.Timedelta(0) + # Day-or-longer intervals use a calendar-aware offset so windows respect + # local wall-clock (DST) boundaries rather than a fixed absolute duration + # (e.g. a "day" around a spring-forward transition is 23 real hours, not + # 24). Tick-based rules like Timedelta can't express that, and pandas + # ignores `offset=` for calendar-aware rules, so this only applies when + # there's no anchor. + _DAY_US = 86_400_000_000 + if anchor is None and interval_us % _DAY_US == 0: + resample_rule = pd.tseries.offsets.Day(interval_us // _DAY_US) + else: + resample_rule = freq + # Convert to local timezone before grouping so window boundaries align # to local calendar time rather than UTC. Reverted to UTC after aggregation. tz_str = normalize_tz(local_timezone) if local_timezone else "UTC" localized = df.copy() localized[TIMESTAMP_COL] = localized[TIMESTAMP_COL].dt.tz_convert(tz_str) + localized = localized.sort_values(TIMESTAMP_COL).reset_index(drop=True) - resampled = ( - localized - .sort_values(TIMESTAMP_COL) - .resample(rule=freq, on=TIMESTAMP_COL, closed="left", label="left", offset=offset) - ) + resampled = _resample(localized, resample_rule, offset) + counts = resampled[RESULT_COL].count() + + if method == TIME_WEIGHTED_MEAN: + values_col = _time_weighted_mean(localized, counts.index.tolist(), resample_rule, offset) + else: + values_col = getattr(resampled[RESULT_COL], method)() - aggregated = pd.DataFrame({ - RESULT_COL: getattr(resampled[RESULT_COL], method)(), - "_count": resampled[RESULT_COL].count(), - }) + aggregated = pd.DataFrame({RESULT_COL: values_col, "_count": counts}) # Drop empty bins (resample includes all bins in the range, even empty ones). aggregated = aggregated[aggregated["_count"] > 0] diff --git a/packages/hydroserverpy/tests/products/test_aggregation.py b/packages/hydroserverpy/tests/products/test_aggregation.py index 96651bde..cba17bfc 100644 --- a/packages/hydroserverpy/tests/products/test_aggregation.py +++ b/packages/hydroserverpy/tests/products/test_aggregation.py @@ -103,6 +103,80 @@ def test_hourly_interval_groups_sub_hour_observations(self): assert result[RESULT_COL].iloc[0] == pytest.approx(2.0) +# --------------------------------------------------------------------------- +# time_weighted_mean (trapezoidal integration) +# --------------------------------------------------------------------------- + +class TestApplyAggregationTimeWeightedMean: + + def test_constant_series_equals_constant(self): + ts = [_utc(2024, 1, 1, 6), _utc(2024, 1, 1, 12), _utc(2024, 1, 1, 18)] + df = _make_df(ts, [5.0, 5.0, 5.0]) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + assert result[RESULT_COL].iloc[0] == pytest.approx(5.0) + + def test_linear_series_equals_midpoint(self): + # Hourly observations 0..23 plus the value at the next day's boundary (24) + # trace a straight line, so the time-weighted mean over the day is the midpoint. + ts = [_utc(2024, 1, 15, h) for h in range(24)] + [_utc(2024, 1, 16, 0)] + vals = [float(h) for h in range(25)] + df = _make_df(ts, vals) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + assert result[RESULT_COL].iloc[0] == pytest.approx(12.0) + + def test_weights_longer_intervals_more_than_simple_mean(self): + # A value held for most of the day should dominate the time-weighted mean, + # unlike the simple arithmetic mean which treats every observation equally. + ts = [_utc(2024, 1, 1, 0), _utc(2024, 1, 1, 6), _utc(2024, 1, 1, 23)] + df = _make_df(ts, [0.0, 12.0, 12.0]) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + simple = apply_aggregation(df, interval="1d", method="mean") + assert result[RESULT_COL].iloc[0] == pytest.approx(10.5) + assert result[RESULT_COL].iloc[0] != pytest.approx(simple[RESULT_COL].iloc[0]) + + def test_extrapolates_flat_when_no_observation_before_window_start(self): + # No observation before 06:00, so the window-start boundary is a flat + # extrapolation of the first observation's value. + ts = [_utc(2024, 1, 1, 6), _utc(2024, 1, 1, 18)] + df = _make_df(ts, [10.0, 20.0]) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + # (10+10)/2*6h + (10+20)/2*12h + (20+20)/2*6h, over 24h + expected = ((10 + 10) / 2 * 6 + (10 + 20) / 2 * 12 + (20 + 20) / 2 * 6) / 24 + assert result[RESULT_COL].iloc[0] == pytest.approx(expected) + + def test_single_observation_in_window_equals_that_value(self): + ts = [_utc(2024, 1, 1, 12)] + df = _make_df(ts, [7.0]) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + assert result[RESULT_COL].iloc[0] == pytest.approx(7.0) + + def test_dst_spring_forward_window_uses_actual_23_hour_duration(self): + # 2024-03-10 is a 23-hour local day in America/Denver (spring forward). + # A naive `window_start + timedelta(days=1)` would overshoot the true + # boundary by an hour and skew the result. + ts = [_utc(2024, 3, 10, 10), _utc(2024, 3, 10, 20)] + df = _make_df(ts, [1.0, 2.0]) + result = apply_aggregation( + df, interval="1d", method="time_weighted_mean", local_timezone="America/Denver" + ) + assert result[RESULT_COL].iloc[0] == pytest.approx(136800 / 82800) + + def test_multiple_windows_computed_independently(self): + ts = _JAN1_TS + [_utc(2024, 1, 2, 6), _utc(2024, 1, 2, 12), _utc(2024, 1, 2, 18)] + df = _make_df(ts, _JAN1_VALS + [5.0, 6.0, 7.0]) + result = apply_aggregation(df, interval="1d", method="time_weighted_mean") + assert len(result) == 2 + + def test_respects_min_values_and_on_sparse(self): + # Jan 1 has 4 obs (passes), Jan 2 has 1 obs (fails min_values=3) + ts = _JAN1_TS + [_utc(2024, 1, 2, 6)] + df = _make_df(ts, _JAN1_VALS + [5.0]) + result = apply_aggregation( + df, interval="1d", method="time_weighted_mean", min_values=3, on_sparse="drop" + ) + assert len(result) == 1 + + # --------------------------------------------------------------------------- # min_values and on_sparse # ---------------------------------------------------------------------------