DEV Community

Cover image for Tagging Manufacturing Data on AWS: From PLC Tags to IoT SiteWise Asset Models
Scott Peterson
Scott Peterson

Posted on Edited on

Tagging Manufacturing Data on AWS: From PLC Tags to IoT SiteWise Asset Models

#manufacturing #aws #iot #data

Every plant has a tag named something like N7_42_TT01. Somebody knew what it meant a decade or more ago. Today it's one of 40,000 tags on an OPC UA server, and your data team is trying to figure out whether it's a temperature, which press it belongs to, and what unit it's in.

Raw tags are cheap to collect and expensive to use. The value shows up when every value carries context: which site, which line, which machine, what it measures, and in what unit. That context is what "tagging" really means on the plant floor.

In this tutorial you'll build that context layer on AWS. You'll define an ISA-95 style naming convention, keep the tag-to-asset mapping in a version-controlled CSV, model your equipment in AWS IoT SiteWise with CloudFormation, ingest tags through a SiteWise Edge gateway, and bind every data stream to the right asset property with a short Python script. At the end you'll query the data with SQL by asset name instead of by tag address.

You don't need a plant or a gateway to follow along. Step 3 includes a small simulator that writes realistic raw tags to SiteWise exactly the way a gateway would.

All the code is on GitHub: scooteronce/sitewise-tag-mapping.

Before you start: you need an AWS account, the AWS CLI, Python 3.10 or later with boto3, and credentials that can manage IoT SiteWise and CloudFormation. Pick one Region where IoT SiteWise is available and use it for everything. The examples use us-west-2:

export AWS_REGION=us-west-2
Enter fullscreen mode Exit fullscreen mode

Please Read This: AWS IoT SiteWise isn't part of the AWS Free Tier. It bills for messages, metric computations, storage, and queries. Running the companion repo end to end with 24 hours of simulated history costs a few cents. Run the teardown at the end when you're done. While it's cheap to run this demo, I don't want to catch anyone off guard.

The problem with raw tags

A PLC tag name is an address, not a description. A controls engineer wrote it to make ladder logic work, not to make analytics easy. A typical OPC UA browse shows a mix like this:

Channel1.Press01.N7_42_TT01
Channel1.Press01.Hyd_Press_PSI
Kepware.Line3.Press_2.CycleCnt
PLC_04.DB12.DBD8
Line3_Press02_Motor_Amps_RAW
Enter fullscreen mode Exit fullscreen mode

Five tags, four naming styles, two vendors, and one tag that's just a memory address. Multiply that across a few thousand machines and three sites, and you get the same set of problems everywhere:

  • No hierarchy. You can't ask "average hydraulic pressure for every press on Line 3" because nothing says which tags belong to Line 3.
  • No units or types. Is Hyd_Press_PSI really PSI? Is RAW scaled?
  • No stability. Re-flash a PLC or migrate to a new OPC server and the addresses change. Every dashboard and model that referenced them breaks.
  • Tribal knowledge. The mapping lives in a spreadsheet on one engineer's laptop.

A good tagging strategy fixes all four. Each value gets a stable, human-readable identity in a hierarchy. The physical address becomes an implementation detail that you can change without breaking consumers. And the mapping lives in Git, where it gets reviewed like code.

Architecture

Raw tags flow left to right into IoT SiteWise as data streams. A mapping file in Git, applied by a small script, binds each stream to a named property on an asset in your hierarchy. Consumers only ever see the hierarchy.

Architecture diagram: an OPC UA server sends PLC tags to SiteWise Edge, which creates data streams in AWS IoT SiteWise. A sync_tags.py script, driven by tag-map.csv in Git and run in CI, binds each stream to a property in the ISA-95 asset hierarchy (Site, Area, Line, Press), which feeds consumers like SQL, ODBC, metrics, dashboards, and ML.

Step 1: Define a naming convention and a mapping file

Start with the hierarchy, not the tags. ISA-95 gives you a vocabulary that most plants already recognize:

ISA-95 level Example Becomes in SiteWise
Site slc Site asset (top of the hierarchy)
Area stamping Area asset
Line line3 Line asset
Equipment press01 Equipment asset
Measurement hydraulic_pressure Measurement property on the equipment model

ISA-95 also defines an Enterprise level above Site. This tutorial starts at Site to keep things short. Add an Enterprise model the same way if you run multiple companies or business units.

A few rules keep the convention usable:

  1. Lowercase snake_case only. No spaces, no vendor prefixes, no memory addresses.
  2. Measurement names describe the physical quantity, not the sensor: hydraulic_pressure, not PT_101.
  3. Units live in the model, not the name. hydraulic_pressure with unit psi, not hyd_press_psi.
  4. Same equipment type, same property names. Every press has oil_temperature, hydraulic_pressure, motor_current, and cycle_count, whatever the PLC calls them.

Now write the mapping file. This CSV is the source of truth that connects each raw data stream to a place in the hierarchy. Commit it to the same repo as your infrastructure code.

stream_alias,site,area,line,equipment,equipment_type,property
/slc-kep01/Channel1/Press01/N7_42_TT01,slc,stamping,line3,press01,press,oil_temperature
/slc-kep01/Channel1/Press01/Hyd_Press_PSI,slc,stamping,line3,press01,press,hydraulic_pressure
/slc-kep01/Channel1/Press01/Motor_Amps_RAW,slc,stamping,line3,press01,press,motor_current
/slc-kep01/Channel1/Press01/CycleCnt,slc,stamping,line3,press01,press,cycle_count
/slc-kep01/Channel1/Press02/TT_OIL,slc,stamping,line3,press02,press,oil_temperature
/slc-kep01/Channel1/Press02/HYD_P,slc,stamping,line3,press02,press,hydraulic_pressure
/slc-kep01/Channel1/Press02/MTR_I,slc,stamping,line3,press02,press,motor_current
/slc-kep01/Channel1/Press02/CYCLES,slc,stamping,line3,press02,press,cycle_count
Enter fullscreen mode Exit fullscreen mode

Notice that Press01 and Press02 use completely different tag names, but both map to the same four property names. That's the whole point. The left column can be ugly. Everything to the right of it is your contract with downstream consumers.

The /slc-kep01 prefix comes from the SiteWise Edge data source configuration in Step 3. It keeps aliases unique when two OPC UA servers expose the same paths.

Step 2: Model the equipment in IoT SiteWise

Asset models are the types: what a press is and what it measures. Assets are the instances: Press01 on Line 3 in Salt Lake City. Models change rarely and deserve infrastructure-as-code. Assets scale with your mapping file, so we'll create those from the CSV in Step 4.

This CloudFormation template defines four models (Site, Area, Line, Press) and the hierarchy between them. Every model, property, and hierarchy gets an external ID. External IDs are identifiers you choose. Most SiteWise APIs, including every call the scripts in this article make, accept them in place of a generated UUID when you add the externalId: prefix. A few operations still require the UUID, so check the API reference before you rely on it elsewhere.

AWSTemplateFormatVersion: "2010-09-09"
Description: ISA-95 asset models for manufacturing tag data

Resources:
  PressModel:
    Type: AWS::IoTSiteWise::AssetModel
    Properties:
      AssetModelName: Press
      AssetModelExternalId: model-press
      AssetModelProperties:
        - Name: oil_temperature
          ExternalId: press-oil_temperature
          DataType: DOUBLE
          Unit: degF
          Type: { TypeName: Measurement }
        - Name: hydraulic_pressure
          ExternalId: press-hydraulic_pressure
          DataType: DOUBLE
          Unit: psi
          Type: { TypeName: Measurement }
        - Name: motor_current
          ExternalId: press-motor_current
          DataType: DOUBLE
          Unit: A
          Type: { TypeName: Measurement }
        - Name: cycle_count
          ExternalId: press-cycle_count
          DataType: INTEGER
          Unit: count
          Type: { TypeName: Measurement }

  LineModel:
    Type: AWS::IoTSiteWise::AssetModel
    Properties:
      AssetModelName: Line
      AssetModelExternalId: model-line
      AssetModelHierarchies:
        - Name: presses
          ExternalId: line-presses
          ChildAssetModelId: !GetAtt PressModel.AssetModelId

  AreaModel:
    Type: AWS::IoTSiteWise::AssetModel
    Properties:
      AssetModelName: Area
      AssetModelExternalId: model-area
      AssetModelHierarchies:
        - Name: lines
          ExternalId: area-lines
          ChildAssetModelId: !GetAtt LineModel.AssetModelId

  SiteModel:
    Type: AWS::IoTSiteWise::AssetModel
    Properties:
      AssetModelName: Site
      AssetModelExternalId: model-site
      AssetModelHierarchies:
        - Name: areas
          ExternalId: site-areas
          ChildAssetModelId: !GetAtt AreaModel.AssetModelId
Enter fullscreen mode Exit fullscreen mode

Deploy it:

aws cloudformation deploy \
  --template-file asset-models.yaml \
  --stack-name sitewise-asset-models
Enter fullscreen mode Exit fullscreen mode

If you add a second equipment type later (a welder, a CNC, an oven), add a model for it and a hierarchy on LineModel. If several equipment types share core measurements, look at SiteWise asset model interfaces, which let you define a standard set of properties once and apply it across models.

Step 3: Get raw tags into SiteWise

In production, a SiteWise Edge gateway browses your OPC UA server and creates one data stream per variable. Each stream's alias is the browse path under the Objects node, with an optional prefix in front. Streams arrive before any asset exists. They sit as unassociated data streams until you bind them to an asset property, which is exactly the gap Step 4 fills.

You don't need a gateway to see this work. Option A simulates one. Option B is the real setup.

Option A: Simulate the gateway (recommended for this tutorial)

simulate_tags.py writes values for the eight raw aliases in the CSV straight to SiteWise by alias, with deliberately inconsistent tag names on each press. To SiteWise, the result looks the same as gateway output: disassociated data streams with ugly names. It only needs boto3 and your credentials.

simulate_tags.py 
"""
Publish simulated press tag data straight to AWS IoT SiteWise, standing in for a
SiteWise Edge gateway. Values are written by property alias, so each tag shows up
as a disassociated data stream, exactly what an OPC UA source on a gateway would
produce. The raw PLC tag names are deliberately inconsistent between presses
(e.g. /slc-kep01/Channel1/Press01/N7_42_TT01 vs /slc-kep01/Channel1/Press02/TT_OIL);
tag-map.csv is what gives them meaning.

  python simulate_tags.py --backfill-hours 24      # history for metrics and queries
  python simulate_tags.py --live                   # keep sending every 10s until Ctrl-C

Uses the standard AWS credential chain (AWS_PROFILE, AWS_REGION, ...).
"""
import argparse
import math
import os
import random
import time
import zlib

import boto3

MAX_ENTRIES = 10        # BatchPutAssetPropertyValue: entries per request
MAX_VALUES = 10         # values per entry
MAX_BACKFILL_HOURS = 167  # SiteWise rejects timestamps older than 7 days

# measurement -> (SiteWise value type, generator(t_seconds, device_seed))
MEASUREMENTS = {
    'oil_temperature':    ('doubleValue',  lambda t, s: 120 + 8 * math.sin(2 * math.pi * (t + s * 3600) / 86400) + random.gauss(0, 1.0)),
    'hydraulic_pressure': ('doubleValue',  lambda t, s: 2200 + 150 * math.sin(2 * math.pi * t / 900) + random.gauss(0, 40)),
    'motor_current':      ('doubleValue',  lambda t, s: 45 + 5 * math.sin(2 * math.pi * t / 600) + random.gauss(0, 3.0)),
    # Derived from the clock so it only ever goes up, even across backfill and live runs
    'cycle_count':        ('integerValue', lambda t, s: int(t // 6) - 290_000_000 + s * 1000),
}

# device -> measurement -> raw PLC tag name, as the OPC UA server would expose it
RAW_TAGS = {
    'Press01': {'oil_temperature': 'N7_42_TT01', 'hydraulic_pressure': 'Hyd_Press_PSI',
                'motor_current': 'Motor_Amps_RAW', 'cycle_count': 'CycleCnt'},
    'Press02': {'oil_temperature': 'TT_OIL', 'hydraulic_pressure': 'HYD_P',
                'motor_current': 'MTR_I', 'cycle_count': 'CYCLES'},
}


def device_seed(device):
    return zlib.crc32(device.encode()) % 24


def property_value(measurement, t, seed):
    value_type, generate = MEASUREMENTS[measurement]
    return {
        'value': {value_type: generate(t, seed)},
        'timestamp': {'timeInSeconds': int(t), 'offsetInNanos': 0},
        'quality': 'GOOD',
    }


def build_entries(prefix, devices, timestamps):
    """ One entry per (alias, chunk of up to MAX_VALUES timestamps), interleaved across
    aliases so a request never piles onto one stream (SiteWise rate-limits per stream) """
    streams = [(device, measurement, tag) for device in devices for measurement, tag in RAW_TAGS[device].items()]
    entries = []
    for i in range(0, len(timestamps), MAX_VALUES):
        chunk = timestamps[i:i + MAX_VALUES]
        for device, measurement, tag in streams:
            seed = device_seed(device)
            entries.append({
                'entryId': f'{device}-{tag}-{i}'[:64],
                'propertyAlias': f'{prefix}/{device}/{tag}',
                'propertyValues': [property_value(measurement, t, seed) for t in chunk],
            })
    return entries


def send(client, entries, max_attempts=5):
    """ Returns the number of entries that still failed after retrying throttled ones """
    errors = {}
    for i in range(0, len(entries), MAX_ENTRIES):
        batch = entries[i:i + MAX_ENTRIES]
        for attempt in range(max_attempts):
            response = client.batch_put_asset_property_value(entries=batch)
            failed = {e['entryId']: e['errors'][0] for e in response.get('errorEntries', [])}
            retry = [entry for entry in batch if failed.get(entry['entryId'], {}).get('errorCode') == 'ThrottlingException']
            for entry_id, error in failed.items():
                if error['errorCode'] != 'ThrottlingException':
                    errors[entry_id] = error
            if not retry:
                break
            batch = retry
            time.sleep(0.5 * 2 ** attempt)
        else:
            errors.update({entry['entryId']: failed[entry['entryId']] for entry in batch})
    for entry_id, error in list(errors.items())[:5]:
        print(f"  {entry_id}: {error['errorCode']} {error['errorMessage']}")
    return len(errors)


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--prefix', default='/slc-kep01/Channel1',
                        help='alias prefix: gateway measurementDataStreamPrefix + OPC UA browse path')
    parser.add_argument('--devices', default=','.join(RAW_TAGS), help=f'comma-separated subset of {",".join(RAW_TAGS)}')
    parser.add_argument('--backfill-hours', type=float, default=0, help=f'write history first (max {MAX_BACKFILL_HOURS})')
    parser.add_argument('--backfill-step', type=int, default=60, help='seconds between backfilled points')
    parser.add_argument('--live', action='store_true', help='keep publishing current values until Ctrl-C')
    parser.add_argument('--interval', type=int, default=10, help='seconds between live publishes')
    args = parser.parse_args()

    if not args.backfill_hours and not args.live:
        parser.error('pass --backfill-hours and/or --live')
    if args.backfill_hours > MAX_BACKFILL_HOURS:
        parser.error(f'--backfill-hours cannot exceed {MAX_BACKFILL_HOURS} (SiteWise 7-day limit)')

    # boto3 only reads AWS_DEFAULT_REGION on its own; also honor AWS_REGION like the CLI does
    client = boto3.client('iotsitewise', region_name=os.environ.get('AWS_REGION'))
    devices = [d.strip() for d in args.devices.split(',') if d.strip()]
    unknown = set(devices) - set(RAW_TAGS)
    if unknown:
        parser.error(f'unknown devices: {", ".join(sorted(unknown))}')
    for device in devices:
        print(f'{args.prefix}/{device}/{{{",".join(RAW_TAGS[device].values())}}}')

    if args.backfill_hours:
        now = int(time.time())
        start = now - int(args.backfill_hours * 3600)
        timestamps = list(range(start, now, args.backfill_step))
        entries = build_entries(args.prefix, devices, timestamps)
        print(f'Backfilling {len(timestamps)} points per stream ({len(entries)} entries)...')
        failures = send(client, entries)
        print(f'Backfill done, {failures} failed entries')

    if args.live:
        print(f'Publishing every {args.interval}s, Ctrl-C to stop')
        try:
            while True:
                failures = send(client, build_entries(args.prefix, devices, [time.time()]))
                print(f"{time.strftime('%H:%M:%S')} sent{'' if not failures else f', {failures} failed'}", flush=True)
                time.sleep(args.interval)
        except KeyboardInterrupt:
            print('Stopped')


if __name__ == '__main__':
    main()
Enter fullscreen mode Exit fullscreen mode

Load some history so the Step 5 query has data:

python simulate_tags.py --backfill-hours 24
Enter fullscreen mode Exit fullscreen mode

SiteWise rejects values older than 7 days, so the script caps backfill at 167 hours. It also retries when SiteWise throttles writes to a single stream.

Option B: SiteWise Edge with an OPC UA server (production)

This path needs a SiteWise Edge gateway (AWS IoT Greengrass V2 on a Linux x86_64 or ARM64 machine) that can reach your OPC UA server. The SiteWise console walks you through creating the gateway. New gateways are MQTT-enabled (V3), so the examples below are V3.

Two settings in the source configuration matter for tagging:

  • measurementDataStreamPrefix prepends a string to every alias from this source. Use it to encode the site and server, such as /slc-kep01. Without it, two servers that both expose /Channel1/Press01/... collide.
  • nodeFilterRules limit which branches of the address space get ingested. Ingest what you've mapped, not the whole server. Every extra variable is a stream you'll pay to ingest and store.

If your OPC UA server requires a login, store the credentials in Secrets Manager:

aws secretsmanager create-secret \
  --name slc-kep01-opcua \
  --secret-string '{"username":"<opc-user>","password":"<opc-password>"}'
Enter fullscreen mode Exit fullscreen mode

The gateway reads the secret through its Greengrass token exchange role, and the console-created role doesn't include that permission. Add it using the role name and secret ARN from your account:

aws iam put-role-policy \
  --role-name <your-token-exchange-role> \
  --policy-name sitewise-opcua-secret \
  --policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "<your-secret-arn>"
    }]
  }'
Enter fullscreen mode Exit fullscreen mode

Also attach the secret to the gateway in the SiteWise console so the edge components can use it.

The opc-ua-configuration.json below isn't in the companion repo because you only need it with a real SiteWise Edge gateway. The repo's simulate_tags.py writes the same stream aliases straight to the cloud, so you can follow along without hardware.

Save the source as opc-ua-configuration.json:

{
  "sources": [
    {
      "name": "slc-kep01",
      "endpoint": {
        "certificateTrust": { "type": "TrustAny" },
        "endpointUri": "opc.tcp://10.20.0.15:49320",
        "securityPolicy": "BASIC256",
        "messageSecurityMode": "SIGN_AND_ENCRYPT",
        "identityProvider": {
          "type": "Username",
          "usernameSecretArn": "<your-secret-arn>"
        },
        "nodeFilterRules": [
          {
            "action": "INCLUDE",
            "definition": {
              "type": "OpcUaRootPath",
              "rootPath": "/Channel1/"
            }
          }
        ]
      },
      "measurementDataStreamPrefix": "/slc-kep01",
      "destination": { "type": "MQTT" }
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode

Apply it to the gateway:

aws iotsitewise update-gateway-capability-configuration \
  --gateway-id <your-gateway-id> \
  --capability-namespace "iotsitewise:opcuacollector:3" \
  --capability-configuration file://opc-ua-configuration.json
Enter fullscreen mode Exit fullscreen mode

TrustAny is fine for a lab. In production, pin the server's certificate. On an older classic-streams (V2) gateway, use the iotsitewise:opcuacollector:2 namespace and a StreamManager destination instead.

Confirm the streams arrived

Whichever option you chose:

aws iotsitewise list-time-series \
  --time-series-type DISASSOCIATED \
  --alias-prefix /slc-kep01
Enter fullscreen mode Exit fullscreen mode

You should see aliases like /slc-kep01/Channel1/Press01/Hyd_Press_PSI. Those strings are the left column of your CSV. If they don't match exactly, fix the CSV now. Alias matching is exact and case-sensitive.

Step 4: Build assets and bind streams from the mapping file

This is where the CSV earns its keep. One script reads it and, for each row:

  1. Creates the site, area, line, and equipment assets if they don't exist, using the ISA-95 path as the external ID (slc.stamping.line3.press01).
  2. Links each child to its parent through the model hierarchy.
  3. Associates the raw data stream alias with the right measurement property.

The script is idempotent, so it runs safely in CI every time the CSV changes. Adding a machine becomes a pull request: add rows, get a review from someone who knows the floor, merge, and the pipeline does the rest.

# sync_tags.py
import csv
import os
import sys

import boto3
from botocore.exceptions import ClientError

# boto3 only reads AWS_DEFAULT_REGION on its own; also honor AWS_REGION like the CLI does
sw = boto3.client("iotsitewise", region_name=os.environ.get("AWS_REGION"))

LEVELS = [
    # (csv column, asset model external ID, hierarchy name on the parent)
    ("site", "model-site", None),
    ("area", "model-area", "areas"),
    ("line", "model-line", "lines"),
    ("equipment", None, "presses"),  # model comes from equipment_type
]
EQUIPMENT_MODELS = {"press": "model-press"}

_assets = {}  # external ID -> describe_asset response


def ensure_asset(ext_id, model_ext_id):
    if ext_id in _assets:
        return _assets[ext_id]
    try:
        asset = sw.describe_asset(assetId=f"externalId:{ext_id}")
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise
        print(f"creating asset {ext_id}")
        created = sw.create_asset(
            assetName=ext_id,
            assetModelId=f"externalId:{model_ext_id}",
            assetExternalId=ext_id,
        )
        sw.get_waiter("asset_active").wait(assetId=created["assetId"])
        asset = sw.describe_asset(assetId=created["assetId"])
    _assets[ext_id] = asset
    return asset


def ensure_child(parent, hierarchy_name, child):
    hierarchy_id = next(
        h["id"] for h in parent["assetHierarchies"] if h["name"] == hierarchy_name
    )
    existing = sw.list_associated_assets(
        assetId=parent["assetId"], hierarchyId=hierarchy_id
    )["assetSummaries"]
    if not any(a["id"] == child["assetId"] for a in existing):
        print(f"  linking {child['assetName']} under {parent['assetName']}")
        sw.associate_assets(
            assetId=parent["assetId"],
            hierarchyId=hierarchy_id,
            childAssetId=child["assetId"],
        )


def ensure_stream(alias, asset, property_name):
    prop_id = next(
        (p["id"] for p in asset["assetProperties"] if p["name"] == property_name), None
    )
    if prop_id is None:
        sys.exit(f"unknown property {property_name!r} on {asset['assetName']} ({alias})")
    try:
        ts = sw.describe_time_series(alias=alias)
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise
        # The gateway hasn't published this tag yet; a later run will bind it
        print(f"  skipping {alias}: no data stream yet")
        return
    if ts.get("assetId") == asset["assetId"] and ts.get("propertyId") == prop_id:
        return
    print(f"  binding {alias} -> {asset['assetName']}.{property_name}")
    sw.associate_time_series_to_asset_property(
        alias=alias, assetId=asset["assetId"], propertyId=prop_id
    )


def main(path):
    with open(path, newline="") as f:
        for row in csv.DictReader(f):
            parent, path_parts = None, []
            for column, model_ext_id, hierarchy in LEVELS:
                path_parts.append(row[column])
                model = model_ext_id or EQUIPMENT_MODELS[row["equipment_type"]]
                asset = ensure_asset(".".join(path_parts), model)
                if parent:
                    ensure_child(parent, hierarchy, asset)
                parent = asset
            ensure_stream(row["stream_alias"], parent, row["property"])


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "tag-map.csv")
Enter fullscreen mode Exit fullscreen mode

Run it:

python sync_tags.py tag-map.csv
Enter fullscreen mode Exit fullscreen mode

If a row's tag hasn't produced data yet, the script skips it with a warning and binds it on a later run. That's safe because the script is idempotent.

A few notes on the design:

  • External IDs do the lookup work. describe_asset(assetId="externalId:slc.stamping.line3.press01") finds the asset without a search or a local state file. External IDs can't be changed once assigned, which is what you want for an identity.
  • Property and hierarchy IDs come from the asset itself, matched by name. Your CSV only ever speaks in the names you chose in Step 1.
  • describe_time_series makes binding idempotent. If the stream already points at the right property, the script skips it.
  • The raw alias never leaves the CSV. Dashboards, queries, and models reference slc.stamping.line3.press01 and hydraulic_pressure. When the PLC gets re-flashed and the tag becomes HP_01, you change one cell, merge, and nothing downstream notices.

A production version should also handle pagination on list_associated_assets and retry on throttling.

Step 5: Add metrics and query by context

Once the streams are bound, the hierarchy starts paying off. You can compute metrics on each press and roll them up the hierarchy, all defined in the model.

Add a five-minute average to the AssetModelProperties list on PressModel:

        - Name: hydraulic_pressure_avg_5m
          ExternalId: press-hydraulic_pressure_avg_5m
          DataType: DOUBLE
          Unit: psi
          Type:
            TypeName: Metric
            Metric:
              Expression: avg(p)
              Variables:
                - Name: p
                  Value:
                    PropertyExternalId: press-hydraulic_pressure
              Window:
                Tumbling:
                  Interval: 5m
Enter fullscreen mode Exit fullscreen mode

Line-level rollups have a rule that's easy to miss: a parent model's metric can only read metrics from child models, not raw measurements. So the press needs its own five-minute max first. Add this to PressModel too:

        - Name: oil_temperature_max_5m
          ExternalId: press-oil_temperature_max_5m
          DataType: DOUBLE
          Unit: degF
          Type:
            TypeName: Metric
            Metric:
              Expression: max(t)
              Variables:
                - Name: t
                  Value:
                    PropertyExternalId: press-oil_temperature
              Window:
                Tumbling:
                  Interval: 5m
Enter fullscreen mode Exit fullscreen mode

Then add an AssetModelProperties list to LineModel that rolls it up. The line takes the max of each press's 5-minute max, which reports the hottest press on the line:

      AssetModelProperties:
        - Name: max_oil_temperature_5m
          ExternalId: line-max_oil_temperature_5m
          DataType: DOUBLE
          Unit: degF
          Type:
            TypeName: Metric
            Metric:
              Expression: max(t)
              Variables:
                - Name: t
                  Value:
                    HierarchyExternalId: line-presses
                    PropertyExternalId: press-oil_temperature_max_5m
              Window:
                Tumbling:
                  Interval: 5m
Enter fullscreen mode Exit fullscreen mode

Redeploy the stack. Every existing and future press and line picks up the new metrics automatically. You didn't touch a single tag.

Metrics only compute for data that arrives after a stream is bound to its asset property. The history you backfilled in Step 3 is visible to queries but never gets metric values. If you're using the simulator, keep live data flowing for at least five minutes before you check them:

python simulate_tags.py --live
Enter fullscreen mode Exit fullscreen mode

Now query with SQL. The SiteWise ExecuteQuery API joins time series data to the asset model, so you filter by the names from your convention instead of by tag address. Replace the timestamps with a shift window inside your data (with the 24-hour backfill, anything in the last day works):

aws iotsitewise execute-query --query-statement "
SELECT a.asset_name, avg(rts.double_value)
FROM raw_time_series rts, asset a, asset_property ap
WHERE a.asset_name LIKE 'slc.stamping.line3.%'
  AND ap.property_name = 'hydraulic_pressure'
  AND rts.event_timestamp > TIMESTAMP '<shift-start, e.g. 2026-11-05 06:00:00>'
  AND rts.event_timestamp < TIMESTAMP '<shift-end, e.g. 2026-11-05 14:00:00>'
GROUP BY a.asset_name
ORDER BY 2 DESC"
Enter fullscreen mode Exit fullscreen mode

That's "average hydraulic pressure for every press on Line 3 during first shift." It's the question from the top of this article, and it works no matter what the PLC programmers named their tags. The same query runs from Tableau, Power BI, or Excel through the SiteWise ODBC driver.

If the FROM clause with no join conditions looks like a cross join, it isn't. SiteWise joins these views implicitly. An explicit JOIN ... ON returns the same results if you prefer to write it that way.

Gotchas and lessons learned

  • Decide the convention before you create anything. External IDs can't be changed once assigned. Renaming line3 to line_03 later means new assets and re-binding every stream.
  • Alias matching is exact. Case, slashes, and the data stream prefix all count. Property aliases max out at 1,000 bytes, and SiteWise doesn't ingest OPC UA variables with longer paths.
  • Rollups read child metrics, not measurements. A line-level metric that references a press measurement directly fails validation and rolls the stack back. Define the per-press metric first.
  • Metrics don't cover history from before the binding. Data ingested before a stream is bound to its asset property is visible to queries but never gets metric values. Bind streams as early as you can.
  • boto3 reads AWS_DEFAULT_REGION, not AWS_REGION. The AWS CLI accepts both. The companion scripts accept either, but if you write your own boto3 code, set AWS_DEFAULT_REGION or pass region_name explicitly.
  • Filter at the source. Browsing an entire OPC UA server can create tens of thousands of streams you'll never use. Use nodeFilterRules to ingest only mapped branches.
  • Watch for duplicate values. AWS notes that duplicate timestamp-quality-value records can be billed twice. Don't point two sources at the same tags.
  • Don't build alarm logic on AWS IoT Events. It reached end of support on May 20, 2026, and SiteWise's built-in alarms ran on it. AWS's recommended replacement is SiteWise MQTT property notifications feeding CloudWatch alarms, optionally imported back into SiteWise as external alarms.
  • Treat the CSV like code. Require review from someone who knows the equipment. A wrong row puts real data on the wrong machine, and nobody notices until a maintenance decision is made from it.

Clean up

Delete things in the reverse order you created them: unbind and delete the data streams, delete assets from the bottom of the hierarchy up, then delete the models. This script handles the first two, driven by the same CSV:

cleanup_tags.py 
# cleanup_tags.py: tear down everything sync_tags.py created, driven by the same CSV
import csv
import os
import sys

import boto3
from botocore.exceptions import ClientError

# boto3 only reads AWS_DEFAULT_REGION on its own; also honor AWS_REGION like the CLI does
sw = boto3.client("iotsitewise", region_name=os.environ.get("AWS_REGION"))
LEVELS = ["site", "area", "line", "equipment"]


def describe(ext_id):
    try:
        return sw.describe_asset(assetId=f"externalId:{ext_id}")
    except ClientError as e:
        if e.response["Error"]["Code"] == "ResourceNotFoundException":
            return None
        raise


def main(path):
    with open(path, newline="") as f:
        rows = list(csv.DictReader(f))

    # 1. Unbind and delete every data stream in the CSV
    for row in rows:
        alias = row["stream_alias"]
        try:
            ts = sw.describe_time_series(alias=alias)
        except ClientError as e:
            if e.response["Error"]["Code"] == "ResourceNotFoundException":
                continue
            raise
        if ts.get("assetId"):
            print(f"unbinding {alias}")
            sw.disassociate_time_series_from_asset_property(
                alias=alias, assetId=ts["assetId"], propertyId=ts["propertyId"]
            )
        print(f"deleting stream {alias}")
        sw.delete_time_series(alias=alias)

    # 2. Delete assets bottom-up: equipment, then line, area, site
    for depth in range(len(LEVELS), 0, -1):
        ext_ids = sorted({".".join(r[c] for c in LEVELS[:depth]) for r in rows})
        for ext_id in ext_ids:
            asset = describe(ext_id)
            if not asset:
                continue
            parent_ext = ext_id.rsplit(".", 1)[0] if depth > 1 else None
            parent = describe(parent_ext) if parent_ext else None
            if parent:
                for h in parent["assetHierarchies"]:
                    children = sw.list_associated_assets(
                        assetId=parent["assetId"], hierarchyId=h["id"]
                    )["assetSummaries"]
                    if any(c["id"] == asset["assetId"] for c in children):
                        sw.disassociate_assets(
                            assetId=parent["assetId"],
                            hierarchyId=h["id"],
                            childAssetId=asset["assetId"],
                        )
            print(f"deleting asset {ext_id}")
            sw.delete_asset(assetId=asset["assetId"])
            sw.get_waiter("asset_not_exists").wait(assetId=asset["assetId"])


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "tag-map.csv")
Enter fullscreen mode Exit fullscreen mode 

python cleanup_tags.py tag-map.csv
aws cloudformation delete-stack --stack-name sitewise-asset-models
Enter fullscreen mode Exit fullscreen mode

If you used Option B, also delete the gateway in the SiteWise console, the Secrets Manager secret, and the inline policy you added to the token exchange role.

Wrapping up

Tagging manufacturing data isn't about renaming tags on the PLC. It's about building a stable layer of context on top of them. With IoT SiteWise, the pieces fit cleanly: SiteWise Edge collects raw streams, asset models encode your ISA-95 hierarchy, external IDs give every asset a readable identity, and a version-controlled mapping file connects the two.

The result is data that your analysts, dashboards, and ML pipelines can use without ever learning what N7_42_TT01 means.

If you're working on industrial data on AWS, I'd like to hear how you handle tag mapping today. Drop a comment, or follow for more posts on AWS and applied AI.

Sources

Top comments (0)