133 lines
5.2 KiB
Plaintext
133 lines
5.2 KiB
Plaintext
def trace_power_flow():
|
|
"""
|
|
Traces power flow through a series of connected transformers (UTs).
|
|
- Reads the state of all switches.
|
|
- Determines which components are energized and from which source.
|
|
- Writes the results back to memory tags.
|
|
"""
|
|
|
|
# 1. Define the network structure in order from left to right
|
|
# This makes it easy to find neighbors.
|
|
# [Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_UT-HS1
|
|
ut_names = ["UT-P5-1", "UT-P5-2", "UT-P5-3", "UT-P5-4", "UT-P5-5"]
|
|
providers = [ "[Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_",
|
|
"[Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_",
|
|
"[Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_",
|
|
"[Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_",
|
|
"[Ignition_Common_IO_Gtwy]Yard/SATB1_YARD_"]
|
|
# 2. Build the list of all tags we need to read
|
|
# Using readBlocking for a single, efficient tag read
|
|
tag_paths_to_read = [
|
|
'[Ignition_Common_IO_Gtwy]Cables/SES/DS3A_W3', # Assumes a tag for the source status
|
|
'[Ignition_Common_IO_Gtwy]Cables/SES/DS3B_W3']
|
|
idx = 0
|
|
for name in ut_names:
|
|
# Assuming your UDT instances are in a folder named 'UTs'
|
|
base_path = providers[idx] + name
|
|
tag_paths_to_read.append("{}/Source 1 Switch".format(base_path))
|
|
tag_paths_to_read.append("{}/Source 2 Switch".format(base_path))
|
|
idx+=1
|
|
|
|
# 3. Read all tags at once
|
|
try:
|
|
tag_values = system.tag.readBlocking(tag_paths_to_read)
|
|
except Exception as e:
|
|
# Log error if tags can't be read
|
|
system.util.getLogger("PowerTrace").error("Error reading tags: {}".format(e))
|
|
return
|
|
|
|
# Create a dictionary for easy access to tag values
|
|
# e.g., values['UT-HS1']['Switch_L_Status']
|
|
values = {}
|
|
values['DS-3A_Live'] = tag_values[0].value
|
|
values['DS-3B_Live'] = tag_values[1].value
|
|
|
|
read_idx = 2
|
|
for name in ut_names:
|
|
values[name] = {
|
|
'Switch_L_Status': tag_values[read_idx].value,
|
|
'Switch_R_Status': tag_values[read_idx + 1].value}
|
|
read_idx += 2
|
|
|
|
# 4. The Tracing Algorithm
|
|
# This dictionary will store the final state: e.g., energized_state['UT P1.2'] = 'DS-1A1'
|
|
energized_state = {}
|
|
|
|
# A queue for our BFS traversal, storing (ut_name, source)
|
|
# Using a list as a queue: append to add, pop(0) to remove from front
|
|
q = []
|
|
|
|
# Initialize the queue with active sources
|
|
if values['DS-3A_Live']:
|
|
q.append( ("UT-P5-1", "DS-1A3") )
|
|
if values['DS-3B_Live']:
|
|
q.append( ("UT-P5-5", "DS-1B3") )
|
|
|
|
# Process the queue until it's empty
|
|
visited = set() # Keep track of UTs we've already processed to prevent infinite loops
|
|
while q:
|
|
current_ut_name, source = q.pop(0)
|
|
|
|
if current_ut_name in visited:
|
|
continue
|
|
|
|
visited.add(current_ut_name)
|
|
energized_state[current_ut_name] = source
|
|
current_ut_index = ut_names.index(current_ut_name)
|
|
|
|
# Check for propagation to the RIGHT (-->)
|
|
if current_ut_index < len(ut_names) - 1:
|
|
neighbor_name = ut_names[current_ut_index + 1]
|
|
# Condition: Current UT's right switch is closed AND Neighbor's left switch is closed
|
|
if values[current_ut_name]['Switch_R_Status']['Value'] and values[neighbor_name]['Switch_L_Status']['Value']:
|
|
if neighbor_name not in visited:
|
|
q.append( (neighbor_name, source) )
|
|
|
|
# Check for propagation to the LEFT (<--)
|
|
if current_ut_index > 0:
|
|
neighbor_name = ut_names[current_ut_index - 1]
|
|
# Condition: Current UT's left switch is closed AND Neighbor's right switch is closed
|
|
if values[current_ut_name]['Switch_L_Status']['Value'] and values[neighbor_name]['Switch_R_Status']['Value']:
|
|
if neighbor_name not in visited:
|
|
q.append( (neighbor_name, source))
|
|
|
|
# 5. Prepare and write the results back to Ignition tags
|
|
tag_paths_to_write = []
|
|
values_to_write = []
|
|
|
|
# Set UT states
|
|
idx = 0
|
|
for name in ut_names:
|
|
is_energized = name in energized_state
|
|
power_source = energized_state.get(name, 'None')
|
|
tag_paths_to_write.append(providers[idx] + name + "/Data/isEnergized")
|
|
values_to_write.append(is_energized)
|
|
tag_paths_to_write.append(providers[idx] + name + "/Data/PowerSource")
|
|
values_to_write.append(power_source)
|
|
idx += 1
|
|
|
|
line_tags_base_path = "[Ignition_Common_IO_Gtwy]Cables/UT/"
|
|
#print energized_state
|
|
# Loop through each connection between UTs
|
|
for i in range(len(ut_names) - 1):
|
|
ut1_name = ut_names[i]
|
|
ut2_name = ut_names[i+1]
|
|
line_energized = (ut1_name in energized_state and
|
|
ut2_name in energized_state and
|
|
#energized_state[ut1_name] == energized_state[ut2_name] and
|
|
values[ut1_name]['Switch_R_Status']['Value'] and
|
|
values[ut2_name]['Switch_L_Status']['Value'])
|
|
#print ut1_name + " : " + str(ut1_name in energized_state) +" & "+str(values[ut1_name]['Switch_R_Status']['Value'])
|
|
#print energized_state[ut1_name] + " = " + energized_state[ut2_name]
|
|
#print ut2_name + " : " + str(ut2_name in energized_state) +" & "+str(values[ut1_name]['Switch_L_Status']['Value'])
|
|
line_tag_name = "Line_{}_to_{}".format(ut1_name, ut2_name)
|
|
# Construct the full tag path directly to the boolean tag
|
|
full_tag_path = line_tags_base_path + line_tag_name
|
|
# Add the path and the calculated value to our list for writing
|
|
tag_paths_to_write.append(full_tag_path)
|
|
values_to_write.append(line_energized)
|
|
#print full_tag_path + " : " + str(line_energized)
|
|
|
|
# Write all values in a single, efficient call
|
|
if tag_paths_to_write:
|
|
system.tag.writeBlocking(tag_paths_to_write, values_to_write) |