Transform Rule Implementation

Transform rules are a special type of rule whose purpose is to “transform” incoming data from a specific field into other data. This transformation can range from simply dividing all incoming numerical data by a factor 1000 (rule:apply_static_transform) or adding a prefix to incoming strings (rule:add_prefix_suffix); to mapping incoming values to entirely different values (rule:map_values) or retrieving specific portions of incoming strings with Regex (rule:apply_regex). Transform rules can be chained together to create transformations that consist of multiple steps. So, for example, if one has the string 'WEATHER_STATION_XXX_KNMI' with 'XXX' being an ID, and wishes for this to be transformed to 'KNMI_XXX', then this result can be achieved using a combination of apply_regex (for retrieving the ID) and add_prefix_suffix (for adding the 'KNMI_' prefix).

As transform rules are applied on incoming data, they are the only type of rule that can be used within Transformation Configurations. While transform rules can be used within flow designs, it is very uncommon for them to be, due to transform rules being required to inherit a different base class than rules of other rule types (TransformRule instead of FlowRule). This separate base class lacks many of the methods that FlowRule has, making it hard to use within flows.

An example of a simple transform rule can be found below (add_prefix_suffix):

    from ewx_public.transform_rule import TransformRule


    class AddPrefixSuffix(TransformRule):
        def apply(self, string_value, prefix_suffix_flag, **kwargs):
            """
            Rule that adds a specified `string_value` as either a prefix or suffix to the channel data.
    
            Args:
                string_value (str): String to add.
                prefix_suffix_flag (str): Add `string_value` as `prefix` or `suffix`.
    
            Returns:
                pandas.Series
    
            """
    
            # Add prefix or suffix
            index = self.dataframe.dropna().index
            if prefix_suffix_flag == 'prefix':
                result = self.dataframe[self.dataframe.columns[0]][index].apply(lambda x: string_value + str(x))
            else:
                result = self.dataframe[self.dataframe.columns[0]][index].apply(lambda x: str(x) + string_value)
    
            # Return it
            self.dataframe[self.dataframe.columns[0]][index] = result
            return self.dataframe

Legacy imports

The legacy import (from energyworx_public.rule import AbstractTransformRule) is still supported on the platform. See ewx-public Package for the full migration guide.

There are several things to note here: A transform rule inherits TransformRule (previously AbstractTransformRule), not FlowRule. Because of this, it uses a different layout than normal rules do (as explained in the [Rule Implementation](./0- Rule%20Implementation.md)). A transform rule has only an apply method — it has no prepare_context, because it cannot load datasources or timeseries data. Nor does it have access to most methods a normal rule has, as those are unique to a flow and cannot be used in a transformation configuration.

Available attributes

The platform instantiates your rule once per configured mapping and sets the following attributes on it before calling apply. They are the complete set of data a transform rule has to work with:

Attribute	Type	Contents
self.dataframe	pd.DataFrame (or pd.Series when chained)	Only the field(s) this mapping points at. This is the data you are asked to transform.
self.adapted_data	pd.DataFrame	The entire parsed payload — every field of the file, whether mapped in the transformation configuration or not.
self.result	pd.DataFrame	The transformation output built so far: everything the transformation configuration has already resolved for this payload.
self.datasources	dict[str, Datasource | None]	The datasources referenced by the payload, keyed by datasource ID. None for IDs that do not exist in the platform yet.
self.mapping	TransformationConfigurationMapping	The mapping this rule was attached to (its field, static value, configured rules and parameters).
self.version	datetime	Ingestion version: the creation timestamp of the ingestion trigger. Used as the default version for tags and channels.
self.billing_account_id	str	The billing account this ingestion belongs to.
self.rule_logger	logger	A logger to be used by rules. Any logs created with this logger will appear in the Audit Events under their own separate header.

self.adapted_data and self.result both carry the row index of the parsed payload, so they can be aligned with the field being transformed — for example self.adapted_data.loc[self.dataframe.index].

Everything is a string

The parsed payload is loaded with dtype="str", so every value in self.dataframe and self.adapted_data is a string, no matter what it represents. Empty values become NaN. A field that was a nested object in the source file arrives as a JSON string. Converting to the right type is the transform rule's job — see Convert timestamp fields.

self.dataframe — the field being transformed

What ends up in self.dataframe depends on how the mapping is configured:

• A single field — one column, named after the field.
• Several fields (a comma-separated field list in the mapping) — one column per field, in the configured order.
• A static value — a single column named static_value, holding that value on every row.
• A chained rule — the return value of the previous rule in the chain. If the previous rule returned a pd.Series, self.dataframe is a Series and not a DataFrame; rules that support chaining therefore often start with a squeeze() or an isinstance check.

self.adapted_data — the whole parsed payload

self.adapted_data holds the complete output of the market adapter: one column per field found in the file, plus a filename column. Nested XML and JSON structures are flattened into one column per path (for example groupA.subgroupB.fieldC) and one row per record; CSV files are flat to begin with. XML and JSON are treated identically.

Use it whenever a transformation needs data that the transformation configuration does not (or cannot) map to the field you are transforming — for example to look up a column whose exact name is not known upfront:

from ewx_public.transform_rule import TransformRule


class StripNamespacePrefix(TransformRule):
    def apply(self, prefix='ns', field_path=None, **kwargs):
        """Return `field_path`'s values, ignoring any XML namespace prefix in the payload.

        The transformation configuration maps the field by its plain path (`groupA.fieldB`),
        while the incoming file may prefix every element (`ns:groupA/ns:fieldB`).
        Matching is done against the unprefixed column names in `self.adapted_data`.

        Args:
            prefix (str): Namespace prefix to ignore.
            field_path (str): Plain (unprefixed) path of the field to return.

        Returns:
            pandas.Series

        """
        for column in self.adapted_data.columns:
            if column.replace(f'{prefix}:', '') == field_path:
                return self.adapted_data[column]

        return self.dataframe.squeeze(axis=1)

note

A rule can only read self.adapted_data if it is executed at all — and a mapping whose field is missing from the payload does not execute its rules. Give such a mapping a static value so the rule always runs, and have it return the real value from self.adapted_data.

self.result — what has been transformed so far

self.result is the dataframe the transformation configuration is building. Its columns are :-separated keys describing the object each value belongs to, so a rule can read anything the transformation has already resolved for this payload.

The datasource-level keys are the most useful ones:

Column	Contents
datasource:id	The resolved datasource ID for the row.
datasource:name, datasource:description	Datasource name and description.
datasource:timezone (also available as timezone)	The datasource timezone.
datasource:classifier	The datasource classifier.
datasource:filter	Virtual-datasource filter.

Tags, channels and flow properties follow the same scheme, indexed by their position in the transformation configuration — datasource:tag:0:argument:tag, datasource:channel:0:argument:classifier, timeseries:channel:0:data:value, flow_property:0:<key>, and so on. Property keys are lower-cased in these column names; datapoint-attribute keys keep their original case.

The built-in local_to_utc rule is a good example: it converts a local timestamp field to UTC by reading the timezone the datasource mapping already resolved.

dataframe = pd.to_datetime(self.dataframe.squeeze(axis=1), format=dateformat)

result = []
for timezone, df in dataframe.groupby(self.result["timezone"]):
    if df.dt.tz is None:
        df = df.dt.tz_localize(timezone, ambiguous="infer")
    result.append(df.dt.tz_convert("UTC"))

return pd.concat(result)

Treat self.result as read-only. It is the same object for every rule in the transformation, so writing to it affects the ingestion beyond your own mapping; return your transformed data instead.

self.datasources — datasources referenced by the payload

self.datasources is keyed by datasource ID, and every datasource ID resolved from the payload has a key:

• the value is a Datasource object (including its tags) for datasources that already exist in the platform;
• the value is None for datasources that this ingestion is about to create.

Keys are the resolved datasource IDs, normalised to upper case — regardless of the casing used in the file. Upper-case the ID you look up (self.datasources.get(datasource_id.upper())) rather than relying on the dict to normalise it for you.

This makes it straightforward to reject or flag data for meters that are not registered yet:

from energyworx.domain import TransformAbortError
from ewx_public.transform_rule import TransformRule


class RequireExistingDatasource(TransformRule):
    def apply(self, **kwargs):
        """Abort the ingestion when the payload references unknown datasources."""
        unknown = [ds_id for ds_id, datasource in self.datasources.items() if datasource is None]
        if unknown:
            raise TransformAbortError(f"Datasource(s) do not exist yet: {', '.join(unknown)}")

        return self.dataframe

Not available on the datasource ID mapping

The datasources are looked up after the datasource IDs have been resolved. A rule attached to the datasource ID mapping therefore runs before the lookup happened and cannot rely on self.datasources being populated. Every other mapping can.

self.mapping — the mapping being transformed

self.mapping is the transformation configuration mapping your rule is attached to. The attributes worth knowing:

• field — the configured field path, or None for a static mapping. A comma-separated string when several fields are mapped.
• static_value — the configured static value, or None.
• rules — the transform rules configured on this mapping, in execution order.
• key — for property, annotation, datapoint-attribute and flow-property mappings: the target key(s), comma-separated.
• get_function_param(key) — the value of a configured parameter of the (legacy) map function.

The built-in composite rule uses it to interleave static values with fields: statics = self.mapping.static_value.split(",").

Rule parameters

The parameters configured on the rule in the transformation configuration are passed to apply as keyword arguments, keyed by parameter key. Always accept **kwargs as well, so the rule keeps working when the platform or the configuration passes something your signature does not name.

Parameter values are strings unless the parameter declares a different value type:

Configured value type	Received as
str (default)	str
int	int
float	float
bool	bool — true, 1, y, yes and + are truthy (case-insensitive)
list	list[str]

Return value

The result of a transform rule is either a pandas Series or a DataFrame (and not an instance of RuleResult, like with normal rules) containing the transformed data. Which one to use depends on whether you wish to be able to chain this transform rule to another transform rule. If you return a pandas Series, then it will not be possible to use a transform rule on the result this transform rule returned. If you return a pandas DataFrame, then a transform rule can be used on the result. The reason to do one or the other depends on the transform rule itself: If you are writing a transform rule like the one above (adding a prefix/suffix), it would not be strange that you would like to perform additional transformations on the data. In this case, it is better to return a dataframe as it makes your transform rules more independent and allows for transform rule chaining. However, if we would write a transform rule that is very specific to a client's use case and one should never attempt to transform the data any further because of that, then it is better to return a series.

Whichever you return, keep the row index intact: the platform writes your result into self.result alongside the other mappings, and aligns it by index.

Two constraints depend on what the mapping feeds:

• Timeseries values (a channel's value mapping) must be convertible to float. Returning something that is not raises One or more values in timeseries data cannot be converted to float — this is deliberate, so that stray strings are not silently interpreted as annotations.
• Property, annotation and datapoint-attribute mappings with several keys expect one column per key, in the same order as the mapping's key list.

Aborting an ingestion

Raising TransformAbortError stops the transformation for the payload and reports it as a warning audit event — use it for expected, business-rule rejections (an unknown datasource, a message that should be ignored):

from energyworx.domain import TransformAbortError

note

TransformAbortError has no ewx-public equivalent yet; the energyworx.domain import above is the supported way to raise it.

Any other exception raised from apply is wrapped in a TransformRuleException that names the rule and the field it was processing, and is reported as an error audit event with the traceback. The transformation configuration itself raises InvalidConfigurationException when a rule's configuration cannot produce a usable result — the built-in to_date rule does this when the values do not match the configured date format.

The platform also logs one audit entry per executed transform rule, so a rule appearing in the audit events confirms it ran.

What is not available

Transform rules run during ingestion, before any flow exists. The following are not available, and are the usual reason for an AttributeError:

• prepare_context, self.context — transform rules only have apply.
• self.datasource, self.channel_data, self.flow_properties, self.destination_column — flow-only attributes.
• store_timeseries, store_annotations, add_tags, self.timeseries_service — a transform rule cannot load or persist platform data. It transforms values; the transformation configuration decides where they are written. See Rule framework functions for what flow rules can do instead.

The standalone helpers in ewx_public.standard_utils are usable, and several are a good fit for ingestion work — for example parse_date, validate_and_parse_date_string, format_date, csl_to_list, set_tz and tz_from_name_or_offset. See the ewx-public Package reference.