diff --git a/.gitignore b/.gitignore index 53a6bf94..e9222661 100644 --- a/.gitignore +++ b/.gitignore @@ -84,4 +84,5 @@ yarn-error.log* # Claude Code .claude/ +.isaac/ CLAUDE.md \ No newline at end of file diff --git a/docs/impulse/docs/data_model/index.md b/docs/impulse/docs/data_model/index.md index e18b0c24..fc123425 100644 --- a/docs/impulse/docs/data_model/index.md +++ b/docs/impulse/docs/data_model/index.md @@ -60,7 +60,7 @@ The Gold layer uses a **star schema** with fact and dimension tables. All table | `event_instance_fact` | One row per event instance per container | Materialized time windows where an event condition holds. | | `histogram_fact` | One row per bin per container | 1D histogram bin values, duration-weighted. | | `histogram2d_fact` | One row per (x, y) bin per container | 2D histogram bin values, duration-weighted. | -| `stats_aggregator_fact` | One row per signal per event instance | Descriptive statistics (min, max, mean, median). | +| `stats_aggregator_fact` | One row per statistic label per signal per event instance | Descriptive statistics (built-in min/max/mean/median and any custom statistics). | | `calculated_channel_fact` | One row per sample interval per container | Materialized derived signal (a *channel*, not a summary), in the silver `channels` shape. | ### Dimension tables diff --git a/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md b/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md new file mode 100644 index 00000000..86e529db --- /dev/null +++ b/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md @@ -0,0 +1,119 @@ +--- +sidebar_label: custom_statistic +title: impulse_query_engine.analyze.query.aggregations.custom_statistic +--- + +Descriptors for custom statistics (per-channel and cross-channel). + + +## PerChannelStatistic + +```python +class PerChannelStatistic() +``` + +Descriptor for a single per-channel custom statistic. + +**Arguments**: + +- `func` (`Callable`): Function with signature ``func(series: SampleSeries, t_start: float, +t_end: float, **params) -> Sequence[float]``. Called once per input +channel and event interval with the channel's series clipped to the +interval. It must return a sequence of scalars whose length equals +``aggregation_labels`` (a single label still requires a one-element +sequence, e.g. ``[value]``). The series may be empty; return +``float("nan")`` entries for undefined results. The function is +cloudpickled to Spark executors, so a module-level importable function is +recommended; never capture Spark objects. +- `aggregation_labels` (`list of str`): Output labels this statistic produces. The values returned by ``func`` +are mapped positionally to these labels, which become the keys of the +statistic's result maps. Labels must be non-empty, unique strings; +changing them changes the aggregation's definition hash. +- `params` (`dict`): Keyword arguments passed to ``func`` on every invocation +(``func(series, t_start, t_end, **params)``). Keys must be valid Python +identifiers matching parameter names of ``func``. Changing params +changes the aggregation's definition hash. + +## CrossChannelStatistic + +```python +class CrossChannelStatistic() +``` + +Descriptor for a single cross-channel custom statistic. + +**Arguments**: + +- `func` (`Callable`): Function with signature ``func(series: list[SampleSeries], t_start: float, +t_end: float, **params) -> Sequence[float]``. Called once per event +interval with the series listed in ``inputs`` (clipped to the interval, +in declared order). It must return a sequence of scalars whose length +equals ``aggregation_labels`` (a single label still requires a +one-element sequence, e.g. ``[value]``). Any series may be empty; return +``float("nan")`` entries for undefined results. The function is +cloudpickled to Spark executors, so a module-level importable function is +recommended; never capture Spark objects. +- `aggregation_labels` (`list of str`): Output labels this statistic produces. The values returned by ``func`` +are mapped positionally to these labels, which become the keys of the +statistic's result maps. Labels must be non-empty, unique strings; +changing them changes the aggregation's definition hash. +- `inputs` (`list of str`): Names of the input channels the function requires, resolved against the +aggregator's ``input_names``. ``None`` (default) passes all input +channels in input order. +- `channel_name` (`str`): A channel name applied to all of the statistic's output rows. Consumed by +downstream consumers (e.g. the reporting layer) only; ignored by the +query engine. ``None`` (default) leaves it to the consumer, which +typically falls back to each output's ``aggregation_label``. +- `params` (`dict`): Keyword arguments passed to ``func`` on every invocation +(``func(series, t_start, t_end, **params)``). Keys must be valid Python +identifiers matching parameter names of ``func``. Changing params +changes the aggregation's definition hash. + +#### normalize\_per\_channel\_statistics + +```python +def normalize_per_channel_statistics( + per_channel_custom_statistics: list[PerChannelStatistic] | None +) -> list[PerChannelStatistic] +``` + +Validate a per-channel statistics list. + +**Arguments**: + +- `per_channel_custom_statistics` (`list or None`): List of ``PerChannelStatistic`` descriptors. + +**Raises**: + +- `TypeError`: If the value is not a list, an item is not a ``PerChannelStatistic`` +with a callable ``func``, or params/labels are invalid. +- `ValueError`: If a descriptor's ``aggregation_labels`` are not unique. + +**Returns**: + +`list of PerChannelStatistic`: The validated list; empty when the input is ``None``. + +#### normalize\_cross\_channel\_statistics + +```python +def normalize_cross_channel_statistics( + cross_channel_custom_statistics: list[CrossChannelStatistic] | None +) -> list[CrossChannelStatistic] +``` + +Validate a cross-channel statistics list. + +**Arguments**: + +- `cross_channel_custom_statistics` (`list or None`): List of ``CrossChannelStatistic`` descriptors. + +**Raises**: + +- `TypeError`: If the value is not a list, an item is not a ``CrossChannelStatistic`` +with a callable ``func``, or params/labels are invalid. +- `ValueError`: If a descriptor's ``aggregation_labels`` are not unique. + +**Returns**: + +`list of CrossChannelStatistic`: The validated list; empty when the input is ``None``. + diff --git a/docs/impulse/docs/references/api/impulse_reporting/aggregations/stats_aggregator.md b/docs/impulse/docs/references/api/impulse_reporting/aggregations/stats_aggregator.md index 441df870..9782ce85 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/aggregations/stats_aggregator.md +++ b/docs/impulse/docs/references/api/impulse_reporting/aggregations/stats_aggregator.md @@ -28,7 +28,11 @@ def __init__(name: str, event: Event | None = None, desc: str = None, agg_type: str = "stats_aggregator", - values_unit: str = None) + values_unit: str = None, + cross_channel_custom_statistics: list[CrossChannelStatistic] + | None = None, + per_channel_custom_statistics: list[PerChannelStatistic] + | None = None) ``` Initialize a StatsAggregator object. @@ -44,6 +48,16 @@ are computed over the entire time series. - `desc` (`str`): Description of the aggregation. - `agg_type` (`str`): Type of aggregation, defaults to "stats_aggregator". - `values_unit` (`str`): Unit of the statistic values. +- `cross_channel_custom_statistics` (`list of CrossChannelStatistic`): Custom statistics computed once per event interval across each +descriptor's declared input channels (referencing ``channel_names``; +all channels when none are declared). Fact rows carry the +descriptor's ``channel_name`` (applied to all its output labels), +defaulting to each output's ``aggregation_label``. See +``impulse_query_engine`` ``StatsAggregator`` for the callable contract. +- `per_channel_custom_statistics` (`list of PerChannelStatistic`): Custom statistics computed once per input channel and event interval, +exactly like a built-in statistic. Fact rows carry the real channel +names. Descriptor ``params`` are passed to the callable as keyword +arguments. #### get\_id @@ -176,8 +190,14 @@ Only includes computation-affecting attributes: - input_expressions - statistics to be calculated - event expression if there is any - -Excludes: name, desc, signal_name, units, page_number, report_id +- custom statistics (name, kind, declared input indices, and function + bytecode, so implementation or input-wiring changes invalidate cached + results; only appended when custom statistics are configured so + aggregators without them keep their previous hash) + +Excludes: name, desc, signal_name, units, page_number, report_id, and the +cross-channel descriptors' channel_name (presentation metadata, like +channel_names). **Returns**: diff --git a/docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md b/docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md index 44947f91..f2eadaf8 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md +++ b/docs/impulse/docs/references/api/impulse_reporting/channels/calculated_channel.md @@ -20,8 +20,8 @@ the narrow result to a gold fact table, and update it incrementally. Structurally parallels :class:`BasicEvent` (holds an aliased expression, name-derived id, SHA-256 definition hash) but — like ``ContainerEvent`` — it drives its own solve via ``QueryBuilder.solve_calculated_channels`` rather than -riding the centralized wide ``solved_df``. Accordingly :meth:`get_expression` -returns ``None`` so it is excluded from the batch solve. +riding the centralized wide ``solved_df``. It is dispatched separately from +the batch solve (never passed to ``collect_solvable_expressions``). **Arguments**: @@ -78,13 +78,10 @@ Return the deterministic entity id (also the fact/dimension ``channel_id``). #### get\_expression ```python -def get_expression() -> TimeSeriesExpression | None +def get_expression() -> TimeSeriesExpression ``` -Return ``None`` — calculated channels drive their own narrow solve. - -Returning ``None`` keeps this channel out of the centralized wide batch -solve (``collect_solvable_expressions``), mirroring ``ContainerEvent``. +Return the wrapped query-engine ``CalculatedChannel`` expression. #### get\_expression\_str diff --git a/docs/impulse/docs/references/api/sidebar.json b/docs/impulse/docs/references/api/sidebar.json index 0c79f03c..f919f467 100644 --- a/docs/impulse/docs/references/api/sidebar.json +++ b/docs/impulse/docs/references/api/sidebar.json @@ -17,6 +17,7 @@ "items": [ { "items": [ + "references/api/impulse_query_engine/analyze/query/aggregations/custom_statistic", "references/api/impulse_query_engine/analyze/query/aggregations/statistic_type" ], "label": "impulse_query_engine.analyze.query.aggregations", diff --git a/docs/impulse/docs/references/report/aggregation.md b/docs/impulse/docs/references/report/aggregation.md index fc5f561f..33bc424b 100644 --- a/docs/impulse/docs/references/report/aggregation.md +++ b/docs/impulse/docs/references/report/aggregation.md @@ -254,6 +254,8 @@ page.add_aggregation(stats) | `desc` | `str` | No | Description. | | `agg_type` | `str` | No | Aggregation type identifier. Defaults to `"stats_aggregator"`. | | `values_unit` | `str` | No | Unit of the statistic values. | +| `per_channel_custom_statistics` | `list[PerChannelStatistic]` | No | Custom statistics computed once per channel per interval (see below). | +| `cross_channel_custom_statistics`| `list[CrossChannelStatistic]`| No | Custom statistics computed once per interval across a set of channels (see below). | ### Supported statistics @@ -268,6 +270,77 @@ page.add_aggregation(stats) A `ValueError` is raised if unsupported statistics are provided. +### Custom statistics + +Beyond the built-in labels, you can inject your own statistic functions. Each is declared as a descriptor +(`PerChannelStatistic` / `CrossChannelStatistic`, see the +[custom_statistic API reference](../api/impulse_query_engine/analyze/query/aggregations/custom_statistic.md)) +and returns a **sequence of scalars** mapped positionally to its `aggregation_labels`; a single-output +statistic returns a one-element sequence. Custom outputs land in the fact table as ordinary +`aggregation_label` / `statistic_value` rows — no schema change — and their labels are added to the +dimension row's `statistics` array. + +There are two kinds: + +- **Per-channel** (`PerChannelStatistic`) — computed once per input channel per interval, exactly like a + built-in. Fact rows carry the **real channel name**. Signature: + `func(series: SampleSeries, t_start: float, t_end: float, **params) -> Sequence[float]`. +- **Cross-channel** (`CrossChannelStatistic`) — computed once per interval across a set of channels. Fact + rows carry the descriptor's `channel_name` (or the output label when it is `None`). Signature: + `func(series: list[SampleSeries], t_start: float, t_end: float, **params) -> Sequence[float]`. + +```python +import numpy as np +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + PerChannelStatistic, + CrossChannelStatistic, +) + +def percentiles(series, t_start, t_end): # per-channel, multi-output + return [float(np.nanpercentile(series.values, 50)), + float(np.nanpercentile(series.values, 90))] + +def spread(series, t_start, t_end): # cross-channel, single output + values = [v for s in series for v in s.values] + return [float(max(values) - min(values))] if values else [float("nan")] + +stats = StatsAggregator( + name="custom_stats", + input_expressions=[eng_rpm, veh_spd], + channel_names=["Engine RPM", "Vehicle Speed"], + statistics=["min", "max"], + event=eng_rpm_event, + per_channel_custom_statistics=[ + PerChannelStatistic(func=percentiles, aggregation_labels=["p50", "p90"]), + ], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=spread, aggregation_labels=["spread"], + channel_name="rpm_speed_spread"), + ], +) +``` + +**Descriptor fields** + +| Field | Applies to | Description | +|----------------------|-------------------|-----------------------------------------------------------------------------------------| +| `func` | both | Callable returning a sequence of scalars (length must equal `aggregation_labels`). | +| `aggregation_labels` | both (required) | Output labels; become `aggregation_label`s in the fact table. Non-empty, unique. | +| `params` | both | Keyword arguments passed to `func` on every call. Keys must be valid identifiers. | +| `inputs` | cross-channel | Names (from `channel_names`) of the channels to pass, in order. `None` passes all. | +| `channel_name` | cross-channel | `channel_name` for all output rows. `None` uses each output's `aggregation_label`. | + +Notes: + +- Labels must be **globally unique** across built-ins and all custom statistics, and must not shadow a + built-in label. +- `func` is cloudpickled to Spark executors: use a module-level importable function and never capture + Spark objects. +- A statistic's `func`, labels, resolved inputs, `params`, and (for cross-channel) `channel_name` are part + of the definition hash, so changing any of them triggers reprocessing. `channel_name` is included because + it is the fact table's merge key — a rename must recompute so old-name rows are pruned rather than left + stale in incremental mode. + ### Output schema **stats_aggregator_fact:** @@ -276,10 +349,10 @@ A `ValueError` is raised if unsupported statistics are provided. |---------------------|----------|--------------------------------------------------| | `container_id` | `int` | Container identifier. | | `visual_id` | `int` | Foreign key to `stats_aggregator_dimension`. | -| `channel_name` | `str` | Signal display name. | +| `channel_name` | `str` | Signal display name (or a cross-channel statistic's `channel_name`). | | `event_id` | `int` | Event identifier. | | `event_instance_id` | `long` | Foreign key to `event_instance_fact`. | -| `aggregation_label` | `str` | Statistic label (e.g. `"mean"`). | +| `aggregation_label` | `str` | Statistic label — a built-in (e.g. `"mean"`) or a custom `aggregation_labels` entry. | | `statistic_value` | `double` | Computed statistic value. | **stats_aggregator_dimension:** @@ -292,7 +365,7 @@ A `ValueError` is raised if unsupported statistics are provided. | `page_number` | `int` | Page number. | | `description` | `str` | Description. | | `agg_type` | `str` | Aggregation type identifier. | -| `statistics` | `array[str]` | Requested statistic labels. | +| `statistics` | `array[str]` | All output labels: built-ins plus every custom statistic's `aggregation_labels`. | | `channel_names` | `array[str]` | Signal display names. | | `signal_expressions` | `array[str]` | TSAL expression strings. | | `values_unit` | `str` | Unit of statistic values. | diff --git a/docs/impulse/docs/references/report/index.md b/docs/impulse/docs/references/report/index.md index 7dbb0ed5..82700fdf 100644 --- a/docs/impulse/docs/references/report/index.md +++ b/docs/impulse/docs/references/report/index.md @@ -114,10 +114,10 @@ Only the hashed attributes matter. Anything else is cosmetic and won't trigger r | `ContainerEvent` | `name` | | `Histogram` | `base_expr`, `bins`, `event` | | `Histogram2D` | `x_expr`, `y_expr`, `x_bins`, `y_bins`, `event` | -| `StatsAggregator` | `input_expressions`, `statistics`, `event` | +| `StatsAggregator` | `input_expressions`, `statistics`, `event`, `channel_names`, custom statistics (labels, functions, inputs, params, cross-channel `channel_name`) | | `CalculatedChannel` | `expr` string + `identity` | -Renaming an aggregation, tweaking the description, or swapping `channel_name` or units keeps the hash stable. No reprocessing. +Renaming an aggregation, tweaking the description, or changing units keeps the hash stable. No reprocessing. `channel_names` (and a cross-channel statistic's `channel_name`) **do** affect the hash for `StatsAggregator` / `PointValueAggregator`, because they are the fact table's `channel_name` merge key — renaming forces a recompute so old-name rows are pruned rather than left stale. #### Container-update detection diff --git a/docs/impulse/pydoc-markdown.yml b/docs/impulse/pydoc-markdown.yml index 3a07927c..aa28ace6 100644 --- a/docs/impulse/pydoc-markdown.yml +++ b/docs/impulse/pydoc-markdown.yml @@ -19,6 +19,7 @@ loaders: - impulse_query_engine.analyze.metadata.tag_expression - impulse_query_engine.analyze.metadata.metric_expression - impulse_query_engine.analyze.query.aggregations.statistic_type + - impulse_query_engine.analyze.query.aggregations.custom_statistic - impulse_query_engine.analyze.query.solvers.default_solver - impulse_query_engine.analyze.query.solvers.solver_config - impulse_query_engine.analyze.query.solvers.query_solver diff --git a/skills/impulse-aggregations/SKILL.md b/skills/impulse-aggregations/SKILL.md index ffcd77db..999295ec 100644 --- a/skills/impulse-aggregations/SKILL.md +++ b/skills/impulse-aggregations/SKILL.md @@ -128,10 +128,54 @@ page.add_aggregation(stats) | `event` | `Event` | No | Scope; if omitted, covers the entire series. | | `desc` | `str` | No | Description. | | `values_unit` | `str` | No | Unit of the statistic values. | +| `per_channel_custom_statistics` | `list[PerChannelStatistic]` | No | Custom stats computed per channel per interval. | +| `cross_channel_custom_statistics` | `list[CrossChannelStatistic]` | No | Custom stats computed per interval across channels. | -Supported statistics: `"min"`, `"max"`, `"mean"` (duration-weighted), `"median"` (duration-weighted), +Supported built-in statistics: `"min"`, `"max"`, `"mean"` (duration-weighted), `"median"` (duration-weighted), `"start"` (first value), `"end"` (last value). Unsupported labels raise `ValueError`. +### Custom statistics + +Inject your own statistic functions via descriptors from +`impulse_query_engine.analyze.query.aggregations.custom_statistic`. Each `func` returns a **sequence** of +scalars mapped positionally to its `aggregation_labels` (a single-output stat returns a one-element +sequence, e.g. `[value]`). Custom outputs become ordinary `aggregation_label` / `statistic_value` fact rows +(no schema change) and their labels join the dimension's `statistics` array. + +- **`PerChannelStatistic(func, aggregation_labels, params={})`** — once per channel per interval; fact rows + keep the real channel name. `func(series: SampleSeries, t_start, t_end, **params) -> Sequence[float]`. +- **`CrossChannelStatistic(func, aggregation_labels, inputs=None, channel_name=None, params={})`** — once + per interval over `inputs` (all channels when `None`); fact rows use `channel_name` (or the label when + `None`). `func(series: list[SampleSeries], t_start, t_end, **params) -> Sequence[float]`. + +```python +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + PerChannelStatistic, CrossChannelStatistic, +) + +def percentiles(series, t_start, t_end): + return [float(np.nanpercentile(series.values, 50)), + float(np.nanpercentile(series.values, 90))] + +stats = StatsAggregator( + name="custom_stats", + input_expressions=[eng_rpm, veh_spd], + channel_names=["Engine RPM", "Vehicle Speed"], + statistics=["min", "max"], + event=container_event, + per_channel_custom_statistics=[ + PerChannelStatistic(func=percentiles, aggregation_labels=["p50", "p90"]), + ], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=corr, aggregation_labels=["corr"], channel_name="rpm_vs_speed"), + ], +) +``` + +Labels must be globally unique and must not shadow a built-in. `func` is cloudpickled to executors, so use +a module-level importable function and never capture Spark objects. Labels, `func`, `inputs`, and `params` +are part of the definition hash. + ## PointValueAggregator Samples channels **at the instants of a `PointsInTimeEvent`** (see `impulse-events`) — one value per diff --git a/skills/impulse-data-model/SKILL.md b/skills/impulse-data-model/SKILL.md index a4e26a78..e26bf674 100644 --- a/skills/impulse-data-model/SKILL.md +++ b/skills/impulse-data-model/SKILL.md @@ -94,7 +94,7 @@ A star schema. Every table is prefixed with your configured `table_prefix` (e.g. | `event_instance_fact` | One row per event instance per container | | `histogram_fact` | One row per bin per container | | `histogram2d_fact` | One row per (x, y) bin per container | -| `stats_aggregator_fact` | One row per signal per event instance | +| `stats_aggregator_fact` | One row per statistic label per signal per event instance | | `calculated_channel_fact` | One row per sample interval per container (a derived signal, silver `channels` shape) | **Dimension tables** @@ -105,7 +105,7 @@ A star schema. Every table is prefixed with your configured `table_prefix` (e.g. | `event_dimension` | Event definitions (name, TSAL expression, required channels). | | `histogram_dimension` | Histogram metadata (bins, signal info, units). | | `histogram2d_dimension` | 2D histogram metadata. | -| `stats_aggregator_dimension` | Statistics metadata (channel names, labels). | +| `stats_aggregator_dimension` | Statistics metadata (channel names, statistic labels incl. custom). | | `calculated_channel_dimension` | Calculated-channel definitions (name, expression, identity). See `impulse-channels`. | **Join pattern** — key columns connect facts to dimensions: diff --git a/skills/impulse-reporting/SKILL.md b/skills/impulse-reporting/SKILL.md index 1894ef73..d4b32d3f 100644 --- a/skills/impulse-reporting/SKILL.md +++ b/skills/impulse-reporting/SKILL.md @@ -129,7 +129,7 @@ are cosmetic and do not trigger reprocessing: | `ContainerEvent` | `name` | | `Histogram` | `base_expr`, `bins`, `event` | | `Histogram2D` | `x_expr`, `y_expr`, `x_bins`, `y_bins`, `event` | -| `StatsAggregator` | `input_expressions`, `statistics`, `event` | +| `StatsAggregator` | `input_expressions`, `statistics`, `event`, `channel_names`, custom stats (incl. cross-channel `channel_name`) | | `CalculatedChannel` | `expr` string + `identity` | **Container-update detection** unions two sets: new containers (silver rows absent from gold) and diff --git a/src/impulse_query_engine/analyze/query/aggregations/custom_statistic.py b/src/impulse_query_engine/analyze/query/aggregations/custom_statistic.py new file mode 100644 index 00000000..62d081b9 --- /dev/null +++ b/src/impulse_query_engine/analyze/query/aggregations/custom_statistic.py @@ -0,0 +1,203 @@ +"""Descriptors for custom statistics (per-channel and cross-channel).""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class PerChannelStatistic: + """ + Descriptor for a single per-channel custom statistic. + + Parameters + ---------- + func : Callable + Function with signature ``func(series: SampleSeries, t_start: float, + t_end: float, **params) -> Sequence[float]``. Called once per input + channel and event interval with the channel's series clipped to the + interval. It must return a sequence of scalars whose length equals + ``aggregation_labels`` (a single label still requires a one-element + sequence, e.g. ``[value]``). The series may be empty; return + ``float("nan")`` entries for undefined results. The function is + cloudpickled to Spark executors, so a module-level importable function is + recommended; never capture Spark objects. + aggregation_labels : list of str + Output labels this statistic produces. The values returned by ``func`` + are mapped positionally to these labels, which become the keys of the + statistic's result maps. Labels must be non-empty, unique strings; + changing them changes the aggregation's definition hash. + params : dict, optional + Keyword arguments passed to ``func`` on every invocation + (``func(series, t_start, t_end, **params)``). Keys must be valid Python + identifiers matching parameter names of ``func``. Changing params + changes the aggregation's definition hash. + """ + + func: Callable + aggregation_labels: list[str] + params: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class CrossChannelStatistic: + """ + Descriptor for a single cross-channel custom statistic. + + Parameters + ---------- + func : Callable + Function with signature ``func(series: list[SampleSeries], t_start: float, + t_end: float, **params) -> Sequence[float]``. Called once per event + interval with the series listed in ``inputs`` (clipped to the interval, + in declared order). It must return a sequence of scalars whose length + equals ``aggregation_labels`` (a single label still requires a + one-element sequence, e.g. ``[value]``). Any series may be empty; return + ``float("nan")`` entries for undefined results. The function is + cloudpickled to Spark executors, so a module-level importable function is + recommended; never capture Spark objects. + aggregation_labels : list of str + Output labels this statistic produces. The values returned by ``func`` + are mapped positionally to these labels, which become the keys of the + statistic's result maps. Labels must be non-empty, unique strings; + changing them changes the aggregation's definition hash. + inputs : list of str, optional + Names of the input channels the function requires, resolved against the + aggregator's ``input_names``. ``None`` (default) passes all input + channels in input order. + channel_name : str, optional + A channel name applied to all of the statistic's output rows. Consumed by + downstream consumers (e.g. the reporting layer) only; ignored by the + query engine. ``None`` (default) leaves it to the consumer, which + typically falls back to each output's ``aggregation_label``. + params : dict, optional + Keyword arguments passed to ``func`` on every invocation + (``func(series, t_start, t_end, **params)``). Keys must be valid Python + identifiers matching parameter names of ``func``. Changing params + changes the aggregation's definition hash. + """ + + func: Callable + aggregation_labels: list[str] + inputs: list[str] | None = None + channel_name: str | None = None + params: dict[str, Any] = field(default_factory=dict) + + +def _validate_params(param_name: str, labels: list[str], params: dict[str, Any]) -> None: + """Validate a statistic's params mapping (dict with identifier keys; None allowed).""" + if params is None: + return + if not isinstance(params, dict): + raise TypeError( + f"{param_name} statistic {labels!r}: params must be a dict, " + f"got {type(params).__name__}" + ) + for key in params: + if not isinstance(key, str) or not key.isidentifier(): + raise TypeError( + f"{param_name} statistic {labels!r}: params keys must be valid Python " + f"identifiers (they are passed as keyword arguments), got {key!r}" + ) + + +def _validate_aggregation_labels(param_name: str, labels: Any) -> None: + """Validate a statistic's aggregation_labels (non-empty list of unique, non-empty strings).""" + if not isinstance(labels, list) or not labels: + raise TypeError( + f"{param_name}: aggregation_labels is required and must be a non-empty " + f"list of strings, got {labels!r}" + ) + for label in labels: + if not isinstance(label, str) or not label: + raise TypeError( + f"{param_name}: aggregation_labels must contain non-empty strings, " + f"got {label!r}" + ) + if len(set(labels)) != len(labels): + raise ValueError(f"{param_name}: aggregation_labels must be unique, got {labels!r}") + + +def normalize_per_channel_statistics( + per_channel_custom_statistics: list[PerChannelStatistic] | None, +) -> list[PerChannelStatistic]: + """ + Validate a per-channel statistics list. + + Parameters + ---------- + per_channel_custom_statistics : list or None + List of ``PerChannelStatistic`` descriptors. + + Returns + ------- + list of PerChannelStatistic + The validated list; empty when the input is ``None``. + + Raises + ------ + TypeError + If the value is not a list, an item is not a ``PerChannelStatistic`` + with a callable ``func``, or params/labels are invalid. + ValueError + If a descriptor's ``aggregation_labels`` are not unique. + """ + return _normalize_statistics( + "per_channel_custom_statistics", per_channel_custom_statistics, PerChannelStatistic + ) + + +def normalize_cross_channel_statistics( + cross_channel_custom_statistics: list[CrossChannelStatistic] | None, +) -> list[CrossChannelStatistic]: + """ + Validate a cross-channel statistics list. + + Parameters + ---------- + cross_channel_custom_statistics : list or None + List of ``CrossChannelStatistic`` descriptors. + + Returns + ------- + list of CrossChannelStatistic + The validated list; empty when the input is ``None``. + + Raises + ------ + TypeError + If the value is not a list, an item is not a ``CrossChannelStatistic`` + with a callable ``func``, or params/labels are invalid. + ValueError + If a descriptor's ``aggregation_labels`` are not unique. + """ + return _normalize_statistics( + "cross_channel_custom_statistics", cross_channel_custom_statistics, CrossChannelStatistic + ) + + +def _normalize_statistics(param_name: str, statistics: list | None, descriptor_type: type) -> list: + """Validate a list of custom-statistic descriptors of a single kind.""" + if statistics is None: + return [] + if not isinstance(statistics, list): + raise TypeError( + f"{param_name} must be a list of {descriptor_type.__name__} descriptors, " + f"got {type(statistics).__name__}" + ) + for statistic in statistics: + if not isinstance(statistic, descriptor_type): + raise TypeError( + f"{param_name} items must be {descriptor_type.__name__} descriptors, " + f"got {type(statistic).__name__}" + ) + _validate_aggregation_labels(param_name, statistic.aggregation_labels) + if not callable(statistic.func): + raise TypeError( + f"{param_name} statistic {statistic.aggregation_labels!r}: func must be " + f"callable, got {type(statistic.func).__name__}" + ) + _validate_params(param_name, statistic.aggregation_labels, statistic.params) + return statistics diff --git a/src/impulse_query_engine/analyze/query/aggregations/statistic_type.py b/src/impulse_query_engine/analyze/query/aggregations/statistic_type.py index 667f92b7..27718fc8 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/statistic_type.py +++ b/src/impulse_query_engine/analyze/query/aggregations/statistic_type.py @@ -17,9 +17,15 @@ class StatisticType(Enum): Mean (average) value statistic. MEDIAN : str Median value statistic. + START : str + First value in the interval. + END : str + Last value in the interval. """ MIN = "min" MAX = "max" MEAN = "mean" MEDIAN = "median" + START = "start" + END = "end" diff --git a/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py b/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py index 19b99f00..18c7f5af 100644 --- a/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py +++ b/src/impulse_query_engine/analyze/query/aggregations/stats_aggregator.py @@ -1,5 +1,7 @@ """StatsAggregator class for computing statistics within event intervals.""" +from collections.abc import Callable, Sequence + import numpy as np import pyspark.sql.types as T @@ -14,10 +16,18 @@ from impulse_query_engine.model.series.sample_series import SampleSeries from .aggregation import Aggregation +from .custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, + normalize_cross_channel_statistics, + normalize_per_channel_statistics, +) -# Define supported statistics and their types +# Define supported statistics and their types. StatisticType covers the numeric +# built-ins (min/max/mean/median and start/end); string statistics are reserved. NUMERIC_STATISTICS = {stat.value for stat in StatisticType} STRING_STATISTICS = {} +BUILTIN_STATISTIC_NAMES = NUMERIC_STATISTICS | set(STRING_STATISTICS) class StatsAggregator(Aggregation): @@ -27,13 +37,24 @@ class StatsAggregator(Aggregation): This aggregator evaluates input expressions to get SampleSeries instances, filters them by event intervals, and computes the requested statistics for each interval. + + Besides the built-in statistics, two kinds of custom statistics can be + injected: + + - ``per_channel_custom_statistics`` are computed once per input channel and + interval, exactly like built-ins, and ride in ``numeric_values``. + - ``cross_channel_custom_statistics`` are computed once per interval across + their declared input channels and are returned in ``cross_channel_values``. """ def __init__( self, input_expressions: list[TimeSeriesExpression], - statistics: list[str], + statistics: list[str] | None = None, event_expression: TimeSeriesExpression = None, + cross_channel_custom_statistics: list[CrossChannelStatistic] | None = None, + per_channel_custom_statistics: list[PerChannelStatistic] | None = None, + input_names: list[str] | None = None, ): """ Initialize a StatsAggregator. @@ -43,7 +64,7 @@ def __init__( input_expressions : list of TimeSeriesExpression List of TimeSeriesExpression instances to compute statistics on. When evaluated, each expression will yield a SampleSeries. - statistics : list of str + statistics : list of str, optional List of statistic types to compute (e.g., ['min', 'max', 'mean', 'median']). Supported numeric statistics: 'min', 'max', 'mean', 'median'. Supported string statistics: 'mode', 'unique_count'. @@ -51,16 +72,135 @@ def __init__( event_expression : TimeSeriesExpression TimeSeriesExpression defining event intervals for statistics computation. When evaluated, it yields an instance of Intervals. + cross_channel_custom_statistics : list of CrossChannelStatistic, optional + Custom statistics computed once per event interval: + ``func(series: list[SampleSeries], t_start: float, t_end: float, + **params) -> Sequence[float]``. The series are clipped to the + interval and ordered like the descriptor's ``inputs`` declaration + (all input expressions in input order when no inputs are declared). + ``func`` returns a sequence mapped positionally to the descriptor's + ``aggregation_labels``. Series may be empty; return ``float("nan")`` + entries for undefined results. Exceptions propagate and fail the + query. Callables are cloudpickled to Spark executors — use + module-level importable functions and never capture Spark objects. + per_channel_custom_statistics : list of PerChannelStatistic, optional + Custom statistics computed once per input channel and event interval, + exactly like a built-in statistic: ``func(series: SampleSeries, + t_start: float, t_end: float, **params) -> Sequence[float]``. The + series is clipped to the interval and may be empty (unlike built-ins, + the callable is invoked even for empty/all-NaN intervals). The same + return and pickling guidance as for cross-channel statistics applies. + input_names : list of str, optional + Names of the input expressions, parallel to ``input_expressions``. + Required when any cross-channel statistic declares ``inputs``. """ self.input_expressions = input_expressions self.event_expression = event_expression - self.statistics = statistics + self.statistics = statistics if statistics is not None else [] + self.input_names = input_names + self.cross_channel_custom_statistics = normalize_cross_channel_statistics( + cross_channel_custom_statistics + ) + self.per_channel_custom_statistics = normalize_per_channel_statistics( + per_channel_custom_statistics + ) + self._validate_custom_statistic_labels() + self._validate_input_names() + self._cross_channel_input_indices = self._resolve_cross_channel_inputs() # Separate numeric and string statistics for processing - self._numeric_stats = [ - s for s in statistics if s in NUMERIC_STATISTICS or s in {"start", "end"} - ] - self._string_stats = [s for s in statistics if s in STRING_STATISTICS] + self._numeric_stats = [s for s in self.statistics if s in NUMERIC_STATISTICS] + self._string_stats = [s for s in self.statistics if s in STRING_STATISTICS] + + def _validate_custom_statistic_labels(self) -> None: + """ + Ensure custom statistic output labels collide neither with built-ins nor + each other. + + Collisions are rejected because two outputs sharing an + ``aggregation_label`` would produce ambiguous, colliding result-map keys. + + Raises + ------ + ValueError + If a custom output label shadows a built-in statistic or is produced + by more than one custom statistic. + """ + all_labels: list[str] = [] + for statistic in self.per_channel_custom_statistics: + all_labels.extend(statistic.aggregation_labels) + for statistic in self.cross_channel_custom_statistics: + all_labels.extend(statistic.aggregation_labels) + + builtin_collisions = set(all_labels) & BUILTIN_STATISTIC_NAMES + if builtin_collisions: + raise ValueError( + "Custom statistic output labels collide with built-in statistics: " + f"{sorted(builtin_collisions)}" + ) + duplicates = sorted({label for label in all_labels if all_labels.count(label) > 1}) + if duplicates: + raise ValueError( + "Custom statistic output labels must be unique across all custom " + f"statistics; duplicated: {duplicates}" + ) + + def _validate_input_names(self) -> None: + """ + Validate ``input_names`` against ``input_expressions``. + + Raises + ------ + ValueError + If ``input_names`` has a different length than ``input_expressions`` + or contains duplicate names. + """ + if self.input_names is None: + return + if len(self.input_names) != len(self.input_expressions): + raise ValueError( + f"Length mismatch: input_names has {len(self.input_names)} elements, " + f"but input_expressions has {len(self.input_expressions)} elements." + ) + if len(set(self.input_names)) != len(self.input_names): + raise ValueError(f"input_names must be unique, got {self.input_names}") + + def _resolve_cross_channel_inputs(self) -> list[list[int] | None]: + """ + Resolve each cross-channel statistic's declared inputs to expression indices. + + Returns + ------- + list of (list of int or None) + Parallel to ``cross_channel_custom_statistics``: per statistic, the + indices into ``input_expressions`` in declared order, or ``None`` + when the statistic consumes all inputs. + + Raises + ------ + ValueError + If inputs are declared without ``input_names`` or reference a name + that is not in ``input_names``. + """ + indices: list[list[int] | None] = [] + for statistic in self.cross_channel_custom_statistics: + if statistic.inputs is None: + indices.append(None) + continue + if self.input_names is None: + raise ValueError( + f"Cross-channel statistic {statistic.aggregation_labels!r} declares " + f"inputs {statistic.inputs}, but no input_names were provided." + ) + unknown = [ch for ch in statistic.inputs if ch not in self.input_names] + if unknown: + raise ValueError( + f"Cross-channel statistic {statistic.aggregation_labels!r} references " + f"unknown input channels {unknown}; available input_names: " + f"{self.input_names}" + ) + indices.append([self.input_names.index(ch) for ch in statistic.inputs]) + return indices def __str__(self) -> str: """ @@ -71,9 +211,18 @@ def __str__(self) -> str: str String representation of the StatsAggregator object. """ + cross_channel = [ + {"labels": statistic.aggregation_labels, "inputs": statistic.inputs} + for statistic in self.cross_channel_custom_statistics + ] + per_channel = [ + statistic.aggregation_labels for statistic in self.per_channel_custom_statistics + ] return ( f"" + f"event_expression={self.event_expression}, statistics={self.statistics}, " + f"cross_channel_custom_statistics={cross_channel}, " + f"per_channel_custom_statistics={per_channel}>" ) def dtype(self) -> T.StructType: @@ -83,6 +232,8 @@ def dtype(self) -> T.StructType: The schema supports a dynamic number of statistics with different types: - Numeric statistics (min, max, mean, median, start, end) as DoubleType - String statistics (mode, unique_count) as StringType + - Cross-channel custom statistics as one map per event interval + (interval-ordered; empty array when none are configured) Returns ------- @@ -106,12 +257,20 @@ def dtype(self) -> T.StructType: T.ArrayType(T.ArrayType(T.MapType(T.StringType(), T.StringType()))), nullable=True, ), + T.StructField( + "cross_channel_values", + T.ArrayType(T.MapType(T.StringType(), T.DoubleType())), + nullable=True, + ), ] ) - def build( - self, cache: SeriesCache - ) -> tuple[list[list[float]], list[list[dict[str, float]]], list[list[dict[str, str]]]]: + def build(self, cache: SeriesCache) -> tuple[ + list[list[float]], + list[list[dict[str, float]]], + list[list[dict[str, str]]], + list[dict[str, float]], + ]: """ Build the statistics aggregation from the cache. @@ -129,12 +288,16 @@ def build( Returns ------- tuple - A 3-tuple containing: - - event_timestamps: List of [start, end] pairs for each event interval - - numeric_values: List of lists of dicts with numeric statistics per - input expression and interval + A 4-tuple containing: + - event_timestamps: List of [start, end] pairs, one per non-degenerate + event interval (canonical interval order) + - numeric_values: List (per input expression) of lists of dicts with + numeric statistics, each inner list aligned with ``event_timestamps`` - string_values: List of lists of dicts with string statistics per input expression and interval (if any) + - cross_channel_values: One dict of cross-channel statistics per + non-degenerate event interval, aligned with ``event_timestamps``; + empty when no cross-channel statistics are configured """ # Step 1: Evaluate input expressions to get SampleSeries instances sample_series_list: list[SampleSeries] = [] @@ -164,25 +327,33 @@ def build( intervals = self.event_expression.build(cache) sample_series_filtered = [s.where(intervals) for s in sample_series_list] - event_timestamps = [] + # Canonical list of non-degenerate intervals, in order. event_timestamps, + # each numeric_values[signal], and cross_channel_values are all aligned to it. + non_degenerate = [ + (interval[0], interval[1]) + for interval in intervals.get_data() + if interval[1] != interval[0] + ] + event_timestamps = [[t_start, t_end] for t_start, t_end in non_degenerate] + numeric_values = [] string_values = [] - for series in sample_series_filtered: - numeric_values_in_series = [] - for interval in intervals.get_data(): - t_start = interval[0] - t_end = interval[1] - - if t_end == t_start: - continue - event_timestamps.append([t_start, t_end]) - numeric_values_in_series.append( + numeric_values.append( + [ self._calculate_aggregations(series, t_start, t_end) - ) + for t_start, t_end in non_degenerate + ] + ) + + cross_channel_values = [] + if self.cross_channel_custom_statistics: + cross_channel_values = [ + self._calculate_cross_channel_statistics(sample_series_filtered, t_start, t_end) + for t_start, t_end in non_degenerate + ] - numeric_values.append(numeric_values_in_series) - return (event_timestamps, numeric_values, string_values) + return (event_timestamps, numeric_values, string_values, cross_channel_values) def _calculate_aggregations(self, sample_series, t_start, t_end) -> dict[str, float]: """ @@ -191,6 +362,10 @@ def _calculate_aggregations(self, sample_series, t_start, t_end) -> dict[str, fl Samples that fall in ``[t_start, t_end]`` are expected to already lie inside those bounds (clipped upstream by ``SampleSeries.where`` in ``build``, or naturally within them when no event expression is set). + + Built-in statistics are NaN for empty/all-NaN intervals; per-channel custom + statistics are always invoked (possibly with an empty series) and decide + their own undefined-value handling. """ mask = (sample_series.tends > t_start) & (sample_series.tstarts < t_end) @@ -203,31 +378,148 @@ def _calculate_aggregations(self, sample_series, t_start, t_end) -> dict[str, fl results = {} if values.size == 0 or np.all(np.isnan(values)): - return {stat: np.nan for stat in self.statistics} - - for stat in self.statistics: - if stat == "start": - results["start"] = sample_series.values[mask][0] - elif stat == "end": - results["end"] = sample_series.values[mask][-1] - elif stat == "min": - results["min"] = np.nanmin(sample_series.values[mask]) - elif stat == "max": - results["max"] = np.nanmax(sample_series.values[mask]) - elif stat == "mean": - mean = np.divide(np.nansum(values * durations), np.nansum(durations)) - results["mean"] = mean - elif stat == "median": - results["median"] = float(self.weighted_median(durations=durations, values=values)) - else: - raise ValueError( - f"Unsupported statistic type: {stat}\n" - "Available options are 'min', 'max', 'mean', " - "'median', 'start', 'end'." + results = {stat: np.nan for stat in self.statistics} + else: + for stat in self.statistics: + if stat == "start": + results["start"] = sample_series.values[mask][0] + elif stat == "end": + results["end"] = sample_series.values[mask][-1] + elif stat == "min": + results["min"] = np.nanmin(sample_series.values[mask]) + elif stat == "max": + results["max"] = np.nanmax(sample_series.values[mask]) + elif stat == "mean": + mean = np.divide(np.nansum(values * durations), np.nansum(durations)) + results["mean"] = mean + elif stat == "median": + results["median"] = float( + self.weighted_median(durations=durations, values=values) + ) + else: + raise ValueError( + f"Unsupported statistic type: {stat}\n" + "Available options are 'min', 'max', 'mean', " + "'median', 'start', 'end'." + ) + + if self.per_channel_custom_statistics: + channel_series = SampleSeries(tstarts=t_starts, tends=t_ends, values=values) + for statistic in self.per_channel_custom_statistics: + results.update( + self._map_labeled_result( + statistic, + statistic.func(channel_series, t_start, t_end, **(statistic.params or {})), + ) ) return results + def _calculate_cross_channel_statistics( + self, series_list: list[SampleSeries], t_start: float, t_end: float + ) -> dict[str, float]: + """ + Compute all cross-channel custom statistics for the interval ``[t_start, t_end]``. + + Each statistic receives the series of its declared inputs (all inputs when + none are declared), clipped to the interval and in declared order. Clipping + is memoized per channel so channels shared between statistics are clipped + only once per interval. + + Parameters + ---------- + series_list : list of SampleSeries + The evaluated input expressions, already filtered by the event intervals. + t_start : float + Interval start time. + t_end : float + Interval end time. + + Returns + ------- + dict of str to float + One value per cross-channel statistic. + """ + interval = Intervals(tstarts=[t_start], tends=[t_end]) + clipped: dict[int, SampleSeries] = {} + + def clip(index: int) -> SampleSeries: + if index not in clipped: + clipped[index] = series_list[index].where(interval) + return clipped[index] + + results = {} + for statistic, indices in zip( + self.cross_channel_custom_statistics, + self._cross_channel_input_indices, + strict=True, + ): + if indices is None: + indices = range(len(series_list)) + statistic_series = [clip(i) for i in indices] + results.update( + self._map_labeled_result( + statistic, + statistic.func(statistic_series, t_start, t_end, **(statistic.params or {})), + ) + ) + return results + + def _map_labeled_result( + self, + statistic: PerChannelStatistic | CrossChannelStatistic, + value, + ) -> dict[str, float]: + """ + Map a custom statistic's returned sequence to its output labels. + + The statistic must return a sequence of scalars whose length equals its + ``aggregation_labels``; the values are mapped positionally. + + Raises + ------ + TypeError + If the statistic returns a non-sequence (or a string). + ValueError + If the returned sequence length does not match the declared labels. + """ + labels = statistic.aggregation_labels + # Accept any sized, ordered container (list, tuple, numpy array, ...) but + # not scalars, strings, or bytes. + if isinstance(value, (str, bytes)) or not isinstance(value, (Sequence, np.ndarray)): + raise TypeError( + f"Custom statistic {labels!r} must return a sequence of " + f"{len(labels)} scalars, got {type(value).__name__}" + ) + if len(value) != len(labels): + raise ValueError( + f"Custom statistic {labels!r} returned {len(value)} values but declares " + f"{len(labels)} aggregation_labels" + ) + return { + label: self._coerce_stat_result(label, v) + for label, v in zip(labels, value, strict=True) + } + + @staticmethod + def _coerce_stat_result(name: str, value) -> float: + """ + Coerce a custom statistic's return value to ``float``. + + Raises + ------ + TypeError + If the value is not convertible to ``float``; the message names the + offending statistic. + """ + try: + return float(value) + except (TypeError, ValueError) as exc: + raise TypeError( + f"Custom statistic '{name}' must return a float-convertible scalar, " + f"got {type(value).__name__}" + ) from exc + def required_tags(self) -> set[str]: """ Return the union of required tags across all input expressions and event expression. diff --git a/src/impulse_reporting/aggregations/point_value_aggregator.py b/src/impulse_reporting/aggregations/point_value_aggregator.py index 8b9c205d..790796aa 100644 --- a/src/impulse_reporting/aggregations/point_value_aggregator.py +++ b/src/impulse_reporting/aggregations/point_value_aggregator.py @@ -363,8 +363,10 @@ def determine_definition_hash(self) -> int: """ Calculate the definition hash for the aggregation. - Only includes computation-affecting attributes: input expressions and the event - expression. Excludes name, description, units, page_number, report_id. + Only includes computation-affecting attributes: input expressions, the event + expression, and channel_names (the fact table's ``channel_name`` merge key, so + a rename forces a recompute). Excludes name, description, units, page_number, + report_id. Returns ------- @@ -383,6 +385,7 @@ def determine_definition_hash(self) -> int: input_expr_strs, # Input expressions "value", # constant aggregation label event_expr_str, # Event (points) expression + repr(self.channel_names), # fact-table channel_name merge key ] hash_input = "::".join(hash_components) diff --git a/src/impulse_reporting/aggregations/stats_aggregator.py b/src/impulse_reporting/aggregations/stats_aggregator.py index ca5b83c2..1bc38d3a 100644 --- a/src/impulse_reporting/aggregations/stats_aggregator.py +++ b/src/impulse_reporting/aggregations/stats_aggregator.py @@ -2,6 +2,7 @@ from __future__ import annotations +import functools import hashlib from collections.abc import Callable @@ -15,6 +16,12 @@ from impulse_query_engine.analyze.metadata.time_series_expression import ( TimeSeriesExpression, ) +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, + normalize_cross_channel_statistics, + normalize_per_channel_statistics, +) from impulse_query_engine.analyze.query.aggregations.stats_aggregator import ( StatsAggregator as QueryEngineStatsAggregator, ) @@ -47,6 +54,8 @@ def __init__( desc: str = None, agg_type: str = "stats_aggregator", values_unit: str = None, + cross_channel_custom_statistics: list[CrossChannelStatistic] | None = None, + per_channel_custom_statistics: list[PerChannelStatistic] | None = None, ): """ Initialize a StatsAggregator object. @@ -70,6 +79,18 @@ def __init__( Type of aggregation, defaults to "stats_aggregator". values_unit : str, optional Unit of the statistic values. + cross_channel_custom_statistics : list of CrossChannelStatistic, optional + Custom statistics computed once per event interval across each + descriptor's declared input channels (referencing ``channel_names``; + all channels when none are declared). Fact rows carry the + descriptor's ``channel_name`` (applied to all its output labels), + defaulting to each output's ``aggregation_label``. See + ``impulse_query_engine`` ``StatsAggregator`` for the callable contract. + per_channel_custom_statistics : list of PerChannelStatistic, optional + Custom statistics computed once per input channel and event interval, + exactly like a built-in statistic. Fact rows carry the real channel + names. Descriptor ``params`` are passed to the callable as keyword + arguments. """ Aggregation.__init__(self, name) self.input_expressions = input_expressions @@ -80,6 +101,13 @@ def __init__( SampleSeries, owner="StatsAggregator", example="a channel selection" ) self.statistics = statistics + self.cross_channel_custom_statistics = normalize_cross_channel_statistics( + cross_channel_custom_statistics + ) + self._validate_cross_channel_channel_names() + self.per_channel_custom_statistics = normalize_per_channel_statistics( + per_channel_custom_statistics + ) self.event = event self._validate_event_evaluation_type( self.event, Intervals, example="(channel > 2000) & (channel < 5000)" @@ -89,6 +117,25 @@ def __init__( self.values_unit = values_unit self.expression = self._set_expression() + def _validate_cross_channel_channel_names(self) -> None: + """ + Validate the ``channel_name`` of each cross-channel statistic descriptor. + + Raises + ------ + ValueError + If a descriptor's ``channel_name`` is set but not a non-empty string. + """ + for statistic in self.cross_channel_custom_statistics: + if statistic.channel_name is None: + continue + if not isinstance(statistic.channel_name, str) or not statistic.channel_name: + raise ValueError( + f"channel_name of cross-channel statistic " + f"{statistic.aggregation_labels!r} must be a non-empty string, " + f"got {statistic.channel_name!r}" + ) + def _validate_channel_names(self) -> None: """ Validate that channel_names and input_expressions have the same length. @@ -172,10 +219,27 @@ def _set_expression(self) -> TimeSeriesExpression: input_expressions=self.input_expressions, statistics=self.statistics, event_expression=self.event.get_expression() if self.event else None, + cross_channel_custom_statistics=self.cross_channel_custom_statistics, + per_channel_custom_statistics=self.per_channel_custom_statistics, + input_names=self.channel_names, ).alias(self.name) return query_eng_stats_agg + @staticmethod + def _custom_statistic_labels( + statistics: list[PerChannelStatistic | CrossChannelStatistic], + ) -> list[str]: + """ + Flatten custom statistics to the output labels they produce. + + These are the values that appear as ``aggregation_label`` in the fact rows. + """ + labels: list[str] = [] + for statistic in statistics: + labels.extend(statistic.aggregation_labels) + return labels + def as_dict(self) -> dict: """ Get a dictionary representation of the statistics aggregation. @@ -192,7 +256,11 @@ def as_dict(self) -> dict: "page_number": self.page_number, "description": self.desc, "agg_type": self.agg_type if self.agg_type else "stats_aggregator", - "statistics": self.statistics, + "statistics": ( + list(self.statistics) + + self._custom_statistic_labels(self.per_channel_custom_statistics) + + self._custom_statistic_labels(self.cross_channel_custom_statistics) + ), "channel_names": self.channel_names, "signal_expressions": [expr.__str__() for expr in self.input_expressions], "values_unit": self.values_unit, @@ -254,6 +322,12 @@ def determine_aggregations( result = solved_df.select("container_id", *stats_names) + # Single pass over the solved struct: per-channel and cross-channel stats + # are exploded together (see _explode_stats_values), so ``result`` and the + # upstream solve are traversed only once. The two channel-name passes run in + # sequence over the one frame: _add_channel_name_column names signal_index + # >= 0 rows, then _add_cross_channel_name_column names the signal_index == -1 + # rows while preserving the per-channel names already set. df = ( result.transform(StatsAggregator._unpivot_measurement_info(stats_names)) .transform(StatsAggregator._extract_stats_info) @@ -261,6 +335,7 @@ def determine_aggregations( .transform(StatsAggregator._add_event_name_column(aggregations)) .transform(StatsAggregator._explode_stats_values) .transform(StatsAggregator._add_channel_name_column(aggregations)) + .transform(StatsAggregator._add_cross_channel_name_column(aggregations)) .transform(StatsAggregator._add_event_instance_id_column) .transform(StatsAggregator._add_visual_id_column(aggregations)) .select(STATS_AGGREGATOR_FACT_SCHEMA.fieldNames()) @@ -312,6 +387,7 @@ def _extract_stats_info(df: DataFrame) -> DataFrame: df.withColumn("event_timestamps", f.col("value.event_timestamps")) .withColumn("numeric_values", f.col("value.numeric_values")) .withColumn("string_values", f.col("value.string_values")) + .withColumn("cross_channel_values", f.col("value.cross_channel_values")) ) @staticmethod @@ -388,35 +464,50 @@ def _(df: DataFrame) -> DataFrame: @staticmethod def _explode_stats_values(df: DataFrame) -> DataFrame: """ - Explode the statistics values into individual rows per signal and interval. + Explode per-channel and cross-channel statistics into one row per + (signal, interval, statistic) in a single pass. + + Per-channel rows carry ``signal_index`` 0..N-1 (mapped to a real channel + name downstream); cross-channel rows carry ``signal_index = -1``. Both are + exploded from a single combined signal axis so ``df`` (and the upstream + solve) is traversed only once. Parameters ---------- df : pyspark.sql.DataFrame - DataFrame containing nested statistics values. + DataFrame containing ``event_timestamps``, ``numeric_values`` and + ``cross_channel_values``. Returns ------- pyspark.sql.DataFrame DataFrame with exploded statistics for each signal and interval. """ - # Step 1: Explode by signal index to get one row per signal - # numeric_values is array of arrays: [[{stats for interval 0}, {stats for interval 1}], ...] - # Each outer array element corresponds to a signal + # Step 1: Explode by signal index to get one row per signal. + # + # Per-channel stats live in ``numeric_values`` (array>, one inner + # list per signal). Cross-channel stats live in ``cross_channel_values`` + # (array, one map per interval) — exactly one signal's worth. Prepend + # them as the first entry of the signal axis so both explode in a single + # pass; ``signal_index = pos - 1`` then maps cross-channel to -1 and real + # signals to 0..N-1. This avoids forking ``df`` (which would re-run the + # upstream solve). An empty ``cross_channel_values`` prepends ``[[]]`` — a + # length-0 interval list that zips to zero rows, so no spurious -1 row. + all_signal_values = f.concat( + f.array(f.col("cross_channel_values")), f.col("numeric_values") + ) df_with_signal = df.select( "container_id", "stats_name", "event_id", "event_name", "event_timestamps", - f.posexplode(f.col("numeric_values")).alias( - "signal_index", "signal_stats_per_interval" - ), - ) + f.posexplode(all_signal_values).alias("pos", "signal_stats_per_interval"), + ).withColumn("signal_index", f.col("pos") - 1) - # Step 2: Explode by interval - zip event_timestamps with signal_stats_per_interval - # event_timestamps: [[start, end], [start, end], ...] - # signal_stats_per_interval: [{stats}, {stats}, ...] + # Step 2: Explode by interval - zip event_timestamps with signal_stats_per_interval. + # Both are aligned per interval (event_timestamps is canonical, one entry per + # interval), so the zip lines up 1:1. df_with_interval = df_with_signal.select( "container_id", "stats_name", @@ -452,6 +543,72 @@ def _explode_stats_values(df: DataFrame) -> DataFrame: f.explode(f.col("statistics")).alias("aggregation_label", "statistic_value"), ) + @staticmethod + def _add_cross_channel_name_column( + aggregations: list[StatsAggregator], + ) -> Callable[..., DataFrame]: + """ + Add a channel_name column for cross-channel statistic rows. + + This pass runs after ``_add_channel_name_column`` over the same frame, so + it only touches cross-channel rows (``signal_index == -1``) and preserves + the per-channel ``channel_name`` already set on ``signal_index >= 0`` rows. + + For a cross-channel row, a descriptor with an explicit ``channel_name`` + applies it to all of that descriptor's output rows (matched on its + ``aggregation_labels``); cross-channel rows without an explicit + ``channel_name`` default to their ``aggregation_label``, which is non-null + and stable as required by the fact table's merge keys. A ``channel_name`` + equal to a real input channel name is allowed and pivots the statistic + into that channel's rows. + + Parameters + ---------- + aggregations : list of StatsAggregator + List of StatsAggregator visual aggregations. + + Returns + ------- + function + Function that adds the channel_name column to a DataFrame. + """ + + def _(df: DataFrame) -> DataFrame: + is_cross_channel = f.col("signal_index") == f.lit(-1) + col_expr = None + for agg in aggregations: + if agg is None: + continue + agg_name = agg.get_name() + for statistic in agg.cross_channel_custom_statistics: + if statistic.channel_name is None: + continue + labels = statistic.aggregation_labels + condition = ( + is_cross_channel + & (f.col("stats_name") == f.lit(agg_name)) + & (f.col("aggregation_label").isin(labels)) + ) + if col_expr is None: + col_expr = f.when(condition, f.lit(statistic.channel_name)) + else: + col_expr = col_expr.when(condition, f.lit(statistic.channel_name)) + + # Cross-channel rows without an explicit descriptor channel_name default + # to their aggregation_label; per-channel rows keep the channel_name the + # previous pass set (never overwritten here). + cross_channel_default = f.when(is_cross_channel, f.col("aggregation_label")).otherwise( + f.col("channel_name") + ) + channel_name_column = ( + col_expr.otherwise(cross_channel_default) + if col_expr is not None + else cross_channel_default + ) + return df.withColumn("channel_name", channel_name_column) + + return _ + @staticmethod def _add_channel_name_column( aggregations: list[StatsAggregator], @@ -572,8 +729,16 @@ def determine_definition_hash(self) -> int: - input_expressions - statistics to be calculated - event expression if there is any + - channel_names, and each cross-channel descriptor's channel_name. These + are the fact table's ``channel_name`` merge key, so a rename must force + a recompute (a changed definition recomputes and prunes all containers); + otherwise, in incremental mode, already-processed containers would keep + rows under the old name. + - custom statistics (labels, kind, declared input indices, params, and + function bytecode, so implementation or input-wiring changes invalidate + cached results; only appended when custom statistics are configured) - Excludes: name, desc, signal_name, units, page_number, report_id + Excludes: name, desc, units, page_number, report_id. Returns ------- @@ -594,9 +759,110 @@ def determine_definition_hash(self) -> int: input_expr_strs, # Input expressions stats_strs, # statistics aggregation types event_expr_str, # Event expression + repr(self.channel_names), # fact-table channel_name merge key ] + + custom_fingerprints = [ + self._fingerprint_custom_statistic( + "per_channel", + statistic.aggregation_labels, + statistic.func, + inputs_repr="", + params=statistic.params, + ) + for statistic in sorted( + self.per_channel_custom_statistics, key=lambda s: tuple(s.aggregation_labels) + ) + ] + for statistic in sorted( + self.cross_channel_custom_statistics, key=lambda s: tuple(s.aggregation_labels) + ): + if statistic.inputs is None: + inputs_repr = "all" + else: + # indices, not names: consistent renames keep the hash stable + inputs_repr = repr([self.channel_names.index(ch) for ch in statistic.inputs]) + custom_fingerprints.append( + self._fingerprint_custom_statistic( + "cross_channel", + statistic.aggregation_labels, + statistic.func, + inputs_repr=inputs_repr, + params=statistic.params, + channel_name=statistic.channel_name, + ) + ) + if custom_fingerprints: + hash_components.append("|".join(custom_fingerprints)) + hash_input = "::".join(hash_components) # Use SHA-256 and return as int (truncated to fit LongType) hash_bytes = hashlib.sha256(hash_input.encode()).digest() return int.from_bytes(hash_bytes[:8], byteorder="big", signed=True) + + @staticmethod + def _fingerprint_custom_statistic( + kind: str, + aggregation_labels: list[str], + func: Callable, + inputs_repr: str, + params: dict | None = None, + channel_name: str | None = None, + ) -> str: + """ + Build a stable fingerprint for a custom statistic. + + The fingerprint covers the statistic's kind, output labels, declared input + indices, provisioned params, the fact-table ``channel_name`` (cross-channel + only), and the function's bytecode, constants, and default argument values + (``__defaults__`` / ``__kwdefaults__``). ``functools.partial`` wrappers are + unwrapped with their bound arguments included, so changed parameters change + the fingerprint. Note that bytecode hashing does not detect changes inside + helper functions called by the statistic or in captured closure variables, + and is sensitive to the Python version. Callables without ``__code__`` fall + back to a labels-only fingerprint. + + Parameters + ---------- + kind : str + Either ``per_channel`` or ``cross_channel``. + aggregation_labels : list of str + The statistic's output labels (its identity). + func : Callable + The statistic's callable. + inputs_repr : str + Representation of the declared input indices. + params : dict, optional + The statistic's provisioned params. + channel_name : str, optional + The cross-channel descriptor's ``channel_name`` (the fact-table merge + key); ``None`` for per-channel statistics and cross-channel statistics + that default to their label. + + Returns + ------- + str + The fingerprint string. + """ + params_repr = repr(sorted((params or {}).items())) + partial_reprs = [] + while isinstance(func, functools.partial): + partial_reprs.append(f"{func.args!r}:{sorted((func.keywords or {}).items())!r}") + func = func.func + code = getattr(func, "__code__", None) + if code is None: + digest = "no-code" + else: + defaults_repr = repr(func.__defaults__) + kwdefaults_repr = repr(sorted((func.__kwdefaults__ or {}).items())) + digest = hashlib.sha256( + code.co_code + + repr(code.co_consts).encode() + + defaults_repr.encode() + + kwdefaults_repr.encode() + + "|".join(partial_reprs).encode() + ).hexdigest() + return ( + f"{kind}:{aggregation_labels!r}:{inputs_repr}:{params_repr}:{channel_name!r}:{digest}" + ) diff --git a/tests/impulse_query_engine/integration/default_solver_test.py b/tests/impulse_query_engine/integration/default_solver_test.py index 61e03581..19486b56 100644 --- a/tests/impulse_query_engine/integration/default_solver_test.py +++ b/tests/impulse_query_engine/integration/default_solver_test.py @@ -143,7 +143,7 @@ def test_solve_with_event_expression_and_stats( assert result.count() == 3 for row in result.collect(): - event_timestamps, numeric_values, _ = row["rpm_when_fast"] + event_timestamps, numeric_values, _, _ = row["rpm_when_fast"] assert len(numeric_values) == 1 for event_stats in numeric_values[0]: assert {"min", "max", "mean"}.issubset(event_stats.keys()) diff --git a/tests/impulse_query_engine/integration/statistics_aggregator_test.py b/tests/impulse_query_engine/integration/statistics_aggregator_test.py index 40e31ac2..80b6bdce 100644 --- a/tests/impulse_query_engine/integration/statistics_aggregator_test.py +++ b/tests/impulse_query_engine/integration/statistics_aggregator_test.py @@ -33,7 +33,7 @@ def test_statistics_single_channel_no_event(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["engine_rpm_stats"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data # Should have one event (entire series) assert len(event_timestamps) == 1 @@ -81,11 +81,11 @@ def test_statistics_multiple_channels_no_event(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["multi_channel_stats"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data - # event_timestamps is appended inside the per-expression loop, - # so N=2 expressions with one synthetic event each give 2 entries. - assert len(event_timestamps) == 2 + # event_timestamps is canonical (one entry per interval), so the single + # synthetic interval spanning both channels gives 1 entry. + assert len(event_timestamps) == 1 # Should have two expressions (eng_rpm and veh_speed) assert len(numeric_values) == 2 @@ -127,7 +127,7 @@ def test_statistics_with_event_expression(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["high_speed_rpm_stats"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data # May have zero or more events depending on data # (some containers may never exceed 50 km/h) @@ -163,7 +163,7 @@ def test_statistics_subset_only_mean(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["rpm_mean_only"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data stats = numeric_values[0][0] # Should only have mean @@ -202,7 +202,7 @@ def test_statistics_with_compound_event(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["temp_during_driving"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data # Should have one expression assert len(numeric_values) == 1 @@ -225,11 +225,13 @@ def test_statistics_aggregator_dtype_in_query(self, spark, basic_narrow_db): schema = result.schema stats_field = schema["rpm_stats"] - # Should be a struct with event_timestamps, numeric_values, string_values + # Should be a struct with event_timestamps, numeric_values, string_values, + # and cross_channel_values struct_fields = {f.name for f in stats_field.dataType.fields} assert "event_timestamps" in struct_fields assert "numeric_values" in struct_fields assert "string_values" in struct_fields + assert "cross_channel_values" in struct_fields def test_compare_statistics_with_native_methods(self, spark, basic_narrow_db): """Compare StatisticsAggregator results with native SampleSeries methods.""" @@ -258,7 +260,7 @@ def test_compare_statistics_with_native_methods(self, spark, basic_narrow_db): rows = result.collect() for row in rows: stats_data = row["agg_stats"] - event_timestamps, numeric_values, string_values = stats_data + event_timestamps, numeric_values, string_values, cross_channel_values = stats_data agg_stats = numeric_values[0][0] diff --git a/tests/impulse_query_engine/integration/stats_aggregator_test.py b/tests/impulse_query_engine/integration/stats_aggregator_test.py index ec03eb63..a34131f0 100644 --- a/tests/impulse_query_engine/integration/stats_aggregator_test.py +++ b/tests/impulse_query_engine/integration/stats_aggregator_test.py @@ -1,11 +1,33 @@ """Integration tests for StatsAggregator with end-to-end usage.""" +import math + +import numpy as np import pytest +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, +) from impulse_query_engine.analyze.query.aggregations.stats_aggregator import StatsAggregator from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver +def _value_spread(series, t_start, t_end): + """Cross-channel statistic: max - min over the values of all series.""" + values = [v for s in series for v in s.values if not np.isnan(v)] + if not values: + return [float("nan")] + return [float(max(values) - min(values))] + + +def _rms(series, t_start, t_end): + """Per-channel statistic: root mean square of the series values.""" + if len(series) == 0: + return [float("nan")] + return [float(np.sqrt(np.nanmean(series.values**2)))] + + def test_stats_case_check_numeric_values(spark, basic_narrow_db): """Test StatsAggregator with single input expression and all statistics.""" query = basic_narrow_db.query @@ -212,3 +234,67 @@ def test_stats_aggregator_with_median(spark, basic_narrow_db): assert ( event_stats["min"] <= event_stats["median"] <= event_stats["max"] ), "median should be between min and max" + + +def test_stats_aggregator_with_custom_statistics(spark, basic_narrow_db): + """End-to-end solve with cross-channel and per-channel custom statistics.""" + query = basic_narrow_db.query + + eng_rpm = query.channel(channel_name="Engine RPM") + veh_spd = query.channel(channel_name="Vehicle Speed Sensor") + air_temp = query.channel(channel_name="Intake Air Temperature") + air_temp_event = air_temp >= 0 + + stats_aggregator = StatsAggregator( + input_expressions=[eng_rpm, veh_spd], + statistics=["min", "max"], + event_expression=air_temp_event, + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_value_spread, + aggregation_labels=["spread"], + inputs=["engine_rpm", "vehicle_speed"], + ), + ], + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + input_names=["engine_rpm", "vehicle_speed"], + ) + + metric_container_id = query.metric("container_id") + df = ( + query.where(metric_container_id == 1) + .select(stats_aggregator.alias("custom_stats")) + .solve(spark, solver=DefaultSolver(spark)) + ) + + rows = df.select("custom_stats").collect() + assert len(rows) == 1 + + my_stats = rows[0]["custom_stats"] + numeric_values = my_stats["numeric_values"] + cross_channel_values = my_stats["cross_channel_values"] + + # one cross-channel map per interval, aligned with the per-channel interval count + assert len(numeric_values) == 2 + assert len(cross_channel_values) > 0 + assert len(cross_channel_values) == len(numeric_values[0]) + + for interval_index, interval_map in enumerate(cross_channel_values): + spread = interval_map["spread"] + assert not math.isnan(spread) + # spread across both channels dominates each channel's own value range + for channel_values in numeric_values: + event_stats = channel_values[interval_index] + assert spread >= event_stats["max"] - event_stats["min"] - 1e-9 + + # per-channel custom statistic lands next to the built-ins, per channel + rms_values = [] + for channel_values in numeric_values: + for event_stats in channel_values: + assert "rms" in event_stats + if not math.isnan(event_stats["rms"]): + rms_values.append(event_stats["rms"]) + # rms of non-negative samples lies within [min, max] + assert event_stats["min"] - 1e-9 <= event_stats["rms"] <= event_stats["max"] + 1e-9 + assert rms_values, "expected at least one non-NaN rms value" + assert all(v > 0 for v in rms_values) diff --git a/tests/impulse_query_engine/unit/analyze/query/aggregations/stats_aggregator_test.py b/tests/impulse_query_engine/unit/analyze/query/aggregations/stats_aggregator_test.py index af552d28..3586d5b2 100644 --- a/tests/impulse_query_engine/unit/analyze/query/aggregations/stats_aggregator_test.py +++ b/tests/impulse_query_engine/unit/analyze/query/aggregations/stats_aggregator_test.py @@ -9,14 +9,20 @@ import numpy as np import numpy.testing as nptest +import pytest from impulse_query_engine.analyze.metadata.tag_expression import TagSelector from impulse_query_engine.analyze.metadata.time_series_expression import ( TimeSeriesSelector, ) +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, +) from impulse_query_engine.analyze.query.aggregations.stats_aggregator import ( StatsAggregator, ) +from impulse_query_engine.model.series.intervals import Intervals from impulse_query_engine.model.series.sample_series import SampleSeries @@ -309,6 +315,13 @@ def test_dtype_contains_expected_fields(): assert "event_timestamps" in field_names assert "numeric_values" in field_names assert "string_values" in field_names + assert "cross_channel_values" in field_names + assert len(dtype.fields) == 4 + + import pyspark.sql.types as T + + cross_channel_field = dtype["cross_channel_values"] + assert cross_channel_field.dataType == T.ArrayType(T.MapType(T.StringType(), T.DoubleType())) def test_required_tags_single_expression(): @@ -498,17 +511,20 @@ def test_build_with_none_event_expression_uses_synced_series_bounds(): statistics=["start", "end"], ) - event_timestamps, numeric_values, string_values = stats_agg.build(cache=None) + event_timestamps, numeric_values, string_values, cross_channel_values = stats_agg.build( + cache=None + ) - # event_timestamps is appended inside the per-expression loop in build(), - # so for N=2 expressions with one synthetic event each the list has 2 entries. - assert event_timestamps == [[0.0, 5.0], [0.0, 5.0]] + # event_timestamps is canonical (one entry per non-degenerate interval), so the + # single synthetic interval spanning both series yields exactly one entry. + assert event_timestamps == [[0.0, 5.0]] assert len(numeric_values) == 2 assert len(numeric_values[0]) == 1 assert len(numeric_values[1]) == 1 assert numeric_values[0][0] == {"start": 1.0, "end": 3.0} assert numeric_values[1][0] == {"start": 10.0, "end": 20.0} assert string_values == [] + assert cross_channel_values == [] def test_has_required_methods(): @@ -578,3 +594,668 @@ def test_stats_aggregator_get_selectors_with_event(): assert len(result) == 2 assert sel in result assert evt in result + + +# --------------------------------------------------------------------------- +# Custom statistics (cross-channel and per-channel) +# --------------------------------------------------------------------------- + + +def _mock_expr(tstarts, tends, values): + """Mock TimeSeriesExpression whose build() returns a SampleSeries.""" + expr = MagicMock() + expr.build.return_value = SampleSeries( + tstarts=np.array(tstarts), tends=np.array(tends), values=np.array(values) + ) + return expr + + +def _mock_event(tstarts, tends): + """Mock event expression whose build() returns Intervals.""" + event = MagicMock() + event.build.return_value = Intervals(tstarts=tstarts, tends=tends) + return event + + +def _spread(series, t_start, t_end): + """Cross-channel: max - min over all values of all series.""" + values = np.concatenate([s.values for s in series]) if series else np.array([]) + valid = values[~np.isnan(values)] + if valid.size == 0: + return [float("nan")] + return [float(np.max(valid) - np.min(valid))] + + +def _total_sample_count(series, t_start, t_end): + """Cross-channel: total number of samples across all series.""" + return [float(sum(len(s) for s in series))] + + +def _rms(series, t_start, t_end): + """Per-channel: root mean square of the series values.""" + if len(series) == 0: + return [float("nan")] + return [float(np.sqrt(np.nanmean(series.values**2)))] + + +def _sample_count(series, t_start, t_end): + """Per-channel: number of samples in the series.""" + return [float(len(series))] + + +def test_build_cross_channel_stat_two_channels_two_intervals(): + expr1 = _mock_expr([0.0, 2.0], [1.0, 3.0], [10.0, 30.0]) + expr2 = _mock_expr([0.0, 2.0], [1.0, 3.0], [20.0, 50.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr1, expr2], + event_expression=event, + statistics=["min"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ], + ) + + _, numeric_values, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 2 + assert len(cross_channel_values) == len(numeric_values[0]) + nptest.assert_almost_equal(cross_channel_values[0]["spread"], 10.0) + nptest.assert_almost_equal(cross_channel_values[1]["spread"], 20.0) + # built-ins still computed per channel + assert numeric_values[0][0] == {"min": 10.0} + assert numeric_values[1][1] == {"min": 50.0} + + +def test_build_multiple_cross_channel_stats(): + expr = _mock_expr([0.0, 2.0], [1.0, 3.0], [10.0, 30.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]), + CrossChannelStatistic(func=_total_sample_count, aggregation_labels=["count"]), + ], + ) + + _, _, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 2 + for interval_map in cross_channel_values: + assert set(interval_map.keys()) == {"spread", "count"} + assert interval_map["count"] == 1.0 + + +def test_build_cross_channel_stat_without_event_expression(): + expr1 = _mock_expr([0.0, 2.0], [2.0, 4.0], [1.0, 3.0]) + expr2 = _mock_expr([1.0, 3.0], [3.0, 5.0], [10.0, 20.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr1, expr2], + event_expression=None, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ], + ) + + _, _, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 1 + nptest.assert_almost_equal(cross_channel_values[0]["spread"], 19.0) + + +def test_cross_channel_inputs_subset_order_and_clipping(): + expr_a = _mock_expr([0.0], [1.0], [1.0]) + expr_b = _mock_expr([0.0], [1.0], [2.0]) + # channel c has a sample spanning the interval boundary at t=1.0 + expr_c = _mock_expr([0.5], [1.5], [3.0]) + event = _mock_event([0.0], [1.0]) + + received = [] + + def capture(series, t_start, t_end): + received.append((series, t_start, t_end)) + return [0.0] + + stats_agg = StatsAggregator( + input_expressions=[expr_a, expr_b, expr_c], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=capture, aggregation_labels=["captured"], inputs=["c", "a"]) + ], + input_names=["a", "b", "c"], + ) + + stats_agg.build(cache=None) + + assert len(received) == 1 + series, t_start, t_end = received[0] + assert (t_start, t_end) == (0.0, 1.0) + # only the declared inputs, in declared order: c first, then a + assert len(series) == 2 + nptest.assert_almost_equal(series[0].values, [3.0]) + nptest.assert_almost_equal(series[1].values, [1.0]) + # the boundary-spanning sample of c is truncated to the interval + assert np.all(series[0].tstarts >= t_start) + assert np.all(series[0].tends <= t_end) + + +def test_cross_channel_stat_called_with_empty_series(): + # channel has samples only within the first interval + expr = _mock_expr([0.0], [1.0], [10.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]), + CrossChannelStatistic(func=_total_sample_count, aggregation_labels=["count"]), + ], + ) + + _, _, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 2 + nptest.assert_almost_equal(cross_channel_values[0]["spread"], 0.0) + assert np.isnan(cross_channel_values[1]["spread"]) + # count-like statistics can return 0 instead of NaN for empty intervals + assert cross_channel_values[1]["count"] == 0.0 + + +def test_build_degenerate_interval_skipped_for_cross_channel(): + expr = _mock_expr([0.0], [1.0], [10.0]) + # trailing zero-length interval survives the Intervals constructor + event = _mock_event([0.0, 2.0], [1.0, 2.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + statistics=["min"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_total_sample_count, aggregation_labels=["count"]) + ], + ) + + _, numeric_values, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 1 + assert len(cross_channel_values) == len(numeric_values[0]) + + +def test_cross_channel_error_propagates_from_build(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + def broken(series, t_start, t_end): + raise RuntimeError("boom") + + stats_agg = StatsAggregator( + input_expressions=[expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=broken, aggregation_labels=["broken"]) + ], + ) + + with pytest.raises(RuntimeError, match="boom"): + stats_agg.build(cache=None) + + +def test_cross_channel_non_sequence_return_raises_type_error(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + def returns_none(series, t_start, t_end): + return None + + stats_agg = StatsAggregator( + input_expressions=[expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=returns_none, aggregation_labels=["bad_stat"]) + ], + ) + + with pytest.raises(TypeError, match="bad_stat"): + stats_agg.build(cache=None) + + +def test_build_custom_only_without_builtin_statistics(): + expr = _mock_expr([0.0], [1.0], [10.0]) + event = _mock_event([0.0], [1.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_total_sample_count, aggregation_labels=["count"]) + ], + ) + + _, numeric_values, _, cross_channel_values = stats_agg.build(cache=None) + + assert numeric_values == [[{}]] + assert cross_channel_values == [{"count": 1.0}] + + +def test_per_channel_custom_stat_alongside_builtins(): + expr1 = _mock_expr([0.0, 2.0], [1.0, 3.0], [3.0, 4.0]) + expr2 = _mock_expr([0.0, 2.0], [1.0, 3.0], [6.0, 8.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr1, expr2], + event_expression=event, + statistics=["min", "max"], + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + ) + + _, numeric_values, _, _ = stats_agg.build(cache=None) + + assert len(numeric_values) == 2 + for signal_values in numeric_values: + assert len(signal_values) == 2 + for interval_map in signal_values: + assert set(interval_map.keys()) == {"min", "max", "rms"} + # single-sample intervals: rms equals the absolute sample value + nptest.assert_almost_equal(numeric_values[0][0]["rms"], 3.0) + nptest.assert_almost_equal(numeric_values[0][1]["rms"], 4.0) + nptest.assert_almost_equal(numeric_values[1][0]["rms"], 6.0) + nptest.assert_almost_equal(numeric_values[1][1]["rms"], 8.0) + + +def test_per_channel_custom_stat_called_for_empty_interval(): + # samples only in the first interval; built-ins NaN-fill the second interval + expr = _mock_expr([0.0], [1.0], [10.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + statistics=["min"], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_sample_count, aggregation_labels=["count"]) + ], + ) + + _, numeric_values, _, _ = stats_agg.build(cache=None) + + assert np.isnan(numeric_values[0][1]["min"]) + assert numeric_values[0][1]["count"] == 0.0 + assert numeric_values[0][0]["count"] == 1.0 + + +def test_per_channel_custom_stat_receives_clipped_series(): + expr = _mock_expr([0.5], [1.5], [3.0]) + event = _mock_event([0.0], [1.0]) + + received = [] + + def capture(series, t_start, t_end): + received.append((series, t_start, t_end)) + return [0.0] + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + per_channel_custom_statistics=[ + PerChannelStatistic(func=capture, aggregation_labels=["captured"]) + ], + ) + + stats_agg.build(cache=None) + + assert len(received) == 1 + series, t_start, t_end = received[0] + assert isinstance(series, SampleSeries) + assert np.all(series.tstarts >= t_start) + assert np.all(series.tends <= t_end) + + +def test_custom_statistics_validation_errors(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + # not a list + with pytest.raises(TypeError, match="cross_channel_custom_statistics"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=CrossChannelStatistic( + func=_spread, aggregation_labels=["spread"] + ), + ) + + # bare callable rejected (must be a descriptor) + with pytest.raises(TypeError, match="descriptors"): + StatsAggregator([expr], cross_channel_custom_statistics=[_spread]) + + with pytest.raises(TypeError, match="descriptors"): + StatsAggregator([expr], per_channel_custom_statistics=[_rms]) + + # built-in label collisions + with pytest.raises(ValueError, match="built-in"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["mean"]) + ], + ) + + with pytest.raises(ValueError, match="built-in"): + StatsAggregator( + [expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_rms, aggregation_labels=["max"]) + ], + ) + + # duplicate label across statistics + with pytest.raises(ValueError, match="unique"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["dup"]) + ], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_rms, aggregation_labels=["dup"]) + ], + ) + + +def test_missing_aggregation_labels_rejected(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + # descriptor built without labels fails at construction (required field) + with pytest.raises(TypeError): + PerChannelStatistic(func=_rms) + + # explicit None is rejected by the normalizer + with pytest.raises(TypeError, match="aggregation_labels is required"): + StatsAggregator( + [expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_rms, aggregation_labels=None) + ], + ) + + +def test_cross_channel_inputs_validation_errors(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + with pytest.raises(ValueError, match="no input_names"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["x"], inputs=["a"]) + ], + ) + + with pytest.raises(ValueError, match="unknown input"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["x"], inputs=["missing"]) + ], + input_names=["a"], + ) + + with pytest.raises(ValueError, match="Length mismatch"): + StatsAggregator([expr], input_names=["a", "b"]) + + with pytest.raises(ValueError, match="unique"): + StatsAggregator([expr, expr], input_names=["a", "a"]) + + +def _count_above(series, t_start, t_end, threshold=0.0): + """Cross-channel: number of samples above a configurable threshold.""" + return [float(sum((s.values > threshold).sum() for s in series))] + + +def _scaled_rms(series, t_start, t_end, scale=1.0): + """Per-channel: scaled root mean square.""" + if len(series) == 0: + return [float("nan")] + return [float(scale * np.sqrt(np.nanmean(series.values**2)))] + + +def test_cross_channel_stat_with_params(): + expr1 = _mock_expr([0.0, 1.0], [1.0, 2.0], [10.0, 30.0]) + expr2 = _mock_expr([0.0, 1.0], [1.0, 2.0], [20.0, 40.0]) + event = _mock_event([0.0], [2.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr1, expr2], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_count_above, aggregation_labels=["count_default"]), + CrossChannelStatistic( + func=_count_above, aggregation_labels=["count_hi"], params={"threshold": 25.0} + ), + ], + ) + + _, _, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 1 + # default threshold 0.0 counts all four samples + assert cross_channel_values[0]["count_default"] == 4.0 + # provisioned threshold 25.0 counts only 30.0 and 40.0 + assert cross_channel_values[0]["count_hi"] == 2.0 + + +def test_per_channel_stat_with_params(): + expr = _mock_expr([0.0], [1.0], [3.0]) + event = _mock_event([0.0], [1.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + per_channel_custom_statistics=[ + PerChannelStatistic(func=_scaled_rms, aggregation_labels=["rms"]), + PerChannelStatistic( + func=_scaled_rms, aggregation_labels=["rms_x10"], params={"scale": 10.0} + ), + ], + ) + + _, numeric_values, _, _ = stats_agg.build(cache=None) + + nptest.assert_almost_equal(numeric_values[0][0]["rms"], 3.0) + nptest.assert_almost_equal(numeric_values[0][0]["rms_x10"], 30.0) + + +def test_custom_statistic_params_validation_errors(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + with pytest.raises(TypeError, match="params must be a dict"): + StatsAggregator( + [expr], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_count_above, aggregation_labels=["x"], params=[1, 2]) + ], + ) + + with pytest.raises(TypeError, match="identifiers"): + StatsAggregator( + [expr], + per_channel_custom_statistics=[ + PerChannelStatistic( + func=_scaled_rms, aggregation_labels=["x"], params={"not a name": 1.0} + ) + ], + ) + + +def test_str_contains_custom_statistics(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + statistics=["min"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"], inputs=["a"]) + ], + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + input_names=["a"], + ) + + str_repr = str(stats_agg) + assert "'labels': ['spread']" in str_repr + assert "'inputs': ['a']" in str_repr + assert "per_channel_custom_statistics=[['rms']]" in str_repr + + +def _min_max_pair(series, t_start, t_end): + """Per-channel multi-output: returns (min, max) as a tuple.""" + if len(series) == 0: + return (float("nan"), float("nan")) + return (float(np.nanmin(series.values)), float(np.nanmax(series.values))) + + +def _spread_and_count(series, t_start, t_end): + """Cross-channel multi-output: returns [spread, total_sample_count].""" + values = np.concatenate([s.values for s in series]) if series else np.array([]) + valid = values[~np.isnan(values)] + spread = float(np.max(valid) - np.min(valid)) if valid.size else float("nan") + return [spread, float(sum(len(s) for s in series))] + + +def test_per_channel_multi_output_maps_labels_positionally(): + expr = _mock_expr([0.0, 2.0], [1.0, 3.0], [10.0, 30.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + statistics=["mean"], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_min_max_pair, aggregation_labels=["lo", "hi"]) + ], + ) + + _, numeric_values, _, _ = stats_agg.build(cache=None) + + for interval_map in numeric_values[0]: + assert set(interval_map.keys()) == {"mean", "lo", "hi"} + nptest.assert_almost_equal(numeric_values[0][0]["lo"], 10.0) + nptest.assert_almost_equal(numeric_values[0][0]["hi"], 10.0) + nptest.assert_almost_equal(numeric_values[0][1]["lo"], 30.0) + nptest.assert_almost_equal(numeric_values[0][1]["hi"], 30.0) + + +def test_cross_channel_multi_output_maps_labels_positionally(): + expr1 = _mock_expr([0.0, 2.0], [1.0, 3.0], [10.0, 30.0]) + expr2 = _mock_expr([0.0, 2.0], [1.0, 3.0], [20.0, 50.0]) + event = _mock_event([0.0, 2.0], [1.0, 3.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr1, expr2], + event_expression=event, + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread_and_count, aggregation_labels=["spread", "n"]) + ], + ) + + _, _, _, cross_channel_values = stats_agg.build(cache=None) + + assert len(cross_channel_values) == 2 + for interval_map in cross_channel_values: + assert set(interval_map.keys()) == {"spread", "n"} + nptest.assert_almost_equal(cross_channel_values[0]["spread"], 10.0) + assert cross_channel_values[0]["n"] == 2.0 + + +def test_single_label_requires_sequence_return(): + expr = _mock_expr([0.0], [1.0], [10.0]) + event = _mock_event([0.0], [1.0]) + + stats_agg = StatsAggregator( + input_expressions=[expr], + event_expression=event, + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + ) + + _, numeric_values, _, _ = stats_agg.build(cache=None) + + nptest.assert_almost_equal(numeric_values[0][0]["rms"], 10.0) + + +def test_multi_output_length_mismatch_raises(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + def one_value(series, t_start, t_end): + return (1.0,) + + stats_agg = StatsAggregator( + input_expressions=[expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=one_value, aggregation_labels=["lo", "hi"]) + ], + ) + + with pytest.raises(ValueError, match="lo.*hi|hi.*lo"): + stats_agg.build(cache=None) + + +def test_multi_output_non_sequence_return_raises(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + def scalar_only(series, t_start, t_end): + return 1.0 + + stats_agg = StatsAggregator( + input_expressions=[expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=scalar_only, aggregation_labels=["lo", "hi"]) + ], + ) + + with pytest.raises(TypeError, match="sequence"): + stats_agg.build(cache=None) + + +def test_multi_output_label_collision_with_builtin_raises(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + with pytest.raises(ValueError, match="built-in"): + StatsAggregator( + input_expressions=[expr], + statistics=["min"], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_min_max_pair, aggregation_labels=["min", "hi"]) + ], + ) + + +def test_multi_output_label_collision_across_statistics_raises(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + with pytest.raises(ValueError, match="unique"): + StatsAggregator( + input_expressions=[expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_min_max_pair, aggregation_labels=["lo", "hi"]) + ], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread_and_count, aggregation_labels=["hi", "n"]) + ], + ) + + +def test_multi_output_label_validation_errors(): + expr = _mock_expr([0.0], [1.0], [10.0]) + + with pytest.raises(TypeError, match="non-empty"): + StatsAggregator( + input_expressions=[expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_min_max_pair, aggregation_labels=[]) + ], + ) + + with pytest.raises(ValueError, match="unique"): + StatsAggregator( + input_expressions=[expr], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_min_max_pair, aggregation_labels=["dup", "dup"]) + ], + ) diff --git a/tests/impulse_reporting/integration/__init__.py b/tests/impulse_reporting/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/impulse_reporting/integration/custom_statistics_test.py b/tests/impulse_reporting/integration/custom_statistics_test.py new file mode 100644 index 00000000..d48adbec --- /dev/null +++ b/tests/impulse_reporting/integration/custom_statistics_test.py @@ -0,0 +1,230 @@ +"""End-to-end report test for custom statistics (cross-channel and per-channel).""" + +import math +import os +from unittest.mock import create_autospec + +import numpy as np +import pyspark.sql.functions as F +from databricks.sdk import WorkspaceClient + +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, +) +from impulse_reporting.aggregations.stats_aggregator import StatsAggregator +from impulse_reporting.core.page import Page +from impulse_reporting.core.report import Report +from impulse_reporting.events.basic_event import BasicEvent + + +def _value_spread(series, t_start, t_end): + """Cross-channel: max - min over the values of all series.""" + values = [v for s in series for v in s.values if not np.isnan(v)] + if not values: + return [float("nan")] + return [float(max(values) - min(values))] + + +def _count_above(series, t_start, t_end, threshold=0.0): + """Cross-channel: number of samples above a configurable threshold.""" + return [float(sum((s.values > threshold).sum() for s in series))] + + +def _rms(series, t_start, t_end): + """Per-channel: root mean square of the series values.""" + if len(series) == 0: + return [float("nan")] + return [float(np.sqrt(np.nanmean(series.values**2)))] + + +def _percentiles(series, t_start, t_end): + """Per-channel multi-output: returns (p50, p90) of the series values.""" + if len(series) == 0: + return (float("nan"), float("nan")) + return ( + float(np.nanpercentile(series.values, 50)), + float(np.nanpercentile(series.values, 90)), + ) + + +def _min_max(series, t_start, t_end): + """Cross-channel multi-output: returns [min, max] over all series values.""" + values = [v for s in series for v in s.values if not np.isnan(v)] + if not values: + return [float("nan"), float("nan")] + return [float(min(values)), float(max(values))] + + +def test_custom_statistics_report(spark): + """Custom statistics flow end-to-end into the stats aggregator fact rows.""" + base_path = os.path.dirname(os.path.abspath(__file__)) + base_path = base_path[: base_path.find("tests")] + config_path = os.path.join(base_path, "tests", "data", "config", "config.json") + + my_report: Report = Report( + name="custom_stats_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config_path=config_path, + ) + + query = my_report.get_db().query + c1 = query.channel(channel_name="Engine RPM") + c2 = query.channel(channel_name="Vehicle Speed Sensor") + + rpm_event = BasicEvent(name="rpm_event", expr=c1 > 0, desc="Engine RPM > 0") + my_report.add_event(rpm_event) + + page = Page(page_number=1) + my_report.add_page(page) + + stats = StatsAggregator( + name="custom_stats", + input_expressions=[c1, c2], + channel_names=["Engine RPM", "Vehicle Speed"], + statistics=["min", "max"], + event=rpm_event, + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_value_spread, + aggregation_labels=["spread"], + inputs=["Engine RPM", "Vehicle Speed"], + channel_name="rpm_speed_spread", + ), + CrossChannelStatistic( + func=_count_above, aggregation_labels=["count"], params={"threshold": 0.0} + ), + ], + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + ) + page.add_aggregation(stats) + + my_report.determine_report() + + assert "STATS_AGGREGATOR" in my_report.aggregation_dfs + stats_df = my_report.aggregation_dfs["STATS_AGGREGATOR"]["changed"] + assert stats_df.count() > 0 + + rows = stats_df.collect() + by_label = {} + for row in rows: + by_label.setdefault(row["aggregation_label"], []).append(row) + + # cross-channel rows: descriptor channel_name and stat-name default + assert {row["channel_name"] for row in by_label["spread"]} == {"rpm_speed_spread"} + assert {row["channel_name"] for row in by_label["count"]} == {"count"} + for row in by_label["spread"]: + assert not math.isnan(row["statistic_value"]) + assert row["statistic_value"] > 0 + for row in by_label["count"]: + assert row["statistic_value"] > 0 + + # per-channel custom rows land under the real channel names, like built-ins + assert {row["channel_name"] for row in by_label["rms"]} == {"Engine RPM", "Vehicle Speed"} + assert {row["channel_name"] for row in by_label["min"]} == {"Engine RPM", "Vehicle Speed"} + + # rms is pivotable alongside the built-ins and lies within [min, max] + # (non-negative sample values) + pivoted = ( + stats_df.groupBy("container_id", "channel_name", "event_instance_id") + .pivot("aggregation_label", ["min", "max", "rms"]) + .agg(F.first("statistic_value")) + .where(F.col("rms").isNotNull() & ~F.isnan("rms")) + .collect() + ) + assert len(pivoted) > 0 + for row in pivoted: + assert row["min"] - 1e-9 <= row["rms"] <= row["max"] + 1e-9 + + # spread dominates each channel's own value range within the same interval + per_channel_range = {} + for row in pivoted: + key = (row["container_id"], row["event_instance_id"]) + value_range = row["max"] - row["min"] + per_channel_range[key] = max(per_channel_range.get(key, 0.0), value_range) + spread_by_instance = { + (row["container_id"], row["event_instance_id"]): row["statistic_value"] + for row in by_label["spread"] + } + joinable = set(spread_by_instance) & set(per_channel_range) + assert joinable, "cross-channel rows must join per-channel rows on event_instance_id" + for key in joinable: + assert spread_by_instance[key] >= per_channel_range[key] - 1e-9 + + # metadata: custom statistic names are documented in the dimension row + metadata_df = my_report.aggregation_metadata_dfs["STATS_AGGREGATOR"] + metadata = metadata_df.collect()[0] + assert metadata["statistics"] == ["min", "max", "rms", "spread", "count"] + + +def test_multi_output_custom_statistics_report(spark): + """Multi-output custom statistics fan out to several fact rows end-to-end.""" + base_path = os.path.dirname(os.path.abspath(__file__)) + base_path = base_path[: base_path.find("tests")] + config_path = os.path.join(base_path, "tests", "data", "config", "config.json") + + my_report: Report = Report( + name="multi_output_stats_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config_path=config_path, + ) + + query = my_report.get_db().query + c1 = query.channel(channel_name="Engine RPM") + c2 = query.channel(channel_name="Vehicle Speed Sensor") + + rpm_event = BasicEvent(name="rpm_event", expr=c1 > 0, desc="Engine RPM > 0") + my_report.add_event(rpm_event) + + page = Page(page_number=1) + my_report.add_page(page) + + stats = StatsAggregator( + name="multi_output_stats", + input_expressions=[c1, c2], + channel_names=["Engine RPM", "Vehicle Speed"], + statistics=["min", "max"], + event=rpm_event, + per_channel_custom_statistics=[ + PerChannelStatistic(func=_percentiles, aggregation_labels=["p50", "p90"]), + ], + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_min_max, aggregation_labels=["lo", "hi"], channel_name="rpm_speed_bounds" + ), + ], + ) + page.add_aggregation(stats) + + my_report.determine_report() + + stats_df = my_report.aggregation_dfs["STATS_AGGREGATOR"]["changed"] + rows = stats_df.collect() + by_label = {} + for row in rows: + by_label.setdefault(row["aggregation_label"], []).append(row) + + # per-channel multi-output labels land under the real channel names + assert {row["channel_name"] for row in by_label["p50"]} == {"Engine RPM", "Vehicle Speed"} + assert {row["channel_name"] for row in by_label["p90"]} == {"Engine RPM", "Vehicle Speed"} + + # cross-channel multi-output labels share the descriptor's channel_name + assert {row["channel_name"] for row in by_label["lo"]} == {"rpm_speed_bounds"} + assert {row["channel_name"] for row in by_label["hi"]} == {"rpm_speed_bounds"} + + # values are finite and hi >= lo for each interval + for label in ("p50", "p90", "lo", "hi"): + for row in by_label[label]: + assert not math.isnan(row["statistic_value"]) + + # per interval instance, hi >= lo + lo_by_instance = {row["event_instance_id"]: row["statistic_value"] for row in by_label["lo"]} + hi_by_instance = {row["event_instance_id"]: row["statistic_value"] for row in by_label["hi"]} + for instance_id, lo in lo_by_instance.items(): + assert hi_by_instance[instance_id] >= lo - 1e-9 + + # the label set is documented in the dimension row + metadata = my_report.aggregation_metadata_dfs["STATS_AGGREGATOR"].collect()[0] + assert metadata["statistics"] == ["min", "max", "p50", "p90", "lo", "hi"] diff --git a/tests/impulse_reporting/unit/aggregations/definition_hash_test.py b/tests/impulse_reporting/unit/aggregations/definition_hash_test.py index 560b105a..c901f3b0 100644 --- a/tests/impulse_reporting/unit/aggregations/definition_hash_test.py +++ b/tests/impulse_reporting/unit/aggregations/definition_hash_test.py @@ -1,6 +1,13 @@ -"""Unit tests for definition hash methods in Histogram and Histogram2D.""" +"""Unit tests for definition hash methods in Histogram, Histogram2D, and StatsAggregator.""" + +import functools +import hashlib from impulse_query_engine.analyze.metadata.time_series_expression import TimeSeriesSelector +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, +) from impulse_reporting.aggregations.histogram import ( HistogramDistance, HistogramDuration, @@ -8,6 +15,7 @@ from impulse_reporting.aggregations.histogram2d import ( Histogram2DDuration, ) +from impulse_reporting.aggregations.stats_aggregator import StatsAggregator from impulse_reporting.events.basic_event import BasicEvent @@ -284,3 +292,333 @@ def test_as_dict_includes_definition_hash(self): assert "definition_hash" in result assert result["definition_hash"] == hist.determine_definition_hash() + + +def _spread(series, t_start, t_end): + """Cross-channel test statistic.""" + values = [v for s in series for v in s.values] + return [float(max(values) - min(values)) if values else float("nan")] + + +def _spread_other_body(series, t_start, t_end): + """Same signature as _spread but a different implementation.""" + values = [v for s in series for v in s.values] + return [float(sum(values)) if values else float("nan")] + + +def _thresholded_count(series, t_start, t_end, threshold=0.0): + """Parameterized statistic for functools.partial tests.""" + return [float(sum((s.values > threshold).sum() for s in series))] + + +def _scaled_default_2(series, t_start, t_end, scale=2.0): + """Statistic whose behavior lives in a default argument (scale=2.0).""" + values = [v for s in series for v in s.values] + return [float(scale * sum(values))] + + +def _scaled_default_3(series, t_start, t_end, scale=3.0): + """Identical body to _scaled_default_2 but a different default (scale=3.0).""" + values = [v for s in series for v in s.values] + return [float(scale * sum(values))] + + +class TestStatsAggregatorDefinitionHash: + """Test suite for StatsAggregator.determine_definition_hash() with custom statistics.""" + + @staticmethod + def _make(name="stats", channel_names=None, **kwargs): + channel_names = channel_names if channel_names is not None else ["ch_a", "ch_b"] + return StatsAggregator( + name=name, + input_expressions=[TimeSeriesSelector(None) for _ in channel_names], + channel_names=channel_names, + statistics=["min", "max"], + event=BasicEvent(name="test_event", expr=TimeSeriesSelector(None) > 0), + **kwargs, + ) + + def test_hash_without_custom_stats_matches_formula(self): + """Aggregators without custom statistics hash input exprs, stats, event, channel_names.""" + stats_agg = self._make() + + event_expr_str = str(stats_agg.event.get_expression()) + input_expr_strs = ",".join(str(expr) for expr in stats_agg.input_expressions) + stats_strs = ",".join(stats_agg.statistics) + hash_input = "::".join( + [input_expr_strs, stats_strs, event_expr_str, repr(stats_agg.channel_names)] + ) + expected = int.from_bytes( + hashlib.sha256(hash_input.encode()).digest()[:8], byteorder="big", signed=True + ) + + assert stats_agg.determine_definition_hash() == expected + + def test_renaming_channel_names_changes_hash(self): + """channel_names is the fact-table merge key, so a rename must force recompute.""" + agg1 = self._make(channel_names=["ch_a", "ch_b"]) + agg2 = self._make(channel_names=["ch_a", "ch_renamed"]) + + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_cross_channel_channel_name_changes_hash(self): + """A cross-channel descriptor's channel_name (a fact merge key) affects the hash.""" + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_spread, aggregation_labels=["spread"], channel_name="combined" + ) + ] + ) + agg2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_spread, aggregation_labels=["spread"], channel_name="other_name" + ) + ] + ) + + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_adding_custom_stat_changes_hash(self): + plain = self._make() + with_cross = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ] + ) + with_per_channel = self._make( + per_channel_custom_statistics=[ + PerChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ] + ) + + assert plain.determine_definition_hash() != with_cross.determine_definition_hash() + assert plain.determine_definition_hash() != with_per_channel.determine_definition_hash() + + def test_same_custom_stats_produce_same_hash(self): + agg1 = self._make( + name="a", + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ], + ) + agg2 = self._make( + name="b", + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ], + ) + + assert agg1.determine_definition_hash() == agg2.determine_definition_hash() + + def test_relabeled_custom_stat_changes_hash(self): + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ] + ) + agg2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread_v2"]) + ] + ) + + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_different_function_body_changes_hash(self): + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"]) + ] + ) + agg2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread_other_body, aggregation_labels=["spread"]) + ] + ) + + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_partial_arguments_change_hash(self): + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=functools.partial(_thresholded_count, threshold=1.0), + aggregation_labels=["count"], + ) + ] + ) + agg2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=functools.partial(_thresholded_count, threshold=2.0), + aggregation_labels=["count"], + ) + ] + ) + + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_kind_is_part_of_hash(self): + """The same label registered per-channel vs cross-channel hashes differently.""" + + def spread(series, t_start, t_end): + return [0.0] + + agg_cross = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=spread, aggregation_labels=["x"]) + ] + ) + agg_per_channel = self._make( + per_channel_custom_statistics=[ + PerChannelStatistic(func=spread, aggregation_labels=["x"]) + ] + ) + + assert agg_cross.determine_definition_hash() != agg_per_channel.determine_definition_hash() + + def test_params_change_hash(self): + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_thresholded_count, + aggregation_labels=["count"], + params={"threshold": 1.0}, + ) + ] + ) + agg1_same = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_thresholded_count, + aggregation_labels=["count"], + params={"threshold": 1.0}, + ) + ] + ) + agg2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_thresholded_count, + aggregation_labels=["count"], + params={"threshold": 2.0}, + ) + ] + ) + + assert agg1.determine_definition_hash() == agg1_same.determine_definition_hash() + assert agg1.determine_definition_hash() != agg2.determine_definition_hash() + + def test_rewiring_cross_channel_inputs_changes_hash(self): + """Pointing a cross-channel statistic at a different input channel changes the hash.""" + agg1 = self._make( + channel_names=["ch_a", "ch_b"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"], inputs=["ch_b"]) + ], + ) + rewired = self._make( + channel_names=["ch_a", "ch_b"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"], inputs=["ch_a"]) + ], + ) + + assert agg1.determine_definition_hash() != rewired.determine_definition_hash() + + def test_cross_channel_inputs_fingerprint_uses_indices(self): + """The inputs portion of the fingerprint is index-based, not name-based.""" + # Same declared-input index (1) under different channel names -> identical + # inputs_repr in the fingerprint, isolating the index-based behavior from the + # channel_names component of the full hash. + fp_b = StatsAggregator._fingerprint_custom_statistic( + "cross_channel", ["spread"], _spread, inputs_repr=repr([1]) + ) + fp_y = StatsAggregator._fingerprint_custom_statistic( + "cross_channel", ["spread"], _spread, inputs_repr=repr([1]) + ) + fp_other = StatsAggregator._fingerprint_custom_statistic( + "cross_channel", ["spread"], _spread, inputs_repr=repr([0]) + ) + + assert fp_b == fp_y + assert fp_b != fp_other + + def test_aggregation_labels_change_hash(self): + labels_ab = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["a", "b"]) + ] + ) + labels_ab_same = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["a", "b"]) + ] + ) + labels_cd = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["c", "d"]) + ] + ) + + assert labels_ab.determine_definition_hash() == labels_ab_same.determine_definition_hash() + assert labels_ab.determine_definition_hash() != labels_cd.determine_definition_hash() + + def test_default_argument_change_alters_hash(self): + """A behavior change carried only by a default argument must change the hash.""" + agg_default_2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_scaled_default_2, aggregation_labels=["scaled"]) + ] + ) + agg_default_3 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_scaled_default_3, aggregation_labels=["scaled"]) + ] + ) + + assert ( + agg_default_2.determine_definition_hash() != agg_default_3.determine_definition_hash() + ) + + def test_same_default_argument_is_stable(self): + agg1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_scaled_default_2, aggregation_labels=["scaled"]) + ] + ) + agg2 = self._make( + name="other", + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_scaled_default_2, aggregation_labels=["scaled"]) + ], + ) + + assert agg1.determine_definition_hash() == agg2.determine_definition_hash() + + def test_positional_only_partial_does_not_crash(self): + """A partial with only positional bound args must hash without raising.""" + # partial bound positionally (no keywords) -> func.keywords is {} on cpython, + # but the fingerprint must tolerate it regardless. + agg_pos_1 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=functools.partial(_thresholded_count, 0.0), + aggregation_labels=["count"], + ) + ] + ) + agg_pos_2 = self._make( + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=functools.partial(_thresholded_count, 5.0), + aggregation_labels=["count"], + ) + ] + ) + + assert isinstance(agg_pos_1.determine_definition_hash(), int) + # different positional bound value -> different fingerprint + assert agg_pos_1.determine_definition_hash() != agg_pos_2.determine_definition_hash() diff --git a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py b/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py index d2eddb43..b3cb22b9 100644 --- a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py +++ b/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py @@ -4,13 +4,19 @@ Tests follow the same pattern as histogram_test.py. """ +import pyspark.sql.types as T import pytest from impulse_query_engine.analyze.metadata.time_series_expression import TimeSeriesSelector +from impulse_query_engine.analyze.query.aggregations.custom_statistic import ( + CrossChannelStatistic, + PerChannelStatistic, +) from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver from impulse_reporting.aggregations.stats_aggregator import StatsAggregator from impulse_reporting.events.basic_event import BasicEvent from impulse_reporting.events.points_in_time_event import PointsInTimeEvent +from impulse_reporting.persist.dimension_schema import STATS_AGGREGATOR_DIMENSION_SCHEMA def test_as_spark_row(): @@ -673,3 +679,308 @@ def test_init_rejects_points_in_time_event(): statistics=["min"], event=PointsInTimeEvent(name="p", expr=TimeSeriesSelector(None).rising_edges()), ) + + +# --------------------------------------------------------------------------- +# Custom statistics (cross-channel and per-channel) +# --------------------------------------------------------------------------- + + +def _spread(series, t_start, t_end): + """Cross-channel test statistic.""" + values = [v for s in series for v in s.values] + return [float(max(values) - min(values)) if values else float("nan")] + + +def _ratio(series, t_start, t_end): + """Cross-channel test statistic.""" + return [0.5] + + +def _rms(series, t_start, t_end): + """Per-channel test statistic.""" + import numpy as np + + return [float(np.sqrt(np.nanmean(series.values**2))) if len(series) else float("nan")] + + +def _make_custom_stats_aggregator(name="my_stats"): + return StatsAggregator( + name=name, + input_expressions=[TimeSeriesSelector(None), TimeSeriesSelector(None)], + channel_names=["ch_a", "ch_b"], + statistics=["min"], + event=BasicEvent(name="test_event", expr=TimeSeriesSelector(None) > 0), + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_spread, + aggregation_labels=["spread"], + inputs=["ch_a", "ch_b"], + channel_name="combined", + ), + CrossChannelStatistic(func=_ratio, aggregation_labels=["ratio"]), + ], + per_channel_custom_statistics=[PerChannelStatistic(func=_rms, aggregation_labels=["rms"])], + ) + + +def test_custom_statistics_normalization_and_as_dict(): + stats_agg = _make_custom_stats_aggregator() + + # descriptors are stored as validated lists + assert stats_agg.cross_channel_custom_statistics[0].channel_name == "combined" + assert stats_agg.cross_channel_custom_statistics[1].channel_name is None + + stats_dict = stats_agg.as_dict() + assert stats_dict["statistics"] == ["min", "rms", "spread", "ratio"] + + # as_spark_row still matches the dimension schema + row = stats_agg.as_spark_row() + assert set(row.asDict().keys()) == set(STATS_AGGREGATOR_DIMENSION_SCHEMA.fieldNames()) + + +def test_cross_channel_empty_channel_name_rejected(): + with pytest.raises(ValueError, match="non-empty"): + StatsAggregator( + name="bad_channel_name", + input_expressions=[TimeSeriesSelector(None)], + channel_names=["ch_a"], + statistics=["min"], + cross_channel_custom_statistics=[ + CrossChannelStatistic(func=_spread, aggregation_labels=["spread"], channel_name="") + ], + ) + + +def test_cross_channel_unknown_input_rejected(): + with pytest.raises(ValueError, match="unknown input"): + StatsAggregator( + name="bad_inputs", + input_expressions=[TimeSeriesSelector(None)], + channel_names=["ch_a"], + statistics=["min"], + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_spread, aggregation_labels=["spread"], inputs=["missing"] + ) + ], + ) + + +def _build_solved_df(spark, stats_name): + """Hand-built solved wide DataFrame with the 4-field stats struct.""" + value_type = T.StructType( + [ + T.StructField("event_timestamps", T.ArrayType(T.ArrayType(T.DoubleType()))), + T.StructField( + "numeric_values", + T.ArrayType(T.ArrayType(T.MapType(T.StringType(), T.DoubleType()))), + ), + T.StructField( + "string_values", + T.ArrayType(T.ArrayType(T.MapType(T.StringType(), T.StringType()))), + ), + T.StructField( + "cross_channel_values", T.ArrayType(T.MapType(T.StringType(), T.DoubleType())) + ), + ] + ) + schema = T.StructType( + [ + T.StructField("container_id", T.IntegerType()), + T.StructField(stats_name, value_type), + ] + ) + # 2 signals x 2 intervals; event_timestamps canonical (one entry per interval) + event_timestamps = [[0.0, 10.0], [10.0, 20.0]] + numeric_values = [ + [{"min": 1.0, "rms": 1.5}, {"min": 2.0, "rms": 2.5}], + [{"min": 3.0, "rms": 3.5}, {"min": 4.0, "rms": 4.5}], + ] + string_values = [] + cross_channel_values = [{"spread": 10.0, "ratio": 0.5}, {"spread": 20.0, "ratio": 0.6}] + return spark.createDataFrame( + [(1, (event_timestamps, numeric_values, string_values, cross_channel_values))], + schema, + ) + + +def test_determine_aggregations_with_custom_statistics(spark): + stats_agg = _make_custom_stats_aggregator() + solved_df = _build_solved_df(spark, stats_agg.get_name()) + + df = StatsAggregator.determine_aggregations(spark, [stats_agg], solved_df=solved_df) + rows = df.collect() + + by_label = {} + for row in rows: + by_label.setdefault(row["aggregation_label"], []).append(row) + + # per-channel rows (built-in + per-channel custom) under real channel names + assert {row["channel_name"] for row in by_label["min"]} == {"ch_a", "ch_b"} + assert {row["channel_name"] for row in by_label["rms"]} == {"ch_a", "ch_b"} + assert sorted(row["statistic_value"] for row in by_label["rms"]) == [1.5, 2.5, 3.5, 4.5] + + # cross-channel rows: descriptor channel_name and stat-name default + assert {row["channel_name"] for row in by_label["spread"]} == {"combined"} + assert sorted(row["statistic_value"] for row in by_label["spread"]) == [10.0, 20.0] + assert {row["channel_name"] for row in by_label["ratio"]} == {"ratio"} + assert sorted(row["statistic_value"] for row in by_label["ratio"]) == [0.5, 0.6] + + # cross-channel rows share the event_instance_id of the same interval's + # per-channel rows (joinable per interval instance) + instance_of_spread_0 = next( + row["event_instance_id"] for row in by_label["spread"] if row["statistic_value"] == 10.0 + ) + instance_of_min_interval_0 = { + row["event_instance_id"] for row in by_label["min"] if row["statistic_value"] in (1.0, 3.0) + } + assert instance_of_min_interval_0 == {instance_of_spread_0} + assert len({row["event_instance_id"] for row in rows}) == 2 + + +def test_determine_aggregations_without_custom_statistics_has_no_extra_rows(spark): + stats_agg = StatsAggregator( + name="plain_stats", + input_expressions=[TimeSeriesSelector(None), TimeSeriesSelector(None)], + channel_names=["ch_a", "ch_b"], + statistics=["min"], + event=BasicEvent(name="test_event", expr=TimeSeriesSelector(None) > 0), + ) + value_type = stats_agg.get_expression().dtype() + schema = T.StructType( + [ + T.StructField("container_id", T.IntegerType()), + T.StructField("plain_stats", value_type), + ] + ) + event_timestamps = [[0.0, 10.0]] + numeric_values = [[{"min": 1.0}], [{"min": 3.0}]] + solved_df = spark.createDataFrame([(1, (event_timestamps, numeric_values, [], []))], schema) + + df = StatsAggregator.determine_aggregations(spark, [stats_agg], solved_df=solved_df) + rows = df.collect() + + assert len(rows) == 2 + assert {row["channel_name"] for row in rows} == {"ch_a", "ch_b"} + assert {row["aggregation_label"] for row in rows} == {"min"} + # single-pass explode: an empty cross_channel_values must not leak a + # signal_index == -1 row, and every row must carry a non-null channel_name. + assert all(row["channel_name"] is not None for row in rows) + + +def test_determine_aggregations_single_pass_preserves_channel_names(spark): + """The cross-channel name pass must not clobber per-channel channel_name. + + Runs both a per-channel custom stat and cross-channel stats through the single + explode+union-free path and asserts per-channel rows keep their real channel + names while cross-channel rows take the descriptor name / label default. + """ + stats_agg = _make_custom_stats_aggregator() + solved_df = _build_solved_df(spark, stats_agg.get_name()) + + rows = StatsAggregator.determine_aggregations( + spark, [stats_agg], solved_df=solved_df + ).collect() + + # No row is left without a channel_name (the merge key must be non-null). + assert all(row["channel_name"] is not None for row in rows) + # Per-channel labels (built-in "min" + per-channel custom "rms") only ever + # appear under the real channel names — never the cross-channel default. + per_channel = [r for r in rows if r["aggregation_label"] in ("min", "rms")] + assert {r["channel_name"] for r in per_channel} == {"ch_a", "ch_b"} + # Cross-channel labels only appear under the descriptor name / label default. + cross_channel = [r for r in rows if r["aggregation_label"] in ("spread", "ratio")] + assert {r["channel_name"] for r in cross_channel} == {"combined", "ratio"} + + +def _bounds(series, t_start, t_end): + """Cross-channel multi-output test statistic.""" + values = [v for s in series for v in s.values] + return [float(min(values)), float(max(values))] if values else [float("nan"), float("nan")] + + +def _quantiles(series, t_start, t_end): + """Per-channel multi-output test statistic.""" + import numpy as np + + if len(series) == 0: + return (float("nan"), float("nan")) + return (float(np.nanpercentile(series.values, 50)), float(np.nanpercentile(series.values, 90))) + + +def _make_multi_output_stats_aggregator(name="multi_stats"): + return StatsAggregator( + name=name, + input_expressions=[TimeSeriesSelector(None), TimeSeriesSelector(None)], + channel_names=["ch_a", "ch_b"], + statistics=["min"], + event=BasicEvent(name="test_event", expr=TimeSeriesSelector(None) > 0), + cross_channel_custom_statistics=[ + CrossChannelStatistic( + func=_bounds, aggregation_labels=["lo", "hi"], channel_name="combined" + ), + ], + per_channel_custom_statistics=[ + PerChannelStatistic(func=_quantiles, aggregation_labels=["p50", "p90"]), + ], + ) + + +def test_multi_output_as_dict_lists_effective_labels(): + stats_agg = _make_multi_output_stats_aggregator() + # per-channel labels then cross-channel labels, after built-ins + assert stats_agg.as_dict()["statistics"] == ["min", "p50", "p90", "lo", "hi"] + + +def test_determine_aggregations_with_multi_output_statistics(spark): + stats_agg = _make_multi_output_stats_aggregator() + + value_type = stats_agg.get_expression().dtype() + schema = T.StructType( + [ + T.StructField("container_id", T.IntegerType()), + T.StructField(stats_agg.get_name(), value_type), + ] + ) + # 2 signals x 2 intervals; event_timestamps canonical (one entry per interval) + event_timestamps = [[0.0, 10.0], [10.0, 20.0]] + numeric_values = [ + [{"min": 1.0, "p50": 1.5, "p90": 1.9}, {"min": 2.0, "p50": 2.5, "p90": 2.9}], + [{"min": 3.0, "p50": 3.5, "p90": 3.9}, {"min": 4.0, "p50": 4.5, "p90": 4.9}], + ] + cross_channel_values = [{"lo": 1.0, "hi": 4.0}, {"lo": 2.0, "hi": 5.0}] + solved_df = spark.createDataFrame( + [(1, (event_timestamps, numeric_values, [], cross_channel_values))], schema + ) + + df = StatsAggregator.determine_aggregations(spark, [stats_agg], solved_df=solved_df) + rows = df.collect() + + by_label = {} + for row in rows: + by_label.setdefault(row["aggregation_label"], []).append(row) + + # per-channel multi-output labels land under real channel names + assert {row["channel_name"] for row in by_label["p50"]} == {"ch_a", "ch_b"} + assert {row["channel_name"] for row in by_label["p90"]} == {"ch_a", "ch_b"} + assert sorted(row["statistic_value"] for row in by_label["p50"]) == [1.5, 2.5, 3.5, 4.5] + + # cross-channel multi-output labels share the descriptor's channel_name + assert {row["channel_name"] for row in by_label["lo"]} == {"combined"} + assert {row["channel_name"] for row in by_label["hi"]} == {"combined"} + assert sorted(row["statistic_value"] for row in by_label["lo"]) == [1.0, 2.0] + assert sorted(row["statistic_value"] for row in by_label["hi"]) == [4.0, 5.0] + + # both cross-channel outputs of interval 0 join the interval's per-channel rows + instance_interval_0 = { + row["event_instance_id"] for row in by_label["min"] if row["statistic_value"] in (1.0, 3.0) + } + lo_hi_instance_0 = { + row["event_instance_id"] + for label in ("lo", "hi") + for row in by_label[label] + if row["statistic_value"] in (1.0, 4.0) + } + assert lo_hi_instance_0 == instance_interval_0 + assert len(instance_interval_0) == 1