2026-04-24 FE Backup
This commit is contained in:
@@ -0,0 +1,128 @@
|
||||
import pprint, time
|
||||
from com.inductiveautomation.ignition.common.tags.config.types import TagObjectType
|
||||
defaultTgtProvider= utils.admin.tags.helper.defaultProvider
|
||||
defaultSrcProv = utils.admin.tags.helper.defaultSourceProvider
|
||||
thisTenant = "Valley"
|
||||
|
||||
Logger = system.util.getLogger("utils.admin.tags.utils.instances")
|
||||
|
||||
|
||||
def transformObjects(startPath):
|
||||
logger = Logger.createSubLogger("transformObjects")
|
||||
newTags = []
|
||||
for udtInst in recBrowse(startPath):
|
||||
newTags.append(createSingleInstance(udtInst))
|
||||
|
||||
system.tag.configure(defaultTgtProvider, newTags, "o")
|
||||
|
||||
def recBrowse(path):
|
||||
""" Recursively process the folder
|
||||
"""
|
||||
logger = Logger.createSubLogger("recBrowse")
|
||||
_tStartTime = time.time()
|
||||
|
||||
|
||||
listOfPaths = []
|
||||
for tag in system.tag.browse(path):
|
||||
if tag["tagType"] == TagObjectType.Folder:
|
||||
if tag["name"] not in ["Parent", "AlarmSummary"]: # don't browse the parent folder when we're doing the root Objects folder
|
||||
listOfPaths.extend(recBrowse(str(tag["fullPath"])))
|
||||
elif tag["tagType"] == TagObjectType.UdtInstance:
|
||||
listOfPaths.append(str(tag["fullPath"]))
|
||||
logger.trace("recursive browse completed in : %s"%(time.time()-_tStartTime))
|
||||
|
||||
return listOfPaths
|
||||
|
||||
|
||||
def createSingleInstance(udtPath):
|
||||
""" create a reference tag equiavalent of a UDT found
|
||||
udtPath: str the udt instance path
|
||||
|
||||
"""
|
||||
logger = Logger.createSubLogger("createSingleInstance")
|
||||
|
||||
tagConfig = system.tag.getConfiguration(udtPath, False)[0]
|
||||
|
||||
path = tagConfig["path"]
|
||||
udtType = tagConfig["typeId"]
|
||||
refPath = str(tagConfig["path"])
|
||||
pathSegments = refPath.split("/")
|
||||
prov = refPath[refPath.find("[")+1 : refPath.find("]")]
|
||||
tagName = tagConfig["name"]
|
||||
if len(tagConfig["name"].split("_")) < 3:
|
||||
# tagName = "%s_%s_%s"%(prov.replace("_","-"), pathSegments[-2],tagName)
|
||||
tagName = "%s_%s"%(pathSegments[-2],tagName)
|
||||
else:
|
||||
tagName = "%s"%(tagName)
|
||||
|
||||
newTag= {"name":tagName,
|
||||
"typeId":tagConfig["typeId"], "tagType":"UdtInstance",
|
||||
"parameters":{"srcPath":{"dataType":"String", "value":refPath}}
|
||||
}
|
||||
|
||||
return newTag
|
||||
|
||||
|
||||
def generateTagExport():
|
||||
""" Generate a dataset representing all of the discovered tags in the MQTT provider
|
||||
|
||||
"""
|
||||
def transformRow(path):
|
||||
metaProps = getMetaProp(path, ["EqName"])
|
||||
def buildRow(udtDefPath, path, tagdict, srcPath=""):
|
||||
instNameArr = path.split("/")[0].split("_")
|
||||
hasProvInPath= True if "Ignition_" in path else False
|
||||
if hasProvInPath:
|
||||
loc = instNameArr[5]
|
||||
devName = "_".join(instNameArr[4:])
|
||||
else:
|
||||
loc = instNameArr[1]
|
||||
devName = "_".join(instNameArr[2:])
|
||||
|
||||
# return [udtDefPath[-2], udtDefPath[-1], "%s/%s"%(path, tagdict["name"]), tagdict.get("engUnit",""), devName, loc, tagdict["name"].replace("_"," "),
|
||||
# tagdict["name"].replace(" ","_"), path]
|
||||
localPath = "%s/%s"%(path, tagdict["name"])
|
||||
remotePath, tagTenant = getCustomTenantProp(localPath)
|
||||
return [udtDefPath[-2], udtDefPath[-1], remotePath, tagdict.get("engUnit",""),
|
||||
devName if metaProps.get("EqName","") == "" else metaProps["EqName"], # DeviceName
|
||||
loc,
|
||||
tagTenant if tagTenant is not None else tagdict["name"].replace("_"," "), # Description
|
||||
tagdict["name"].replace(" ","_"),
|
||||
remotePath] # Notes
|
||||
|
||||
rows = []
|
||||
config = system.tag.getConfiguration(path, True)[0]
|
||||
udtDefPath = str(config.get("typeId","/")).split("/")
|
||||
srcPath = config["parameters"].get("srcPath").value
|
||||
for tag in config["tags"]:
|
||||
if tag["tagType"] == TagObjectType.Folder and tag["name"] not in ["Meta"]:
|
||||
for fTag in tag["tags"]:
|
||||
# rows.append(buildRow(udtDefPath, "%s/%s"%(path, tag["name"]), fTag))
|
||||
rows.append(buildRow(udtDefPath, "%s/%s"%(path, tag["name"]), fTag, "%s/%s"%(srcPath, tag["name"])))
|
||||
else:
|
||||
if tag["name"] not in ["Meta"]:
|
||||
rows.append(buildRow(udtDefPath, path, tag, srcPath))
|
||||
|
||||
return rows
|
||||
_udtInstList= utils.admin.tags.utils.instances.recBrowse(defaultTgtProvider)
|
||||
header = ["componentType", "topic", "tag", "units", "deviceName", "hallId", "description", "dataPointType", "Notes"]
|
||||
allRows = []
|
||||
for udtInstPath in _udtInstList:
|
||||
allRows.extend(transformRow(udtInstPath))
|
||||
|
||||
fullDS = system.dataset.toDataSet(header, allRows)
|
||||
return fullDS
|
||||
# return system.dataset.toExcel(True, [fullDS], ["All Points"])
|
||||
|
||||
def getMetaProp(instPath, propNames):
|
||||
""" Get the meta properties of a tag
|
||||
returns: dictionary of the propertyname and value
|
||||
"""
|
||||
pathsToRead = ["%s/Meta/%s"%(instPath,p) for p in propNames]
|
||||
qvs = system.tag.readBlocking(pathsToRead)
|
||||
return {p:qv.value for p,qv in zip(propNames, qvs)}
|
||||
|
||||
def getCustomTenantProp(localTagPath):
|
||||
print localTagPath
|
||||
tagpath = system.tag.readBlocking("%s.%s"%(localTagPath,"SourceTagPath"))[0].value
|
||||
return tagpath, system.tag.readBlocking("%s.%s"%(tagpath,thisTenant))[0].value
|
||||
Reference in New Issue
Block a user