120 lines
5.0 KiB
Plaintext
120 lines
5.0 KiB
Plaintext
import pprint
|
|
from com.inductiveautomation.ignition.common.tags.config.types import TagObjectType
|
|
defaultProvider= utils.admin.tags.helper.defaultProvider
|
|
defaultSrcProv = utils.admin.tags.helper.defaultSourceProvider
|
|
|
|
def getUDTBaseParameters():
|
|
return {'srcPath': {"dataType":"String", "value":None},
|
|
'tagGroup': {"dataType":"String", "value":"Default"}}
|
|
|
|
def transformBase():
|
|
""" Read the [default]Base UDT and convert it into the Reference Tag equivalent
|
|
New logic to create an additional expression tag bound to the srcPath parameter
|
|
Returns: Nothing, will create the UDT in the global defaultProvider
|
|
"""
|
|
newTagConfig= (transformSingleObject("%s_types_/Base"%(defaultSrcProv)))
|
|
devPathTag={'dataType': "String",
|
|
'expression':'{srcPath}',
|
|
'name': 'DeviceTagPath',
|
|
'tagType': "AtomicTag",
|
|
'valueSource': 'expr'}
|
|
newTagConfig["parameters"] = getUDTBaseParameters()
|
|
newTagConfig["tags"][0]["tags"].append(devPathTag)
|
|
|
|
|
|
pprint.pprint(newTagConfig)
|
|
system.tag.configure("%s/_types_"%(defaultProvider), [newTagConfig], "o")
|
|
|
|
def transformObjects(udtFolder= "Objects"):
|
|
""" Scan for all of the Object UDTs in [default] and create the Reference tag version
|
|
Returns: Nothing
|
|
"""
|
|
metaTagNames = [t["name"] for t in system.tag.browse("[MQTT]_types_/Base/Meta")]
|
|
def normalizeTagConfig(tagConfig):
|
|
# tagConfig.pop("path")
|
|
tagConfig.pop("parameters")
|
|
# Clean up the meta tags inherited from base
|
|
for t in tagConfig["tags"]:
|
|
if t["name"] == "Meta":
|
|
t["tags"] = [item for item in t["tags"] if item['name'] not in metaTagNames]
|
|
|
|
return tagConfig
|
|
allTagConfig = {}
|
|
for udtPath in recBrowse("%s_types_/%s"%(defaultSrcProv, udtFolder)):
|
|
newTagConfig = normalizeTagConfig(transformSingleObject(udtPath))
|
|
if "parameters" not in newTagConfig.keys():
|
|
newTagConfig["parameters"] = getUDTBaseParameters()
|
|
rootFolder = "/".join(udtPath.split("/")[:-1]).replace(defaultSrcProv,defaultProvider)
|
|
allTagConfig.setdefault(rootFolder, [])
|
|
|
|
allTagConfig[rootFolder].append(newTagConfig)
|
|
|
|
for k,v in allTagConfig.iteritems():
|
|
system.tag.configure(k, v, "o")
|
|
|
|
|
|
def recBrowse(path):
|
|
""" Recursively process the folder
|
|
"""
|
|
listOfPaths = []
|
|
for tag in system.tag.browse(path):
|
|
if tag["tagType"] == TagObjectType.Folder:
|
|
if tag["name"] != "Parent": # don't browse the parent folder when we're doing the root Objects folder
|
|
listOfPaths.extend(recBrowse(str(tag["fullPath"])))
|
|
elif tag["tagType"] == TagObjectType.UdtType:
|
|
listOfPaths.append(str(tag["fullPath"]))
|
|
return listOfPaths
|
|
|
|
|
|
def transformSingleObject(udtPath):
|
|
""" Convert a single UDT Object to a reference tag equivalent. Reusable in individual scenarios
|
|
Returns: JSON configuration for the new tag object
|
|
"""
|
|
curTagConfig = system.tag.getConfiguration(udtPath, True)[0]
|
|
# curTagConfig.pop("path") # don't need this since we're creating new
|
|
|
|
curTagConfig["tags"] = recFolderTransform(curTagConfig["tags"], curTagConfig["name"])
|
|
return curTagConfig
|
|
|
|
|
|
def recFolderTransform(tagObj, rootObjName= "", folderPath=""):
|
|
""" Recursively process the folder's contents to convert the existing tags to Reference Tags
|
|
Returns: Newly constructed JSON
|
|
"""
|
|
alarmSummIdx = -1
|
|
for idx, obj in enumerate(tagObj):
|
|
# print obj["name"], obj["tagType"]
|
|
if obj["tagType"]== TagObjectType.Folder:
|
|
# print " %s -- folder found"%(obj["name"])
|
|
if "tags" in obj.keys():
|
|
recFolderTransform(obj["tags"], folderPath="/".join([folderPath, obj["name"]]))
|
|
elif str(obj["tagType"]) not in ["UdtInstance"] and obj["name"] != "AlarmSummary":
|
|
obj["valueSource"] = "reference"
|
|
# obj["sourceTagPath"] = {"bindType":"parameter", "binding":"{srcPath}/%s{TagName}"%(folderPath[:]+"/" if folderPath != "" else "")}
|
|
obj["sourceTagPath"] = {"bindType":"parameter", "binding":"{srcPath}/%s%s"%(folderPath[1:]+"/" if folderPath != "" else "", obj["name"])} # hardcode tagpath, but change name to underscore
|
|
obj["tagGroup"] = {"bindType":"parameter", "binding":"{tagGroup}"}
|
|
obj["name"] = obj["name"].replace(" ", "_")
|
|
# remove keys that are no longer relative to the MQTT transmission
|
|
for k in ["value", "opcServer", "opcItemPath", "sampleMode", "alarms", "sampleMode", "scaleMode", "scaledHigh"]:
|
|
obj.pop(k, None)
|
|
# Remove historian settings since we don't need to log this
|
|
obj["historyEnabled"]= False
|
|
for setting in ["historyMaxAge", "historyProvider", "historyTagGroup"]:
|
|
obj.pop(setting, None)
|
|
elif obj["tagType"] == TagObjectType.UdtInstance:
|
|
print "1"
|
|
# Special handler for the Base Object
|
|
if str(obj["name"]) == "AlarmSummary" and rootObjName=="Base" :
|
|
replacementTag = obj["tags"][0]
|
|
replacementTag["valueSource"] = "reference"
|
|
replacementTag["sourceTagPath"] = {"bindType":"parameter", "binding":"{srcPath}/{TagName}/ActiveAlarms"}
|
|
replacementTag["tagGroup"] = {"bindType":"parameter", "binding":"{tagGroup}"}
|
|
replacementTag.pop("name")
|
|
replacementTag.pop("path")
|
|
replacementTag.pop("value")
|
|
obj.pop("tags")
|
|
obj.pop("typeId")
|
|
|
|
obj.update(replacementTag)
|
|
|
|
return tagObj |