Connect To Your Tags
In this section, we'll connect equipment state and count tags to the assets.
The Kanoa MES module must be installed before you start. It provides the Kanoa/timer and Kanoa/assetInputs UDTs that the simulator tags use. To check, look under UDT Definitions → Kanoa in the [Kanoa] provider and confirm timer and assetInputs are there.
- Download
CoreForge Sim_UDTs_and_Tags.json - In the Designer, open the Tag Browser and select the [Kanoa] tag provider.
- Select
Import Tagsby clicking on the three dots to the right of the Kanoa Tag provider and chooseCoreForge Sim (UDTs + Tags).jsonusing the Direct method. If asked, set Collision Policy to Overwrite, then click Import.
Import from the top-level Tags folder of the [Kanoa] provider.


If you import into a subfolder, the simulator UDTs won't be created as UDT definitions, and every simulator tag will fail with "does not have item 'timer' for overrides". If you see this error, the simulator UDTs weren't created in the right place. Delete any stray types folder under Tags, then repeat the import from the top-level Tags folder with Overwrite. Your existing tags are repaired in place.
- Check the result:
- UDT Definitions → custom contains simState and simCounter.
- Tags → CoreForge Sim → Site contains the Machining, Mixing, Finishing and Coating assets.
- Expand any asset's sim/state. It should show timer, values and simParams with no error icon.
- Expand the
CoreForgetags (not the CoreForgeSim tags) and set the MQTT path parameter on each asset's line tag (assetOEE UDT instance) to the matchingCoreForgeSimtag path:
| Line UDT | MQTT path value |
|---|---|
| Machine 1 | [Kanoa]CoreForge Sim/Site/Machining/Machine 1/assetInputs |
| Spray Line 1 | [Kanoa]CoreForge Sim/Site/Coating/Spray Line 1/assetInputs |
| Manual Pack 1 | [Kanoa]CoreForge Sim/Site/Finishing/Manual Pack 1/assetInputs |
Your setup should look like something below...

For our training, we set the MQTTPath parameter of the line UDT instance and we can do this because our simulator tags are named identical to the tags in the Inputs folder i.e `Infeed Count', 'State'. This allows us to set the path once. In reality, you'll be connecting to different named plc or mqtt tags for the source of counts and states, in which case you can simply change the reference of the Inputs tags for Infeed Count, Outfeed Count, Waste Count and State.
Copy this script into the Ignition Designer Script Console (Tools → Script Console) and run it. It checks that both files were imported into the Kanoa provider, that the simulator tags are running, and that each line's assetOEE UDT has the right MQTT path. Each check prints PASS or FAIL with what to fix, plus a NOTE showing whether the line's inputs are reading the simulator. It doesn't change anything.
Show the check script
# Kanoa Ops Training - check "Connect To Your Tags" (Base Configuration, lesson 7)
#
# Run in the Ignition Designer Script Console (Tools > Script Console), in the
# project that inherits KANOA_SCRIPTS. It only reads; it changes nothing.
# Every check prints PASS or FAIL with what to fix, then a summary.
PROVIDER = '[Kanoa]'
SIM_ROOT = PROVIDER + 'CoreForge Sim'
UDT_TYPES = ['custom/simState', 'custom/simCounter']
LINES = [
# asset path, MQTT path, a simulator tag that only works if the sim UDTs imported
('CoreForge\\Philadelphia\\Machining\\Machine 1',
SIM_ROOT + '/Site/Machining/Machine 1/assetInputs',
SIM_ROOT + '/Site/Machining/Machine 1/sim/state/values/stateCode'),
('CoreForge\\Philadelphia\\Coating\\Spray Line 1',
SIM_ROOT + '/Site/Coating/Spray Line 1/assetInputs',
SIM_ROOT + '/Site/Coating/Spray Line 1/Infeed/sim/infeed/values/countValue'),
('CoreForge\\Philadelphia\\Finishing\\Manual Pack 1',
SIM_ROOT + '/Site/Finishing/Manual Pack 1/assetInputs',
None),
]
results = []
def check(ok, what, fix=None):
results.append(bool(ok))
if ok:
print 'PASS %s' % what
else:
print 'FAIL %s' % what
if fix:
print ' -> %s' % fix
def note(what):
print 'NOTE %s' % what
def text(v):
return '' if v is None else unicode(v).strip()
def normTag(path):
# tag paths are case-insensitive; [~] means the tag's own provider
p = text(path).replace('\\', '/').rstrip('/').lower()
if p.startswith('[~]'):
p = PROVIDER.lower() + p[3:]
return p
def exists(path):
try:
return bool(path) and system.tag.exists(path)
except:
return False
def isGood(qv):
try:
return qv.quality.isGood()
except:
return False
def unwrap(v):
# getConfiguration parameter values can be wrapped in a dict or property object
for _ in range(3):
if v is None:
return None
if hasattr(v, 'get') and not isinstance(v, basestring):
try:
inner = v.get('value')
except:
inner = None
if inner is None:
return v
v = inner
elif hasattr(v, 'getValue'):
v = v.getValue()
else:
return v
return v
def mqttPath(udtPath):
qv = system.tag.readBlocking([udtPath + '/Parameters.MQTTPath'])[0]
if isGood(qv):
return text(qv.value)
try:
params = system.tag.getConfiguration(udtPath, False)[0].get('parameters') or {}
for k in params.keys():
if str(k).lower() == 'mqttpath':
return text(unwrap(params[k]))
except:
pass
return None
print '=' * 70
print 'Kanoa Ops Training - Connect To Your Tags check'
print '=' * 70
# 1. Simulator UDT definitions (simulator UDTs.json)
print '\n-- Simulator UDT definitions'
for t in UDT_TYPES:
check(exists(PROVIDER + '_types_/' + t), 'UDT definition %s exists in the Kanoa provider' % t,
'In the Tag Browser, select the Kanoa provider, click Import Tags, and import simulator UDTs.json.')
# 2. Simulation tag instances (CoreForge Sim Tags.json)
print '\n-- CoreForge Sim tags'
rootOk = exists(SIM_ROOT)
check(rootOk, 'CoreForge Sim folder is at the root of the Kanoa provider',
'Import CoreForge Sim Tags.json into the Kanoa provider with nothing selected, so the folder lands at the root '
'(not inside another folder, and not in the default provider).')
for assetPath, source, simTag in LINES:
name = assetPath.split('\\')[-1]
if rootOk:
check(exists(source + '/Inputs/State'), 'Simulator inputs exist for %s' % name,
'%s is missing. Re-import CoreForge Sim Tags.json.' % source)
if simTag and exists(simTag):
qv = system.tag.readBlocking([simTag])[0]
check(isGood(qv), '%s simulator tags are running' % name,
'%s reads %s. Import simulator UDTs.json (the definitions), then re-import CoreForge Sim Tags.json.'
% (simTag, qv.quality))
# 3. MQTT path on each line's assetOEE UDT
print '\n-- MQTT path on each assetOEE UDT'
for assetPath, source, simTag in LINES:
name = assetPath.split('\\')[-1]
try:
base = text(system.kanoa.asset.getAssetTagPath({'assetPath': assetPath}))
except Exception:
base = ''
udt = None
for c in (base + '/Line', base):
if base and exists(c + '/Inputs/State'):
udt = c
break
check(udt is not None, '%s has an assetOEE UDT' % name,
'No assetOEE tag found for %s. In Asset Configuration, make sure %s exists and is OEE Enabled.' % (assetPath, name))
if udt is None:
continue
actual = mqttPath(udt)
if not actual:
check(False, '%s MQTT path is set' % name,
'In the Tag Browser, edit %s and set the MQTT path parameter to %s' % (udt, source))
continue
if normTag(actual) == normTag(source):
check(True, '%s MQTT path is %s' % (name, source))
elif not actual.startswith('['):
check(False, '%s MQTT path includes the tag provider' % name,
"It's '%s'. Add the provider at the start: %s" % (actual, source))
else:
check(False, '%s MQTT path is %s' % (name, source),
"It's '%s'. Edit %s and set the MQTT path parameter to %s" % (actual, udt, source))
continue
# 4. Data reaches the assetOEE Inputs (informational: values move every tick)
pairs = ['Inputs/State', 'Inputs/Infeed Count', 'Inputs/Outfeed Count']
vals = system.tag.readBlocking([udt + '/' + p for p in pairs] + [source + '/' + p for p in pairs])
same = all(text(vals[i].value) == text(vals[i + len(pairs)].value) for i in range(len(pairs)))
if same:
note('%s assetOEE Inputs match the simulator (State %s, Infeed %s, Outfeed %s)'
% (name, text(vals[0].value) or '-', text(vals[1].value) or '-', text(vals[2].value) or '-'))
else:
note('%s assetOEE Inputs differ from the simulator right now (State %s vs %s). '
'Counts move every minute; re-run in a minute if this persists.'
% (name, text(vals[0].value) or '-', text(vals[len(pairs)].value) or '-'))
passed = results.count(True)
print '\n' + '=' * 70
print 'Result: %d of %d checks passed' % (passed, len(results))
if passed == len(results):
print 'All good. Your tags are connected.'
else:
print 'Fix the FAIL items above, then run this again.'
print '=' * 70
