diff --git a/docs/impulse/docs/config/configuration.md b/docs/impulse/docs/config/configuration.md index 55f10185..e84857b8 100644 --- a/docs/impulse/docs/config/configuration.md +++ b/docs/impulse/docs/config/configuration.md @@ -352,6 +352,40 @@ mode-resolution rules and what counts as a definition change. --- +## calculated_channels (optional) + +Controls the optional `calculated_channel_metrics` output. By default a report +writes calculated channels to `calculated_channel_fact` (the derived signal) and +`calculated_channel_dimension` (the definitions). Setting `emit_channel_metrics` +adds a third table, `calculated_channel_metrics`, shaped like the silver +`channel_metrics` table so the fact + metrics pair can serve as an Impulse silver +source. See the [Channels reference](../references/report/channel.md) for the +output schema. + +| Field | Type | Default | Description | +|-----------------------|-------------|--------------------------------------|---------------------------------------------------------------------------------------------------| +| `emit_channel_metrics`| `bool` | `false` | Turns on the `calculated_channel_metrics` table. | +| `attribute_columns` | `list[str]` | `[]` | Calculated-channel `attributes` keys to surface as columns on the metrics table (e.g. `["unit"]`). | +| `kpis` | `list[str]` | `["duration", "min", "max", "mean"]` | KPIs computed per `(container_id, channel_id)`, one column each. Must be registered KPI names. | + +When enabled, each row of `calculated_channel_metrics` is one +`(container_id, channel_id)` pair, carrying the selected `kpis` plus dynamic +identity columns (the union of `identity` keys across the report's channels) and +the configured `attribute_columns`. A channel that omits an identity or attribute +key gets `null` for that column; an identity key wins over an attribute key of the +same name. + +The available `kpis` are `duration`, `min`, `max`, and `mean` (all +duration-weighted, matching the silver ingestion semantics). An unknown KPI name is +rejected at config validation with a `ValueError` naming the valid KPIs. + +:::note Off by default +When `emit_channel_metrics` is `false` (the default), no metrics table is written +and `attribute_columns` / `kpis` have no effect. +::: + +--- + ## measurement_dimensions (optional) List of `container_metrics` column names to surface into the gold-layer diff --git a/docs/impulse/docs/data_model/gold_layer_event_normalized.md b/docs/impulse/docs/data_model/gold_layer_event_normalized.md index df45d8b7..cafcbfaf 100644 --- a/docs/impulse/docs/data_model/gold_layer_event_normalized.md +++ b/docs/impulse/docs/data_model/gold_layer_event_normalized.md @@ -186,6 +186,18 @@ calculated_channel_fact { timestamp _created_at } +calculated_channel_metrics { + int container_id + long channel_id + string type + string data_type + double duration + double min + double max + double mean + timestamp _created_at +} + histogram_fact }o--|| event_dimension: event_id histogram2d_fact }o--|| event_dimension: event_id stats_aggregator_fact }o--|| event_instance_fact: event_instance_id @@ -221,7 +233,8 @@ guaranteed. | `{prefix}_histogram2d_fact` | `container_id`, `visual_id`, `event_id`, `x_bin_id`, `y_bin_id` | 2D histogram bin values per container. | | `{prefix}_stats_aggregator_fact` | `container_id`, `visual_id`, `event_instance_id`, `channel_name`, `aggregation_label` | Statistics values per signal, event instance, and container. | | `{prefix}_event_instance_fact` | `container_id`, `event_id`, `event_instance_id` | Materialized event occurrences with start/end timestamps. | -| `{prefix}_calculated_channel_fact` | `container_id`, `channel_id`, `tstart` | Materialized derived signal — one row per sample interval, in the silver `channels` shape (`tstart`, `tend`, `value`). The channel's identity lives on `calculated_channel_dimension`, joined via `channel_id`. | +| `{prefix}_calculated_channel_fact` | `container_id`, `channel_id`, `tstart` | Materialized derived signal, one row per sample interval, in the silver `channels` shape (`tstart`, `tend`, `value`). The channel's identity lives on `calculated_channel_dimension`, joined via `channel_id`. | +| `{prefix}_calculated_channel_metrics` | `container_id`, `channel_id` | Optional per-channel metrics in the silver `channel_metrics` shape, so the fact + metrics pair can serve as an Impulse silver source. Written only when [`config.calculated_channels.emit_channel_metrics`](../config/configuration.md#calculated_channels-optional) is set. Carries the configured `kpis` plus dynamic identity/attribute columns. See [Channels](../references/report/channel.md). | --- diff --git a/docs/impulse/docs/data_model/index.md b/docs/impulse/docs/data_model/index.md index fc123425..7dc4373d 100644 --- a/docs/impulse/docs/data_model/index.md +++ b/docs/impulse/docs/data_model/index.md @@ -62,6 +62,7 @@ The Gold layer uses a **star schema** with fact and dimension tables. All table | `histogram2d_fact` | One row per (x, y) bin per container | 2D histogram bin values, duration-weighted. | | `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. | +| `calculated_channel_metrics` | One row per calculated channel per container | Optional per-channel metrics in the silver `channel_metrics` shape. Written only when `config.calculated_channels.emit_channel_metrics` is set. | ### Dimension tables diff --git a/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/statistic_type.md b/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/statistic_type.md index 4420d143..6646dd62 100644 --- a/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/statistic_type.md +++ b/docs/impulse/docs/references/api/impulse_query_engine/analyze/query/aggregations/statistic_type.md @@ -20,4 +20,6 @@ Enumeration of supported statistic types for aggregations. - `MAX` (`str`): Maximum value statistic. - `MEAN` (`str`): Mean (average) value statistic. - `MEDIAN` (`str`): Median value statistic. +- `START` (`str`): First value in the interval. +- `END` (`str`): Last value in the interval. 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 9782ce85..14b7a933 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 @@ -190,14 +190,16 @@ Only includes computation-affecting attributes: - input_expressions - statistics to be calculated - event expression if there is any -- 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). +- 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, units, page_number, report_id. **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 f2eadaf8..e9bfaacc 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 @@ -178,3 +178,50 @@ Create the dimension DataFrame for the given channels. which ``createDataFrame`` builds directly from the plain dict returned by +#### determine\_channel\_metrics + +```python +def determine_channel_metrics( + cls, + spark: SparkSession, + channels: list[CalculatedChannel], + fact_df: DataFrame | None, + *, + attribute_columns: list[str] | None = None, + kpis: list[str] | None = None) -> DataFrame | None +``` + +Derive a silver-shaped ``channel_metrics`` DataFrame from the fact rows. + +The calculated-channel fact table already matches the silver ``channels`` +table; this builds its companion ``channel_metrics`` so the pair can serve +as an Impulse silver source. Metrics are aggregated **directly from the +narrow fact rows** (``container_id, channel_id, tstart, tend, value``), +grouped by ``(container_id, channel_id)``. + +The output schema is **dynamic**: fixed columns ``container_id, +channel_id, type, data_type`` plus one column per configured KPI (see +``kpis``), one per identity key (the union across all ``channels``), and one +per configured attribute key. Identity/attribute values are pulled from +each channel's in-memory ``identity`` / ``attributes`` dicts (null where a +channel omits a key). On an identity/attribute key collision, identity wins +and the attribute is skipped. + +**Arguments**: + +- `spark` (`SparkSession`): Session used to build the per-channel metadata frame. +- `channels` (`list of CalculatedChannel`): The channels whose fact rows are in ``fact_df``; supply identity and +attributes. +- `fact_df` (`DataFrame or None`): Narrow fact DataFrame (output of :meth:`determine_calculated_channels`). +``None`` returns ``None``. +- `attribute_columns` (`list of str`): Attribute keys to surface as columns. Default/empty → no attribute +columns. A key no channel defines yields an all-null column. +- `kpis` (`list of str`): KPI names to compute (see ``calculated_channel_kpis.KPI_BUILDERS``); the +output carries one column per name, in order. ``None`` → the default +KPIs (``duration, min, max, mean``). + +**Returns**: + +`DataFrame or None`: The dynamic-schema metrics DataFrame, or ``None`` when ``fact_df`` is +``None``. + diff --git a/docs/impulse/docs/references/api/impulse_reporting/channels/channel_types.md b/docs/impulse/docs/references/api/impulse_reporting/channels/channel_types.md index 46e7e05d..1446ad39 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/channels/channel_types.md +++ b/docs/impulse/docs/references/api/impulse_reporting/channels/channel_types.md @@ -67,6 +67,28 @@ Get the dimension table name for the channel type. `str`: The name of the dimension table associated with this channel type. +#### get\_metrics\_table\_name + +```python +def get_metrics_table_name() -> str +``` + +Get the (optional) channel-metrics table name for the channel type. + +This table mirrors the silver-layer ``channel_metrics`` table so the +calculated-channel fact + metrics pair can serve as an Impulse silver +source. Unlike the fact/dimension tables it has **no** fixed schema +constant: identity/attribute columns are derived dynamically per report +(see :meth:`CalculatedChannel.determine_channel_metrics`). + +**Raises**: + +- `ValueError`: If the channel type is not supported. + +**Returns**: + +`str`: The name of the channel-metrics table associated with this channel type. + #### get\_dimension\_schema ```python @@ -103,6 +125,26 @@ Return the first ChannelType whose fact table name matches. `ChannelType`: +#### get\_any\_for\_metrics\_table + +```python +def get_any_for_metrics_table(cls, table_name: str) -> "ChannelType" +``` + +Return the first ChannelType whose metrics table name matches. + +**Arguments**: + +- `table_name` (`str`): Metrics table name to look up. + +**Raises**: + +- `ValueError`: If no ChannelType matches the given table name. + +**Returns**: + +`ChannelType`: + #### get\_any\_for\_dimension\_table ```python diff --git a/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md b/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md index 3658b3a5..cf3b0c21 100644 --- a/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md +++ b/docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md @@ -237,6 +237,27 @@ Configuration for incremental processing behavior. - `silver_last_modified_column` (`str, default="timestamp"`): Column name in the silver layer used for freshness comparison. - `gold_last_modified_column` (`str, default="last_modified"`): Column name in the gold layer used for freshness comparison. +## CalculatedChannels + +```python +class CalculatedChannels(BaseModel) +``` + +Configuration for calculated-channel outputs. + +**Arguments**: + +- `emit_channel_metrics` (`bool, default=False`): When True, also emit a ``calculated_channel_metrics`` gold table (silver +``channel_metrics`` shape) alongside the calculated-channel fact table, so +the fact + metrics pair can serve as an Impulse silver source. +- `attribute_columns` (`list of str, default=[]`): Calculated-channel attribute keys to surface as columns on the metrics +table (e.g. ``["unit"]``). Empty (the default) → no attribute columns. +Identity keys are always surfaced dynamically and win over an +attribute key of the same name. +- `kpis` (`list of str, default=["duration", "min", "max", "mean"]`): KPIs computed on the metrics table, one column per name. Each must be a +registered KPI (see ``calculated_channel_kpis.KPI_BUILDERS``); an unknown +name is rejected at validation. Duplicates are removed (order preserved). + ## ImpulseConfig ```python @@ -257,6 +278,9 @@ Attributes Optional query engine configuration. Defaults to Solvers.DEFAULT_SOLVER. incremental : IncrementalConfig, optional Optional incremental processing configuration. Defaults to IncrementalConfig(). + calculated_channels : CalculatedChannels, optional + Optional calculated-channel output configuration (e.g. opting in to the + ``calculated_channel_metrics`` table). Defaults to CalculatedChannels(). measurement_dimensions : list of str, optional Column names to surface from ``container_metrics`` into the gold-layer ``measurement_dimension`` table. Names are matched diff --git a/docs/impulse/docs/references/report/channel.md b/docs/impulse/docs/references/report/channel.md index fb2c6342..3193291f 100644 --- a/docs/impulse/docs/references/report/channel.md +++ b/docs/impulse/docs/references/report/channel.md @@ -138,3 +138,33 @@ run-length-encoded shape as the silver `channels` table. Both tables carry the configurable `table_prefix` (e.g. `{prefix}_calculated_channel_fact`). See the [gold layer schema](../../data_model/gold_layer_event_normalized.md) for how they fit the star schema, and [incremental processing](./index.md#incremental-processing) for how definition changes are reprocessed. + +--- + +## Optional channel metrics table + +Because `calculated_channel_fact` already matches the silver `channels` shape, a calculated channel needs +only a companion `channel_metrics` table to serve as an Impulse silver source in its own right. Set +[`config.calculated_channels.emit_channel_metrics`](../../config/configuration.md#calculated_channels-optional) +to also write a `calculated_channel_metrics` table, shaped like the silver `channel_metrics` table. + +### calculated_channel_metrics + +One row per `(container_id, channel_id)`, derived directly from the fact rows. The schema is **dynamic**: +fixed columns plus one column per configured KPI, one per identity key (the union of `identity` keys across +the report's channels), and one per configured attribute key. + +| Column | Type | Description | +|----------------|----------|-------------------------------------------------------------------------------------------------| +| `container_id` | `int` | Container identifier. Type is inherited from the silver source. | +| `channel_id` | `long` | Calculated-channel identifier (matches the fact and dimension). | +| *identity cols*| `str` | One column per identity key (e.g. `channel_name`, `data_key`); `null` where a channel omits it. | +| *attribute cols*| `str` | One column per `attribute_columns` entry (e.g. `unit`); `null` where a channel omits it. | +| `type` | `str` | `"CALC"`. | +| `data_type` | `str` | `"double"`. | +| *kpi cols* | `double` | One column per configured KPI, in order (default `duration`, `min`, `max`, `mean`). | + +The KPIs are duration-weighted (matching the silver ingestion semantics): `duration` is the span +`max(tend) - min(tstart)`, `min` / `max` ignore NaN values, and `mean` is the duration-weighted average. +Select which KPIs to compute via `config.calculated_channels.kpis`; an identity key wins over an attribute +key of the same name. diff --git a/skills/impulse-channels/SKILL.md b/skills/impulse-channels/SKILL.md index b568c13b..f2e47067 100644 --- a/skills/impulse-channels/SKILL.md +++ b/skills/impulse-channels/SKILL.md @@ -5,8 +5,8 @@ description: > channels and materialized at the same per-sample grain. Use when the user wants to "add a calculated channel", derive/persist a signal (e.g. "speed in km/h", "power = rpm × torque"), materialize a virtual signal into a queryable table, or run `solve_calculated_channels`. Covers the reporting-layer - CalculatedChannel, the ad-hoc `QueryBuilder.solve_calculated_channels` endpoint, and the - calculated_channel_fact/dimension gold output. + CalculatedChannel, the ad-hoc `QueryBuilder.solve_calculated_channels` endpoint, the + calculated_channel_fact/dimension gold output, and the optional calculated_channel_metrics table. --- # Impulse — calculated channels @@ -98,6 +98,30 @@ across selections. identity is **not** on the fact — it lives on the dimension; join on `channel_id` (and to `measurement_dimension` on `container_id`; see `impulse-data-model`). +## Optional channel metrics table + +`calculated_channel_fact` already matches the silver `channels` shape, so a calculated channel needs only a +companion `channel_metrics` table to become an Impulse silver source. Set +`config.calculated_channels.emit_channel_metrics = True` to also write `calculated_channel_metrics`, shaped +like silver `channel_metrics` (one row per `(container_id, channel_id)`, derived from the fact rows). + +```python +from impulse_reporting.config.config_parser import CalculatedChannels + +# in the ImpulseConfig: +calculated_channels = CalculatedChannels( + emit_channel_metrics=True, + attribute_columns=["unit"], # attribute keys to surface as columns; default [] + kpis=["duration", "min", "max", "mean"], # default; each must be a registered KPI +) +``` + +The schema is **dynamic**: fixed `container_id`, `channel_id`, `type` (`"CALC"`), `data_type` (`"double"`), +one column per configured KPI, one per identity key (union across the report's channels), and one per +`attribute_columns` entry. `null` fills a key a channel omits; an identity key wins over an attribute key of +the same name. KPIs are duration-weighted; an unknown KPI name is rejected at config validation. Adding a +new KPI is a one-line entry in `impulse_reporting.channels.calculated_channel_kpis.KPI_BUILDERS`. + ## Incremental Calculated channels reuse the report's incremental engine (see `impulse-reporting`). A definition change — diff --git a/skills/impulse-config/SKILL.md b/skills/impulse-config/SKILL.md index 912a7be5..ae0e8c86 100644 --- a/skills/impulse-config/SKILL.md +++ b/skills/impulse-config/SKILL.md @@ -6,7 +6,7 @@ description: > "configure an Impulse report", set the source/sink tables, filter which containers are processed, choose RLE vs RAW, turn on incremental processing, run without writing (sinkless), remap column names, or scope by project. Covers source, unity_sink, container_filters, query_engine, solver_config, - incremental, and measurement_dimensions, all validated by Pydantic. + incremental, measurement_dimensions, and calculated_channels, all validated by Pydantic. --- # Impulse — configuration @@ -40,6 +40,7 @@ config = { }, "incremental": {"enabled": True}, "measurement_dimensions": ["container_id", "vehicle_key", "start_ts", "stop_ts"], + "calculated_channels": {"emit_channel_metrics": True, "attribute_columns": ["unit"]}, # optional } ``` @@ -172,3 +173,19 @@ List of `container_metrics` columns (post-mapping **internal** names) to surface Default: `["container_id", "start_ts", "stop_ts"]`. Keep `container_id` — it is the incremental upsert key and the join key to fact tables. Any column present in your post-mapping `container_metrics` DataFrame is valid; a missing one fails the run fast with a `ValueError` naming it. + +## calculated_channels (optional) + +Controls the optional `calculated_channel_metrics` table. Off by default; when on, it is written alongside +`calculated_channel_fact` / `calculated_channel_dimension` in the silver `channel_metrics` shape, so the +fact + metrics pair can serve as an Impulse silver source. See `impulse-channels`. + +| Field | Default | Description | +|------------------------|--------------------------------------|-----------------------------------------------------------------------------| +| `emit_channel_metrics` | `false` | Turn on the `calculated_channel_metrics` table. | +| `attribute_columns` | `[]` | Calculated-channel `attributes` keys to surface as columns (e.g. `["unit"]`). | +| `kpis` | `["duration", "min", "max", "mean"]` | KPIs computed per `(container_id, channel_id)`, one column each. | + +Each metrics row is one `(container_id, channel_id)` pair with the selected `kpis` (duration-weighted) plus +dynamic identity columns (union of `identity` keys) and the configured `attribute_columns`; identity wins +over an attribute of the same name. An unknown KPI name is rejected at config validation. diff --git a/skills/impulse-data-model/SKILL.md b/skills/impulse-data-model/SKILL.md index e26bf674..bdee84d8 100644 --- a/skills/impulse-data-model/SKILL.md +++ b/skills/impulse-data-model/SKILL.md @@ -96,6 +96,7 @@ A star schema. Every table is prefixed with your configured `table_prefix` (e.g. | `histogram2d_fact` | One row per (x, y) bin per container | | `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) | +| `calculated_channel_metrics` | One row per calculated channel per container (optional; silver `channel_metrics` shape). Written only when `config.calculated_channels.emit_channel_metrics` is set. See `impulse-channels`. | **Dimension tables** diff --git a/src/impulse_reporting/channels/calculated_channel.py b/src/impulse_reporting/channels/calculated_channel.py index 4bba5319..9ad03c63 100644 --- a/src/impulse_reporting/channels/calculated_channel.py +++ b/src/impulse_reporting/channels/calculated_channel.py @@ -3,6 +3,8 @@ import hashlib from collections.abc import Mapping +import pyspark.sql.functions as F +import pyspark.sql.types as T from pyspark.sql import DataFrame, Row, SparkSession from impulse_query_engine.analyze.metadata.time_series_expression import ( @@ -14,10 +16,27 @@ from impulse_query_engine.analyze.query.query_builder import QueryBuilder from impulse_query_engine.analyze.query.solvers.query_solver import QuerySolver from impulse_query_engine.model.series.sample_series import SampleSeries +from impulse_reporting.channels.calculated_channel_kpis import ( + DEFAULT_KPIS, + build_kpi_columns, +) from impulse_reporting.persist.dimension_schema import CALCULATED_CHANNEL_DIMENSION_SCHEMA from impulse_reporting.persist.fact_schema import CALCULATED_CHANNEL_FACT_SCHEMA +def _union_identity_keys(channels: list[CalculatedChannel]) -> list[str]: + """Return the sorted union of identity keys across ``channels``. + + Used to build the dynamic identity columns of the calculated-channel metrics + table: every key any channel declares becomes a column (null on channels that + omit it). Sorting gives a stable, order-independent column layout. + """ + keys: set[str] = set() + for channel in channels: + keys.update(channel.identity) + return sorted(keys) + + class CalculatedChannel: """A reporting-layer calculated (derived) channel. @@ -206,3 +225,101 @@ def determine_metadata_df(cls, spark: SparkSession, channels: list[CalculatedCha """ rows = [channel.as_spark_row() for channel in channels] return spark.createDataFrame(rows, schema=CALCULATED_CHANNEL_DIMENSION_SCHEMA) + + @classmethod + def determine_channel_metrics( + cls, + spark: SparkSession, + channels: list[CalculatedChannel], + fact_df: DataFrame | None, + *, + attribute_columns: list[str] | None = None, + kpis: list[str] | None = None, + ) -> DataFrame | None: + """Derive a silver-shaped ``channel_metrics`` DataFrame from the fact rows. + + The calculated-channel fact table already matches the silver ``channels`` + table; this builds its companion ``channel_metrics`` so the pair can serve + as an Impulse silver source. Metrics are aggregated **directly from the + narrow fact rows** (``container_id, channel_id, tstart, tend, value``), + grouped by ``(container_id, channel_id)``. + + The output schema is **dynamic**: fixed columns ``container_id, + channel_id, type, data_type`` plus one column per configured KPI (see + ``kpis``), one per identity key (the union across all ``channels``), and one + per configured attribute key. Identity/attribute values are pulled from + each channel's in-memory ``identity`` / ``attributes`` dicts (null where a + channel omits a key). On an identity/attribute key collision, identity wins + and the attribute is skipped. + + Parameters + ---------- + spark : SparkSession + Session used to build the per-channel metadata frame. + channels : list of CalculatedChannel + The channels whose fact rows are in ``fact_df``; supply identity and + attributes. + fact_df : DataFrame or None + Narrow fact DataFrame (output of :meth:`determine_calculated_channels`). + ``None`` returns ``None``. + attribute_columns : list of str, optional + Attribute keys to surface as columns. Default/empty → no attribute + columns. A key no channel defines yields an all-null column. + kpis : list of str, optional + KPI names to compute (see ``calculated_channel_kpis.KPI_BUILDERS``); the + output carries one column per name, in order. ``None`` → the default + KPIs (``duration, min, max, mean``). + + Returns + ------- + DataFrame or None + The dynamic-schema metrics DataFrame, or ``None`` when ``fact_df`` is + ``None``. + """ + if fact_df is None: + return None + + attribute_columns = list(attribute_columns or []) + kpis = list(kpis) if kpis is not None else list(DEFAULT_KPIS) + + # KPI aggregations are built from the registry (single extension point), so + # the computed columns and the projection below both follow ``kpis``. + agg_df = fact_df.groupBy("container_id", "channel_id").agg(*build_kpi_columns(kpis)) + + # Union of identity keys (stable order); attribute columns lose to identity + # on a key collision. + identity_keys = _union_identity_keys(channels) + effective_attribute_keys = [k for k in attribute_columns if k not in identity_keys] + + # Per-channel metadata frame. channel_id is cast to the fact's channel_id + # type so the join keys line up regardless of int/long width. + channel_id_type = fact_df.schema["channel_id"].dataType + meta_schema = T.StructType( + [T.StructField("channel_id", channel_id_type, False)] + + [T.StructField(k, T.StringType(), True) for k in identity_keys] + + [T.StructField(k, T.StringType(), True) for k in effective_attribute_keys] + ) + meta_rows = [] + for channel in channels: + row: dict = {"channel_id": channel.get_id()} + for key in identity_keys: + row[key] = channel.identity.get(key) + for key in effective_attribute_keys: + row[key] = channel.attributes.get(key) + meta_rows.append(Row(**row)) + meta_df = spark.createDataFrame(meta_rows, schema=meta_schema) + + result = ( + agg_df.join(meta_df, on="channel_id", how="left") + .withColumn("type", F.lit("CALC")) + .withColumn("data_type", F.lit("double")) + ) + + ordered_columns = ( + ["container_id", "channel_id"] + + identity_keys + + effective_attribute_keys + + ["type", "data_type"] + + kpis + ) + return result.select(*ordered_columns) diff --git a/src/impulse_reporting/channels/calculated_channel_kpis.py b/src/impulse_reporting/channels/calculated_channel_kpis.py new file mode 100644 index 00000000..4f1db7a4 --- /dev/null +++ b/src/impulse_reporting/channels/calculated_channel_kpis.py @@ -0,0 +1,104 @@ +"""Registry of calculated-channel metric KPIs. + +Each KPI is a named builder that returns a Spark aggregation :class:`Column` +(already ``.alias(name)``-ed) computed over the narrow calculated-channel fact +rows grouped by ``(container_id, channel_id)``. The registry is the single +extension point for the ``calculated_channel_metrics`` table: **adding a KPI means +adding one entry to** :data:`KPI_BUILDERS` — it then becomes selectable via +``CalculatedChannels.kpis`` config with no other changes. + +Semantics match :class:`SampleSeries` (duration-weighted where relevant), with +``dur = tend - tstart``. NaN values are excluded from ``min``/``max``/``mean`` +(matching ``np.nanmin`` / ``nanmax`` / ``nansum``); the ``mean`` denominator keeps +every interval's duration and uses ``try_divide`` so a zero total duration (a group +of only zero-duration point-in-time samples) yields ``null`` instead of failing the +run under ANSI mode (Spark 4.0 default). +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +import pyspark.sql.functions as F +from pyspark.sql import Column + + +@dataclass(frozen=True) +class KpiColumns: + """Reusable per-group column expressions shared by the KPI builders. + + Built once per aggregation from the fact columns so each builder does not + recompute the duration / NaN-masking expressions. + """ + + dur: Column + value: Column + non_nan_value: Column + weighted_value: Column + + @classmethod + def from_fact(cls) -> "KpiColumns": + """Build the shared expressions from the fixed fact column names.""" + dur = F.col("tend") - F.col("tstart") + value = F.col("value") + non_nan_value = F.when(~F.isnan(value), value) + weighted_value = F.when(~F.isnan(value), value * dur).otherwise(F.lit(0.0)) + return cls( + dur=dur, value=value, non_nan_value=non_nan_value, weighted_value=weighted_value + ) + + +def _duration(cols: KpiColumns) -> Column: + return (F.max("tend") - F.min("tstart")).alias("duration") + + +def _min(cols: KpiColumns) -> Column: + return F.min(cols.non_nan_value).alias("min") + + +def _max(cols: KpiColumns) -> Column: + return F.max(cols.non_nan_value).alias("max") + + +def _mean(cols: KpiColumns) -> Column: + # Duration-weighted; try_divide → null (not error) when total duration is zero. + return F.try_divide(F.sum(cols.weighted_value), F.sum(cols.dur)).alias("mean") + + +# Name → aggregation-Column builder. Adding a KPI = adding one entry here. +KPI_BUILDERS: dict[str, Callable[[KpiColumns], Column]] = { + "duration": _duration, + "min": _min, + "max": _max, + "mean": _mean, +} + +# Default KPIs computed when config does not narrow the selection. +DEFAULT_KPIS: list[str] = ["duration", "min", "max", "mean"] + + +def build_kpi_columns(names: list[str]) -> list[Column]: + """Return the aggregation columns for ``names`` in the given order. + + Parameters + ---------- + names : list of str + KPI names to compute; each must be a key of :data:`KPI_BUILDERS`. + + Returns + ------- + list of Column + Aliased aggregation columns, one per name, ready to splat into ``.agg``. + + Raises + ------ + ValueError + If any name is not a registered KPI. + """ + unknown = [n for n in names if n not in KPI_BUILDERS] + if unknown: + valid = ", ".join(sorted(KPI_BUILDERS)) + raise ValueError(f"Unknown calculated-channel KPI(s): {unknown}. Valid KPIs: {valid}.") + cols = KpiColumns.from_fact() + return [KPI_BUILDERS[name](cols) for name in names] diff --git a/src/impulse_reporting/channels/channel_types.py b/src/impulse_reporting/channels/channel_types.py index bcb63330..db89319d 100644 --- a/src/impulse_reporting/channels/channel_types.py +++ b/src/impulse_reporting/channels/channel_types.py @@ -83,6 +83,32 @@ def get_dimension_table_name(self) -> str: case _: raise ValueError(f"Unsupported channel type: {self}") + def get_metrics_table_name(self) -> str: + """ + Get the (optional) channel-metrics table name for the channel type. + + This table mirrors the silver-layer ``channel_metrics`` table so the + calculated-channel fact + metrics pair can serve as an Impulse silver + source. Unlike the fact/dimension tables it has **no** fixed schema + constant: identity/attribute columns are derived dynamically per report + (see :meth:`CalculatedChannel.determine_channel_metrics`). + + Returns + ------- + str + The name of the channel-metrics table associated with this channel type. + + Raises + ------ + ValueError + If the channel type is not supported. + """ + match self: + case ChannelType.CALCULATED_CHANNEL: + return "calculated_channel_metrics" + case _: + raise ValueError(f"Unsupported channel type: {self}") + def get_dimension_schema(self) -> StructType: """ Get the dimension schema for the channel type. @@ -126,6 +152,29 @@ def get_any_for_fact_table(cls, table_name: str) -> "ChannelType": return ct raise ValueError(f"No ChannelType found for fact table: {table_name}") + @classmethod + def get_any_for_metrics_table(cls, table_name: str) -> "ChannelType": + """Return the first ChannelType whose metrics table name matches. + + Parameters + ---------- + table_name : str + Metrics table name to look up. + + Returns + ------- + ChannelType + + Raises + ------ + ValueError + If no ChannelType matches the given table name. + """ + for ct in cls: + if ct.get_metrics_table_name() == table_name: + return ct + raise ValueError(f"No ChannelType found for metrics table: {table_name}") + @classmethod def get_any_for_dimension_table(cls, table_name: str) -> "ChannelType": """Return the first ChannelType whose dimension table name matches. diff --git a/src/impulse_reporting/config/config_parser.py b/src/impulse_reporting/config/config_parser.py index 6bff951e..536c18b9 100644 --- a/src/impulse_reporting/config/config_parser.py +++ b/src/impulse_reporting/config/config_parser.py @@ -6,6 +6,7 @@ from pydantic import AfterValidator, BaseModel, field_validator, model_validator from impulse_query_engine.analyze.query.solvers.solver_config import RawEncoder, SolverConfig +from impulse_reporting.channels.calculated_channel_kpis import DEFAULT_KPIS, KPI_BUILDERS def is_valid_table_name(table_name: str) -> str: @@ -436,6 +437,58 @@ class IncrementalConfig(BaseModel): gold_last_modified_column: str = "_created_at" +class CalculatedChannels(BaseModel): + """ + Configuration for calculated-channel outputs. + + Attributes + ---------- + emit_channel_metrics : bool, default=False + When True, also emit a ``calculated_channel_metrics`` gold table (silver + ``channel_metrics`` shape) alongside the calculated-channel fact table, so + the fact + metrics pair can serve as an Impulse silver source. + attribute_columns : list of str, default=[] + Calculated-channel attribute keys to surface as columns on the metrics + table (e.g. ``["unit"]``). Empty (the default) → no attribute columns. + Identity keys are always surfaced dynamically and win over an + attribute key of the same name. + kpis : list of str, default=["duration", "min", "max", "mean"] + KPIs computed on the metrics table, one column per name. Each must be a + registered KPI (see ``calculated_channel_kpis.KPI_BUILDERS``); an unknown + name is rejected at validation. Duplicates are removed (order preserved). + """ + + emit_channel_metrics: bool = False + attribute_columns: list[str] = [] + kpis: list[str] = list(DEFAULT_KPIS) + + @field_validator("attribute_columns", mode="after") + @classmethod + def _normalize_attribute_columns(cls, value: list[str]) -> list[str]: + seen: set[str] = set() + normalized: list[str] = [] + for name in value: + is_valid_unity_entity_name(name) + if name not in seen: + seen.add(name) + normalized.append(name) + return normalized + + @field_validator("kpis", mode="after") + @classmethod + def _normalize_kpis(cls, value: list[str]) -> list[str]: + seen: set[str] = set() + normalized: list[str] = [] + for name in value: + if name not in KPI_BUILDERS: + valid = ", ".join(sorted(KPI_BUILDERS)) + raise ValueError(f"Unknown calculated-channel KPI: {name}. Valid KPIs: {valid}.") + if name not in seen: + seen.add(name) + normalized.append(name) + return normalized + + class ImpulseConfig(BaseModel): """ Main configuration model. @@ -452,6 +505,9 @@ class ImpulseConfig(BaseModel): Optional query engine configuration. Defaults to Solvers.DEFAULT_SOLVER. incremental : IncrementalConfig, optional Optional incremental processing configuration. Defaults to IncrementalConfig(). + calculated_channels : CalculatedChannels, optional + Optional calculated-channel output configuration (e.g. opting in to the + ``calculated_channel_metrics`` table). Defaults to CalculatedChannels(). measurement_dimensions : list of str, optional Column names to surface from ``container_metrics`` into the gold-layer ``measurement_dimension`` table. Names are matched @@ -526,6 +582,7 @@ class ImpulseConfig(BaseModel): container_filters: ContainerFilters | None = None query_engine: QueryEngine = QueryEngine(solver=Solvers.DEFAULT_SOLVER) incremental: IncrementalConfig | None = None + calculated_channels: CalculatedChannels = CalculatedChannels() measurement_dimensions: list[str] = list(DEFAULT_MEASUREMENT_DIMENSIONS) diff --git a/src/impulse_reporting/core/report.py b/src/impulse_reporting/core/report.py index a0e4b6f2..e355cdb3 100644 --- a/src/impulse_reporting/core/report.py +++ b/src/impulse_reporting/core/report.py @@ -106,6 +106,7 @@ def __init__( self.aggregation_metadata_dfs = {} self.calculated_channel_dfs = {} self.calculated_channel_metadata_dfs = {} + self.calculated_channel_metrics_dfs = {} self.container_dimension_df = None self.channel_mapping_resolution_dimension_df = None self._is_incremental = None @@ -598,6 +599,10 @@ def _persist_full(self): persist_facts_full(self.calculated_channel_dfs, ChannelType, storage_factory) persist_dimensions_full(self.calculated_channel_metadata_dfs, ChannelType, storage_factory) + # optional calculated channel metrics table (dynamic schema — stored + # directly without the fixed-schema projecting writer) + self._persist_channel_metrics(incremental=False) + # persist measurement dimensions if self.container_dimension_df: writer = storage_factory.create_container_dimension_writer() @@ -708,6 +713,13 @@ def _transform(df, schema): merge_keys=["channel_id"], ) + # Optional calculated channel metrics table (dynamic schema — merged + # directly, scoping the delete-by-source to updated containers). + self._persist_channel_metrics( + incremental=True, + updated_container_ids=updated_container_ids, + ) + # Persist the measurement dimension LAST (as ``_persist_full`` does). It # holds the gold timestamp that container-update detection compares # against; the fact solves above are lazy, so writing it earlier would @@ -739,6 +751,72 @@ def _transform(df, schema): ], ) + def _persist_channel_metrics( + self, + *, + incremental: bool, + updated_container_ids: list | None = None, + ): + """Persist the optional calculated-channel metrics table(s). + + The metrics schema is dynamic (identity/attribute columns vary per + report), so this bypasses the fixed-schema ``DefaultReportEntityWriter`` + and stores the already-shaped DataFrame directly — mirroring the + ``container_dimension`` special-case. Full mode overwrites; incremental + mode upserts on ``(container_id, channel_id)`` and prunes stale rows from + updated containers via ``merge_incremental``. + + Parameters + ---------- + incremental : bool + Whether to merge (True) or overwrite (False). + updated_container_ids : list, optional + Ids of updated containers, scoping the incremental delete-by-source. + """ + from functools import reduce + + import pyspark.sql.functions as F + + if not self.calculated_channel_metrics_dfs: + return + + transformer = ReportEntityTransformer() + updated_container_ids = updated_container_ids or [] + + # Group per-type metrics dfs by output table (parallels persist_facts_*). + dfs_by_table: dict[str, list[DataFrame]] = {} + for type_name, dfs in self.calculated_channel_metrics_dfs.items(): + if isinstance(dfs, dict): + candidates = [dfs.get(key) for key in ("changed", "unchanged")] + else: + candidates = [dfs] + table_dfs = [df for df in candidates if df is not None] + if not table_dfs: + continue + table_name = ChannelType[type_name].get_metrics_table_name() + dfs_by_table.setdefault(table_name, []).extend(table_dfs) + + for table_name, dfs_list in dfs_by_table.items(): + entity_type = ChannelType.get_any_for_metrics_table(table_name) + # Resolve the metrics URI directly (no fixed-schema writer). + uri = self.sink.config.get_output_uri_channel_metrics_table(entity_type) + combined = reduce(lambda a, b: a.unionByName(b), dfs_list) + df_enriched = combined.transform(transformer.add_meta_information) + if incremental: + delete_conditions = [] + if updated_container_ids: + delete_conditions.append( + F.col("target.container_id").isin(updated_container_ids) + ) + self.sink.merge_incremental( + df_enriched, + uri, + ["container_id", "channel_id"], + delete_conditions=delete_conditions, + ) + else: + self.sink.store(df_enriched, uri) + def _transform_for_persistence( self, df: DataFrame, @@ -1022,6 +1100,43 @@ def determine_report(self, is_incremental: bool = None): channels_by_type, ChannelType, self.spark ) + # Optionally derive a silver-shaped channel_metrics table from the fact + # rows so the fact + metrics pair can serve as an Impulse silver source. + # Identity/attribute columns are derived dynamically; the full per-type + # channel list is passed to both buckets so changed/unchanged share a + # schema (extra channels are ignored by the fact-driven left join). + self.calculated_channel_metrics_dfs = {} + if self.config.calculated_channels.emit_channel_metrics: + attribute_columns = self.config.calculated_channels.attribute_columns + kpis = self.config.calculated_channels.kpis + changed_metrics: dict = {} + unchanged_metrics: dict = {} + for type_name, full_channels in channels_by_type.items(): + if not full_channels: + continue + cls = ChannelType[type_name].value + changed_fact = changed_channel_dfs.get(type_name) + unchanged_fact = unchanged_channel_dfs.get(type_name) + if changed_fact is not None: + changed_metrics[type_name] = cls.determine_channel_metrics( + self.spark, + full_channels, + changed_fact, + attribute_columns=attribute_columns, + kpis=kpis, + ) + if unchanged_fact is not None: + unchanged_metrics[type_name] = cls.determine_channel_metrics( + self.spark, + full_channels, + unchanged_fact, + attribute_columns=attribute_columns, + kpis=kpis, + ) + self.calculated_channel_metrics_dfs = merge_changed_unchanged( + changed_metrics, unchanged_metrics + ) + # Determine container dimension self.container_dimension_df = ContainerDimension.get_dimension( spark=self.spark, diff --git a/src/impulse_reporting/persist/report_storage.py b/src/impulse_reporting/persist/report_storage.py index 6e6c7169..d09414b5 100644 --- a/src/impulse_reporting/persist/report_storage.py +++ b/src/impulse_reporting/persist/report_storage.py @@ -86,6 +86,23 @@ def get_output_uri_channel_mapping_resolution_dimension_table(self) -> str: """ pass + @abstractmethod + def get_output_uri_channel_metrics_table(self, element: ChannelType) -> str: + """ + Get the output URI for the (optional) channel-metrics table. + + Parameters + ---------- + element : ChannelType + The channel type to get the URI for. + + Returns + ------- + str + The output URI for the channel-metrics table. + """ + pass + @dataclass() class UnitySinkConfig(SinkConfig): @@ -185,6 +202,27 @@ def get_output_uri_channel_mapping_resolution_dimension_table(self) -> str: uri = f"{self.catalog_name}.{self.schema_name}." "channel_mapping_resolution_dimension" return uri + def get_output_uri_channel_metrics_table(self, element: ChannelType) -> str: + """ + Get the output URI for the channel-metrics table in Unity Catalog format. + + Parameters + ---------- + element : ChannelType + The channel type to get the URI for. + + Returns + ------- + str + The Unity Catalog URI for the channel-metrics table. + """ + table_name = element.get_metrics_table_name() + if self.table_prefix: + uri = f"{self.catalog_name}.{self.schema_name}.{self.table_prefix}_{table_name}" + else: + uri = f"{self.catalog_name}.{self.schema_name}.{table_name}" + return uri + class Sink(ABC): """ diff --git a/tests/impulse_reporting/integration/calculated_channel_test.py b/tests/impulse_reporting/integration/calculated_channel_test.py index 1f57b205..2fea3f3c 100644 --- a/tests/impulse_reporting/integration/calculated_channel_test.py +++ b/tests/impulse_reporting/integration/calculated_channel_test.py @@ -14,8 +14,10 @@ import pytest from databricks.sdk import WorkspaceClient +from impulse_query_engine.measurement_db import MeasurementDB, MeasurementDBConfig from impulse_reporting.channels.calculated_channel import CalculatedChannel from impulse_reporting.config.config_parser import ( + CalculatedChannels, DataType, ImpulseConfig, IncrementalConfig, @@ -31,10 +33,11 @@ _FACT = "spark_catalog.gold.evaluation_calculated_channel_fact" _DIM = "spark_catalog.gold.evaluation_calculated_channel_dimension" +_METRICS = "spark_catalog.gold.evaluation_calculated_channel_metrics" -def _config(silver_table="container_metrics", is_enabled=False): - return ImpulseConfig( +def _config(silver_table="container_metrics", is_enabled=False, calculated_channels=None): + kwargs = dict( source=Source( container_metrics_table=f"spark_catalog.silver.{silver_table}", channel_metrics_table="spark_catalog.silver.channel_metrics", @@ -51,6 +54,9 @@ def _config(silver_table="container_metrics", is_enabled=False): gold_last_modified_column="_created_at", ), ) + if calculated_channels is not None: + kwargs["calculated_channels"] = calculated_channels + return ImpulseConfig(**kwargs) def _add_channel(report, factor=3.6, name="speed_kmh", identity=None): @@ -380,3 +386,159 @@ def test_calculated_channel_raw_mode(spark, setup_raw_channels_db): assert "identity" not in df.columns ids = {r["channel_id"] for r in df.select("channel_id").distinct().collect()} assert ids == {ch.get_id()} + + +def test_channel_metrics_not_emitted_by_default(spark): + # Flag off (default) → no calculated_channel_metrics table is written even + # when channels carry attributes. + report = Report( + name="calc_channel_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config=dict(_config(is_enabled=False)), + ) + _add_channel(report, factor=3.6) + report.determine_report() + report.persist_results() + + assert spark.catalog.tableExists(_FACT) + assert not spark.catalog.tableExists(_METRICS) + + +def test_channel_metrics_emitted_and_usable_as_impulse_source(spark): + # Opt in to the metrics table, with `unit` surfaced as an attribute column. + report = Report( + name="calc_channel_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config=dict( + _config( + is_enabled=False, + calculated_channels=CalculatedChannels( + emit_channel_metrics=True, attribute_columns=["unit"] + ), + ) + ), + ) + q = report.get_db().query + ch = CalculatedChannel( + name="speed_kmh", + expr=q.channel(channel_name="Vehicle Speed Sensor") * 3.6, + identity={"channel_name": "speed_kmh", "data_key": "CALC"}, + attributes={"unit": "km/h"}, + ) + report.add_calculated_channel(ch) + + report.determine_report() + report.persist_results() + + assert spark.catalog.tableExists(_METRICS) + metrics = spark.read.table(_METRICS) + # Dynamic identity columns (channel_name, data_key) + configured attribute (unit) + # + fixed metric columns; identity keys always present, attribute opt-in. + for col in [ + "container_id", + "channel_id", + "channel_name", + "data_key", + "unit", + "type", + "data_type", + "duration", + "min", + "max", + "mean", + ]: + assert col in metrics.columns, col + + row = metrics.filter(F.col("channel_id") == ch.get_id()).first() + assert row["type"] == "CALC" + assert row["data_type"] == "double" + assert row["channel_name"] == "speed_kmh" + assert row["data_key"] == "CALC" + assert row["unit"] == "km/h" + # One metrics row per (container, channel). + fact = spark.read.table(_FACT) + n_pairs = fact.select("container_id", "channel_id").distinct().count() + assert metrics.count() == n_pairs + + # Round-trip: the fact + metrics pair is a valid Impulse silver source. Feed + # them back as `channels` + `channel_metrics` (wide model: channel_name is a + # column on channel_metrics) and resolve the channel by name. A minimal + # `container_metrics` (one row per container) satisfies the filter pipeline. + fact_as_channels = spark.read.table(_FACT).select( + "container_id", "channel_id", "tstart", "tend", "value" + ) + container_metrics = fact_as_channels.select("container_id").distinct() + db = MeasurementDB( + MeasurementDBConfig.for_debug( + { + "channels": fact_as_channels, + "channel_metrics": metrics, + "container_metrics": container_metrics, + } + ), + ws=report.ws, + ) + solved = ( + db.query.channel(channel_name="speed_kmh").alias("v").solve(spark, report.get_solver()) + ) + assert solved.count() > 0 + + +def test_channel_metrics_incremental_is_idempotent(spark): + # Seed gold (full) with the metrics table, then re-run incrementally with the + # same definition/data: the metrics upsert on (container_id, channel_id) must + # keep the row count stable (one row per container/channel). + cc = CalculatedChannels(emit_channel_metrics=True) + + r1 = Report( + name="calc_channel_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config=dict(_config(is_enabled=False, calculated_channels=cc)), + ) + _add_channel(r1, factor=3.6) + r1.determine_report() + r1.persist_results() + count_before = spark.read.table(_METRICS).count() + assert count_before > 0 + + r2 = Report( + name="calc_channel_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config=dict(_config(is_enabled=True, calculated_channels=cc)), + ) + _add_channel(r2, factor=3.6) + r2.determine_report() + r2.persist_results() + + assert spark.read.table(_METRICS).count() == count_before + + +def test_channel_metrics_custom_kpis(spark): + # A non-default KPI selection controls which KPI columns are emitted. + report = Report( + name="calc_channel_report", + spark=spark, + workspace_client=create_autospec(WorkspaceClient), + config=dict( + _config( + is_enabled=False, + calculated_channels=CalculatedChannels( + emit_channel_metrics=True, kpis=["min", "max"] + ), + ) + ), + ) + _add_channel(report, factor=3.6) + report.determine_report() + report.persist_results() + + metrics = spark.read.table(_METRICS) + # Only the selected KPIs are present; the dropped defaults are absent. + assert "min" in metrics.columns + assert "max" in metrics.columns + assert "mean" not in metrics.columns + assert "duration" not in metrics.columns diff --git a/tests/impulse_reporting/unit/channels/calculated_channel_test.py b/tests/impulse_reporting/unit/channels/calculated_channel_test.py index 8d5856e3..c8510089 100644 --- a/tests/impulse_reporting/unit/channels/calculated_channel_test.py +++ b/tests/impulse_reporting/unit/channels/calculated_channel_test.py @@ -2,6 +2,7 @@ """Unit tests for the reporting-layer CalculatedChannel class.""" import pytest +import pyspark.sql.types as T from impulse_query_engine.analyze.metadata.time_series_expression import TimeSeriesSelector from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver @@ -164,3 +165,197 @@ def test_returns_fact_columns_with_matching_channel_id(self, spark, basic_narrow ] ids = {r["channel_id"] for r in df.select("channel_id").distinct().collect()} assert ids == {ch.get_id()} + + +_FACT_SCHEMA = T.StructType( + [ + T.StructField("container_id", T.LongType(), False), + T.StructField("channel_id", T.LongType(), False), + T.StructField("tstart", T.LongType(), False), + T.StructField("tend", T.LongType(), False), + T.StructField("value", T.DoubleType(), True), + ] +) + + +class TestDetermineChannelMetrics: + def test_returns_none_when_fact_none(self, spark): + assert ( + CalculatedChannel.determine_channel_metrics(spark, [], None, attribute_columns=[]) + is None + ) + + def test_duration_weighted_values(self, spark): + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + cid = ch.get_id() + # Two intervals with different durations: [0,1) value 10, [1,3) value 20. + # duration-weighted mean = (10*1 + 20*2) / (1+2) = 50/3. + fact = spark.createDataFrame( + [(1, cid, 0, 1, 10.0), (1, cid, 1, 3, 20.0)], schema=_FACT_SCHEMA + ) + out = CalculatedChannel.determine_channel_metrics(spark, [ch], fact, attribute_columns=[]) + rows = out.collect() + assert len(rows) == 1 + r = rows[0] + assert r["container_id"] == 1 + assert r["channel_id"] == cid + assert r["type"] == "CALC" + assert r["data_type"] == "double" + assert r["duration"] == 3 # max(tend) - min(tstart) = 3 - 0 + assert r["min"] == 10.0 + assert r["max"] == 20.0 + assert r["mean"] == pytest.approx(50.0 / 3.0) + assert r["channel_name"] == "s" + + def test_nan_values_ignored_in_min_max_mean(self, spark): + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + cid = ch.get_id() + # [0,1) value 10, [1,2) NaN → NaN excluded from min/max/weighted-sum, but its + # duration still counts in the denominator (matches SampleSeries.mean). + fact = spark.createDataFrame( + [(1, cid, 0, 1, 10.0), (1, cid, 1, 2, float("nan"))], schema=_FACT_SCHEMA + ) + r = CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=[] + ).collect()[0] + assert r["min"] == 10.0 + assert r["max"] == 10.0 + # (10*1) / (1+1) = 5.0 + assert r["mean"] == pytest.approx(5.0) + + def test_zero_total_duration_mean_is_null_not_error(self, spark): + # A group of only zero-duration point-in-time samples (tstart == tend) has + # sum(dur) == 0. Under ANSI mode (Spark 4.0 default) plain division would + # raise DIVIDE_BY_ZERO; try_divide yields a null mean instead. + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + cid = ch.get_id() + fact = spark.createDataFrame( + [(1, cid, 5, 5, 10.0), (1, cid, 7, 7, 20.0)], schema=_FACT_SCHEMA + ) + r = CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=[] + ).collect()[0] + assert r["duration"] == 2 # max(tend) - min(tstart) = 7 - 5 + assert r["mean"] is None + # min/max still resolve from the values. + assert r["min"] == 10.0 + assert r["max"] == 20.0 + + def test_dynamic_identity_columns_union(self, spark): + # Two channels with DIFFERENT identity keys → output has the union, null + # where a channel omits a key. + ch1 = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s1"}) + ch2 = CalculatedChannel( + "b", TimeSeriesSelector(None) * 1.0, {"channel_name": "s2", "data_key": "K"} + ) + fact = spark.createDataFrame( + [(1, ch1.get_id(), 0, 1, 1.0), (1, ch2.get_id(), 0, 1, 2.0)], + schema=_FACT_SCHEMA, + ) + out = CalculatedChannel.determine_channel_metrics( + spark, [ch1, ch2], fact, attribute_columns=[] + ) + assert "channel_name" in out.columns + assert "data_key" in out.columns + by_id = {r["channel_id"]: r for r in out.collect()} + assert by_id[ch1.get_id()]["channel_name"] == "s1" + assert by_id[ch1.get_id()]["data_key"] is None # ch1 has no data_key + assert by_id[ch2.get_id()]["data_key"] == "K" + + def test_attribute_columns_config_selected(self, spark): + # A configured attribute key surfaces as a column; unconfigured attributes + # do not. A channel omitting the key gets null. + ch1 = CalculatedChannel( + "a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s1"}, attributes={"unit": "kmh"} + ) + ch2 = CalculatedChannel( + "b", TimeSeriesSelector(None) * 1.0, {"channel_name": "s2"}, attributes={"scale": "2"} + ) + fact = spark.createDataFrame( + [(1, ch1.get_id(), 0, 1, 1.0), (1, ch2.get_id(), 0, 1, 2.0)], + schema=_FACT_SCHEMA, + ) + out = CalculatedChannel.determine_channel_metrics( + spark, [ch1, ch2], fact, attribute_columns=["unit"] + ) + assert "unit" in out.columns + assert "scale" not in out.columns # not configured + by_id = {r["channel_id"]: r for r in out.collect()} + assert by_id[ch1.get_id()]["unit"] == "kmh" + assert by_id[ch2.get_id()]["unit"] is None + + def test_no_attribute_columns_by_default(self, spark): + ch = CalculatedChannel( + "a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}, attributes={"unit": "kmh"} + ) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 1, 1.0)], schema=_FACT_SCHEMA) + out = CalculatedChannel.determine_channel_metrics(spark, [ch], fact, attribute_columns=[]) + assert out.columns == [ + "container_id", + "channel_id", + "channel_name", + "type", + "data_type", + "duration", + "min", + "max", + "mean", + ] + + def test_identity_wins_on_attribute_collision(self, spark): + # A key present in BOTH identity and attribute_columns yields the identity + # value, and only one column. + ch = CalculatedChannel( + "a", + TimeSeriesSelector(None) * 1.0, + {"channel_name": "s", "unit": "identity_unit"}, + attributes={"unit": "attr_unit"}, + ) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 1, 1.0)], schema=_FACT_SCHEMA) + out = CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=["unit"] + ) + assert out.columns.count("unit") == 1 + assert out.collect()[0]["unit"] == "identity_unit" + + def test_kpis_subset_only_emits_selected(self, spark): + # Selecting a subset yields only those KPI columns (plus the fixed + # container/channel/identity/type/data_type columns). + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 2, 10.0)], schema=_FACT_SCHEMA) + out = CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=[], kpis=["mean"] + ) + assert out.columns == [ + "container_id", + "channel_id", + "channel_name", + "type", + "data_type", + "mean", + ] + assert out.collect()[0]["mean"] == pytest.approx(10.0) + + def test_kpis_order_is_preserved(self, spark): + # The KPI columns appear in the configured order (tail of the schema). + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 2, 10.0)], schema=_FACT_SCHEMA) + out = CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=[], kpis=["max", "min", "duration"] + ) + assert out.columns[-3:] == ["max", "min", "duration"] + + def test_kpis_default_is_the_four(self, spark): + # kpis=None → default duration, min, max, mean (in order). + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 2, 10.0)], schema=_FACT_SCHEMA) + out = CalculatedChannel.determine_channel_metrics(spark, [ch], fact, attribute_columns=[]) + assert out.columns[-4:] == ["duration", "min", "max", "mean"] + + def test_unknown_kpi_raises(self, spark): + ch = CalculatedChannel("a", TimeSeriesSelector(None) * 1.0, {"channel_name": "s"}) + fact = spark.createDataFrame([(1, ch.get_id(), 0, 2, 10.0)], schema=_FACT_SCHEMA) + with pytest.raises(ValueError, match="Unknown calculated-channel KPI"): + CalculatedChannel.determine_channel_metrics( + spark, [ch], fact, attribute_columns=[], kpis=["bogus"] + ) diff --git a/tests/impulse_reporting/unit/config/config_parser_test.py b/tests/impulse_reporting/unit/config/config_parser_test.py index ec9936b7..9398bb3f 100644 --- a/tests/impulse_reporting/unit/config/config_parser_test.py +++ b/tests/impulse_reporting/unit/config/config_parser_test.py @@ -2,6 +2,7 @@ from pydantic import ValidationError from impulse_reporting.config.config_parser import ( + CalculatedChannels, CastType, Comparator, ContainerFilters, @@ -955,3 +956,22 @@ def test_impulse_config_source_rejects_invalid_channel_mapping_table(): config_json["source"]["channel_mapping_table"] = "invalid_table_name" with pytest.raises(ValidationError): ImpulseConfig.model_validate(config_json) + + +def test_calculated_channels_default_kpis(): + """CalculatedChannels defaults to the four built-in KPIs.""" + config = CalculatedChannels() + assert config.emit_channel_metrics is False + assert config.kpis == ["duration", "min", "max", "mean"] + + +def test_calculated_channels_kpis_dedupe_preserves_order(): + """Duplicate KPI names are removed, insertion order preserved.""" + config = CalculatedChannels(kpis=["mean", "mean", "min"]) + assert config.kpis == ["mean", "min"] + + +def test_calculated_channels_unknown_kpi_rejected(): + """An unknown KPI name is rejected at validation with a helpful message.""" + with pytest.raises(ValidationError, match="Unknown calculated-channel KPI"): + CalculatedChannels(kpis=["bogus"])