Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0.5.2
0.6.0
Original file line number Diff line number Diff line change
Expand Up @@ -41,17 +41,15 @@ Initialize a TimeSeriesSelector.

- `expr` (`TagExpression`): Tag expression to select.
- `uses_alias` (`bool`): Whether the channel resolves via the channel-alias table.
- `series_type` (`SeriesType`): How the selected channel's samples are interpreted. ``CONTINUOUS``
(default) builds a :class:`SampleSeries` — today's behavior,
unchanged. ``POINTS_IN_TIME`` builds a :class:`PointsInTimeSeries`
(values valid only at their timestamps); identification / matching is
identical, only the built object and its result dtype differ. This is
the plan-time source of truth for the series type (so ``dtype()`` is
correct for a bare POI selection with no per-channel metadata lookup).
- `value_type` (`SeriesValueType`): For a ``POINTS_IN_TIME`` selection, the declared value data type
(``DOUBLE`` / ``STRING``). Ignored for ``CONTINUOUS``. Drives plan-time
typing and string-op gating; validated against the silver
``poi_channels.dtype`` at solve time (assertion contract).
- `series_type` (`SeriesType`): Which object this selector builds. The default (``CONTINUOUS``)
builds a :class:`SampleSeries`; ``POINTS_IN_TIME`` builds a
:class:`PointsInTimeSeries`. Matching is identical; only the built
object and its result dtype differ. This is the plan-time source of
truth for the series type.
- `value_type` (`SeriesValueType`): For ``POINTS_IN_TIME``, the declared value data type
(``DOUBLE`` / ``STRING``); ignored otherwise. Drives plan-time typing
and string-op gating; validated against the silver
``poi_channels.dtype`` at solve time.

#### dtype

Expand All @@ -63,9 +61,10 @@ Returns the Spark data type.

**Returns**:

`pyspark.sql.types.DataType`: ``BinaryType`` for a CONTINUOUS selection (serialized ``SampleSeries``),
or the value-type-aware ``PointsInTimeSeries.dtype()`` for a
POINTS_IN_TIME selection (``array<array<double>>`` for numeric,
`pyspark.sql.types.DataType`: ``BinaryType`` when this selector builds a :class:`SampleSeries`
(a serialized blob). When it builds a :class:`PointsInTimeSeries`,
the value-type-aware ``PointsInTimeSeries.dtype()``
(``array<array<double>>`` for numeric,
``array<struct<tstart,value>>`` for string).

#### deserialize
Expand All @@ -74,19 +73,19 @@ POINTS_IN_TIME selection (``array<array<double>>`` for numeric,
def deserialize(d)
```

Deserialize a CONTINUOUS result after collection/toPandas.
Deserialize a :class:`SampleSeries` result after collection/toPandas.

POINTS_IN_TIME results are serialized by ``get_data()`` (a plain
``[[t, v], ...]`` list) and need no deserialization, so they are returned
as-is; only a CONTINUOUS (binary) blob is decoded to a :class:`SampleSeries`.
A :class:`PointsInTimeSeries` result is serialized by ``get_data()``
(a plain ``[[t, v], ...]`` list) and needs no deserialization, so it is
returned as-is; only a :class:`SampleSeries` (binary) blob is decoded.

**Arguments**:

- `d` (`Any`): Data to deserialize.

**Returns**:

`SampleSeries or Any`: Deserialized sample series (CONTINUOUS), else *d* unchanged.
`SampleSeries or Any`: The decoded :class:`SampleSeries`, else *d* unchanged.

#### build

Expand Down Expand Up @@ -238,7 +237,7 @@ Build the time series from cache.

**Returns**:

`SampleSeries or PointsInTimeSeries`: Built series (a ``SampleSeries`` for the CONTINUOUS-only aliases used today).
`SampleSeries or PointsInTimeSeries`: Built series (aliased selectors build a :class:`SampleSeries` today).

#### get\_required\_tag\_exprs

Expand Down Expand Up @@ -411,7 +410,11 @@ class TimeSeriesUDF(TimeSeriesOp)
#### \_\_init\_\_

```python
def __init__(func, *args, **kwargs)
def __init__(func,
*args,
container_tags=None,
container_metrics=None,
**kwargs)
```

Initialize a TimeSeriesUDF.
Expand All @@ -420,6 +423,12 @@ Initialize a TimeSeriesUDF.

- `func` (`callable`): The user-defined function to apply.
- `*args`: Arguments for the UDF.
- `container_tags` (`list of str`): Container-tag keys to inject into *func* as a ``container_tags``
keyword argument at build time (keyword-only; not treated as an
operand).
- `container_metrics` (`list of str`): Container-metric columns to inject into *func* as a
``container_metrics`` keyword argument at build time
(keyword-only; not treated as an operand).
- `**kwargs`: Keyword arguments for the UDF.

#### build
Expand All @@ -430,6 +439,11 @@ def build(cache: SeriesCache)

Build the time series from cache using the UDF.

When the UDF declared ``container_tags`` / ``container_metrics``, the
requested values are resolved from *cache* and passed to *func* as
``container_tags`` / ``container_metrics`` keyword arguments (dicts
keyed by the declared names; missing values are ``None``).

**Arguments**:

- `cache` (`SeriesCache`): Cache containing time series data.
Expand Down Expand Up @@ -459,14 +473,18 @@ class CallableTimeSeriesExpression()
#### \_\_init\_\_

```python
def __init__(func)
def __init__(func, container_tags=None, container_metrics=None)
```

Initialize a CallableTimeSeriesExpression.

**Arguments**:

- `func` (`callable`): Function to wrap.
- `container_tags` (`list of str`): Container-tag keys forwarded to each :class:`TimeSeriesUDF` this
wrapper builds.
- `container_metrics` (`list of str`): Container-metric columns forwarded to each :class:`TimeSeriesUDF`
this wrapper builds.

#### \_\_call\_\_

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,32 @@ the source to the target unit on the fly.

`pyspark.sql.DataFrame`: DataFrame containing results for each container.

#### attach\_container\_metadata

```python
def attach_container_metadata(query,
channels_df,
pre_filtered_containers_df=None) -> DataFrame
```

Left-join the container tags/metrics the selections request onto *channels_df*.

The columns ride into each group's pandas frame via the broadcast join in
:meth:`_prepare_channels_join`, without dropping channel rows. When
*pre_filtered_containers_df* is given, the metadata read is scoped to
those containers.

**Arguments**:

- `query` (`QueryBuilder`): Query object (selections + db info).
- `channels_df` (`pyspark.sql.DataFrame`): Channel-match frame from the filter pipeline.
- `pre_filtered_containers_df` (`pyspark.sql.DataFrame`): Incremental container subset to scope the metadata read to.

**Returns**:

`pyspark.sql.DataFrame`: *channels_df* with one column per requested tag/metric left-joined on
(unchanged when nothing is requested).

#### solve\_calculated\_channels

```python
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,31 @@ aliasing (e.g. ``DefaultSolver``) must override this.

`pyspark.sql.DataFrame`: Merged DataFrame with ``(container_id, channel_id, selector_ids)``.

#### attach\_container\_metadata

```python
def attach_container_metadata(query,
channels_df,
pre_filtered_containers_df=None) -> DataFrame
```

Attach requested container-level tags/metrics onto the channel-match frame.

No-op by default; solvers that inject container metadata (e.g.
``DefaultSolver``) override this to left-join the columns. When
*pre_filtered_containers_df* is given, the metadata read is scoped to
those containers.

**Arguments**:

- `query` (`QueryBuilder`): Query object (selections + db info).
- `channels_df` (`pyspark.sql.DataFrame`): Channel-match frame from the filter pipeline.
- `pre_filtered_containers_df` (`pyspark.sql.DataFrame`): Incremental container subset to scope the metadata read to.

**Returns**:

`pyspark.sql.DataFrame`: *channels_df*, unchanged by default.

#### filter\_candidates

```python
Expand Down
Loading