Skip to main content

impulse_reporting.config.config_parser

is_valid_table_name​

def is_valid_table_name(table_name: str) -> str

Validate if a string is a valid Unity Catalog table name.

Arguments:

  • table_name (str): The table name to validate. Should be in format 'catalog.schema.table'.

Raises:

  • ValueError: If the table name does not match the required format or contains invalid characters.

Returns:

str: The validated table name if valid.

is_valid_unity_entity_name​

def is_valid_unity_entity_name(entity_name: str) -> str

Validate if a string is a valid Unity Catalog entity name.

Arguments:

  • entity_name (str): The entity name to validate (catalog, schema, or table prefix).

Raises:

  • ValueError: If the entity name contains invalid characters.

Returns:

str: The validated entity name if valid.

Solvers​

class Solvers(Enum)

Enumeration of available solver types for the query engine.

DEFAULT_SOLVER is the single, unified solver. DELTA_SOLVER and KEY_VALUE_STORE_SOLVER are deprecated aliases kept so that existing report configs continue to deserialize; both now resolve to the same DefaultSolver. They will be removed in a future release.

Arguments:

  • DEFAULT_SOLVER (str): None
  • DELTA_SOLVER (str): Deprecated alias for DEFAULT_SOLVER.
  • KEY_VALUE_STORE_SOLVER (str): Deprecated alias for DEFAULT_SOLVER.

Source​

class Source(BaseModel)

Configuration for data source tables in Unity Catalog.

Arguments:

  • container_tags_table (str): Full Unity Catalog path to the container tags table (narrow/EAV format). Required when filtering by container tags. Omit for wide-only data models that carry container attributes as columns on container_metrics. project_id scoping is independent of this field — it works in both narrow EAV and wide-only data models because it is applied to container_metrics (and channel_mapping if configured) regardless of whether container_tags_table is set.
  • container_metrics_table (str): Full Unity Catalog path to the container metrics table.
  • channel_metrics_table (str): Full Unity Catalog path to the channel metrics table.
  • channels_uri (str): Full Unity Catalog path to the channels data table.
  • poi_channels_uri (str): Full Unity Catalog path to the Points-in-Time (POI) channel data table. Required only when the report selects POI channels via poi_channel(); omit it for sample-only data models.
  • channel_mapping_table (str): Full Unity Catalog path to the channel mapping table. Required when using channel_with_alias() for logical alias resolution.
  • unit_conversion_table (str): Full Unity Catalog path to the unit conversion table. When set together with a channel_mapping_table whose rows carry source_unit and target_unit columns, the query engine converts time-series values from the source to the target unit during solve().

UnitySink​

class UnitySink(BaseModel)

Configuration for data sink location in Unity Catalog.

Arguments:

  • catalog (str): Target catalog name for output tables.
  • schema (str): Target schema name for output tables.
  • table_prefix (str): Prefix to use for generated output table names.
  • cleanup_temp_tables (bool): When True, the intermediate __impulse_temp_* tables written to this sink during batch solving are dropped after Report.run() persists successfully. Defaults to False (temp tables are retained for inspection and only cleared at the start of the next report run).

Comparator​

class Comparator(str, Enum)

Supported comparison operators for container filters.

CastType​

class CastType(str, Enum)

Supported Spark cast types for tag value columns.

TagFilter​

class TagFilter(BaseModel)

A single tag-based filter applied on the container_tags_table (EAV).

Arguments:

  • tag_name (str): The tag key / element_id to filter on.
  • comparator (Comparator): The comparison operator.
  • value (str | int | float | datetime): The expected value. Must match the cast_type: str for STRING, int for INT, int|float for DOUBLE, ISO-format string for TIMESTAMP (automatically parsed to datetime).
  • cast_type (CastType): Spark type to cast the tag value to before comparison.

MetricFilter​

class MetricFilter(BaseModel)

A single metric-based filter applied on the container_metrics_table.

Arguments:

  • column_name (str): The metric column to filter on.
  • comparator (Comparator): The comparison operator.
  • value (str | int | float | datetime): The expected value. When value_type is provided, must match accordingly.
  • value_type (CastType): When provided, validates and/or converts the value to the expected type.

ContainerFilters​

class ContainerFilters(BaseModel)

Container-level filters in disjunctive normal form (OR of ANDs).

Each outer list element is a group of filters that are AND-combined. The resulting group expressions are then OR-combined.

Arguments:

  • tag_filters (list[list[TagFilter]]): Tag-based filter groups (applied on container_tags_table).
  • metric_filters (list[list[MetricFilter]]): Metric-based filter groups (applied on container_metrics_table).

QueryEngine​

class QueryEngine(BaseModel)

Configuration for the query engine solver.

Arguments:

  • solver (Solvers, default=Solvers.DEFAULT_SOLVER): The solver type to use for query execution.

  • raw_encoder (RawEncoder, optional, default=None): Encoder used to convert RAW point data into intervals. RLE collapses consecutive equal-valued samples into runs; INTERVAL only derives tend and drops exact duplicates. Only takes effect when data_type=RAW; ignored for RLE input. When omitted and data_type=RAW, it is resolved to RLE at validation time; for RLE input the field stays None and is never consulted.

  • solver_config (SolverConfig): Per-table column name mappings and filter configuration for the solver. Use this when your silver-layer tables use non-default column names or when you need project/toolbox scoping. Key sub-fields:

  • project_id (str): Top-level project filter value applied to container_tags, container_metrics, and channel_mapping tables when the corresponding columns exist after column renaming.

  • Per-table sections (container_tags, container_metrics, channel_mapping, channels, etc.) each with column_name_mapping and filters dicts.

When omitted, all default column names are used and no project/toolbox filtering is applied.

  • measurement_db (str, default="MeasurementDB"): Registered name of the MeasurementDB implementation to use.
  • measurement_db_config (dict): Keyword arguments for that implementation's config class, merged with the source tables.

validate_drop_implausible_data_requires_raw​

def validate_drop_implausible_data_requires_raw()

drop_implausible_data=True currently only takes effect with RAW data.

The filter is applied inside the RAW -> interval conversion path by the selected raw_encoder (RleEncoder / IntervalEncoder).

reject_channels_filter_on_is_plausible_in_raw​

def reject_channels_filter_on_is_plausible_in_raw()

Reject an is_plausible channels filter in RAW mode (delegates to SolverConfig).

The shared invariant lives on SolverConfig so direct query-engine use (DefaultSolver) enforces the same rule; here we surface it at config-parse time.

warn_channels_filters_bridge_in_raw​

def warn_channels_filters_bridge_in_raw()

Warn (not reject) on any non-plausibility channels filter in RAW mode.

Such filters run before raw encoding and bridge intervals across dropped samples. That is fine for whole-channel scoping but corrupts per-sample cleaning, and the two cannot be told apart statically, so we warn. is_plausible is hard-rejected above as the one unambiguously per-sample case.

default_raw_encoder_for_raw_data​

def default_raw_encoder_for_raw_data()

When data_type=RAW and raw_encoder is unset, default to RLE.

IncrementalConfig​

class IncrementalConfig(BaseModel)

Configuration for incremental processing behavior.

Arguments:

  • enabled (bool, default=False): Whether incremental processing is enabled.
  • silver_last_modified_column (str, default="timestamp"): Column name in the silver layer used for freshness comparison. This is the name after column_name_mapping is applied (the physical name unless you remap that column).
  • gold_last_modified_column (str, default="last_modified"): Column name in the gold layer used for freshness comparison.

CalculatedChannels​

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).

FullRecalculation​

class FullRecalculation(BaseModel)

Scoped full-recalculation request, by entity name.

Lists aggregations / events / calculated channels — identified by their human-readable name — that should be fully recalculated on the next run, regardless of whether their definition hash changed.

Only meaningful in incremental mode: listed entities recompute over all containers and fully replace their gold rows (the same treatment a definition-hash change already receives), while every other entity stays incremental. In full mode this is a no-op, since everything recomputes over all containers anyway.

Names that do not match any registered entity are rejected at the start of Report.run (fail fast on typos / stale names).

Arguments:

  • aggregations (list of str, default=[]): Names of aggregations to fully recalculate.
  • events (list of str, default=[]): Names of events to fully recalculate.
  • calculated_channels (list of str, default=[]): Names of calculated channels to fully recalculate.

build_measurement_db_config​

def build_measurement_db_config(
    source_dict: dict[str, Any], query_engine: QueryEngine
) -> tuple[type[AbstractMeasurementDB], MeasurementDBConfig]

Resolve the selected MeasurementDB implementation and build its config class from

measurement_db_config merged with the source tables in source_dict.

Raises KeyError for an unregistered measurement_db and TypeError for a missing, unknown or duplicated config key.

ImpulseConfig​

class ImpulseConfig(BaseModel)

Main configuration model.

Attributes​

source : Source Configuration for input data sources. unity_sink : UnitySink Configuration for output data location. container_filters : ContainerFilters, optional Optional container-level filters (tag-based and/or metric-based). query_engine : QueryEngine, optional 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(). full_recalculation : FullRecalculation, optional Optional scoped full-recalculation request naming aggregations / events / calculated channels to fully recompute on the next incremental run regardless of definition-hash changes. Defaults to None (no override). measurement_dimensions : list of str, optional Column names to surface from container_metrics into the gold-layer measurement_dimension table. Names are matched after query_engine.solver_config.container_metrics.column_name_mapping has been applied — i.e. these are the internal (post-mapping) column names, not the physical silver column names. If a silver table uses a physical name like my_measurement_id mapped to container_id, list "container_id" here. Each listed name lands in the gold table verbatim, so the configured name is also the gold column name. Defaults to ["container_id", "start_ts", "stop_ts"]. The framework does not inject any column the user omits — keeping container_id in the list is recommended because it is the upsert key for incremental processing and the join key to event-fact tables, but the choice is the user's. Examples​

config_data = { ... "source": { ... "container_metrics_table": "impulse_demo.silver.container_metric", ... "channel_metrics_table": "impulse_demo.silver.channel_metric", ... "channels_uri": "impulse_demo.silver.channel_data", ... "channel_mapping_table": "impulse_demo.data_model.channel_mapping" ... }, ... "unity_sink": { ... "catalog": "impulse_demo", ... "schema": "silver_refactored", ... "table_prefix": "evaluation" ... }, ... "container_filters": { ... "tag_filters": [ ... [ ... {"tag_name": "uut_id", "comparator": "==", "value": "AA080518", "cast_type": "string"} ... ] ... ], ... "metric_filters": [ ... [ ... {"column_name": "uut_id", "comparator": "==", "value": "AA080518"}, ... {"column_name": "start_ts", "comparator": ">=", "value": "2025-04-27T05:20:54.000Z"} ... ] ... ] ... }, ... "query_engine": { ... "solver": "DefaultSolver", ... "solver_config": { ... "project_id": "my_project", ... "container_tags": { ... "column_name_mapping": {"entity_id": "container_id"}, ... "filters": {"parent_id": "my_parent_id"} ... }, ... "container_metrics": { ... "column_name_mapping": {} ... }, ... "channel_metrics": { ... "column_name_mapping": {} ... }, ... "channel_mapping": { ... "column_name_mapping": {}, ... "filters": {"toolbox_id": "my_toolbox"} ... }, ... "channels": { ... "column_name_mapping": {} ... } ... } ... } ... } config = ImpulseConfig.model_validate(config_data)