Skip to main content
View as Markdown

Create Assets with Scripting

It's easy to migrate an existing asset hierarchy from another system. You'll need to export it into a flat file, and then you have choices. Our recommendation is to use AI to convert the data from the flat file into Kanoa's asset export JSON structure, by telling it to use an asset export file (like the ones you downloaded during Base Configuration) as a template. The more data you can export (assets, items, states, modes, links), the more data can be imported. Then all you need to do is import that file from the Configuration → Ops → Assets screen.

Here, though, we'll show you scripting examples that cover many of the system.kanoa functions you need. The first script rolls out more machines in the Machining area by copying Machine 1. For each new machine it:

  • Copies the first asset in the Machines asset group (Machine 1) with system.kanoa.asset.copyAsset, which copies the asset together with its child assets and its mode, state, item, and attribute links.
  • Renames the copy and sets its sort order.
  • Sets the new asset to the Idle mode.
  • Writes the asset's MQTT path parameter so it reads its own CoreForge Sim simulator tags.
Create Assets from script
  1. Copy the following script to the script console in the designer and execute it.

This script will run a little while as each system.kanoa call in the Script Console is a separate round trip from the Designer to the gateway, and some of these calls do a lot of work on the gateway.

View rolloutMachining script
"""
Roll out Machine 2 to Machine 8 by copying Machine 1.

Run in the Ignition Designer Script Console. For each new machine it:
  1. Copies Machine 1 (the template). The copy brings Machine 1's asset group,
     modes, states, item rates and settings with it.
  2. Renames the copy (e.g. 'Machine 2') and sets its sort order from the number,
     in a single save.
  3. Sets the new asset to Idle, so it starts in a known mode.
Then it points each new asset's assetOEE tag at its own CoreForge Sim tags by
writing the MQTT path parameter, and reports any write that didn't take.

Every gateway call is timed, so you can see which steps take longest.

Safe to run again: machines that already exist are skipped, so nothing is
duplicated. If an MQTT path write fails (the tag wasn't ready yet), set that
path by hand, because a rerun skips machines that already exist.
"""
import time

userId = system.kanoa.security.getIDPUserId({'userName': 'SYSTEM'})
TEMPLATE = 'Machine 1'
GROUP = 'Machines'
newAssets = [
	{'assetName': 'Machine %d' % n, 'mqttTagPath': '[Kanoa]CoreForge Sim/Site/Machining/Machine %d/assetInputs' % n}
	for n in range(2, 9)
]
# Fields updateAsset needs. The copy has the same values as the template apart from its name and sort order.
ASSET_FIELDS = ['assetGroupId', 'assetTypeId', 'enabled', 'oeeEnabled', 'parentId', 'processTypeId', 'timeZone']

timings = {}
def timed(label, fn, *args):
	"""Call fn(*args), print how long it took, and add it to the totals."""
	start = time.time()
	result = fn(*args)
	seconds = time.time() - start
	timings.setdefault(label, []).append(seconds)
	print '   %-22s %5.1fs' % (label, seconds)
	return result

runStart = time.time()
print '=' * 70
print 'Rolling out %d machines from %s' % (len(newAssets), TEMPLATE)
print '=' * 70

# 1. Find the template and the machines that already exist
print '\nLooking up assets in the %s group...' % GROUP
assets = system.kanoa.utilities.convertDatasetToJSON(timed('getAssets', system.kanoa.asset.getAssets, {'assetGroupName': GROUP}))
existing = set(a['assetName'] for a in assets)
print ' Found %d: %s' % (len(assets), ', '.join(sorted(existing)) or 'none')

template = [a for a in assets if a['assetName'] == TEMPLATE]
if not template:
	raise Exception('%s is not in the %s asset group. Add it to the group, then run this again.' % (TEMPLATE, GROUP))
copyMe = template[0]
print ' Template: %s (assetId %s)' % (copyMe['assetName'], copyMe['assetId'])

missingFields = [f for f in ASSET_FIELDS if f not in copyMe]
if missingFields:
	print ' Asset data has no %s, so name and sort order will be saved separately' % ', '.join(missingFields)

modeSourceId = timed('getModeSourceId', system.kanoa.asset.getModeSourceId, 'PLC')
print ' Mode changes will be recorded with mode source PLC (id %s)' % modeSourceId

# 2. Copy, rename and set each new machine to Idle
created, skipped, failed = [], [], []
tagPaths, mqttTagPaths = [], []
for asset in newAssets:
	name = asset['assetName']
	machineStart = time.time()
	print '\n-- %s' % name
	if name in existing:
		print ' Already exists. Skipping'
		skipped.append(name)
		continue

	print ' Copying %s...' % TEMPLATE
	assetId, msg = timed('copyAsset', system.kanoa.asset.copyAsset, copyMe['assetId'], userId)
	if msg:
		print ' Could not copy: %s' % msg
		failed.append(name)
		continue
	print ' Copied as assetId %s' % assetId

	sortOrder = int(name.split()[-1])
	if missingFields:
		timed('updateAssetField name', system.kanoa.asset.updateAssetField, assetId, 'assetName', name, userId)
		timed('updateAssetField sort', system.kanoa.asset.updateAssetField, assetId, 'sortOrder', sortOrder, userId)
	else:
		assetInfo = dict((f, copyMe[f]) for f in ASSET_FIELDS)
		assetInfo.update({'assetId': assetId, 'assetName': name, 'sortOrder': sortOrder})
		count, msg = timed('updateAsset', system.kanoa.asset.updateAsset, assetInfo, userId)
		if msg:
			print ' Could not rename: %s' % msg
	print ' Renamed to %s, sort order %s' % (name, sortOrder)
	created.append(name)

	tagPath = timed('getAssetTagPath', system.kanoa.asset.getAssetTagPath, {'assetId': assetId}) + '/Line/Parameters.MQTTPath'
	tagPaths.append(tagPath)
	mqttTagPaths.append(asset['mqttTagPath'])
	print ' MQTT path to set later: %s' % asset['mqttTagPath']

	modes = timed('getAssetModes', system.kanoa.asset.getAssetModes, {'assetId': assetId, 'modeName': 'Idle'})
	modeCode = system.kanoa.utilities.getFieldValue('modeCode', modes)
	if modeCode is None:
		print ' Idle mode is not linked to %s. Mode not set' % name
	else:
		assetOperationInfo = {
			'assetId': assetId, 'modeCode': modeCode, 'workOrderId': None, 'itemId': None, 'standardRate': None,
			'modeSourceId': modeSourceId, 'tStamp': system.date.now(), 'includeChildren': True, 'setChildModeOnly': True
		}
		modeEventId, msg = timed('assetOperationChange', system.kanoa.event.assetOperationChange, assetOperationInfo, userId)
		if msg:
			print ' Could not set Idle: %s' % msg
		else:
			print ' Set to Idle (mode code %s)' % modeCode
	print ' %s took %.1fs' % (name, time.time() - machineStart)

# 3. Point each new asset at its own simulator tags
print '\n-- MQTT paths'
badWrites = 0
if tagPaths:
	print ' Writing %d MQTT path(s)...' % len(tagPaths)
	qualities = timed('writeBlocking', system.tag.writeBlocking, tagPaths, mqttTagPaths)
	for path, value, quality in zip(tagPaths, mqttTagPaths, qualities):
		if quality.isGood():
			print ' OK    %s' % path
		else:
			badWrites += 1
			print ' FAIL  %s (%s). Set it by hand to %s' % (path, quality, value)
else:
	print ' Nothing to write; no new assets were created'

# 4. Summary
print '\n' + '=' * 70
print 'Created: %s' % (', '.join(created) or 'none')
print 'Skipped (already existed): %s' % (', '.join(skipped) or 'none')
print 'Failed to copy: %s' % (', '.join(failed) or 'none')
if badWrites:
	print '%d MQTT path(s) were not written. See the FAIL lines above.' % badWrites

print '\nTime by step (slowest first):'
for label, times in sorted(timings.items(), key=lambda kv: -sum(kv[1])):
	print ' %-22s %6.1fs total  %5.1fs average  x%d' % (label, sum(times), sum(times) / len(times), len(times))
print ' %-22s %6.1fs' % ('Whole run', time.time() - runStart)
print 'Done!'
print '=' * 70

Check your work

Copy this script into the Ignition Designer Script Console (Tools → Script Console) and run it after the rollout. For Machine 2 to Machine 8 it checks that each exists once under Machining, is in the Machines group, is OEE Enabled, has the right sort order, has the same mode, state and item rate links as Machine 1, and reads its own CoreForge Sim tags through its MQTT path. It also flags unrenamed copies left by an interrupted run. Each check prints PASS or FAIL with what to fix. It doesn't change anything.

Show the check script
checkLesson-scriptAssets.py
# Kanoa Ops Training - check "Create Assets with Scripting" (Scaling across the enterprise, lesson 1)
#
# 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.

AREA = 'CoreForge\\Philadelphia\\Machining'
GROUP = 'Machines'
TEMPLATE = 'Machine 1'
MACHINES = range(2, 9)	# Machine 2 to Machine 8
MQTT = '[Kanoa]CoreForge Sim/Site/Machining/Machine %d/assetInputs'

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 field(row, *names):
	for n in names:
		if n in row and row[n] is not None:
			return row[n]
	return None

def text(v):
	return '' if v is None else unicode(v).strip()

def norm(path):
	return text(path).replace('/', '\\').strip('\\')

def truthy(v):
	return str(v).lower() in ('true', '1', 'yes')

def rows(data):
	# Kanoa returns a dataset, a list of dictionaries, or None
	if data is None:
		return []
	if hasattr(data, 'getRowCount'):
		cols = list(data.getColumnNames())
		return [dict((c, data.getValueAt(i, c)) for c in cols) for i in range(data.getRowCount())]
	return [dict(r) for r in data]

def mqttPath(assetId):
	"""The MQTT path parameter on the asset's assetOEE tag, or None if it can't be read."""
	try:
		udt = system.kanoa.asset.getAssetTagPath({'assetId': assetId}) + '/Line'
	except:
		return None
	qv = system.tag.readBlocking([udt + '/Parameters.MQTTPath'])[0]
	if qv.quality.isGood():
		return text(qv.value)
	try:
		params = system.tag.getConfiguration(udt, False)[0].get('parameters') or {}
		for k in params.keys():
			if str(k).lower() == 'mqttpath':
				v = params[k]
				return text(v.get('value') if hasattr(v, 'get') else v)
	except:
		pass
	return None

def links(assetId):
	"""How many modes, states and item rates are linked to the asset."""
	counts = {}
	for label, fn in [('modes', system.kanoa.asset.getAssetModes), ('states', system.kanoa.asset.getAssetStates),
			('item rates', system.kanoa.item.getAssetItems)]:
		try:
			counts[label] = len(rows(fn({'assetId': assetId})))
		except:
			counts[label] = None
	return counts

print '=' * 70
print 'Kanoa Ops Training - Create Assets with Scripting check'
print '=' * 70

assets = rows(system.kanoa.asset.getAssets({}))
inArea = [a for a in assets if norm(field(a, 'assetPath')).startswith(AREA + '\\')]
byName = {}
for a in inArea:
	byName.setdefault(text(field(a, 'assetName')), []).append(a)

# 1. The template
print '\n-- Template'
template = byName.get(TEMPLATE, [])
check(template, '%s exists in Machining' % TEMPLATE, 'Machine 1 is the template the script copies. Create it first (Base Configuration).')
templateLinks = links(field(template[0], 'assetId')) if template else {}
if template:
	note('%s has %s' % (TEMPLATE, ', '.join('%s %s' % (v, k) for k, v in sorted(templateLinks.items()) if v is not None)))

# 2. Each new machine
for n in MACHINES:
	name = 'Machine %d' % n
	print '\n-- %s' % name
	found = byName.get(name, [])
	if not found:
		check(False, '%s exists' % name, 'Run the rollout script. It skips machines that already exist, so it only adds the missing ones.')
		continue
	check(len(found) == 1, '%s exists once' % name,
		'There are %d assets named %s. Delete the extras in Configuration > Ops > Assets.' % (len(found), name))
	a = found[0]
	assetId = field(a, 'assetId')
	group = text(field(a, 'assetGroupName'))
	check(group == GROUP, '%s is in the %s asset group' % (name, GROUP), "It's in '%s'. Edit the asset and set its group to %s." % (group or 'no group', GROUP))
	if field(a, 'oeeEnabled', 'oeeEnable') is not None:
		check(truthy(field(a, 'oeeEnabled', 'oeeEnable')), '%s is OEE Enabled' % name, 'Edit the asset and tick OEE Enabled.')
	sortOrder = field(a, 'sortOrder')
	check(text(sortOrder) == str(n), '%s has sort order %d' % (name, n), "It's %s. Edit the asset and set Sort Order to %d." % (text(sortOrder) or 'blank', n))

	got = links(assetId)
	short = ['%s (%s of %s)' % (k, got[k], templateLinks[k]) for k in sorted(got)
		if got[k] is not None and templateLinks.get(k) is not None and got[k] < templateLinks[k]]
	if templateLinks:
		check(not short, '%s has the same links as %s' % (name, TEMPLATE),
			'Missing links: %s. Copying Machine 1 should bring these; delete %s and run the script again.' % (', '.join(short), name))

	expected = MQTT % n
	actual = mqttPath(assetId)
	if actual is None:
		check(False, '%s MQTT path is readable' % name, "Couldn't read its assetOEE tag. Check that the asset is OEE Enabled.")
	else:
		check(actual.lower().rstrip('/') == expected.lower(), '%s reads its own simulator tags' % name,
			"Its MQTT path is '%s'. Edit its assetOEE tag (under Machining/%s) and set the MQTT path parameter to %s"
			% (actual or 'blank', name, expected))

# 3. Leftovers from a failed or repeated run
print '\n-- Leftovers'
extra = sorted(set(name for name in byName if 'copy' in name.lower()))
check(not extra, 'No unrenamed copies in Machining',
	'Found %s. A copy was made but not renamed; delete it in Configuration > Ops > Assets.' % ', '.join(extra))
expectedNames = set([TEMPLATE] + ['Machine %d' % n for n in MACHINES])
others = sorted(set(text(field(a, 'assetName')) for a in inArea if text(field(a, 'assetGroupName')) == GROUP) - expectedNames - set(extra))
if others:
	note('Also in the %s group: %s. If any of these came from the script, delete them.' % (GROUP, ', '.join(others)))

passed = results.count(True)
print '\n' + '=' * 70
print 'Result: %d of %d checks passed' % (passed, len(results))
if passed == len(results):
	print 'All good. Machine 2 to Machine 8 are rolled out.'
else:
	print 'Fix the FAIL items above, then run this again.'
print '=' * 70
Last updated on