Began Framework start

This commit is contained in:
2026-03-17 09:10:25 -05:00
parent c1603c24b0
commit 2db1253693
1277 changed files with 66669 additions and 1 deletions

View File

@@ -0,0 +1,16 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"data.bin"
],
"attributes": {
"lastModificationSignature": "cf555718617e348f08a85720ea50b371b3622253d0f2020616bae5ad0a40679a",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,239 @@
logger = system.util.getLogger("Import Tools")
def excelToDataSet(fileName, hasHeaders = False, forceString = False, sheetNum = 0, firstRow = None, lastRow = None, firstCol = None, lastCol = None, customHeaders = None):
import org.apache.poi.ss.usermodel.WorkbookFactory as WorkbookFactory
import org.apache.poi.ss.usermodel.DateUtil as DateUtil
from java.io import FileInputStream
from java.util import Date
from os.path import exists
"""
Description:
Function to create a dataset from an Excel spreadsheet. This is typically used in Vision and the file path is
directly referenced to its location in the filesystem.
Arguments:
fileName: The path to the Excel spreadsheet. (required)
hasHeaders: If true, uses the first row of the spreadsheet as column names.
forceString: If true, forces all cell values to be strings. This can be useful if cell values in columns are not consistent.
To create a dataset, the data type is determined by the first row. If the data types are different, you can get errors.
If you force all cells to strings, then this is not an issue.
sheetNum: Select the sheet to process. Defaults to the first sheet.
firstRow: Select first row to process.
lastRow: Select last row to process.
firstCol: Select first column to process
lastCol: Select last column toprocess
History:
No. Date Author Comment
1.0 2021-01-22 Jordan Clark Initial - https://forum.inductiveautomation.com/t/copying-a-excel-file-to-a-tables-dataset/34942/13
1.1 2023-02-17 James Landwerlen Updated to handle blank cells
and to force strings
"""
if exists(fileName):
fileStream = FileInputStream(fileName)
try:
wb = WorkbookFactory.create(fileStream)
sheet = wb.getSheetAt(sheetNum)
if firstRow is None:
firstRow = sheet.getFirstRowNum()
if lastRow is None:
lastRow = sheet.getLastRowNum()
data = []
for i in range(firstRow , lastRow + 1):
row = sheet.getRow(i)
if i == firstRow:
if firstCol is None:
firstCol = row.getFirstCellNum()
if lastCol is None:
lastCol = row.getLastCellNum()
else:
# if lastCol is specified add 1 to it.
lastCol += 1
if hasHeaders and customHeaders is None:
headers = list(row)[firstCol:lastCol]
# print headers
elif hasHeaders and customHeaders is not None:
headers = customHeaders
else:
headers = ['Col'+str(i) for i in range(firstCol, lastCol)]
# print headers
rowOut = []
for j in range(firstCol, lastCol):
if i == firstRow and hasHeaders:
pass
else:
cell = row.getCell(j)
if cell is not None:
cellType = cell.getCellType().toString()
if cellType == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif cellType == 'STRING':
value = cell.getStringCellValue()
elif cellType == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif cellType == 'BLANK':
value = None
elif cellType == 'FORMULA':
formulatype=str(cell.getCachedFormulaResultType())
if formulatype == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif formulatype == 'STRING':
value = cell.getStringCellValue()
elif formulatype == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif formulatype == 'BLANK':
value = None
else:
value = None
else:
value = None
if forceString:
if value == None:
rowOut.append(value)
else:
rowOut.append(str(value))
else:
rowOut.append(value)
if len(rowOut) > 0:
data.append(rowOut)
fileStream.close()
return system.dataset.toDataSet(headers, data)
except Exception as e:
logString = "Failed - %s" %(e)
logger.warn(logString)
system.perspective.closePopup("progress")
params = {"message" : logString}
title = "Import Error"
view = "Popups/Error/Main"
system.perspective.openPopup("errorPopup", view, params, title, modal = True)
def excelBytesToDataSet(bytesIn, hasHeaders = True, forceString = True, sheetNum = 0, firstRow = None, lastRow = None, firstCol = None, lastCol = None):
import org.apache.poi.ss.usermodel.WorkbookFactory as WorkbookFactory
import org.apache.poi.ss.usermodel.DateUtil as DateUtil
from java.io import ByteArrayInputStream
fileStream = ByteArrayInputStream(bytesIn)
try:
wb = WorkbookFactory.create(fileStream)
sheet = wb.getSheetAt(sheetNum)
if firstRow is None:
firstRow = sheet.getFirstRowNum()
if lastRow is None:
lastRow = sheet.getLastRowNum()
data = []
for i in range(firstRow , lastRow + 1):
row = sheet.getRow(i)
if i == firstRow:
if firstCol is None:
firstCol = row.getFirstCellNum()
if lastCol is None:
lastCol = row.getLastCellNum()
else:
# if lastCol is specified add 1 to it.
lastCol += 1
if hasHeaders:
headers = list(row)[firstCol:lastCol]
print headers
else:
headers = ['Col'+str(i) for i in range(firstCol, lastCol)]
print headers
rowOut = []
for j in range(firstCol, lastCol):
if i == firstRow and hasHeaders:
pass
else:
cell = row.getCell(j)
if cell is not None:
cellType = cell.getCellType().toString()
if cellType == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif cellType == 'STRING':
value = cell.getStringCellValue()
elif cellType == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif cellType == 'BLANK':
value = None
elif cellType == 'FORMULA':
formulatype=str(cell.getCachedFormulaResultType())
if formulatype == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif formulatype == 'STRING':
value = cell.getStringCellValue()
elif formulatype == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif formulatype == 'BLANK':
value = None
else:
value = None
else:
value = None
if forceString:
if value == None:
rowOut.append(value)
else:
rowOut.append(str(value))
else:
rowOut.append(value)
if len(rowOut) > 0:
data.append(rowOut)
fileStream.close()
return system.dataset.toDataSet(headers, data)
except Exception as e:
logString = "Failed - %s" %(e)
logger.warn(logString)
system.perspective.closePopup("progress")
params = {"message" : logString}
title = "Import Error"
view = "Popups/Error/Main"
system.perspective.openPopup("errorPopup", view, params, title, modal = True)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "dd6f9a8f42c769dd5d3a420e7f4a11d82040060fab927e6b4e4f001efa19e044",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-12T15:26:16Z"
}
}
}

View File

@@ -0,0 +1,103 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/Micro/P_AInAdv"
def importTag(pointData):
tagProvider = pct.tools.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Inp_PV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# CV Ranges
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMin"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMin"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_InpRawMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_InpRawMin"])
system.tag.writeBlocking(writePaths, writeVals)
except:
print("No %s Present" %(UDTType))
def runSimulation():
typePathList = basic.listUDTInstances(UDTType)
print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
try:
for i in range(0, len(tagRead)):
print tagRead[i].value["_enable"]
if tagRead[i].value["_enable"]:
input = tagRead[i].value["Inp_PV"]["Data"]
noise = tagRead[i].value["Noise"]
noiseBase = tagRead[i].value["NoiseBase"]
min_val = -noise
max_val = noise
noiseAdd = random.uniform(min_val, max_val)
value = noiseBase + noiseAdd
print typePathList[i] + "/Inp_PV/Data"
tagPaths.append(typePathList[i] + "/Inp_PV/Data")
tagVals.append(value)
system.tag.writeBlocking(tagPaths, tagVals)
except:
print("No Analog Inputs Enabled")

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "d7e5515c9c4091a93b815babbb2bf002f4920796fa9dc69a28035930edbfd330",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,70 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/Micro/P_AOut"
def importTag(pointData):
tagProvider = pct.tools.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Out_CV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# CV Ranges
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMin"])
except:
print("No %s Present" %(UDTType))

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "b75b72a969741e16592bb2d4246109061603304cc0cb5cbac58de72f761458e3",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:08Z"
}
}
}

View File

@@ -0,0 +1,35 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/Micro/P_DIn"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Inp_PV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "ee949c46f2cf590ebfd6811f475cf6c4a57ea14d1329ecd52b65b775e692798b",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,35 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/Micro/P_DOut"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Out" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "2dfd99ae889e6e450850f55c84abdf76df6dcd997e6949b8ab842992796d5224",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,82 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/Micro/P_Motor"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"]
tagName = pointData["deviceTag"]
typeId = UDTType
tagType = "UdtInstance"
extension = pointData["deviceExt"].replace(".","")
params = {}
params[extension] = pointData["plcAddress"]
params["PLC"] = pointData["procName"]
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def runSimulation():
typePathList = basic.listUDTInstances(UDTType)
print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
now = system.date.now()
try:
for i in range(0, len(tagRead)):
if tagRead[i].value["_enable"]:
outRun = tagRead[i].value["Sts_Running"]
outStart = tagRead[i].value["Sts_Starting"]
outStop = tagRead[i].value["Sts_Stopping"] or tagRead[i].value["Sts_Stopped"]
currentTime = system.date.toMillis(system.date.now())
timeDiff = currentTime - tagRead[i].value["StartCommand"]
feedbackTime = tagRead[i].value["Config"]["Cfg_SimFdbkT"] * 1000
if (timeDiff > feedbackTime) and (outStart or outRun):
runFdbk = True
else:
runFdbk = False
tagPaths.append(typePathList[i] + "/Inp_RunFdbk/Data")
tagVals.append(runFdbk)
tagPaths.append(typePathList[i] + "/Out_Start/Data")
tagVals.append(outStart)
tagPaths.append(typePathList[i] + "/Out_Stop/Data")
tagVals.append(outStop)
tagPaths.append(typePathList[i] + "/Out_Run/Data")
tagVals.append(outRun)
if len(tagPaths) > 0:
system.tag.writeBlocking(tagPaths, tagVals)
except:
print("No Motors Enabled")

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "18c6155860b5957222dedbf2e96e961a4427a77e3c3697430e90b739b4434ce5",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,91 @@
def runSimulation():
SIM_TAGS_FOLDER = "Simulation/Vessels"
# Get all vessel instances
vessels = system.tag.browseTags(parentPath=SIM_TAGS_FOLDER, udtParentType="Library/Objects/Vessel")
for vessel in vessels:
path = vessel.fullPath
print path
# Read all needed values in one call
paths = [
path + "/Parameters/UseInletValve",
path + "/Parameters/UseOutletValve",
path + "/Parameters/TotalVolume",
path + "/Parameters/HH_SP",
path + "/Parameters/H_SP",
path + "/Parameters/L_SP",
path + "/Parameters/LL_SP",
path + "/Parameters/ScanRate",
path + "/Links/InletOpen/LinkData",
path + "/Links/OutletOpen/LinkData",
path + "/Volume",
path + "/Level",
path + "/InletFlow",
path + "/OutletFlow"
]
values = system.tag.readBlocking(paths)
print values
use_inlet_valve = values[0].value
use_outlet_valve = values[1].value
capacity = values[2].value
hh = values[3].value
h = values[4].value
l = values[5].value
ll = values[6].value
scan_rate = values[7].value
inlet_open = values[8].value
outlet_open = values[9].value
current_vol = values[10].value
level = values[11].value
inlet_flow = values[12].value
outlet_flow = values[13].value
if inlet_flow is None:
inlet_flow = 0.0
if outlet_flow is None:
outlet_flow = 0.0
print inlet_flow
print outlet_flow
if not use_inlet_valve:
inlet_open = True
if not use_outlet_valve:
outlet_open = True
# Time step in seconds (match your script rate)
dt = scan_rate / 1000.0
# Calculate volume change (flow in GPM, dt in seconds)
inflow = (inlet_flow / 60.0) * dt if inlet_open else 0.0
outflow = (outlet_flow / 60.0) * dt if outlet_open else 0.0
print inflow
print outflow
new_vol = current_vol + inflow - outflow
new_vol = max(0.0, min(new_vol, capacity)) # clamp
level_pct = (new_vol / capacity) * 100.0 if capacity > 0 else 0.0
# Write results
write_paths = [
path + "/Volume",
path + "/Level",
path + "/HH",
path + "/H",
path + "/L",
path + "/LL"
]
write_values = [
new_vol,
level_pct,
level_pct >= hh,
level_pct >= h,
level_pct <= l,
level_pct <= ll
]
system.tag.writeBlocking(write_paths, write_values)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "87f06ad75a64fae119242d7c2a8895041505a29c862897d5527ff47ec3394d0b",
"lastModification": {
"actor": "admin",
"timestamp": "2026-02-19T19:39:40Z"
}
}
}

View File

@@ -0,0 +1,117 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_AInAdv"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Inp_PV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# CV Ranges
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMin"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_PVEUMin"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_InpRawMax"])
writePaths.append(typePathList[i] + "/Inp_PV/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_InpRawMin"])
system.tag.writeBlocking(writePaths, writeVals)
except:
print("No %s Present" %(UDTType))
def runSimulation():
try:
typePathList = basic.listUDTInstances(UDTType)
# print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
for i in range(0, len(tagRead)):
print tagRead[i].value["_enable"]
if tagRead[i].value["_enable"]:
if tagRead[i].value["_enableL2"]:
noise = tagRead[i].value["Noise"]
min_val = -noise
max_val = noise
noiseAdd = random.uniform(min_val, max_val)
value = tagRead[i].value["L2"]["LinkData"] + noiseAdd
tagPaths.append(typePathList[i] + "/Inp_PV/Data")
tagVals.append(value)
else:
input = tagRead[i].value["Inp_PV"]["Data"]
noise = tagRead[i].value["Noise"]
noiseBase = tagRead[i].value["NoiseBase"]
min_val = -noise
max_val = noise
noiseAdd = random.uniform(min_val, max_val)
value = noiseBase + noiseAdd
print typePathList[i] + "/Inp_PV/Data"
tagPaths.append(typePathList[i] + "/Inp_PV/Data")
tagVals.append(value)
system.tag.writeBlocking(tagPaths, tagVals)
except:

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "9c56d798d0cc604698f2560d240cc6d4497059cdea25ff95fa8f2b3b57dfa561",
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:10:21Z"
}
}
}

View File

@@ -0,0 +1,70 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_AOut"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Out_CV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# CV Ranges
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMin"])
except:
print("No %s Present" %(UDTType))

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "e6646190933fbba3807b1bbe264bc27ff74221c9e334609b9fd907129dcc1765",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:08Z"
}
}
}

View File

@@ -0,0 +1,35 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_DIn"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Inp_PV" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "ea22a45180ec9df061a5d88d77673a8083e8d3580f157af19830b8564f9e67de",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,35 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_DOut"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"]
tagName = pointData["pointTag"]
typeId = UDTType
tagType = "UdtInstance"
params = {
"Out" : pointData["plcAddress"],
"PLC" : pointData["procName"]
}
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "8694428e35d4df34cfa6d836cf9102ad482b3902bef33a52a252e89854d6335d",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,112 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_Motor"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"]
tagName = pointData["deviceTag"]
typeId = UDTType
tagType = "UdtInstance"
extension = pointData["deviceExt"].replace(".","")
params = {}
params[extension] = pointData["plcAddress"]
params["PLC"] = pointData["procName"]
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# Set the Type of Limit switch for Open and Close
basePath = typePathList[i] +"/Inp_RunFdbk/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasRunFdbk"]
pct.tools.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Start/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Start"]
pct.tools.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Stop/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Stop"]
pct.tools.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Run/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Run"]
pct.tools.point_opc_memory_swap(basePath, baseVal)
except:
print("No %s Present" %(UDTType))
def runSimulation():
try:
typePathList = basic.listUDTInstances(UDTType)
tagRead = system.tag.readBlocking(typePathList)
tagPaths = []
tagVals = []
now = system.date.now()
for i in range(0, len(tagRead)):
if tagRead[i].value["_enable"]:
outRun = tagRead[i].value["Sts_Running"]
outStart = tagRead[i].value["Sts_Starting"]
outStop = tagRead[i].value["Sts_Stopping"] or tagRead[i].value["Sts_Stopped"]
currentTime = system.date.toMillis(system.date.now())
timeDiff = currentTime - tagRead[i].value["StartCommand"]
feedbackTime = tagRead[i].value["Config"]["Cfg_SimFdbkT"] * 1000
if (timeDiff > feedbackTime) and (outStart or outRun):
runFdbk = True
else:
runFdbk = False
tagPaths.append(typePathList[i] + "/Inp_RunFdbk/Data")
tagVals.append(runFdbk)
tagPaths.append(typePathList[i] + "/Out_Start/Data")
tagVals.append(outStart)
tagPaths.append(typePathList[i] + "/Out_Stop/Data")
tagVals.append(outStop)
tagPaths.append(typePathList[i] + "/Out_Run/Data")
tagVals.append(outRun)
if len(tagPaths) > 0:
system.tag.writeBlocking(tagPaths, tagVals)
except:
print("No Motors Enabled")

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "87cb4238f4c1869eae2389f8e124271d05f689ae1ee77ee675db02621290fa0b",
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:12:14Z"
}
}
}

View File

@@ -0,0 +1,190 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_VSD"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"]
tagName = pointData["deviceTag"]
typeId = UDTType
tagType = "UdtInstance"
extension = pointData["deviceExt"].replace(".","")
params = {}
params[extension] = pointData["plcAddress"]
params["PLC"] = pointData["procName"]
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMax"])
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMin"])
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMax"])
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMin"])
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkRawMax"])
writePaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedFdbkRawMin"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefEUMax"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefEUMin"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefEUMax"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefEUMin"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefRawMax"])
writePaths.append(typePathList[i] + "/Out_SpeedRef/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_SpeedRefRawMin"])
system.tag.writeBlocking(writePaths, writeVals)
# Set the Type of Limit switch for Open and Close
basePath = typePathList[i] +"/Inp_Running/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasRunFdbk"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Inp_SpeedFdbk/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasSpeedFdbk"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Inp_Faulted/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasDriveFaultAlm"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Start/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Start"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Stop/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Stop"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Out_Run/Data"
baseVal = tagRead[i].value["Config"]["Has_Out_Run"]
basic.point_opc_memory_swap(basePath, baseVal)
except:
print("No %s Present" %(UDTType))
def runSimulation():
try:
typePathList = basic.listUDTInstances(UDTType)
print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
now = system.date.now()
for i in range(0, len(tagRead)):
if tagRead[i].value["_enable"]:
outRun = tagRead[i].value["Out_Run"]["Data"]
outRunFwd = tagRead[i].value["Sts_RunningFwd"]
outRunRev = tagRead[i].value["Sts_RunningRev"]
outStart = tagRead[i].value["Sts_StartingFwd"] or tagRead[i].value["Sts_StartingRev"]
outStop = tagRead[i].value["Sts_StoppingFwd"] or tagRead[i].value["Sts_StoppingFwd"] or tagRead[i].value["Sts_Stopped"]
reference = tagRead[i].value["Out_SpeedRef"]["Data"]
rangeScaleHigh = tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMax"]
rangeScaleLow = tagRead[i].value["Config"]["Cfg_SpeedFdbkEUMin"]
ranging = abs(rangeScaleHigh - rangeScaleLow) * 0.01
tagPaths.append(typePathList[i] + "/Out_Start/Data")
tagVals.append(outStart)
tagPaths.append(typePathList[i] + "/Out_Fwd/Data")
tagVals.append(outRunFwd)
tagPaths.append(typePathList[i] + "/Out_Rev/Data")
tagVals.append(outRunRev)
tagPaths.append(typePathList[i] + "/Out_Stop/Data")
tagVals.append(outStop)
newFeedback = reference + random.uniform(-0.09,0.09)
if outStart or outRunFwd or outRunRev:
print "start"
tagPaths.append(typePathList[i] + "/Out_Run/Data")
tagVals.append(True)
tagPaths.append(typePathList[i] + "/Inp_Running/Data")
tagVals.append(True)
newFeedback = reference + random.uniform(-ranging,ranging)
print newFeedback
tagPaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data")
tagVals.append(newFeedback)
else:
newFeedback = 0.0
tagPaths.append(typePathList[i] + "/Inp_SpeedFdbk/Data")
tagVals.append(newFeedback)
tagPaths.append(typePathList[i] + "/Out_Run/Data")
tagVals.append(False)
tagPaths.append(typePathList[i] + "/Inp_Running/Data")
tagVals.append(False)
if len(tagPaths) > 0:
system.tag.writeBlocking(tagPaths, tagVals)
except:
print("No Motors Enabled")

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "f83f058c0a420b7443a85086db9f1c8418089bc8124a1f4b7fa3d7a473622de0",
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:12:47Z"
}
}
}

View File

@@ -0,0 +1,143 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_ValveC"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"]
tagName = pointData["deviceTag"]
typeId = UDTType
tagType = "UdtInstance"
extension = pointData["deviceExt"].replace(".","")
params = {}
params[extension] = pointData["plcAddress"]
params["PLC"] = pointData["procName"]
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# CV Ranges
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMax"])
writePaths.append(typePathList[i] + "/Out_CV/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVRawMin"])
# Feedback Ranges
writePaths.append(typePathList[i] + "/Val_Pos/Data." + tagsList[0])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Feedback/Data." + tagsList[1])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Val_Pos/Data." + tagsList[2])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMax"])
writePaths.append(typePathList[i] + "/Val_Pos/Data." + tagsList[3])
writeVals.append(tagRead[i].value["Config"]["Cfg_CVEUMin"])
writePaths.append(typePathList[i] + "/Val_Pos/Data." + tagsList[4])
writeVals.append(tagRead[i].value["Config"]["Cfg_FdbkRawMax"])
writePaths.append(typePathList[i] + "/Val_Pos/Data." + tagsList[5])
writeVals.append(tagRead[i].value["Config"]["Cfg_FdbkRawMin"])
system.tag.writeBlocking(writePaths, writeVals)
# Set the Type of Limit switch for Open and Close
basePath = typePathList[i] +"/Inp_ClosedLS/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasClosedLS"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Inp_OpenLS/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasOpenLS"]
basic.point_opc_memory_swap(basePath, baseVal)
except:
print("No %s Present" %(UDTType))
def runSimulation():
try:
typePathList = basic.listUDTInstances(UDTType)
print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
for i in range(0, len(tagRead)):
if tagRead[i].value["_enable"]:
command = tagRead[i].value["Out_CV"]["Data"]
newFeedback = command + random.uniform(-0.09,0.09)
if newFeedback < 0.0:
newFeedback = 0.0
tagPaths.append(typePathList[i] + "/Inp_PosFdbk/Data")
tagVals.append(newFeedback)
if not tagRead[i].value["_manualLS"]:
if newFeedback > 1.0:
tagPaths.append(typePathList[i] + "/Inp_OpenLS/Data")
tagVals.append(True)
tagPaths.append(typePathList[i] + "/Inp_ClosedLS/Data")
tagVals.append(False)
if newFeedback < 1.0:
tagPaths.append(typePathList[i] + "/Inp_OpenLS/Data")
tagVals.append(False)
tagPaths.append(typePathList[i] + "/Inp_ClosedLS/Data")
tagVals.append(True)
if len(tagPaths) > 0:
system.tag.writeBlocking(tagPaths, tagVals)
except Exception as e:
print("No Valves Enabled")
print e

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "37454f07c0b3be2d3081db97346a7bfd2a25809f3d72eff119f57e3ad0ba0aa5",
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:15:06Z"
}
}
}

View File

@@ -0,0 +1,107 @@
import Prime.PCTest as pct
import Prime.Tools.basic as basic
import random
UDTType = "Library/PlantPax/41/P_ValveMO"
def importTag(pointData):
tagProvider = basic.getDefaultProvider(True)
baseTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"]
tagName = pointData["deviceTag"]
typeId = UDTType
tagType = "UdtInstance"
extension = pointData["deviceExt"].replace(".","")
params = {}
params[extension] = pointData["plcAddress"]
params["PLC"] = pointData["procName"]
if pointData["device"] is not None:
params["Device"] = pointData["device"]
# Configure the Tag.
tag = {
"name": tagName,
"typeId" : typeId,
"tagType" : tagType,
"parameters" : params
}
# Set the collision policy to Abort. That way if a tag already exists at the base path,
# we will not override the Tag. If you are overwriting an existing Tag, then set this to "o".
collisionPolicy = "m"
# Create the Tag.
system.tag.configure(baseTagPath, [tag], collisionPolicy)
def initialize():
# Variable holding lise of tag properties
tagsList = ["EngHigh","EngLow", "ScaledHigh", "ScaledLow", "RawHigh", "RawLow"]
try:
typePathList = basic.listUDTInstances(UDTType)
# Read entire UDT for transferring ranges
tagRead = system.tag.readBlocking(typePathList)
writePaths = []
writeVals = []
for i in range(0,len(typePathList)):
# Set the Type of Limit switch for Open and Close
basePath = typePathList[i] +"/Inp_ClosedLS/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasClosedLS"]
basic.point_opc_memory_swap(basePath, baseVal)
basePath = typePathList[i] +"/Inp_OpenLS/Data"
baseVal = tagRead[i].value["Config"]["Cfg_HasOpenLS"]
basic.point_opc_memory_swap(basePath, baseVal)
except:
print("No %s Present" %(UDTType))
def runSimulation():
try:
typePathList = basic.listUDTInstances(UDTType)
print typePathList
tagRead = system.tag.readBlocking(typePathList)
print len(tagRead)
tagPaths = []
tagVals = []
now = system.date.now()
for i in range(0, len(tagRead)):
if tagRead[i].value["_enable"]:
commandOpen = tagRead[i].value["Sts_Opening"] + tagRead[i].value["Out_Open"]["Data"]
commandClose = tagRead[i].value["Sts_Closing"] + tagRead[i].value["Out_Close"]["Data"]
currentTime = system.date.toMillis(system.date.now())
timeDiff = currentTime - tagRead[i].value["StartCommand"]
feedbackTime = tagRead[i].value["Config"]["Cfg_SimFdbkT"] * 1000
if timeDiff > feedbackTime:
if not tagRead[i].value["_manualLS"]:
if commandOpen:
tagPaths.append(typePathList[i] + "/Inp_OpenLS/Data")
tagVals.append(True)
print "open"
tagPaths.append(typePathList[i] + "/Inp_ClosedLS/Data")
tagVals.append(False)
if commandClose:
tagPaths.append(typePathList[i] + "/Inp_OpenLS/Data")
tagVals.append(False)
print "close"
tagPaths.append(typePathList[i] + "/Inp_ClosedLS/Data")
tagVals.append(True)
else:
if not tagRead[i].value["_manualLS"]:
tagPaths.append(typePathList[i] + "/Inp_OpenLS/Data")
tagVals.append(False)
tagPaths.append(typePathList[i] + "/Inp_ClosedLS/Data")
tagVals.append(False)
if len(tagPaths) > 0:
system.tag.writeBlocking(tagPaths, tagVals)
except:
print("No Valves Enabled")

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "b015fb592ebe4fb457e50797c83ec8c62dd3e74d47cc2e1c5535c5a35d5623a6",
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:15:53Z"
}
}
}

View File

@@ -0,0 +1,540 @@
# This Tools Library is specifically for the Prime Controls Testing Platform.
# All Generic Tools should be Placed in the Prime Library that this resides in.
from Prime.Tools import basic
from Prime.Tools import ImportTools
from Prime.PCTest import PlantPax41
from Prime.PCTest import MicroLogix
from Prime import PCTest
from java.lang import Exception as JavaException
logger = system.util.getLogger("Test Platform")
def clearSystem(removePLC=False):
deleteTags()
tagProvider = basic.getDefaultProvider(True)
PLCListPath = [tagProvider + "System/PLCList"]
PLCList = system.tag.readBlocking(PLCListPath)[0].value
PLCList = basic.datasetToJSON(PLCList)
rowCount = len(PLCList)
if removePLC:
# This will only delete PLCs in the System/PLCList
for i in range(0,rowCount):
deviceName = PLCList[i]["ProcName"]
try:
system.device.removeDevice(deviceName)
except:
"Doesn't Exist Currently"
processorTag = tagProvider + "System/PLCList"
system.tag.writeBlocking([processorTag], [None])
def deleteTags():
tagPath = basic.getDefaultProvider(True)
results = system.tag.browse(tagPath)
deleteTagPaths = []
for result in results.getResults():
if "]System" not in str(result["fullPath"]):
print str(result["fullPath"])
deleteTagPaths.append( str(result["fullPath"]))
system.tag.deleteTags(deleteTagPaths)
def tagImport(fileData, sheetType=1, initial=True, conflict="m", plcoverride=False):
# Sheet Type Definitions - These are added as the Library Grows
# 1 - PlantPax41 (Default)
# 2 - MicroLogix
payload = {"updateString":"Starting Import"}
system.perspective.sendMessage("ProgressViewText", payload, scope="session", pageId="progress")
payload = {"progVal":0}
system.perspective.sendMessage("ProgressViewBar", payload, scope="session", pageId="progress")
# Select Code Library Type for Dynamic Function Calls
if sheetType == 1:
moduleProps = {
"module" : "PlantPax41",
"AI" : "AInAdv",
"AO" : "AOut",
"DI" : "DIn",
"DO" : "DOut",
"Motor" : "Motor",
"ValveMO" : "ValveMO",
"ValveC" : "ValveC",
"VSD" : "VSD",
"driver" : "LogixDriver"
}
# Verify Sheet position using Developer Tools in Excel
# Some sheets are hidden with VBA and can only be accessed in this manner
procSheet = 0
dataSheet = 5
if sheetType == 2:
moduleProps = {
"module" : "MicroLogix",
"AI" : "AInAdv",
"AO" : "AOut",
"DI" : "DIn",
"DO" : "DOut",
"driver" : "MicroLogix"
}
# Verify Sheet position using Developer Tools in Excel
# Some sheets are hidden with VBA and can only be accessed in this manner
procSheet = 1
dataSheet = 4
print procSheet
tagProvider = basic.getDefaultProvider(True)
processorTag = tagProvider + "System/PLCList"
# existing values
if initial:
system.tag.writeBlocking([processorTag], [None])
existing_processors = system.tag.readBlocking([processorTag])
existing_processors = existing_processors[0].value
try:
logger.info("Import Started")
print "start"
if isinstance(fileData, str):
logger.warn("Local Import Processed")
data = ImportTools.excelToDataSet(fileData, hasHeaders = True, sheetNum = dataSheet, firstRow=1, forceString = True)
new_processors = ImportTools.excelToDataSet(fileData, hasHeaders = True, sheetNum = procSheet, firstRow=0, forceString = True)
print new_processors
else:
logger.warn("Remote Import Processed")
data = ImportTools.excelBytesToDataSet(fileData, hasHeaders = True, sheetNum = dataSheet, firstRow=1, forceString = True)
new_processors = ImportTools.excelBytesToDataSet(fileData, hasHeaders = True, sheetNum = procSheet, firstRow=0, forceString = True)
print new_processors
print data
except (ValueError, JavaException, Exception), e:
logString = "Invalid sheet index - %s" % (e.getMessage())
logger.error(logString)
raise
payload = {"updateString":"File Processed"}
system.perspective.sendMessage("ProgressViewText", payload, scope="session", pageId="progress")
payload = {"progVal":5}
system.perspective.sendMessage("ProgressViewBar", payload, scope="session", pageId="progress")
tagPaths = []
tagValues = []
filteredData = []
headers = ["ProcName", "Style", "PartNumber", "RackName", "Slot", "IPAddress", "addCol", "Driver"]
existingProcessorsDict = {}
newProcessorsDict = {}
# Convert to PyDataSet to allow for 8.1 Compatibility
existingProcessorsDS = system.dataset.toPyDataSet(existing_processors)
# Place Existing Processors in Dictionary for Comparison and Updates
try:
for row in existingProcessorsDS:
existingProcessorsDict[row[0]] = row
existing = True
except:
logger.info("No Existing PLCs")
existing = False
payload = {"updateString":"Creating Device Dataset"}
system.perspective.sendMessage("ProgressViewText", payload, scope="session", pageId="progress")
payload = {"progVal":10}
system.perspective.sendMessage("ProgressViewBar", payload, scope="session", pageId="progress")
# Place Existing Processors in Dictionary for Comparison and Updates
rows = new_processors.getRowCount()
for i in range(0,rows):
proc = new_processors.getValueAt(i,0)
rowValue = []
if proc is not None:
for j in range(0,len(headers)):
if j != (len(headers) - 1):
try:
checkValue = new_processors.getValueAt(i,headers[j])
rowValue.append(checkValue)
except:
rowValue.append(None)
else:
rowValue.append(moduleProps["driver"])
newProcessorsDict[proc] = rowValue
# If there are current PLCs in system update existing PLCs with new data and add new PLCs
# After update section dataset is sorted by Processor Name and stored in PLC List
if existing:
existingProcessorsDict.update(newProcessorsDict)
filteredData = existingProcessorsDict.values()
else:
filteredData = newProcessorsDict.values()
updatedProcessors = system.dataset.toDataSet(headers, filteredData)
sortedProcessors = system.dataset.sort(updatedProcessors, "ProcName",True, False)
system.tag.writeBlocking([processorTag],[sortedProcessors])
# Set the collision policy to merge so that any changes made will be overridden
collisionPolicy = conflict
rowCount = data.getRowCount()
importCount = 0
try:
payload = {"updateString":"Importing Tags"}
system.perspective.sendMessage("ProgressViewText", payload, scope="session", pageId="progress")
payload = {"progVal":25}
system.perspective.sendMessage("ProgressViewBar", payload, scope="session", pageId="progress")
# Parse Through Dataset
for i in range(0,rowCount):
pointData = {}
pointData["tagProvider"] = tagProvider
# Get relevant data from row on sheet
# Processor and Rack Info
pointData["procName"] = data.getValueAt(i, "ProcName")
try:
pointData["RA"] = data.getValueAt(i, "RA")
except:
pointData["RA"] = "Main"
# Point Info
pointData["pointType"] = data.getValueAt(i, "Type")
pointData["pointTag"] = data.getValueAt(i,"PointTag")
pointData["plcAddress"] = data.getValueAt(i, "PlcAddress")
# Device Information
pointData["device"] = data.getValueAt(i,"Device")
if pointData["device"] == "":
pointData["device"] = None
pointData["deviceDesc"] = data.getValueAt(i,"DeviceDescription")
print data.getValueAt(i,"DeviceDescription")
if pointData["deviceDesc"] == "":
pointData["deviceDesc"] = None
pointData["deviceTag"] = data.getValueAt(i,"DeviceTag")
pointData["deviceType"] = data.getValueAt(i, "DeviceType")
pointData["deviceExt"] = data.getValueAt(i, "Extension")
# Description Information
pointData["descA"] = data.getValueAt(i,"DescriptionA")
pointData["descB"] = data.getValueAt(i,"DescriptionB")
pointData["descC"] = data.getValueAt(i,"DescriptionC")
pointData["descD"] = data.getValueAt(i,"DescriptionD")
if (pointData["deviceTag"] is None) or (pointData["deviceTag"] is ""):
if ((pointData["pointTag"] != "SPARE") and (pointData["pointTag"] is not None)):
# Create a tag for the base directory of rack
# This will separate the data for non-device digital and analogs.
tagName = "_device"
valueSource = "memory"
value = "Rack"
dataType = "String"
# Get Path for base directory tag
tag_path = pointData["tagProvider"] + pointData["procName"] + "/" + pointData["RA"]
# Configure the Tag.
tag = {
"name" : tagName,
"valueSource" : valueSource,
"value" : value,
"dataType" : dataType
}
# Create the Tag.
system.tag.configure(tag_path, [tag], collisionPolicy)
try:
# This space is to call the import function based on Module within the PCTest Library
# This is specifically for growth later allowing for other PLC libraries to be utilized
if hasattr(PCTest, moduleProps["module"]):
package = getattr(PCTest, moduleProps["module"])
if hasattr(package, moduleProps[pointData["pointType"]]):
library = getattr(package, moduleProps[pointData["pointType"]])
if hasattr(library, "importTag"):
func = getattr(library, "importTag")
func(pointData)
except Exception as e:
print e
print pointData
fullTagPath = tagProvider + pointData["procName"] + "/" + pointData["RA"] + "/" + pointData["pointTag"] + "/Description"
# Build Description for Tag
if pointData["deviceDesc"] is None:
if pointData["descA"] is not None and not "":
fullDescription = pointData["descA"]
if pointData["descB"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"]
if pointData["descC"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"] + " - " + pointData["descC"]
if pointData["descD"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"] + " - " + pointData["descC"] + " - " + pointData["descD"]
else:
fullDescription = moduleProps[pointData["pointType"]]
else:
fullDescription = pointData["deviceDesc"]
tagPaths.append(fullTagPath)
tagValues.append(fullDescription)
if pointData["deviceTag"] is not None and not (pointData["deviceTag"] == "") and ((pointData["pointTag"] != "SPARE") or (pointData["pointTag"] is not None) or (pointData["pointTag"] is not "")):
# Create a tag for the base directory of Devices
# This will separate the data for non-device digital and analogs.
tagName = "_device"
valueSource = "memory"
value = pointData["deviceType"]
dataType = "String"
# Get Path for base directory tag
tag_path = pointData["tagProvider"] + pointData["procName"] + "/" + pointData["deviceType"]
# Configure the Tag.
tag = {
"name" : tagName,
"valueSource" : valueSource,
"value" : value,
"dataType" : dataType
}
# Create the Tag.
system.tag.configure(tag_path, [tag], collisionPolicy)
try:
# This space is to call the import function based on Module within the PCTest Library
# This is specifically for growth later allowing for other PLC libraries to be utilized
if hasattr(PCTest, moduleProps["module"]):
package = getattr(PCTest, moduleProps["module"])
if hasattr(package, moduleProps[pointData["deviceType"]]):
library = getattr(package, moduleProps[pointData["deviceType"]])
if hasattr(library, "importTag"):
func = getattr(library, "importTag")
func(pointData)
except Exception as e:
print e
print pointData
fullTagPath = tagProvider + pointData["procName"] + "/" + pointData["deviceType"] + "/" + pointData["deviceTag"] + "/Description"
# Build Description for Tag
if pointData["deviceDesc"] is None:
if pointData["descA"] is not None and not "":
fullDescription = pointData["descA"]
if pointData["descB"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"]
if pointData["descC"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"] + " - " + pointData["descC"]
if pointData["descD"] is not None and not "":
fullDescription = pointData["descA"] + " - " + pointData["descB"] + " - " + pointData["descC"] + " - " + pointData["descD"]
else:
fullDescription = moduleProps[pointData["pointType"]]
else:
fullDescription = pointData["deviceDesc"]
tagPaths.append(fullTagPath)
tagValues.append(fullDescription)
progNum = ((i * 1.0) / rowCount) * 60.0
payload = {"progVal": (25 + progNum)}
system.perspective.sendMessage("ProgressViewBar", payload, scope="session", pageId="progress")
except (ValueError, JavaException, Exception), e:
logString = "Error - %s" % (e.getMessage())
logger.error(logString)
raise
# Write data to description and type tags
system.tag.writeBlocking(tagPaths, tagValues)
# This function is change the Point Tag type - This function is specific to the UDTs within the Testing Platform
def point_opc_memory_swap(path, opc):
tags = {"name": "Data"}
if opc:
tags["valueSource"] = "opc"
else:
tags["valueSource"] = "memory"
system.tag.configure(path.replace("/Data",""), tags, collisionPolicy="m")
# This function provides a way to bulk add or delete devices
def setupPLCs(delete = False):
tagProvider = basic.getDefaultProvider(True)
PLCListPath = [tagProvider + "System/PLCList"]
PLCList = system.tag.readBlocking(PLCListPath)[0].value
PLCList = basic.datasetToJSON(PLCList)
rowCount = len(PLCList)
for i in range(0,rowCount):
if (PLCList[i]["IPAddress"] is not None) and (PLCList[i]["IPAddress"] != ""):
print i
print PLCList[i]["ProcName"]
device = PLCList[i]["Driver"]
# Get Appropriate fucntion for device props.
if hasattr(PCTest, "tools"):
package = getattr(PCTest, "tools")
if hasattr(package, device + "DeviceProps"):
func = getattr(package, (device + "DeviceProps"))
# Build needed arguments for adding device
deviceProps = func(PLCList[i])
deviceType = device
deviceName = PLCList[i]["ProcName"]
# There is no mechanism to update existing devices through scripting. Device Must be removed and added again
try:
system.device.removeDevice(deviceName)
except:
"Doesn't Exist Currently"
if not delete:
system.device.addDevice(deviceName=deviceName, deviceType=deviceType, deviceProps=deviceProps)
def deviceConfigure(deviceInfo):
# Check if Device Exists
deviceTagPath = "[System]Gateway/Devices/" + deviceInfo["ProcName"]
exists = system.tag.exists(deviceTagPath)
if (deviceInfo["IPAddress"] is not None) and (deviceInfo["IPAddress"] != ""):
device = deviceInfo["Driver"]
# Get Appropriate fucntion for device props.
if hasattr(PCTest, "tools"):
package = getattr(PCTest, "tools")
if hasattr(package, device + "DeviceProps"):
func = getattr(package, (device + "DeviceProps"))
# Build needed arguments for adding device
deviceProps = func(deviceInfo)
deviceType = device
deviceName = deviceInfo["ProcName"]
# There is no mechanism to update existing devices through scripting. Device Must be removed and added again
try:
system.device.removeDevice(deviceName)
except:
logStr = deviceName + " Doesn't Exist Currently"
logger.info(logStr)
logStr = "Adding -" + deviceName
logger.info(logStr)
system.device.addDevice(deviceName=deviceName, deviceType=deviceType, deviceProps=deviceProps)
def deviceDelete(deviceInfo):
# Check if Device Exists
deviceTagPath = "[System]Gateway/Devices/" + deviceInfo["ProcName"]
exists = system.tag.exists(deviceTagPath)
if exists:
system.device.removeDevice(deviceInfo["ProcName"])
# The following functions use the PLCTag
def LogixDriverDeviceProps(data):
deviceProps = {}
deviceProps["hostname"] = data["IPAddress"]
deviceProps["port"] = "44818"
deviceProps["concurrency"] = "4"
deviceProps["slotnumber"] = data["Slot"]
return deviceProps
def MicroLogixDeviceProps(data):
deviceProps = {}
deviceProps["hostname"] = data["IPAddress"]
deviceProps["port"] = "44818"
deviceProps["concurrency"] = "4"
return deviceProps
def initialize():
PlantPax41.AInAdv.initialize()
PlantPax41.AOut.initialize()
PlantPax41.Motor.initialize()
PlantPax41.ValveC.initialize()
PlantPax41.VSD.initialize()
PlantPax41.ValveMO.initialize()
MicroLogix.AInAdv.initialize()
MicroLogix.AOut.initialize()
def clearPLCFaults(plc):
provider = Prime.Tools.basic.getDefaultProvider(True)
tagPath = provider + plc
faultPaths = []
faultVals = []
tags = system.tag.browse(tagPath, filter ={"recursive" : True, "name":"Fault"})
results = tags.getResults()
for result in results:
faultPath = str(result["fullPath"])
faultPaths.append(faultPath)
faultVals.append(False)
system.tag.writeBlocking(faultPaths, faultVals)
def savePLCScenario(plc, scenarioName):
saveTags = ["_enable", "_enableL2", "Noise", "NoiseBase", "Data", "Fault"]
dsHeaders = ["tagPath", "val", "name"]
dsValues = []
jsonVals = []
provider = Prime.Tools.basic.getDefaultProvider(True)
tagPath = provider + plc
tags = system.tag.browse(tagPath, filter ={"recursive" : True})
results = tags.getResults()
enableList = []
nameList = []
for result in results:
tag_name = result.get('name','')
if tag_name in saveTags:
enableTag = str(result["fullPath"])
enableList.append(enableTag)
nameList.append(tag_name)
tags = system.tag.readBlocking(enableList)
for i in range(0,len(tags)):
appendVal = {}
dsTagPath = enableList[i]
dsVal = tags[i].value
dsName = nameList[i]
appendVal[dsHeaders[0]] = dsTagPath
appendVal[dsHeaders[1]] = dsVal
appendVal[dsHeaders[2]] = dsName
jsonVals.append(appendVal)
jsonVals = sorted(jsonVals, key=lambda x: x["tagPath"])
scenarioPath = tagPath + "/Scenarios/" + scenarioName
if system.tag.exists(scenarioPath):
system.tag.writeBlocking([scenarioPath],[jsonVals])
else:
basePath = tagPath + "/Scenarios/"
tag = {
"name" : scenarioName,
"valueSource": "memory",
"dataType": "Document"
}
system.tag.configure(basePath, [tag], "m")
system.tag.writeBlocking([scenarioPath],[jsonVals])

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 7,
"lastModificationSignature": "12e9c464e7f529e36c0e854f6cae4c910e2bde3b614e38814560665fcff62a32",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,251 @@
import system
import json
from java.util import Date
from java.text import SimpleDateFormat
from java.lang import Exception as JavaException
logger = system.util.getLogger("Docker")
def getDockerAPIEndpoint():
"""Get API endpoint from config tag"""
result = system.tag.readBlocking(["Docker/Config/APIEndpoint"])
return result[0].value
def isMonitoringEnabled():
"""Check if monitoring is enabled"""
result = system.tag.readBlocking(["Docker/Config/Enabled"])
return result[0].value
def parseDockerTimestamp(timestamp):
"""Convert Docker timestamp to Java Date"""
try:
from java.util import TimeZone
# Docker uses ISO 8601 format in UTC
# Example: "2024-12-17T10:30:00.123456789Z"
clean_timestamp = timestamp.split('.')[0] + 'Z' # Remove microseconds
sdf = SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'")
sdf.setTimeZone(TimeZone.getTimeZone("UTC")) # Force UTC parsing
return sdf.parse(clean_timestamp)
except Exception as e:
logger.error("Timestamp parse error: " + str(e))
return Date()
def calculateUptime(state):
"""Calculate container uptime from state info"""
try:
if state.get('Status') == 'running':
started = state.get('StartedAt', '')
if started:
# Parse and calculate duration
startDate = parseDockerTimestamp(started)
now = Date()
diff = now.time - startDate.time
# Convert to readable format
totalMinutes = diff / (1000 * 60)
days = int(totalMinutes / (60 * 24))
hours = int((totalMinutes % (60 * 24)) / 60)
minutes = int(totalMinutes % 60)
if days > 0:
return "%dd %dh %dm" % (days, hours, minutes)
elif hours > 0:
return "%dh %dm" % (hours, minutes)
else:
return "%dm" % minutes
return "Not running"
except Exception as e:
logger.error("Uptime calc error: " + str(e))
return "Unknown"
def getContainerIPAddress(details):
"""Extract IP address from container details"""
try:
networkSettings = details.get('NetworkSettings', {})
networks = networkSettings.get('Networks', {})
# Try to get IP from first available network
if networks:
for networkName, networkInfo in networks.items():
ipAddress = networkInfo.get('IPAddress', '')
if ipAddress:
return ipAddress
# Fallback to deprecated IPAddress field
ipAddress = networkSettings.get('IPAddress', '')
if ipAddress:
return ipAddress
return "N/A"
except Exception as e:
logger.error("Failed to get IP address: " + str(e))
return "N/A"
def getAllContainers(apiEndpoint):
"""Query Docker API for all containers"""
try:
url = "%s/containers/json?all=true" % apiEndpoint
try:
response = system.net.httpClient().get(url)
except (ValueError, JavaException, Exception), e:
logger.error("Failed to get containers: " + str(e))
import traceback
logger.error(traceback.format_exc())
return []
if response.isGood():
containers = response.json
return containers
else:
logger.error("API call failed: HTTP %d" % response.statusCode)
return []
except Exception as e:
logger.error("Failed to get containers: " + str(e))
import traceback
logger.error(traceback.format_exc())
return []
def getContainerDetails(apiEndpoint, containerId):
"""Get detailed info for a specific container"""
try:
url = "%s/containers/%s/json" % (apiEndpoint, containerId)
response = system.net.httpClient().get(url)
if response.isGood():
return response.json
else:
logger.error("Failed to get details for %s: HTTP %d" % (containerId, response.statusCode))
return None
except Exception as e:
logger.error("Failed to get container details: " + str(e))
return None
def buildContainerDataset(containers, apiEndpoint):
"""Build a dataset from container information"""
# Define column headers
headers = [
'ContainerID',
'ContainerName',
'Image',
'IPAddress',
'Status',
'State',
'IsRunning',
'Created',
'UpTime',
'HealthStatus',
'RestartCount',
'LastChecked'
]
# Build data rows
data = []
for containerInfo in containers:
try:
# Extract container name (remove leading slash)
names = containerInfo.get('Names', [])
containerName = names[0].lstrip('/') if names else containerInfo.get('Id', '')[:12]
# Get detailed info
details = getContainerDetails(apiEndpoint, containerInfo['Id'])
if not details:
continue
# Extract values
state = details.get('State', {})
config = details.get('Config', {})
containerID = containerInfo.get('Id', '')[:12]
image = containerInfo.get('Image', '')
ipAddress = getContainerIPAddress(details)
status = containerInfo.get('Status', '')
stateStr = state.get('Status', '')
isRunning = state.get('Running', False)
created = parseDockerTimestamp(details.get('Created', ''))
uptime = calculateUptime(state)
healthStatus = state.get('Health', {}).get('Status', 'none') if state.get('Health') else 'none'
restartCount = state.get('RestartCount', 0)
lastChecked = Date()
# Add row to data
row = [
containerID,
containerName,
image,
ipAddress,
status,
stateStr,
isRunning,
created,
uptime,
healthStatus,
restartCount,
lastChecked
]
data.append(row)
except Exception as e:
logger.error("Failed to process container: " + str(e))
continue
# Sort data by container name (index 1)
data.sort(key=lambda x: x[1])
# Create dataset
dataset = system.dataset.toDataSet(headers, data)
return dataset
def monitorContainers():
"""Main monitoring function - monitors ALL containers and stores in dataset tag"""
# Check if monitoring is enabled
if not isMonitoringEnabled():
logger.trace("Docker monitoring is disabled")
return
try:
apiEndpoint = getDockerAPIEndpoint()
logger.trace("Starting container monitoring...")
# Get all containers
containers = getAllContainers(apiEndpoint)
if not containers:
logger.trace("No containers found or API call failed")
# Write empty dataset
emptyDataset = system.dataset.toDataSet(
['ContainerID', 'ContainerName', 'Image', 'IPAddress', 'Status', 'State',
'IsRunning', 'Created', 'UpTime', 'HealthStatus', 'RestartCount', 'LastChecked'],
[]
)
system.tag.writeBlocking(["Docker/ContainerData"], [emptyDataset])
return
logger.trace("Found %d containers to monitor" % len(containers))
# Build dataset from containers
dataset = buildContainerDataset(containers, apiEndpoint)
# Write dataset to tag
system.tag.writeBlocking(["Docker/ContainerData"], [dataset])
logger.trace("Updated dataset with %d containers" % dataset.rowCount)
except Exception as e:
logger.error("Container monitoring failed: " + str(e))
import traceback
logger.error(traceback.format_exc())
def execute():
# Execute monitoring
monitorContainers()

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "765132eca2d1dbf07bd3369aeb9ad6ca9bfaa72b27673dd7a83614b5ad411fc2",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-19T16:11:40Z"
}
}
}

View File

@@ -0,0 +1,240 @@
def excelToDataSet(fileName, hasHeaders = False, forceString = False, sheetNum = 0, firstRow = None, lastRow = None, firstCol = None, lastCol = None, customHeaders = None):
import org.apache.poi.ss.usermodel.WorkbookFactory as WorkbookFactory
import org.apache.poi.ss.usermodel.DateUtil as DateUtil
from java.io import FileInputStream
from java.util import Date
from os.path import exists
"""
Description:
Function to create a dataset from an Excel spreadsheet. This is typically used in Vision and the file path is
directly referenced to its location in the filesystem.
Arguments:
fileName: The path to the Excel spreadsheet. (required)
hasHeaders: If true, uses the first row of the spreadsheet as column names.
forceString: If true, forces all cell values to be strings. This can be useful if cell values in columns are not consistent.
To create a dataset, the data type is determined by the first row. If the data types are different, you can get errors.
If you force all cells to strings, then this is not an issue.
sheetNum: Select the sheet to process. Defaults to the first sheet.
firstRow: Select first row to process.
lastRow: Select last row to process.
firstCol: Select first column to process
lastCol: Select last column toprocess
History:
No. Date Author Comment
1.0 2021-01-22 Jordan Clark Initial - https://forum.inductiveautomation.com/t/copying-a-excel-file-to-a-tables-dataset/34942/13
1.1 2023-02-17 James Landwerlen Updated to handle blank cells
and to force strings
"""
if exists(fileName):
fileStream = FileInputStream(fileName)
try:
wb = WorkbookFactory.create(fileStream)
sheet = wb.getSheetAt(sheetNum)
if firstRow is None:
firstRow = sheet.getFirstRowNum()
if lastRow is None:
lastRow = sheet.getLastRowNum()
data = []
for i in range(firstRow , lastRow + 1):
row = sheet.getRow(i)
if i == firstRow:
if firstCol is None:
firstCol = row.getFirstCellNum()
if lastCol is None:
lastCol = row.getLastCellNum()
else:
# if lastCol is specified add 1 to it.
lastCol += 1
if hasHeaders and customHeaders is None:
headers = list(row)[firstCol:lastCol]
# print headers
elif hasHeaders and customHeaders is not None:
headers = customHeaders
else:
headers = ['Col'+str(i) for i in range(firstCol, lastCol)]
# print headers
rowOut = []
for j in range(firstCol, lastCol):
if i == firstRow and hasHeaders:
pass
else:
cell = row.getCell(j)
if cell is not None:
cellType = cell.getCellType().toString()
if cellType == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif cellType == 'STRING':
value = cell.getStringCellValue()
elif cellType == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif cellType == 'BLANK':
value = None
elif cellType == 'FORMULA':
formulatype=str(cell.getCachedFormulaResultType())
if formulatype == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif formulatype == 'STRING':
value = cell.getStringCellValue()
elif formulatype == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif formulatype == 'BLANK':
value = None
else:
value = None
else:
value = None
if forceString:
if value == None:
rowOut.append(value)
else:
rowOut.append(str(value))
else:
rowOut.append(value)
if len(rowOut) > 0:
data.append(rowOut)
fileStream.close()
return system.dataset.toDataSet(headers, data)
except Exception as e:
logString = "Failed - %s" %(e)
logger.warn(logString)
system.perspective.closePopup("progress")
params = {"message" : logString}
title = "Import Error"
view = "Popups/Error/Main"
system.perspective.openPopup("errorPopup", view, params, title, modal = True)
def excelBytesToDataSet(bytesIn, hasHeaders = True, forceString = True, sheetNum = 0, firstRow = None, lastRow = None, firstCol = None, lastCol = None):
import org.apache.poi.ss.usermodel.WorkbookFactory as WorkbookFactory
import org.apache.poi.ss.usermodel.DateUtil as DateUtil
from java.io import ByteArrayInputStream
fileStream = ByteArrayInputStream(bytesIn)
try:
wb = WorkbookFactory.create(fileStream)
sheet = wb.getSheetAt(sheetNum)
except Exception as e:
logString = "Failed - %s" %(e)
logger.warn(logString)
system.perspective.closePopup("progress")
params = {"message" : logString}
title = "Import Error"
view = "Popups/Error/Main"
system.perspective.openPopup("errorPopup", view, params, title, modal = True)
if firstRow is None:
firstRow = sheet.getFirstRowNum()
if lastRow is None:
lastRow = sheet.getLastRowNum()
data = []
for i in range(firstRow , lastRow + 1):
row = sheet.getRow(i)
if i == firstRow:
if firstCol is None:
firstCol = row.getFirstCellNum()
if lastCol is None:
lastCol = row.getLastCellNum()
else:
# if lastCol is specified add 1 to it.
lastCol += 1
if hasHeaders:
headers = list(row)[firstCol:lastCol]
print headers
else:
headers = ['Col'+str(i) for i in range(firstCol, lastCol)]
print headers
rowOut = []
for j in range(firstCol, lastCol):
if i == firstRow and hasHeaders:
pass
else:
cell = row.getCell(j)
if cell is not None:
cellType = cell.getCellType().toString()
if cellType == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif cellType == 'STRING':
value = cell.getStringCellValue()
elif cellType == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif cellType == 'BLANK':
value = None
elif cellType == 'FORMULA':
formulatype=str(cell.getCachedFormulaResultType())
if formulatype == 'NUMERIC':
if DateUtil.isCellDateFormatted(cell):
value = cell.dateCellValue
else:
value = cell.getNumericCellValue()
if value == int(value):
value = int(value)
elif formulatype == 'STRING':
value = cell.getStringCellValue()
elif formulatype == 'BOOLEAN':
value = cell.getBooleanCellValue()
elif formulatype == 'BLANK':
value = None
else:
value = None
else:
value = None
if forceString:
if value == None:
rowOut.append(value)
else:
rowOut.append(str(value))
else:
rowOut.append(value)
if len(rowOut) > 0:
data.append(rowOut)
fileStream.close()
return system.dataset.toDataSet(headers, data)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "753798a6778f6b0ef11619392be6fde08c25830a094478cefe85621a01b8f4b5",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,64 @@
def datasetToJSON(data):
pds = system.dataset.toPyDataSet(data)
columnNames = pds.columnNames
retVal = [
{
col: row[col]
for col in columnNames
} for row in pds
]
return retVal
# Function to Get the default provider of the current project
def getDefaultProvider(full = False):
# Full Returns Tag Provider With Brackets
if full:
return "[" + str(system.tag.getConfiguration()[0]['path']).lstrip('[').rstrip(']') + "]"
else:
return str(system.tag.getConfiguration()[0]['path']).lstrip('[').rstrip(']')
# Function to return UDTs of a particular type denoted by variable UDTType
def listUDTInstances(UDTType):
default = "[" + getDefaultProvider() + "]"
tags = system.tag.browse(default, {"recursive":True,"tagType":"UdtInstance", "typeId":UDTType})
results = tags.getResults()
pathList = []
typePathList = []
for result in results:
typePathList.append(str(result["fullPath"]))
return typePathList
def errorPopup(exceptionStr):
params = {"message" : exceptionStr}
title = "System Error"
view = "Popups/Error/Main"
system.perspective.openPopup("errorPopup", view, params, title, modal = True)
def restartPLCTags(plc):
provider = getDefaultProvider(True)
tagPath = provider + plc
tags = system.tag.browse(tagPath, filter ={"recursive" : True,"tagType":"UdtInstance"})
results = tags.getResults()
enableList = []
for result in results:
if "Points" not in str(result["typeId"]):
enableTag = str(result["fullPath"]) + ".Enabled"
enableList.append(enableTag)
trueList = [True for i in range(len(enableList))]
falseList = [False for i in range(len(enableList))]
system.tag.writeBlocking(enableList, falseList)
system.tag.writeBlocking(enableList, trueList)

View File

@@ -0,0 +1,17 @@
{
"scope": "A",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"code.py"
],
"attributes": {
"hintScope": 2,
"lastModificationSignature": "9a050cb3f6afe7af70b3103b6c21d3d27983318ce1f8c5f1c553524109783052",
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-17T16:24:07Z"
}
}
}

View File

@@ -0,0 +1,5 @@
def handleTimerEvent():
from Prime.PCTest import PlantPax41
PlantPax41.AInAdv.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 1000,
"lastModificationSignature": "7b1937fb4d8da3ef82e5378f19cb7af91bcfb9ab455306d281e33309e6a2fe5b",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2026-02-18T14:59:24Z"
},
"enabled": true
}
}

View File

@@ -0,0 +1,2 @@
def handleTimerEvent():
Prime.Tools.Docker.execute()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 10000,
"lastModificationSignature": "089696e990dda62733cd2e893f16635a6822de93eb9f87ef3ffa5110688e7a61",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2026-01-05T20:16:43Z"
},
"enabled": true
}
}

View File

@@ -0,0 +1,5 @@
def handleTimerEvent():
from Prime.PCTest import PlantPax41
PlantPax41.Motor.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 1000,
"lastModificationSignature": "edad3ff840ea51c3627c00d1cf4afad674968eddd45300c23890d14e0638169d",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-19T15:45:39Z"
},
"enabled": false
}
}

View File

@@ -0,0 +1,5 @@
def handleTimerEvent():
from Prime.PCTest import PlantPax41
PlantPax41.VSD.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 1000,
"lastModificationSignature": "31cab582b8c4e1070ce6904a09ca664f4aab653af07b1b8d632073311f33ca96",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2026-02-25T21:41:23Z"
},
"enabled": true
}
}

View File

@@ -0,0 +1,4 @@
def handleTimerEvent():
from Prime.PCTest import PlantPax41
PlantPax41.ValveC.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 1000,
"lastModificationSignature": "a10e481737e15582e5f4d58987f2c18c4370b8d83ab2056008093f4a786d38db",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2026-02-25T21:05:15Z"
},
"enabled": true
}
}

View File

@@ -0,0 +1,5 @@
def handleTimerEvent():
from Prime.PCTest import PlantPax41
PlantPax41.ValveMO.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 1000,
"lastModificationSignature": "f720fa5c86539b318d56a814b35de7392c95fb6fc05648264f7ad6cd6ef1fb0b",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2025-12-19T15:45:39Z"
},
"enabled": false
}
}

View File

@@ -0,0 +1,4 @@
def handleTimerEvent():
from Prime.PCTest import Objects
Objects.Tank.runSimulation()

View File

@@ -0,0 +1,20 @@
{
"scope": "G",
"version": 1,
"restricted": false,
"overridable": true,
"files": [
"handleTimerEvent.py"
],
"attributes": {
"sharedThread": true,
"delay": 250,
"lastModificationSignature": "e7be223279a0f76b4084ec343c6f08743e1d082413267b6c95580a668ea9a405",
"fixedDelay": true,
"lastModification": {
"actor": "admin",
"timestamp": "2026-02-19T19:30:43Z"
},
"enabled": true
}
}