# Data Transformation
open.WARE provides a comprehensive set of transformation operations for data processing within the open.WARE platform. These operations enable you to manipulate, aggregate, analyze, and transform data streams in real-time or batch processing scenarios.
## REST API Usage
Transformations can be invoked via the REST API endpoint:
```
POST /api/transform/pipe
```
### Request Format
All transformation requests use the same JSON structure:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.TransformationClassName",
"params": {
// Transformation-specific parameters
}
}
],
"start": 1609459200000,
"end": 1609545600000,
"reference": "optional_reference_id"
}
```
- **`stages`** (required): Array of transformation operations executed sequentially. Each stage receives the output of the previous stage.
- **`start`** (optional): Global start timestamp in milliseconds (Unix epoch). Automatically passed to all operations that support it.
- **`end`** (optional): Global end timestamp in milliseconds (Unix epoch). Automatically passed to all operations that support it.
- **`reference`** (optional): Reference identifier passed to all operations. Used for data source differentiation or session tracking.
**Note**: The `action` field can use either fully qualified Java class names (e.g., `"de.openinc.opentransformations.LiveDataSource"`) or configured aliases (e.g., `"live"`) for better readability. See [Aliases in Production Environments](#aliases-in-production-environments) for details.
### Response Format
The API returns an `OpenWareDataItem` JSON object:
```json
{
"id": "sensor_001",
"name": "Temperature Sensor",
"source": "temperature",
"user": "temperature",
"meta": {},
"reference": "optional_reference_id",
"valueTypes": [
{
"name": "temperature",
"unit": "°C",
"type": "Number"
}
],
"values": [
{
"date": 1609459200000,
"value": [23.5]
},
{
"date": 1609459260000,
"value": [23.7]
}
]
}
```
**Response Fields:**
- **`id`**: Unique sensor identifier
- **`name`**: Human-readable sensor name
- **`source`**: Data source identifier
- **`user`**: Same as source (legacy field)
- **`meta`**: Metadata object with custom properties
- **`reference`**: Reference identifier if provided
- **`valueTypes`**: Array describing each data dimension
- **`name`**: Dimension name
- **`unit`**: Unit of measurement
- **`type`**: OpenWare type string. Valid values: `"Number"`, `"String"`, `"Boolean"`, `"Geo"`, `"Object"`
- **`values`**: Array of timestamped data points
- **`date`**: Unix timestamp in milliseconds
- **`value`**: Array of values matching the order of `valueTypes`
### Multi-dimensional Data
Data can have multiple dimensions. Each value array contains elements corresponding to each dimension defined in `valueTypes`:
```json
{
"valueTypes": [
{"name": "temperature", "unit": "°C", "type": "Number"},
{"name": "humidity", "unit": "%", "type": "Number"}
],
"values": [
{"date": 1609459200000, "value": [23.5, 65.2]},
{"date": 1609459260000, "value": [23.7, 64.8]}
]
}
```
## Available Operations
The following sections document each available transformation operation with its fully qualified Java class name. For improved readability in production environments, you can configure **aliases** (short names like `"live"`, `"historical"`, `"math"`) instead of full class names - see [Aliases in Production Environments](#aliases-in-production-environments) for configuration details.
**Note**: Some operations (particularly `MapTransformer`) use the open.WARE condition and rule system for advanced filtering and mapping logic. For complete documentation of all available rule types and condition structures, see **[RULES.md](RULES.md)**.
### Data Sources
Data source operations retrieve sensor data from the open.WARE platform. They typically serve as the first stage in a transformation pipeline.
#### Live Data Source
**Class**: `de.openinc.opentransformations.LiveDataSource`
Retrieves the most recent N data points from a sensor, optionally at specific timestamps. Can retrieve data at multiple time intervals when combined with `unit`, `start`, and `end` parameters.
**Required Parameters**:
- **`id`** (string): Sensor identifier
- **`source`** (string): Data source identifier
- **`amount`** (integer): Number of most recent data points to retrieve
**Optional Parameters**:
- **`unit`** (string): Time interval unit for multi-point retrieval. Valid values: `"minute"`, `"hour"`, `"day"`, `"week"`, `"month"`, `"year"`. When provided with `start` and `end`, retrieves `amount` values at each interval boundary.
- **`start`** (integer): Start timestamp in milliseconds. Can be provided globally or per-stage.
- **`end`** (integer): End timestamp in milliseconds. Can be provided globally or per-stage.
- **`reference`** (string): Reference identifier for data source differentiation. Can be provided globally or per-stage.
- **`offset`** (integer): Time offset in milliseconds to shift the retrieval timestamp
- **`useStart`** (boolean): If `true`, uses `start` timestamp instead of `end` for retrieval. Default: `false`
- **`overwriteTime`** (boolean): If `true`, overwrites retrieved data timestamps with the retrieval timestamp. Default: `false`
**Behavior**:
- Without `unit`: Retrieves the `amount` most recent values at the `end` timestamp (or current time if not specified)
- With `unit`, `start`, and `end`: Generates time intervals and retrieves `amount` values at each interval boundary
- Values from previous pipeline stages are appended to the result
**REST API Example (Simple)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_001",
"source": "temperature",
"amount": 10
}
}
],
"end": 1609545600000
}
```
**REST API Example (Interval-based)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_001",
"source": "temperature",
"amount": 1,
"unit": "hour"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with sensor metadata and the requested data points in chronological order.
#### Raw Data Source
**Class**: `de.openinc.opentransformations.RawDataSource`
Retrieves historical raw sensor data within a specified time range. Supports optional data sampling or min-max retrieval to limit result size.
**Required Parameters**:
- **`id`** (string): Sensor identifier
- **`source`** (string): Data source identifier
- **`start`** (integer): Start timestamp in milliseconds. Can be provided globally or per-stage.
- **`end`** (integer): End timestamp in milliseconds. Can be provided globally or per-stage.
**Optional Parameters**:
- **`reference`** (string): Reference identifier for data source differentiation. Can be provided globally or per-stage.
- **`mode`** (string): Data retrieval mode. Valid values:
- `"minmax"`: Returns min and max values for each time bucket
- `"first"`; Returns the first N values
- `"last"`: Returns the last N values
- `"sample"`: Returns sampled data points
- **`maxValues`** (integer): Maximum number of values to retrieve. Required when `mode` is specified.
- **`dimension`** (integer): Dimension index for mode-based retrieval. Required when `mode` is specified.
- **`updateAggregates`** (boolean): Whether to update aggregate statistics during retrieval. Default: `false`
- **`startoverwrite`** (integer): Overrides the global `start` timestamp
- **`endoverwrite`** (integer): Overrides the global `end` timestamp
**Behavior**:
- Retrieves all raw data points within the specified time range
- When `mode` is specified, limits data points using the specified strategy (min-max or sampling)
- Supports analytic sensors (virtual sensors created through analytics)
- Values from previous pipeline stages are appended to the result
**REST API Example (Basic)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**REST API Example (With Sampling)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature",
"mode": "minmax",
"maxValues": 1000,
"dimension": 0
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with all historical data points within the time range, optionally sampled according to the specified mode.
### Mathematical Operations
#### Math Operation
**Class**: `de.openinc.opentransformations.MathOperation`
Applies mathematical expressions to each data point using the [mXparser library](https://mathparser.org/). Supports complex expressions with custom arguments, functions, and filtering conditions.
**Required Parameters**:
- **`operation`** (string): Operation mode. Currently supports `"forEach"` to apply expressions to each value
- **`expressions`** (array): Array of expression objects defining calculations to perform
- **`arguments`** (array): Array of argument objects defining constants or external data sources
**Optional Parameters**:
- **`functions`** (array): Array of custom function definitions
- **`filters`** (array): Array of filter expression strings. Only values where all filters evaluate to 1.0 are processed.
**Expression Object Structure**:
```json
{
"expression": "x0 * 2 + offset",
"name": "doubled_temp",
"unit": "°C",
"owtype": "Number"
}
```
- **`expression`**: Mathematical expression string. Reference input dimensions as `x0`, `x1`, `x2`, etc.
- **`name`**: Name for the resulting dimension
- **`unit`**: Unit of measurement for the result
- **`owtype`**: OpenWare type. Valid values: `"Number"`, `"String"`, `"Boolean"`, `"Geo"`, `"Object"`
**Argument Object Structure**:
```json
{
"name": "offset",
"value": 5.0
}
```
Or reference live data from another sensor:
```json
{
"name": "threshold",
"value": {
"id": "sensor_002",
"source": "temperature",
"dimension": 0
}
}
```
**Function Object Structure**:
```json
{
"name": "myFunc",
"value": "f(x) = 2*x + 1"
}
```
**Behavior**:
- Processes each value from the input data
- Input dimensions are accessible as `x0`, `x1`, `x2`, etc. in expressions
- Boolean dimensions are converted to 1.0 (true) or 0.0 (false)
- Skips values that produce NaN results or fail filter conditions
- Creates new output data with dimensions defined by the expressions
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.MathOperation",
"params": {
"operation": "forEach",
"expressions": [
{
"expression": "(x0 - 32) * 5/9",
"name": "celsius",
"unit": "°C",
"owtype": "Number"
},
{
"expression": "if(x0 > threshold, 1, 0)",
"name": "high_temp_alert",
"unit": "",
"owtype": "Boolean"
}
],
"arguments": [
{
"name": "threshold",
"value": 75.0
}
],
"functions": [],
"filters": ["x0 > 0"]
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with new dimensions created by evaluating the expressions for each input value. The result ID is prefixed with `"Math_"`.
#### Difference Transformer
**Class**: `de.openinc.opentransformations.DiffTransformer`
Calculates the difference between consecutive values in a specified dimension. Useful for computing deltas, rates of change, or consumption from cumulative readings.
**Required Parameters**:
- **`dimension`** (integer): Index of the dimension to calculate differences for
**Optional Parameters**:
- **`laterValuesFirst`** (boolean): If `true`, processes values in reverse chronological order. Default: `false`
- **`includeInitial`** (boolean/number): If provided as a boolean (`true`), includes the first value with an initial difference of 0. If provided as a number, uses that number as the initial baseline value.
**Behavior**:
- Reduces output to a single dimension (the specified dimension)
- First value is skipped unless `includeInitial` is provided
- For each subsequent value, calculates: `currentValue - previousValue`
- Preserves original timestamps
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "power_meter",
"source": "energy"
}
},
{
"action": "de.openinc.opentransformations.DiffTransformer",
"params": {
"dimension": 0,
"laterValuesFirst": false,
"includeInitial": 0.0
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with a single dimension containing the calculated differences between consecutive values.
### Statistical Operations
#### In-Memory Aggregation
**Class**: `de.openinc.opentransformations.InMemAggregation`
Performs statistical aggregations on data. Supports three modes: time-based aggregation, split-based aggregation, and multi-dimensional aggregation.
#### Mode 1: Time-Based Aggregation
Aggregates data into time buckets (e.g., hourly, daily).
**Required Parameters**:
- **`dimension`** (integer): Dimension index to aggregate
- **`start`** (integer): Start timestamp in milliseconds
- **`end`** (integer): End timestamp in milliseconds
- **`timeinterval`** (string): Time interval for bucketing. Valid values: `"minute"`, `"hour"`, `"day"`, `"week"`, `"month"`, `"year"`
- **`operation`** (string): Statistical operation to perform
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.InMemAggregation",
"params": {
"dimension": 0,
"timeinterval": "hour",
"operation": "mean"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
#### Mode 2: Split-Based Aggregation
Divides the time range into N equal splits and aggregates each split.
**Required Parameters**:
- **`dimension`** (integer): Dimension index to aggregate
- **`start`** (integer): Start timestamp in milliseconds
- **`end`** (integer): End timestamp in milliseconds
- **`splits`** (integer): Number of equal time splits (minimum 1)
- **`operation`** (string): Statistical operation to perform
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.InMemAggregation",
"params": {
"dimension": 0,
"splits": 24,
"operation": "max"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
#### Mode 3: Multi-Dimensional Aggregation
Aggregates multi-dimensional data where the first dimension acts as a grouping/split dimension.
**Required Parameters**:
- **`operations`** (array of strings): Array of operations, one for each dimension except the first. Length must equal `number of dimensions - 1`.
**Optional Parameters**:
- **`percentile`** (number): Percentile value (0-100) when using `"percentile"` operation
- **`fillEmptyBuckets`** (array of booleans): For each dimension, whether to fill empty buckets with the last valid value. Length must match `operations` array.
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "multi_sensor"
}
},
{
"action": "de.openinc.opentransformations.InMemAggregation",
"params": {
"operations": ["mean", "max", "sum"],
"percentile": 95,
"fillEmptyBuckets": [false, true, false]
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Supported Operations**:
- **`mean`** or **`avg`**: Average value
- **`min`**: Minimum value
- **`max`**: Maximum value
- **`sum`**: Sum of all values
- **`stdd`**: Standard deviation
- **`variance`**: Variance
- **`count`**: Number of values (changes unit to `"#"`)
- **`percentile`**: Percentile value (requires `percentile` parameter)
- **`diffminmax`**: Difference between max and min
- **`difffirstlast`**: Difference between last and first value
- **`first`**: First value (works with non-numeric dimensions)
- **`last`**: Last value (works with non-numeric dimensions)
**Behavior**:
- Reduces output to the specified dimension(s)
- Each output value represents the aggregate for one time bucket
- Dimension name is prefixed with the operation and interval (e.g., `"mean_hour_temperature"`)
- For non-numeric dimensions, only `first`, `last`, and `count` operations are supported
**Returns**: `OpenWareDataItem` with aggregated values, one per time bucket or split. The name is suffixed with `"_<operation>"`.
#### Moving Average Transformer
**Class**: `de.openinc.opentransformations.MovingAvgTransformer`
Applies a moving average smoothing filter to reduce noise in data. Uses a sliding window to compute the average of the most recent N values.
**Required Parameters**:
- **`dimension`** (integer): Dimension index to smooth
- **`window`** (integer): Window size for moving average (number of data points)
**Behavior**:
- Reduces output to a single dimension (the specified dimension)
- First `window-1` values are skipped (buffer fill period)
- Each output value is the average of the current value and the previous `window-1` values
- Maintains original timestamps
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.MovingAvgTransformer",
"params": {
"dimension": 0,
"window": 5
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with smoothed values. Output contains `N - window + 1` values where N is the input size.
### Data Transformation
#### Map Transformer
**Class**: `de.openinc.opentransformations.MapTransformer`
Maps values in a dimension to new values based on conditional rules. Transforms one dimension type to another (e.g., number to string, number to boolean). Uses the open.WARE alarm/monitoring condition system for flexible rule evaluation.
**Required Parameters**:
- **`dimension`** (integer): Dimension index to map
- **`mappings`** (array): Array of mapping rule objects
- **`defaultMapping`** (any): Default value when no mapping conditions match
- **`valueType`** (object): New value type definition for the mapped dimension
**Mapping Object Structure**:
```json
{
"condition": {
"type": "rule",
"source": "temperature",
"id": "sensor_001",
"dimension": 0,
"trigger": {
"type": "number_in_range",
"min": 0,
"max": 10
}
},
"mapping": "low"
}
```
**Value Type Object Structure**:
```json
{
"name": "status",
"unit": "",
"type": "String"
}
```
**Condition System**:
Conditions use the open.WARE alarm/monitoring system with two main types:
1. **Simple rule conditions** (`type: "rule"`): Evaluate a single rule against a sensor dimension
2. **Compound conditions** (`type: "and"` or `type: "or"`): Combine multiple conditions with logical operators
Each rule condition requires:
- `type`: Must be `"rule"`
- `source`: Data source identifier of the sensor to evaluate
- `id`: Sensor ID to evaluate
- `dimension`: Dimension index in that sensor
- `trigger` (or `rule`): Rule object defining the match criteria
**Available Rule Types**:
See [RULES.md](RULES.md) for the complete reference. Common rule types include:
- **Numeric**: `number_gt`, `number_lt`, `number_gte`, `number_lte`, `number_in_range`, `number_out_of_range`, `number_equals`, `number_change`, `number_increased`, `number_decreased`, etc.
- **String**: `string_equals`, `string_includes`, `string_starts_with`, `string_ends_with`, `string_change`, etc.
- **Boolean**: `boolean_true`, `boolean_false`, `boolean_rising_edge`, `boolean_falling_edge`, `boolean_change`
- **Timestamp**: `ts_minute_of_day`, `ts_day_of_week`, `ts_last_value_changed`, etc.
**Behavior**:
- Evaluates conditions sequentially for each value in the input data
- Uses the first matching mapping
- Falls back to `defaultMapping` if no conditions match
- Replaces the specified dimension with the new mapped dimension
- Preserves all other dimensions
- User must have read access to all sensors referenced in conditions
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.MapTransformer",
"params": {
"dimension": 0,
"mappings": [
{
"condition": {
"type": "rule",
"source": "temperature",
"id": "sensor_001",
"dimension": 0,
"trigger": {
"type": "number_lt",
"value": 10
}
},
"mapping": "cold"
},
{
"condition": {
"type": "rule",
"source": "temperature",
"id": "sensor_001",
"dimension": 0,
"trigger": {
"type": "number_in_range",
"min": 10,
"max": 25
}
},
"mapping": "comfortable"
},
{
"condition": {
"type": "rule",
"source": "temperature",
"id": "sensor_001",
"dimension": 0,
"trigger": {
"type": "number_gte",
"value": 25
}
},
"mapping": "hot"
}
],
"defaultMapping": "unknown",
"valueType": {
"name": "temperature_category",
"unit": "",
"type": "String"
}
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with the specified dimension replaced by mapped values matching the new value type.
#### Dimension Reduce Transformer
**Class**: `de.openinc.opentransformations.DimensionReduceTransformer`
Reduces multi-dimensional data to a single dimension, removing all other dimensions from both metadata and values.
**Required Parameters**:
- **`dimension`** (integer): Index of the dimension to keep
**Behavior**:
- Removes all dimensions except the specified one
- Removes corresponding values from all data points
- Preserves timestamps and the selected dimension values
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "multi_sensor",
"source": "environment"
}
},
{
"action": "de.openinc.opentransformations.DimensionReduceTransformer",
"params": {
"dimension": 1
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with only the specified dimension retained in both `valueTypes` and `values`.
### Geospatial Operations
#### Geo Transformer
**Class**: `de.openinc.opentransformations.GeoTransformer`
Performs geospatial transformations and clustering operations on coordinate data.
**Required Parameters**:
- **`operation`** (string): Type of geospatial operation to perform
#### Operation: Extract Coordinates
Extracts coordinate data from GeoJSON objects and converts them to numeric dimensions.
**Operation Value**: `"extractcoord"`
**Optional Parameters**:
- **`dimension`** (integer): Index of dimension containing GeoJSON data. If not specified, assumes default GeoJSON structure.
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "gps_tracker",
"source": "location"
}
},
{
"action": "de.openinc.opentransformations.GeoTransformer",
"params": {
"operation": "extractcoord",
"dimension": 0
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
#### Operation: K-Means Clustering
Groups coordinate points into clusters using the K-means algorithm.
**Operation Value**: `"clusterkmeans"`
**Additional Parameters**: Passed to the clustering algorithm (refer to GeoHelper implementation for details)
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "gps_tracker",
"source": "location"
}
},
{
"action": "de.openinc.opentransformations.GeoTransformer",
"params": {
"operation": "clusterkmeans"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
#### Operation: DBSCAN Clustering
Groups coordinate points using density-based spatial clustering (DBSCAN algorithm).
**Operation Value**: `"clusterdbscan"`
**Additional Parameters**: Passed to the clustering algorithm (refer to GeoHelper implementation for details)
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "gps_tracker",
"source": "location"
}
},
{
"action": "de.openinc.opentransformations.GeoTransformer",
"params": {
"operation": "clusterdbscan"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with transformed geospatial data. Format depends on the operation:
- **extractcoord**: Numeric coordinate dimensions (latitude, longitude)
- **clusterkmeans/clusterdbscan**: Clustered data with cluster labels
### Data Merging
#### Sync Merge
**Class**: `de.openinc.opentransformations.SyncMerge`
Synchronizes and merges multiple data streams into a single multi-dimensional output based on timestamps. Each input stream becomes additional dimensions in the output.
**Required Parameters**:
- **`items`** (array): Array of transformation pipeline objects to execute and merge
**Optional Parameters**:
- **`mode`** (string): Merge strategy for handling missing values. Valid values:
- `"FILL"` (default): Fills missing values with default values from value types (e.g., 0 for numbers)
- `"FILL_NULL"`: Fills missing values with `null`
- **`fillIndex`** (integer): Controls which item's timestamps are used for output. Default: `-1`
- `-1`: Use all timestamps from all items (union)
- `0, 1, 2, ...`: Use only timestamps from the specified item index
**Behavior**:
- Executes each pipeline in `items` independently
- Collects all unique timestamps (or timestamps from `fillIndex` item)
- For each timestamp, combines values from all sources
- Missing values are handled according to `mode`:
- **FILL**: Uses the most recent value before the timestamp (last observation carried forward), or default value if none exists
- **FILL_NULL**: Inserts `null` for missing values at exact timestamps
- Dimension names are prefixed with the source item name to avoid conflicts
- Global `start`, `end`, and `reference` parameters are automatically passed to all items
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.SyncMerge",
"params": {
"items": [
{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_001",
"source": "temperature",
"amount": 10
}
},
{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_002",
"source": "humidity",
"amount": 10
}
},
{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_003",
"source": "pressure",
"amount": 10
}
}
],
"mode": "FILL",
"fillIndex": -1
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**REST API Example (Only temperature timestamps)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.SyncMerge",
"params": {
"items": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_002",
"source": "humidity"
}
}
],
"mode": "FILL",
"fillIndex": 0
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with merged dimensions from all input items. The ID and source are taken from the first item, and the name is prefixed with `"Merge"`. Each value type name is prefixed with its source item name.
### Specialized Operations
#### Template Transformer
**Class**: `de.openinc.opentransformations.TemplateTransformer`
Reshapes data according to a template specification, changing the sensor metadata (id, source, name, meta) and value type definitions without modifying the actual values. Validates that input data matches the template structure.
**Required Parameters**:
- **`template`** (object): JSON object defining the target data structure
**Template Object Structure**:
```json
{
"id": "transformed_sensor",
"source": "processed_data",
"name": "Transformed Sensor Data",
"meta": {},
"valueTypes": [
{
"name": "temperature",
"unit": "°C",
"type": "Number"
}
]
}
```
**Behavior**:
- Validates that input data structure matches the template's `valueTypes`
- Replaces sensor metadata (id, source, name, meta) with template values
- Preserves all data values and timestamps
- Throws an error if value types don't match between input and template
- If input has no values, returns an empty item with template metadata
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "raw_sensor",
"source": "raw_data"
}
},
{
"action": "de.openinc.opentransformations.MathOperation",
"params": {
"operation": "forEach",
"expressions": [
{
"expression": "x0 * 1.8 + 32",
"name": "fahrenheit",
"unit": "°F",
"owtype": "Number"
}
],
"arguments": []
}
},
{
"action": "de.openinc.opentransformations.TemplateTransformer",
"params": {
"template": {
"id": "temp_fahrenheit",
"source": "processed_temperature",
"name": "Temperature (Fahrenheit)",
"meta": {
"conversion": "celsius_to_fahrenheit",
"version": "1.0"
},
"valueTypes": [
{
"name": "temperature",
"unit": "°F",
"type": "Number"
}
]
}
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with metadata replaced by template values and original data values preserved.
#### Metering Consumption
**Class**: `de.openinc.opentransformations.MeteringConsumption`
Calculates consumption/usage from cumulative metering data by computing differences over time intervals. Handles meter resets and rollbacks when configured.
**Required Parameters**:
- **`id`** (string): Sensor identifier
- **`source`** (string): Data source identifier
- **`dimension`** (integer): Dimension index containing cumulative meter readings
- **`timeinterval`** (string): Time interval for consumption calculation. Valid values: `"minute"`, `"hour"`, `"day"`, `"week"`, `"month"`, `"year"`
- **`start`** (integer): Start timestamp in milliseconds
- **`end`** (integer): End timestamp in milliseconds
**Optional Parameters**:
- **`correct_negative_difference`** (boolean): If `true`, handles meter resets by detecting negative differences and correcting them. Default: `false`
- **`offset`** (integer): Time offset in milliseconds to shift interval boundaries
- **`offsettedTimestamp`** (boolean): If `true`, uses the offset timestamp in output. Default: `false`
- **`start_overwrite`** (integer): Overrides the start timestamp
- **`end_overwrite`** (integer): Overrides the end timestamp
**Behavior**:
- Generates time intervals based on `timeinterval`, `start`, and `end`
- For each interval:
1. Retrieves the meter reading at interval start
2. Retrieves the meter reading at interval end
3. Calculates consumption: `endValue - startValue`
- When `correct_negative_difference` is `true`:
- Detects meter resets (when reading decreases)
- Retrieves historical data to find reset point
- Calculates consumption using values before and after reset
- Reduces output to a single dimension
- Only works with numeric dimensions
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.MeteringConsumption",
"params": {
"id": "power_meter_001",
"source": "electricity",
"dimension": 0,
"timeinterval": "hour",
"correct_negative_difference": true,
"offset": 0,
"offsettedTimestamp": false
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**REST API Example (Daily with offset)**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.MeteringConsumption",
"params": {
"id": "water_meter",
"source": "water_consumption",
"dimension": 0,
"timeinterval": "day",
"offset": 21600000,
"offsettedTimestamp": true,
"correct_negative_difference": false
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with one value per time interval, each representing the consumption during that interval. Timestamps represent the start of each interval (or offset start if `offsettedTimestamp` is `true`).
## Enterprise Edition Operations
**Note**: The following transformation operations require the **openware-EE plugin** to be installed and active. These operations provide advanced analytics capabilities and performance optimizations for enterprise deployments.
### Database-Level Aggregation
#### DB Aggregation
**Class**: `de.openinc.owee.transformation.DBAggregation`
**Common Alias**: `source_aggregation`
Performs statistical aggregations directly in the database (MongoDB) for significantly improved performance on large datasets. Functionally similar to `InMemAggregation` but executes aggregation operations at the database level rather than loading all data into memory.
**Required Parameters**:
- **`id`** (string): Sensor identifier
- **`source`** (string): Data source identifier
- **`dimension`** (integer): Dimension index to aggregate
- **`start`** (integer): Start timestamp in milliseconds
- **`end`** (integer): End timestamp in milliseconds
- **`operation`** (string): Statistical operation to perform
**Aggregation Mode Parameters** (choose one):
- **Time-based**: **`timeinterval`** (string) - Time interval for bucketing: `"second"`, `"minute"`, `"hour"`, `"day"`, `"week"`, `"month"`, `"year"`
- **Split-based**: **`splits`** (integer) - Number of equal time splits
**Optional Parameters**:
- **`reference`** (string): Reference identifier for data filtering
**Supported Operations**:
Same as `InMemAggregation`: `mean`, `avg`, `min`, `max`, `sum`, `stdd`, `variance`, `count`, `diffminmax`, `difffirstlast`, `first`, `last`
**Behavior**:
- Executes aggregation using MongoDB aggregation pipeline for optimal performance
- Automatically falls back to in-memory aggregation for analytic sensors (virtual sensors)
- Handles timezone configuration via `timezone` setting (default: Europe/Berlin)
- Significantly faster than in-memory aggregation for large datasets
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.owee.transformation.DBAggregation",
"params": {
"id": "sensor_001",
"source": "temperature",
"dimension": 0,
"timeinterval": "hour",
"operation": "mean"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**REST API Example (Using Alias)**:
```json
{
"stages": [
{
"action": "source_aggregation",
"params": {
"id": "sensor_001",
"source": "temperature",
"dimension": 0,
"splits": 24,
"operation": "max"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with aggregated values, one per time bucket or split. Same output format as `InMemAggregation`.
**Performance Note**: Recommended for datasets with thousands of data points or more. The database-level aggregation can be orders of magnitude faster than in-memory processing.
### Pattern Recognition
#### DTW Transformer
**Class**: `de.openinc.owee.transformation.DTWTransformer`
Performs Dynamic Time Warping (DTW) analysis to detect pattern occurrences in time series data. DTW measures similarity between two temporal sequences by finding the optimal alignment between them, making it robust to time shifts and stretching.
**Required Parameters**:
- **`pattern`** (object): Pattern definition object
- **`data_dim`** (integer): Dimension index in the input data to analyze. Default: `0`
**Pattern Object Structure**:
```json
{
"source": "temperature",
"id": "reference_sensor",
"start": 1609459200000,
"end": 1609462800000,
"dim": 0
}
```
- **`source`**: Data source of the reference pattern
- **`id`**: Sensor ID containing the reference pattern
- **`start`**: Start timestamp of the pattern window
- **`end`**: End timestamp of the pattern window
- **`dim`**: Dimension index in the pattern data (default: 0)
**Behavior**:
- Retrieves the reference pattern from the specified sensor and time range
- Generates sliding windows over the input data matching the pattern's duration
- Calculates DTW distance for each window against the reference pattern
- User must have read access to the pattern data source
- Lower DTW distance indicates better pattern match
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.owee.transformation.DTWTransformer",
"params": {
"pattern": {
"source": "temperature",
"id": "reference_pattern",
"start": 1609459200000,
"end": 1609462800000,
"dim": 0
},
"data_dim": 0
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with DTW analysis results. Each value represents one window and contains:
- **`distance`** (Number): DTW distance to the pattern (lower = better match)
- **`start`** (Number): Start timestamp of the window (milliseconds)
- **`end`** (Number): End timestamp of the window (milliseconds)
- **`source`** (String): Source of the analyzed data
- **`ID`** (String): ID of the analyzed sensor
- **`dimension`** (Number): Dimension that was analyzed
- **`pattern`** (Object): Copy of the pattern configuration
The result ID is formatted as: `{original_id}_dtw_{pattern_start}_{pattern_end}`
**Use Cases**:
- Detect recurring patterns in sensor data
- Find anomalies by identifying periods unlike the reference pattern
- Quality control by comparing production cycles to a reference
- Energy consumption pattern analysis
### Advanced Statistics
#### Statistic Transformers
**Class**: `de.openinc.owee.transformation.StatisticTransformers`
Provides advanced statistical analysis operations including Principal Component Analysis (PCA), Multiple Linear Regression (MLR), and distribution analysis.
**Required Parameters**:
- **`operation`** (string): Type of statistical operation. Valid values: `"pca"`, `"mlr"`, `"distribution"`
#### Operation: PCA (Principal Component Analysis)
Performs dimensionality reduction and finds principal components (eigenvectors) of multi-dimensional data.
**Operation Value**: `"pca"`
**Additional Parameters**: None
**Behavior**:
- Normalizes all dimensions
- Calculates covariance matrix
- Performs eigenvalue decomposition
- Returns eigenvalues as dimensions (principal components)
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "multi_sensor",
"source": "environment"
}
},
{
"action": "de.openinc.owee.transformation.StatisticTransformers",
"params": {
"operation": "pca"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with one value containing eigenvalues as dimensions. Each dimension represents a principal component's variance. The result name is prefixed with `"PCA_"` and ID with `"PCA"`.
#### Operation: MLR (Multiple Linear Regression)
Performs ordinary least squares multiple linear regression where the last dimension is the dependent variable and all other dimensions are independent variables.
**Operation Value**: `"mlr"`
**Additional Parameters**: None
**Behavior**:
- Uses all dimensions except the last as independent variables (predictors)
- Uses the last dimension as the dependent variable (response)
- Calculates regression parameters (beta coefficients)
- Computes R-squared, residuals, and standard error
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.SyncMerge",
"params": {
"items": [
{"action": "historical", "params": {"id": "temp", "source": "sensors"}},
{"action": "historical", "params": {"id": "humidity", "source": "sensors"}},
{"action": "historical", "params": {"id": "energy", "source": "sensors"}}
],
"mode": "FILL"
}
},
{
"action": "de.openinc.owee.transformation.StatisticTransformers",
"params": {
"operation": "mlr"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with one value containing a single dimension of type `Object` with regression results:
- **`beta`**: Array of regression coefficients
- **`residuals`**: Array of residual values
- **`rSquared`**: R-squared value (coefficient of determination)
- **`sigma`**: Regression standard error
The result name is prefixed with `"MLR_"` and ID with `"MLR"`.
#### Operation: Distribution / Histogram
Calculates the empirical distribution of data and returns histogram bin statistics.
**Operation Value**: `"distribution"`
**Required Parameters**:
- **`dimension`** (integer): Dimension index to analyze
**Optional Parameters**:
- **`binCount`** (integer): Number of histogram bins. Default: `10`
**Behavior**:
- Creates histogram with specified number of bins
- Calculates statistics for each bin
- Uses empirical distribution from Apache Commons Math
**REST API Example**:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.owee.transformation.StatisticTransformers",
"params": {
"operation": "distribution",
"dimension": 0,
"binCount": 20
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Returns**: `OpenWareDataItem` with histogram data. Each value represents one bin with three dimensions:
- **`Count`** (Number): Number of values in the bin
- **`min`** (Number): Minimum value in the bin (0 if empty)
- **`max`** (Number): Maximum value in the bin (0 if empty)
The result name is suffixed with `"_Histogram"`. Values are indexed sequentially (0, 1, 2, ...) rather than by timestamp.
### Time Series Decomposition
#### STL Transformer
**Class**: `de.openinc.owee.transformation.STLTransformer`
Performs a Seasonal-Trend decomposition using LOESS (STL) on one numeric dimension of the previous result. The series is split into a seasonal component (the pattern that repeats every `period`), a trend component (the slowly changing level) and a residual (everything else). For every value, `value = seasonal + trend + residual` holds.
STL needs equally spaced values. With `interval`, the transformer aggregates the input into fixed-length buckets itself, so raw sensor data can be passed in directly.
**Optional Parameters**:
- **`dimension`** (integer): Index of the numeric dimension to decompose. Default: `0`
- **`interval`** (integer): Bucket length in milliseconds. The input values are aggregated per bucket, and empty buckets are filled by linear interpolation. Buckets are aligned to multiples of `interval` since the epoch (UTC), like `InMemAggregation` with `interval`. If omitted, the input is used as is and must already be equally spaced. The spacing is then derived from the median distance between timestamps.
- **`operation`** (string): Aggregation per bucket when `interval` is set. Valid values: `"mean"`, `"avg"`, `"median"`, `"min"`, `"max"`, `"sum"`, `"first"`, `"last"`, `"stdd"`, `"variance"`. Default: `"mean"`
- **`period`** (integer): Length of the season in milliseconds, e.g. `86400000` for a daily pattern. It must span at least two intervals, and the data must cover at least two full periods. The default depends on the interval:
| Interval | Default period |
|----------|----------------|
| Below 1 minute | 1 minute |
| 1 minute to below 15 minutes | 1 hour |
| 15 minutes to below 1 day | 1 day |
| 1 day to below 1 week | 1 week |
| 1 week and longer | 1 year (365 days) |
- **`robust`** (boolean): Uses robust LOESS fitting that reduces the influence of outliers. Outliers then show up in the residual rather than distorting trend and season. Default: `true`
- **`seasonalWidth`** (integer): LOESS window of the seasonal smoothing, measured in periods; odd values ≥ 7 are recommended. This lets the seasonal pattern change slowly over time. If omitted, the seasonal pattern is strictly periodic (the same in every period).
**Behavior**:
- Sorts the input chronologically
- Values whose dimension is missing or not a finite number are skipped when resampling. Without `interval` they are interpolated like empty buckets
- Leading and trailing gaps take the nearest available value
- Rejects more than 1,000,000 buckets; use a larger `interval` for long ranges
- Throws an error if the dimension is not of type `Number`, if the period is shorter than two intervals, or if the data covers less than two periods
**REST API Example** (daily pattern of a temperature, hourly buckets):
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.owee.transformation.STLTransformer",
"params": {
"dimension": 0,
"interval": 3600000,
"operation": "mean",
"period": 86400000
}
}
],
"start": 1609459200000,
"end": 1610668800000
}
```
**REST API Example** (weekly pattern of a daily consumption computed by a previous stage):
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.MeteringConsumption",
"params": {
"id": "meter_001",
"source": "energy",
"dimension": 0,
"timeinterval": "day"
}
},
{
"action": "de.openinc.owee.transformation.STLTransformer",
"params": {
"period": 604800000
}
}
],
"start": 1609459200000,
"end": 1617235200000
}
```
**Returns**: `OpenWareDataItem` with the `id` and `source` of the input and the name suffixed with `"_STL"`. There is one value per bucket, time-stamped with the start of the bucket. Each value has four dimensions, all with the unit of the input dimension:
- **`value`** (Number): Aggregated (and possibly interpolated) input value
- **`seasonal`** (Number): Seasonal component
- **`trend`** (Number): Trend component
- **`residual`** (Number): Residual component
The item's `meta.stl` object describes the decomposition:
```json
{
"dimension": 0,
"interval": 3600000,
"period": 86400000,
"valuesPerPeriod": 24,
"operation": "mean",
"seasonalStrength": 0.96,
"trendStrength": 0.92
}
```
`seasonalStrength` and `trendStrength` range from 0 (no seasonality/trend) to 1 (the component explains nearly all variation apart from the residual). They are calculated as `max(0, 1 - Var(residual) / Var(component + residual))`, following Hyndman & Athanasopoulos, *Forecasting: Principles and Practice*.
**Use Cases**:
- Daily or weekly load profiles of energy consumption
- Separating the seasonal pattern from long-term drift, e.g. of temperatures or production output
- Anomaly detection on the residual: large residuals mark values that neither the trend nor the season explains
- Seasonally adjusted series (`trend + residual`) for comparing periods
## Usage in Transformation Pipelines
Transformation operations are executed sequentially through the REST API endpoint `/api/transform/pipe`. Each operation (stage) receives the output of the previous operation and applies its transformation to produce new output.
### Pipeline Execution Model
1. **Sequential Processing**: Stages execute in order, with each stage receiving the previous stage's output
2. **Global Parameters**: `start`, `end`, and `reference` are automatically passed to all stages that support them
3. **Access Control**: If enabled, verifies user has read access to data produced by each stage
4. **Error Handling**: Pipeline stops on first error and returns error details
### REST API Endpoint
```
POST /api/transform/pipe
Content-Type: application/json
```
**Authentication**: Requires valid session or API token (implementation-specific)
### Pipeline Structure
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.OperationClass",
"params": {
// Operation-specific parameters
}
}
],
"start": 1609459200000,
"end": 1609545600000,
"reference": "optional_reference_id"
}
```
**Root Properties**:
- **`stages`** (required): Array of transformation operations
- **`start`** (optional): Global start timestamp (milliseconds since Unix epoch)
- **`end`** (optional): Global end timestamp (milliseconds since Unix epoch)
- **`reference`** (optional): Reference identifier for tracking or data source differentiation
**Stage Properties**:
- **`action`** (required): Fully qualified class name or alias of the transformation operation
- **`params`** (required): Parameters object specific to the operation
### Complete Pipeline Example
This example demonstrates a complete data analysis pipeline:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.MovingAvgTransformer",
"params": {
"dimension": 0,
"window": 5
}
},
{
"action": "de.openinc.opentransformations.InMemAggregation",
"params": {
"dimension": 0,
"timeinterval": "hour",
"operation": "mean"
}
}
],
"start": 1609459200000,
"end": 1609545600000,
"reference": "analysis_session_001"
}
```
**What this pipeline does**:
1. Retrieves raw temperature data from sensor_001
2. Applies 5-point moving average smoothing
3. Aggregates to hourly means
**Same pipeline using aliases** (when configured):
```json
{
"stages": [
{"action": "historical", "params": {"id": "sensor_001", "source": "temperature"}},
{"action": "moving_avg", "params": {"dimension": 0, "window": 5}},
{"action": "transform_aggregation", "params": {"dimension": 0, "timeinterval": "hour", "operation": "mean"}}
],
"start": 1609459200000,
"end": 1609545600000,
"reference": "analysis_session_001"
}
```
### Multi-Source Pipeline Example
Merge data from multiple sensors:
```json
{
"stages": [
{
"action": "de.openinc.opentransformations.SyncMerge",
"params": {
"items": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "temp_sensor",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "humidity_sensor",
"source": "humidity"
}
}
],
"mode": "FILL",
"fillIndex": 0
}
},
{
"action": "de.openinc.opentransformations.MathOperation",
"params": {
"operation": "forEach",
"expressions": [
{
"expression": "x0",
"name": "temperature",
"unit": "°C",
"owtype": "Number"
},
{
"expression": "x1",
"name": "humidity",
"unit": "%",
"owtype": "Number"
},
{
"expression": "x0 - (100 - x1) / 5",
"name": "heat_index",
"unit": "°C",
"owtype": "Number"
}
],
"arguments": []
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**What this pipeline does**:
1. Merges temperature and humidity data streams
2. Calculates a heat index from both values
### Access Control
When access control is enabled:
- User must have read permission for source sensors in data source operations
- User must have read permission for data produced by each pipeline stage
- Unauthorized access returns HTTP 403 Forbidden
### Error Responses
**HTTP 400 Bad Request**: Invalid parameters or malformed pipeline
```json
{
"error": "Missing one of the following parameters: id, source, start, end"
}
```
**HTTP 403 Forbidden**: Access denied
```json
{
"error": "Not allowed to read data temperature---sensor_001"
}
```
**HTTP 500 Internal Server Error**: Pipeline execution failure
```json
{
"error": "Could not process transform pipeline: <error details>"
}
```
## Aliases in Production Environments
In production environments, transformation operations can be referenced using aliases instead of their full class names. These aliases are configured through the environment variable `OW_DE_OPENINC_OW_MIDDLEWARE_SERVICES_TRANSFORMATIONSERVICE`, which contains a JSON mapping structure or these values are provided as JSON config file at startup.
### Configuring Aliases
Aliases are defined in a JSON configuration that maps short names to full class names (as ENV or config via ConfigLoader):
```json
{
"services": {
"live": "de.openinc.opentransformations.LiveDataSource",
"historical": "de.openinc.opentransformations.RawDataSource",
"math": "de.openinc.opentransformations.MathOperation",
"difference": "de.openinc.opentransformations.DiffTransformer",
"transform_aggregation": "de.openinc.opentransformations.InMemAggregation",
"template": "de.openinc.opentransformations.TemplateTransformer",
"sync_merge": "de.openinc.opentransformations.SyncMerge",
"dimension_reduce": "de.openinc.opentransformations.DimensionReduceTransformer",
"diffFirstLast": "de.openinc.opentransformations.MeteringConsumption",
"source_aggregation": "de.openinc.owee.transformation.DBAggregation"
}
}
```
### Standard Aliases
Common aliases used in production deployments:
| Alias | Full Class Name | Description |
|-------|----------------|-------------|
| `live` | `de.openinc.opentransformations.LiveDataSource` | Retrieve recent live data points |
| `historical` | `de.openinc.opentransformations.RawDataSource` | Retrieve historical raw data |
| `math` | `de.openinc.opentransformations.MathOperation` | Apply mathematical expressions |
| `difference` | `de.openinc.opentransformations.DiffTransformer` | Calculate consecutive differences |
| `transform_aggregation` | `de.openinc.opentransformations.InMemAggregation` | Statistical aggregations |
| `moving_avg` | `de.openinc.opentransformations.MovingAvgTransformer` | Moving average smoothing |
| `map` | `de.openinc.opentransformations.MapTransformer` | Conditional value mapping |
| `dimension_reduce` | `de.openinc.opentransformations.DimensionReduceTransformer` | Reduce to single dimension |
| `geo` | `de.openinc.opentransformations.GeoTransformer` | Geospatial operations |
| `sync_merge` | `de.openinc.opentransformations.SyncMerge` | Merge multiple data streams |
| `template` | `de.openinc.opentransformations.TemplateTransformer` | Apply metadata template |
| `diffFirstLast` | `de.openinc.opentransformations.MeteringConsumption` | Calculate consumption from meters |
| `source_aggregation` | `de.openinc.owee.transformation.DBAggregation` | Database-level aggregation (EE) |
### Using Aliases in REST API
When aliases are configured, use them in the `action` field instead of full class names:
```json
{
"stages": [
{
"action": "live",
"params": {
"id": "sensor_001",
"source": "temperature",
"amount": 10
}
},
{
"action": "transform_aggregation",
"params": {
"dimension": 0,
"timeinterval": "hour",
"operation": "mean"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}
```
**Benefits of Using Aliases**:
- Shorter, more readable pipeline definitions
- Easier to type and remember
- Abstraction from implementation details
- Consistent naming across deployments
## Quick Reference
### Common Use Cases
#### 1. Get Latest Sensor Values
```bash
curl -X POST https://your-openware-instance/api/transform/pipe \
-H "Content-Type: application/json" \
-d '{
"stages": [{
"action": "de.openinc.opentransformations.LiveDataSource",
"params": {
"id": "sensor_001",
"source": "temperature",
"amount": 1
}
}]
}'
```
#### 2. Calculate Hourly Averages
```bash
curl -X POST https://your-openware-instance/api/transform/pipe \
-H "Content-Type: application/json" \
-d '{
"stages": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {
"id": "sensor_001",
"source": "temperature"
}
},
{
"action": "de.openinc.opentransformations.InMemAggregation",
"params": {
"dimension": 0,
"timeinterval": "hour",
"operation": "mean"
}
}
],
"start": 1609459200000,
"end": 1609545600000
}'
```
#### 3. Merge Multiple Sensors
```bash
curl -X POST https://your-openware-instance/api/transform/pipe \
-H "Content-Type: application/json" \
-d '{
"stages": [{
"action": "de.openinc.opentransformations.SyncMerge",
"params": {
"items": [
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {"id": "temp_sensor", "source": "temperature"}
},
{
"action": "de.openinc.opentransformations.RawDataSource",
"params": {"id": "humidity_sensor", "source": "humidity"}
}
],
"mode": "FILL"
}
}],
"start": 1609459200000,
"end": 1609545600000
}'
```
#### 4. Calculate Energy Consumption
```bash
curl -X POST https://your-openware-instance/api/transform/pipe \
-H "Content-Type: application/json" \
-d '{
"stages": [{
"action": "de.openinc.opentransformations.MeteringConsumption",
"params": {
"id": "power_meter",
"source": "energy",
"dimension": 0,
"timeinterval": "day",
"correct_negative_difference": true
}
}],
"start": 1609459200000,
"end": 1609545600000
}'
```
### Parameter Quick Reference
| Parameter | Type | Used By | Description |
|-----------|------|---------|-------------|
| `id` | string | Data sources, MeteringConsumption | Sensor identifier |
| `source` | string | Data sources, MeteringConsumption | Data source identifier |
| `dimension` | integer | Most operations | Dimension index to process |
| `start` | integer | Time-based operations | Start timestamp (ms) |
| `end` | integer | Time-based operations | End timestamp (ms) |
| `amount` | integer | LiveDataSource | Number of data points |
| `operation` | string | InMemAggregation, MathOperation, GeoTransformer | Operation type |
| `timeinterval` | string | InMemAggregation, MeteringConsumption | Time interval (minute/hour/day/week/month/year) |
| `window` | integer | MovingAvgTransformer | Window size for moving average |
| `expressions` | array | MathOperation | Mathematical expressions to apply |
| `items` | array | SyncMerge | Sub-pipelines to merge |
| `template` | object | TemplateTransformer | Target data structure |
### OpenWare Data Types
Standard types available for value dimensions:
| Type (in JSON) | Java Class | Description |
|------|-----------|-------------|
| `Number` | `de.openinc.model.data.OpenWareNumber` | Numeric values (double precision) |
| `String` | `de.openinc.model.data.OpenWareString` | Text values |
| `Boolean` | `de.openinc.model.data.OpenWareBoolValue` | Boolean values (true/false) |
| `Geo` | `de.openinc.model.data.OpenWareGeo` | Geographic coordinates (GeoJSON) |
| `Object` | `de.openinc.model.data.OpenWareValueDimension` | Generic object type |
**Note**: When an `OpenWareDataItem` is serialized to JSON, the `type` field in `valueTypes` contains the simple type strings shown in the first column (e.g., `"Number"`), not the fully qualified Java class names.
## Error Handling
All transformation operations implement comprehensive error handling:
### Common Errors
**Missing Required Parameters**
```json
{
"error": "Missing one of the following parameters: id, source, amount"
}
```
**Solution**: Ensure all required parameters are provided in the `params` object.
**Invalid Dimension Index**
```json
{
"error": "Item has no Dimension 5"
}
```
**Solution**: Check that the dimension index is within the range `[0, number_of_dimensions - 1]`.
**Type Mismatch**
```json
{
"error": "Only operations first, last and count are allowed for non numeric dimensions"
}
```
**Solution**: Use operations compatible with the dimension type, or convert the dimension type first.
**Access Denied**
```json
{
"error": "Not allowed to read data temperature---sensor_001"
}
```
**Solution**: Ensure the authenticated user has read permission for the requested sensor.
**Invalid Time Range**
```json
{
"error": "Start timestamp must be before end timestamp"
}
```
**Solution**: Verify that `start < end` and timestamps are valid Unix epoch milliseconds.
### Best Practices
1. **Validate Parameters**: Always verify required parameters before submitting requests
2. **Check Dimensions**: Ensure dimension indices exist in your data
3. **Handle Empty Results**: Check for empty value arrays in responses
4. **Use Type Safety**: Match operation requirements with data types
5. **Test Incrementally**: Build complex pipelines one stage at a time
6. **Monitor Performance**: Large time ranges or complex operations may take time
7. **Use Sampling**: For large datasets, consider using `mode: "minmax"` or `mode: "sample"` in RawDataSource
## Development
### Adding New Transformation Operations
To create a custom transformation operation:
1. **Extend the Base Class**:
```java
public class MyCustomTransformer extends TransformationOperation {
private OpenWareDataItem result;
@Override
public TransformationOperation apply(User user, OpenWareDataItem old, JSONObject params)
throws Exception {
// Validate parameters
if (!params.has("requiredParam")) {
throw new IllegalArgumentException("Missing required parameter: requiredParam");
}
// Perform transformation
result = transformData(old, params);
return this;
}
@Override
public OpenWareDataItem getResult() {
return result;
}
@Override
public void setReference(String ref) {
// Store reference if needed
}
// Optionally override for time-based operations
@Override
public void setStart(long startTS) {
this.start = startTS;
}
@Override
public void setEnd(long endTS) {
this.end = endTS;
}
}
```
2. **Register the Operation**:
Add your class to `META-INF/services/de.openinc.api.TransformationOperation`:
```
de.openinc.opentransformations.MyCustomTransformer
```
3. **Build and Deploy**:
```bash
mvn clean package
# Copy the resulting JAR to your open.WARE plugins directory
```
4. **Configure Alias (Optional)**:
Add an alias to your environment configuration:
```json
{
"services": {
"my_custom": "de.openinc.opentransformations.MyCustomTransformer"
}
}
```
### Testing Transformations
Test your transformations using curl or any HTTP client:
```bash
curl -X POST http://localhost:8080/api/transform/pipe \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_TOKEN" \
-d '{
"stages": [{
"action": "de.openinc.opentransformations.MyCustomTransformer",
"params": {
"requiredParam": "value"
}
}]
}'
```
## Additional Resources
- [open.WARE Platform Documentation](https://openinc.de)
- [mXparser Mathematics Library](https://mathparser.org/)
- [OpenWare Data Model Reference](BasicDataFormat.md)
For more information, refer to the open.WARE platform documentation.