The Tag Processor is designed to prepare incoming data for the UMH data model. It processes messages through three configurable stages: defaults, conditional transformations, and advanced processing, all using a Node-RED style JavaScript environment.
Use the tag_processor compared to the nodered_js when you are processing tags or time series data and converting them to the UMH data model within the _historian data contract. This processor is optimized for handling structured time series data, automatically formats messages, and generates appropriate metadata.
Message Formatting Behavior
The processor automatically formats different input types into a consistent structure with a "value" field:
Simple Values (numbers, strings, booleans)
Input:
42
Output:
{
"value": 42
}
Input:
"test string"
Output:
{
"value": "test string"
}
Input:
true
Output:
{
"value": true
}
Arrays (converted to string representation)
Input:
["a", "b", "c"]
Output:
{
"value": "[a b c]"
}
Objects (preserved as JSON objects)
Input:
{
"key1": "value1",
"key2": 42
}
Output:
{
"value": {
"key1": "value1",
"key2": 42
}
}
Numbers (preserved as numbers)
Input:
23.5
Output:
{
"value": 23.5
}
Input:
42
Output:
{
"value": 42
}
This consistent formatting ensures that:
All messages have a "value" field
Simple types (numbers, strings, booleans) are preserved as-is
Complex types (arrays, objects) are converted to their string representations
Numbers are always preserved as numeric types (integers or floats)
Configuration
pipeline:
processors:
- tag_processor:
defaults: |
// Set default location hierarchy and datacontract
msg.meta.location_path = "enterprise.plant1.machiningArea.cnc-line.cnc5.plc123";
msg.meta.data_contract = "_historian";
msg.meta.tag_name = "value";
msg.payload = msg.payload; //does not modify the payload
return msg;
conditions:
- if: msg.meta.opcua_node_id === "ns=1;i=2245"
then: |
// Set path hierarchy and tag name for specific OPC UA node
msg.meta.virtual_path = "axis.x.position";
msg.meta.tag_name = "actual";
return msg;
advancedProcessing: |
// Optional advanced message processing
// Example: double numeric values
msg.payload = parseFloat(msg.payload) * 2;
return msg;
Processing Stages
Defaults
Sets initial metadata values
Runs first on every message
Must return a message object
Conditions
List of conditional transformations
Each condition has an if expression and a then code block
Runs after defaults
Must return a message object
Advanced Processing
Optional final processing stage
Can modify both metadata and payload
Must return a message object
Metadata Fields
The processor uses the following metadata fields:
Required Fields:
location_path: Hierarchical location path in dot notation (e.g., "enterprise.site.area.line.workcell.plc123")
data_contract: Data schema identifier (e.g., "_historian", "_analytics")
tag_name: Name of the tag/variable (e.g., "temperature", "pressure")
Optional Fields:
virtual_path: Logical, non-physical grouping path in dot notation (e.g., "axis.x.position")
Generated Fields:
umh_topic: Automatically generated from the above fields in the format:
Match an entire folder structure using includes("DataAccess_AnalogType")
Move all matching nodes into a new virtual path prefix (axis.x)
Preserve the original folder hierarchy under the new location
Apply consistent location path for the entire folder structure
Advanced Processing with getLastPayload
getLastPayload is a function that returns the last payload of a message that was avaialble in Kafka. Remember that you will get the full payload, and might still need to extract the value you need.
This is not yet implemented, but will be available in the future.
Note: In the tag_processor, the resulting payload will always include timestamp_ms and one additional key corresponding to the tag_name. If you need to fully control the resulting payload structure, consider using the nodered_js processor instead. You can set the topic and payload manually, as shown below:
pipeline:
processors:
- nodered_js:
code: |
// set kafka topic manually
msg.meta.umh_topic = "umh.v1.enterprise.site.area._workorder.new"
// only take two fields from the payload
msg.payload = {
"maintenanceSchedule": {
"eventType": msg.payload.maintenanceSchedule.eventType,
"description": msg.payload.maintenanceSchedule.description
}
}
return msg;
