Skip to content

converters

Classes#

Automatic #

Bases: _ADIBaseTransformer

Attributes#

transformation_type class-attribute instance-attribute #
transformation_type: str = 'TRANS_AUTO'

BinAverage #

Bases: _ADIBaseTransformer

Attributes#

transformation_type class-attribute instance-attribute #
transformation_type: str = 'TRANS_BIN_AVERAGE'

CreateTimeGrid #

Bases: DataConverter

Attributes#

interval instance-attribute #
interval: str

The frequency of time points. This is passed to pd.timedelta_range as the 'freq' argument. E.g., '30s', '5min', '10min', '1H', etc.

Functions#

convert #
convert(
    data: xr.DataArray,
    variable_name: str,
    dataset_config: Optional[DatasetConfig] = None,
    retrieved_dataset: Optional[RetrievedDataset] = None,
    retriever: Optional[StorageRetriever] = None,
    time_span: Optional[Tuple[datetime, datetime]] = None,
    input_key: Optional[str] = None,
    **kwargs: Any
) -> Optional[xr.DataArray]
Source code in tsdat/transform/converters.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def convert(
    self,
    data: xr.DataArray,
    variable_name: str,
    dataset_config: Optional["DatasetConfig"] = None,
    retrieved_dataset: Optional[RetrievedDataset] = None,
    retriever: Optional["StorageRetriever"] = None,
    time_span: Optional[Tuple[datetime, datetime]] = None,
    input_key: Optional[str] = None,
    **kwargs: Any,
) -> Optional[xr.DataArray]:
    if time_span is None:
        raise ValueError("time_span argument required for CreateTimeGrid variable")

    # TODO: if not time_span, then get the time range from the retrieved data

    start, end = time_span[0], time_span[1]
    time_deltas = pd.timedelta_range(
        start="0 days",
        end=end - start,
        freq=self.interval,
        closed="left",
    )
    date_times = time_deltas + start

    time_grid = xr.DataArray(
        name=variable_name,
        data=date_times,
        dims=variable_name,
        attrs={"units": "Seconds since 1970-01-01 00:00:00"},
    )

    if (
        retrieved_dataset is not None
        and input_key is not None
        and retriever is not None
        and retriever.parameters is not None
        and retriever.parameters.trans_params is not None
        and retriever.parameters.trans_params.select_parameters(input_key)
    ):
        params = retriever.parameters.trans_params.select_parameters(input_key)
        width = params["width"].get(variable_name)
        alignment = params["alignment"].get(variable_name)
        if width is not None and alignment is not None:
            bounds = _create_bounds(time_grid, alignment=alignment, width=width)
            retrieved_dataset.data_vars[f"{variable_name}_bound"] = bounds

    return time_grid

Interpolate #

Bases: _ADIBaseTransformer

Attributes#

transformation_type class-attribute instance-attribute #
transformation_type: str = 'TRANS_INTERPOLATE'

NearestNeighbor #

Bases: _ADIBaseTransformer

Attributes#

transformation_type class-attribute instance-attribute #
transformation_type: str = 'TRANS_SUBSAMPLE'

Functions#

error_traceback #

error_traceback(error)
Source code in tsdat/transform/converters.py 
28
29
30
31
32
33
def error_traceback(error):
    warnings.warn(
            "\n\nEncountered an error running transformer. Please ensure necessary"
            " dependencies are installed."
        )
    logger.exception(error)