Impulse Tsal

Impulse — TSAL expressions

TSAL is the Python DSL every Impulse mode is built on. You select physical channels and combine them with operators and methods; the result is a TimeSeriesExpression that stays lazy until a query is solved (impulse-analyze) or a report is computed (impulse-reporting).

You reach the builder through a MeasurementDB, which you get from a Report (report.get_db()) or construct directly (see impulse-analyze):

db = report.get_db()
query = db.query    # a QueryBuilder

Selecting channels

Every keyword argument to channel() is a tag filter; all must match. Which tags are valid depends on your silver layer — either columns on channel_metrics (wide model) or (key, value) rows in a channel_tags table (see impulse-data-model).

eng_rpm = db.query.channel(channel_name="Engine RPM", brand="Seat", model="Leon")
veh_spd = db.query.channel(channel_name="Vehicle Speed Sensor")

channel() returns a TimeSeriesSelector, which is a TimeSeriesExpression — use it directly in arithmetic, comparisons, and methods.

Logical aliases. When a stable logical name should resolve to one of many physical channels via a separately maintained mapping table, use channel_with_alias() instead. It requires a channel_mapping_table in source (see impulse-config):

rpm = db.query.channel_with_alias(channel_name="Engine RPM")

Operators

Arithmetic (+ - * / %) — combine two signals or a signal and a scalar; produces a SampleSeries. Two signals are automatically synchronized to overlapping validity intervals first.

avg_temp   = (amb_air_temp + intake_air_temp) / 2
power      = voltage * current

Comparison (> >= < <= == !=) — produces Intervals, the time windows where the condition holds. This is the primary building block for events.

high_rpm = eng_rpm > 5000            # Intervals where RPM exceeds 5000

Logical (& intersection / | union) — combine Intervals. Parenthesize each comparison because of Python precedence.

rpm_band = (eng_rpm > 2000) & (eng_rpm < 5000)
cold_or_hot = (temp < 0) | (temp > 90)

Signal methods

All methods are available on any TimeSeriesExpression.

Resampling and integration

distance_km = veh_spd.resample(1e6).cumtrapz() / 3600 / 1e6   # speed (km/h) -> cumulative km

Filtering

rpm_in_band  = eng_rpm.where((eng_rpm > 2000) & (eng_rpm < 5000))   # SampleSeries
rpm_at_starts = eng_rpm.where(veh_spd.rising_edges())               # PointsInTimeSeries

Reducers (scalar results) — .sum(), .min(), .max(), .mean(). On a SampleSeries these are duration-weighted (a value that is current for 2 s counts twice as much as one current for 1 s).

Edge detection

Histograms (lower-level; for reports prefer the classes in impulse-aggregations)

Signal manipulation

Rolling windows — .rolling_average(window_size); .rolling_stats(window_size) returns a tuple of three SampleSeries (min, max, average).

User-defined functions

from impulse_query_engine.model.series import SampleSeries
from impulse_query_engine.analyze.metadata.time_series_expression import TimeSeriesExpression

@TimeSeriesExpression.udf
def squared(series):
    return SampleSeries(series.tstarts, series.tends, series.values ** 2)

squared_rpm = squared(eng_rpm)          # a reusable TSAL expression
# or apply inline: eng_rpm.apply(some_callable)

A UDF can also declare container-level tags/metrics it needs; the engine injects their per-container values as container_tags / container_metrics keyword arguments (dicts keyed by the names you declared, value None when absent). Declare them on @udf(...) or .apply(...):

@TimeSeriesExpression.udf(container_metrics=["nominal_power"], container_tags=["vehicle_type"])
def derate(series, container_metrics, container_tags):
    return series / container_metrics["nominal_power"]

derated = derate(power)
# inline equivalent: power.apply(derate_fn, container_metrics=["nominal_power"])

• container_metrics=[...] are read as columns on container_metrics (a missing column raises ValueError).
• container_tags=[...] are read from the EAV container_tags table and require a container_tags_table (else ValueError). See impulse-data-model.
• Values are constant per container. The declaration propagates when the UDF is wrapped in an aggregation, event, or calculated channel — the wrapper does not redeclare anything.

What an expression evaluates to (the core data model)

Every expression resolves — per container — to one of four in-memory types. Knowing the type tells you what you can do next and which event/aggregation accepts it.

Transitions between them:

Intervals also supports expand() / shrink(), merge_overlaps() / merge_intervals(gap), debounce(d), and filter(d) (drop windows shorter than d). where(PointsInTime) drops any instant that falls in a gap where the signal is not valid, so the result may be shorter than the input.

String-valued PointsInTimeSeries (from poi_channel(dtype="string"), e.g. DTC codes) is restricted: only equality (== "code" / != → PointsInTime) and sampling via .where(...) apply. Arithmetic (+ - * /), ordering (> >= < <=), and numeric reductions (sum/mean/min/max) raise a TypeError at build time — the PointsInTimeSeries reduction/ordering rows above apply to numeric POI series only.

Which type does each consumer need

• Events (impulse-events): BasicEvent and SequenceOfEvents need Intervals; PointsInTimeEvent needs PointsInTime. Validated at construction — a wrong type raises ValueError.
• Aggregations (impulse-aggregations): histogram/stats signal inputs must be SampleSeries.
• Calculated channels (impulse-channels): the wrapped expression must evaluate to SampleSeries (it materializes a derived signal). Validated at construction.

Common virtual-signal recipes

# derived signal
delta_temp = intake_air_temp - amb_air_temp

# integration-based signal (distance from speed)
distance_km = veh_spd.resample(1e6).cumtrapz() / 3600 / 1e6

# distance/cycle binning: one interval per 10 km travelled
every_10km = (distance_km % 10).intervals_between_falling_edges()