import json import re import os import sys # ============================================================================ # DEVICE RULES FOR NAME GENERATION # ============================================================================ DEVICE_RULES = [ # ------------------------- # Station devices # ------------------------- { "name": "EPC", "pattern": r'_EPC\d+$', "path": "System/{system}/Station/EPC/{id}", "strip_suffix": "_Assembly", }, { "name": "JR", "pattern": r'_JR\d+$', "path": "System/{system}/Station/JR/{id}", }, { "name": "SS_PB", "pattern": r'_SS\d+$', "path": "System/{system}/Station/SS_PB/{id}", }, { "name": "S_PB", "pattern": r'_S\d+$', "path": "System/{system}/Station/S_PB/{id}", }, # ------------------------- # Photoeyes # ------------------------- { "name": "PE", "pattern": r'_(TPE|PE|JPE|FPE)\d+$', "path": "System/{system}/PE/{id}", }, # ------------------------- # Conveyor # ------------------------- { "name": "VFD", "pattern": r'_VFD\d*$', "path": "System/{system}/Conveyor/VFD/{id}", }, { "name": "EXTENDO", "pattern": r'_EX\d+$', "path": "System/{system}/Conveyor/EXTENDO/{id}", }, # ------------------------- # IO blocks # ------------------------- { "name": "FIO", "pattern": r'_FIO\d+$', "path": "System/{system}/IO_BLOCK/FIO/{id}", }, { "name": "FIOH", "pattern": r'_FIOH\d+$', "path": "System/{system}/IO_BLOCK/FIOH/{id}", }, { "name": "SIO", "pattern": r'_SIO\d+$', "path": "System/{system}/IO_BLOCK/SIO/{id}", }, { "name": "DPM", "pattern": r'_DPM\d+$', "path": "System/{system}/IO_BLOCK/DPM/{id}", }, # ------------------------- # Encoder # ------------------------- { "name": "ENCODER", "pattern": r'_ENCODER$', "path": "System/{system}/ENCODER/{id}", "case_insensitive": True, }, # ------------------------- # Chutes & sorter # ------------------------- { "name": "CHUTE", "pattern": r'^Chute_\d+$', "path": "System/SMC/Chute/{id}", }, { "name": "SORTER", "pattern": r'sorter', "path": "System/SMC/Sorter", "contains": True, }, # ------------------------- # MCM root : MCM and PMM # ------------------------- { "name": "MCM", "pattern": r'^MCM\d+$', "path": "System/{system}/{id}", }, { "name": "PMM", "pattern": r'_PMM\d+$', "path": "System/{system}/{id}", }, { "name": "CONVEYOR", "pattern": r'^[A-Z0-9_]+$', "path": "System/{system}/CMC/Conveyors/{id}", # Extra guards (optional, declarative) "requires": [ {"contains": "_"}, {"has_digit": True}, ], }, ] # ============================================================================ # MODIFY THIS LINE ACCORDING TO YOUR ENVIRONMENT # ============================================================================ BASE_NEW = r"C:\Program Files\Inductive Automation\Ignition\data\projects\TPA8_SCADA\com.inductiveautomation.perspective\views\DetailedView" # Binding Configurations STATE_BINDING_TEMPLATE = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/State", "references": { "0": "{this.%s.name}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},999)", "type": "expression" }, { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": 0, "output": "Closed"}, {"input": 1, "output": "Actuated"}, {"input": 2, "output": "Communication Faulted"}, {"input": 3, "output": "Conveyor Running In Maintenance Mode"}, {"input": 4, "output": "Disabled"}, {"input": 5, "output": "Disconnected"}, {"input": 6, "output": "Stopped"}, {"input": 7, "output": "Enabled Not Running"}, {"input": 8, "output": "Encoder Fault"}, {"input": 9, "output": "Energy Management"}, {"input": 10, "output": "ESTOP Was Actuated"}, {"input": 11, "output": "EStopped"}, {"input": 12, "output": "EStopped Locally"}, {"input": 13, "output": "Extended Faulted"}, {"input": 14, "output": "Full"}, {"input": 15, "output": "Gaylord Start Pressed"}, {"input": 16, "output": "Jam Fault"}, {"input": 17, "output": "Jammed"}, {"input": 18, "output": "Loading Allowed"}, {"input": 19, "output": "Loading Not Allowed"}, {"input": 20, "output": "Low Air Pressure Fault Was Present"}, {"input": 21, "output": "Maintenance Mode"}, {"input": 22, "output": "Conveyor Stopped In Maintenance Mode"}, {"input": 23, "output": "Motor Faulted"}, {"input": 24, "output": "Motor Was Faulted"}, {"input": 25, "output": "Normal"}, {"input": 26, "output": "Off Inactive"}, {"input": 27, "output": "Open"}, {"input": 28, "output": "PLC Ready To Run"}, {"input": 29, "output": "Package Release Pressed"}, {"input": 30, "output": "Power Branch Was Faulted"}, {"input": 31, "output": "Pressed"}, {"input": 32, "output": "Ready To Receive"}, {"input": 33, "output": "Running"}, {"input": 34, "output": "Started"}, {"input": 35, "output": "Stopped"}, {"input": 36, "output": "System Started"}, {"input": 37, "output": "Unknown"}, {"input": 38, "output": "VFD Fault"}, {"input": 39, "output": "Conveyor Running In Power Saving Mode"}, {"input": 40, "output": "Conveyor Jogging In Maintenance Mode"}, {"input": 41, "output": "VFD Reset Required"}, {"input": 42, "output": "Jam Reset Push Button Pressed"}, {"input": 43, "output": "Start Push Button Pressed"}, {"input": 44, "output": "Stop Push Button Pressed"}, {"input": 45, "output": "No Container"}, {"input": 46, "output": "Ready To Be Enabled"}, {"input": 47, "output": "Half Full"}, {"input": 48, "output": "Enabled"}, {"input": 49, "output": "Tipper Faulted"}, {"input": 50, "output": "Diverted"}, {"input": 51, "output": "Retracted"}, {"input": 52, "output": "Diverting"}, {"input": 66, "output": "Inch And Store Mode"} ], "fallback": "Unknown", "type": "map" } ] } } PRIORITY_BINDING_TEMPLATE = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Priority", "references": { "0": "{this.%s.name}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},0)", "type": "expression" }, { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": 0, "output": "No Active Alarms"}, {"input": 1, "output": "High"}, {"input": 2, "output": "Medium"}, {"input": 3, "output": "Low"}, {"input": 4, "output": "Diagnostic"} ], "fallback": "No Active Alarms", "type": "map" } ] } } COLOR_BINDING_TEMPLATE = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Color", "references": { "0": "{this.%s.name}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},999)", "type": "expression" }, { "inputType": "scalar", "outputType": "color", "mappings": [ {"input": 0, "output": "#C2C2C2"}, {"input": 1, "output": "#FF0000"}, {"input": 2, "output": "#FFA500"}, {"input": 3, "output": "#0008FF"}, {"input": 4, "output": "#00FF00"}, {"input": 5, "output": "#FFF700"}, {"input": 6, "output": "#87CEEB"}, {"input": 7, "output": "#90EE90"}, {"input": 8, "output": "#964B00"}, {"input": 9, "output": "#FFFFFF"}, {"input": 10, "output": "#000000"}, {"input": 11, "output": "#8B0000"}, {"input": 12, "output": "#808080"}, {"input": 13, "output": "#8B8000"}, {"input": 14, "output": "#006400"}, {"input": 15, "output": "#90FF90"}, {"input": 16, "output": "#00008B"}, {"input": 17, "output": "#FF7276"}, {"input": 18, "output": "#556B2F"}, {"input": 19, "output": "#B43434"}, {"input": 20, "output": "#4682B4"}, {"input": 21, "output": "#FFD700"} ], "fallback": "#000000", "type": "map" } ] } } INDUCTION_COLOR_SCRIPT = """ if value is None: return '#000000' # Priority-based status determination (lower number = higher priority) active_status = 'default' # Highest-priority checks first if value.get('awInduction', 1) == 0: active_status = 'inactive' elif value.get('Common_Error', False): active_status = 'common_error' elif value.get('In_Test_Mode', False): active_status = 'test_mode' elif value.get('Jammed', False): active_status = 'jammed' elif value.get('Energy_Saving', False): active_status = 'energy_saving' elif value.get('Running', False): active_status = 'running' elif value.get('Starting', False) or value.get('Stopping', False): active_status = 'transitional' elif value.get('Stopped', False) or value.get('Disabled', False): active_status = 'inactive' elif value.get('Blocked', False): active_status = 'blocked' # Dictionary-based switch (cleaner than elif chain) color_map = { 'common_error': '#FF0000', 'inactive': '#C2C2C2', 'blocked': '#C2C2C2', 'running': '#00FF00', 'transitional': '#90FF90', 'jammed': '#FF8C00', 'energy_saving': '#87CEFA', 'test_mode': '#AC5F00', 'default': '#000000' } # Return color using dictionary lookup (switch-like behavior) return color_map.get(active_status, '#000000')""" INDUCTION_PRIORITY_SCRIPT = """ data = dict(value) if value else {} if data.get('Common_Error') or data.get('Jammed') : return "High" elif data.get('Blocked'): return "High" elif data.get('Test_Mode'): return "Medium" elif data.get('Disabled'): return "Low" elif data.get('Starting') or data.get('Stopping') or data.get('Running') or data.get('Energy_Saving') or data.get('Stopped') or data.get('Disabled') == False: return "No Active Alarms" return "No Active Alarms" """ INDUCTION_STATE_SCRIPT = """ data = dict(value) if value else {} if data.get('awInduction', 1) == 0: return 'Not Running' elif data.get('Common_Error'): return 'Common Error' elif data.get('Jammed'): return 'Jammed' elif data.get('Test_Mode'): return 'In Test Mode' elif data.get('Energy_Saving'): return 'Energy Saving' elif data.get('Running'): return 'Running' elif data.get('Starting'): return 'Starting' elif data.get('Stopping'): return 'Stopping' elif data.get('Stopped'): return 'Stopped' elif data.get('Disabled') == False: return 'Enabled' elif data.get('Blocked'): return 'Blocked' else: return 'Not Running' """ CONVEYOR_COLOR_SCRIPT = """ if value is None: return '#000000' # Determine conveyor status (priority order) active_status = 'default' if value.get('E_Stop', False): active_status = 'e_stop' elif value.get('Technical_Fault', False): active_status = 'technical_fault' elif value.get('Operational_Fault', False): active_status = 'operational_fault' elif value.get('Dieback', False): active_status = 'dieback' elif value.get('Running', False): active_status = 'running' elif value.get('Energy_Saving', False): active_status = 'energy_saving' elif value.get('Stopped', False): active_status = 'stopped' elif value.get('awConveyor', 1) == 0: active_status = 'stopped' # Color mapping color_map = { 'e_stop': '#FF0000', # Red — Emergency stop 'technical_fault': '#FF0000', # Red— Technical fault 'operational_fault': '#FF8C00',# Orange — Operational fault 'dieback': '#90FF90', # Light green — Dieback (line full) 'running': '#00FF00', # Green — Running 'energy_saving': '#87CEFA', # Light blue — Energy saving 'stopped': '#C2C2C2', # Dark gray — Inactive 'default': '#000000' # Black — Default } return color_map.get(active_status, '#000000')""" CONVEYOR_PRIORITY_SCRIPT = """ data = dict(value) if value else {} if data.get('E_Stop') or data.get('Technical_Fault') or data.get('Operational_Fault'): return "High" elif data.get('Running') or data.get('Energy_Saving') or data.get('Stopped') or data.get('Dieback'): return "No Active Alarms" return "No Active Alarms" """ CONVEYOR_STATE_SCRIPT = """ data = dict(value) if value else {} if data.get('E_Stop'): return 'Estopped' elif data.get('Technical_Fault'): return 'Faulted' elif data.get('Operational_Fault'): return 'Jammed' elif data.get('Dieback'): return 'Dieback' elif data.get('Running'): return 'Running' elif data.get('Energy_Saving'): return 'Energy Saving' elif data.get('Stopped'): return 'Stopped' elif data.get('awConveyor', 1) == 0: return 'Not Running' else: return 'Not Running'""" # Chute scripts for Chute_{nnn} elements CHUTE_COLOR_SCRIPT = """ data = dict(value) if value else {} if data.get("Jammed"): return "#FFA500" elif data.get("Full"): return "#0000FF" elif data.get("Half_Full"): return "#FFFF00" elif data.get("No_Container"): return "#C2C2C2" elif data.get("Blocked_By_Operator"): return "#C2C2C2" elif data.get("Blocked_From_SCADA"): return "#C2C2C2" elif data.get("Disabled") == False: return "#00FF00" else: return "#000000" """ CHUTE_PRIORITY_SCRIPT = """ data = dict(value) if value else {} if data.get("Jammed"): return "High" elif data.get("Full") or data.get("Half_Full") or data.get("No_Container") or data.get("Blocked_From_SCADA") or data.get("Blocked_By_Operator"): return "Low" elif data.get("Disabled") == False: return "No Active Alarms" return "No Active Alarms" """ CHUTE_STATE_SCRIPT = """ data = dict(value) if value else {} if data.get("Jammed"): return "Jammed" elif data.get("Full"): return "Full" elif data.get("Half_Full"): return "Half Full" elif data.get("No_Container"): return "No Container" elif data.get("Blocked_By_Operator"): return "Blocked By Operator" elif data.get("Blocked_From_SCADA"): return "Blocked From SCADA" elif data.get("Disabled") == False: return "Enabled" else: return "Inactive" """ # Sorter scripts SORTER_COLOR_SCRIPT = """ if value is None: return '#000000' active_status = 'default' if value.get('awInduction', 1) == 0: active_status = 'inactive' elif value.get('Common_Error', False): active_status = 'common_error' elif value.get('In_Test_Mode', False): active_status = 'test_mode' elif value.get('Jammed', False): active_status = 'jammed' elif value.get('Energy_Saving', False): active_status = 'energy_saving' elif value.get('Running', False): active_status = 'running' elif value.get('Starting', False) or value.get('Stopping', False): active_status = 'transitional' elif value.get('Stopped', False) or value.get('Disabled', False): active_status = 'inactive' elif value.get('Blocked', False): active_status = 'blocked' color_map = { 'common_error': '#FF0000', 'inactive': '#C2C2C2', 'blocked': '#C2C2C2', 'running': '#00FF00', 'transitional': '#90FF90', 'jammed': '#FF8C00', 'energy_saving': '#87CEFA', 'test_mode': '#AC5F00', 'default': '#000000' } return color_map.get(active_status, '#000000') """ SORTER_PRIORITY_SCRIPT = """ data = dict(value) if value else {} if data.get('Common_Error'): return "High" elif data.get('Test_Mode') or data.get('Discharge_Test_Mode') or data.get('Lamp_Test_Mode'): return "Medium" elif data.get('Disabled'): return "Low" elif data.get('Starting') or data.get('Stopping') or data.get('Running') or data.get('Energy_Saving') or data.get('Stopped'): return "No Active Alarms" return "No Active Alarms" """ SORTER_STATE_SCRIPT = """ data = dict(value) if value else {} if data.get('wSorter', 1) == 0: return 'Not Running' if data.get('Common_Error'): return 'Common Error' elif data.get('Test_Mode'): return 'Test Mode' elif data.get('Discharge_Test_Mode'): return 'Discharge Test Mode' elif data.get('Lamp_Test_Mode'): return 'Lamp Test Mode' elif data.get('Energy_Saving'): return 'Energy Saving' elif data.get('Running'): return 'Running' elif data.get('Starting'): return 'Starting' elif data.get('Stopping'): return 'Stopping' elif data.get('Stopped'): return 'Stopped' elif data.get('Disabled'): return 'Disabled' elif data.get('Blocked'): return 'Blocked' return 'Not Running' """ INDUCTION_BINDING_TEMPLATE = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}", "references": { "0": "{this.%s.name}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "code": "", "type": "script" } ] } } DISPLAY_BINDING_TEMPLATE = { "binding": { "type": "expr", "config": { "expression": "" }, "transforms": [ { "inputType": "scalar", "outputType": "scalar", "mappings": [ { "input": False, "output": "none" } ], "fallback": "block", "type": "map" } ] } } INDUCTION_DATA = { "Induction_0": { "d": "m 0,0 h 53.89 l -0.88,0.88 13.47,13.47 32.13,-32.13 -14.08,-14.08 -12.87,12.87 h -71.71 z", "transform": "translate(1411.47,855.11) translate(-0.01,-0.05)" }, "Induction_1": { "d": "m 0,0 h 53.89 l -0.88,0.88 13.47,13.47 32.13,-32.13 -14.08,-14.08 -12.87,12.87 h -71.71 z", "transform": "translate(1413.24,784.98) translate(-0.01,-0.05)" }, "Induction_14": { "d": "m 0,0 h -53.89 l 0.88,0.88 -13.47,13.47 -32.13,-32.13 14.08,-14.08 12.87,12.87 h 71.71 z", "transform": "translate(1600.55,564.1) translate(-0.01,-0.05)" }, "Induction_17": { "d": "m 0,0 h -53.89 l 0.88,0.88 -13.47,13.47 -32.13,-32.13 14.08,-14.08 12.87,12.87 h 71.71 z", "transform": "translate(1487.17,484.1) translate(-0.01,-0.05)" } } def rule_matches(rule, core): # Regex match flags = re.IGNORECASE if rule.get("case_insensitive") else 0 if not re.search(rule["pattern"], core, flags): return False # Optional constraints for req in rule.get("requires", []): if "contains" in req and req["contains"] not in core: return False if req.get("has_digit") and not any(c.isdigit() for c in core): return False return True def generate_name(identifier, system_name): if not identifier: return None core = identifier for rule in DEVICE_RULES: # Optional suffix stripping (EPC _Assembly etc.) if rule.get("strip_suffix") and core.endswith(rule["strip_suffix"]): core = core[: -len(rule["strip_suffix"])] if rule_matches(rule, core): return rule["path"].format( system=system_name, id=core ) return None # ============================================================================ # JSON HELPERS # ============================================================================ def find_svg_component(node): if isinstance(node, dict): if node.get('type') == 'ia.shapes.svg': return node for key in ('children', 'root'): child = node.get(key) if isinstance(child, list): for c in child: found = find_svg_component(c) if found: return found elif isinstance(child, dict): found = find_svg_component(child) if found: return found return None def extract_mcm_id(folder_name): """ Extract MCM ID from folder name (e.g., "MCM04" from "MCM04 North Bulk Inbound...") Uses first 5 characters or extracts MCM pattern (MCM followed by digits) """ # Try to extract MCM pattern (MCM followed by digits) match = re.search(r'(MCM\d+)', folder_name, re.IGNORECASE) if match: return match.group(1).upper() # Fallback: use first 5 characters return folder_name[:5].upper() def load_json(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error decoding JSON from {filepath}: {e}") return None def update_scada_svg(scada_view_data, system_name): count_updated = 0 prop_config = {} svg_comp = find_svg_component(scada_view_data) if svg_comp: # Clear existing propConfig to remove old/nested bindings svg_comp['propConfig'] = {} prop_config = svg_comp['propConfig'] if 'props' in svg_comp and 'elements' in svg_comp['props']: # Return value is the number of TOP LEVEL SVG elements we processed # (more meaningful than guessing from propConfig length, since some elements # produce more than 3 bindings, e.g. SS buttons). count_updated = traverse_elements( svg_comp['props']['elements'], "props.elements", prop_config, is_top_level=True, system_name=system_name ) return count_updated def traverse_elements(elements, current_path, prop_config, is_top_level, system_name): processed_count = 0 for i, el in enumerate(elements): element_path = f"{current_path}[{i}]" # Cleanup nested properties if they exist if not is_top_level: for key in ['state', 'priority', 'color']: if key in el: del el[key] # Only process bindings for TOP LEVEL items if is_top_level: # Skip defs if el.get('type') == 'defs' or el.get('name') == 'defs1': continue # Update Name Logic # Prefer SVG 'id' when present; otherwise use the leaf of the existing name. # (Existing name can be a full tag path like "System/MCM01/..." which we do NOT # want to feed into regexes.) name = el.get('name') el_id = el.get('id') name_leaf = name.split('/')[-1] if isinstance(name, str) and name else "" identifier = el_id or name_leaf or "" # ------------------------------------------------------------------ # NAME GENERATION (NO OLD VIEW DEPENDENCY) # ------------------------------------------------------------------ new_name = generate_name(identifier, system_name) if new_name and el.get('name') != new_name: el['name'] = new_name # Refresh name after modification name = el.get('name') # Determine check name (strip _Assembly for pattern matching) check_name = identifier or "" if check_name.endswith('_Assembly'): check_name = check_name[:-9] # Convert Induction elements from use to path if identifier in INDUCTION_DATA: el['type'] = 'path' el['d'] = INDUCTION_DATA[identifier]['d'] el['transform'] = INDUCTION_DATA[identifier]['transform'] el['fill'] = {"paint": "#ffffff", "opacity": "1"} el['stroke'] = {"paint": "#000000", "dasharray": "none", "opacity": "1", "width": "1"} for key in ['x', 'y', 'href']: if key in el: del el[key] # 2. Add Bindings based on Type # Helper to check pattern in both check_name (ID) and name def matches_pattern(pattern): """Check if pattern matches in either check_name or name""" return re.search(pattern, check_name) or (name and re.search(pattern, name)) def name_contains(substring): """Check if name contains a substring""" return name and substring in name def name_or_id_startswith(prefix): """Check if name or ID starts with prefix""" name_part = name.split('/')[-1] if name else "" return check_name.startswith(prefix) or name_part.startswith(prefix) def name_or_id_endswith(suffix): """Check if name or ID ends with suffix""" return check_name.endswith(suffix) or (name and name.endswith(suffix)) # Helper to create binding def create_binding(template, subpath=""): b = json.loads(json.dumps(template)) ref_key = list(b['binding']['config']['references'].keys())[0] b['binding']['config']['references'][ref_key] = b['binding']['config']['references'][ref_key] % element_path # If subpath provided, insert it into tagPath # Template tagPath: "[{fc}_SCADA_TAG_PROVIDER]{0}/State" (or /Color, /Priority) # We want to replace "/State" with "/Subpath/State" if subpath is "Start" if subpath: tp = b['binding']['config']['tagPath'] # Split by last slash base, suffix = tp.rsplit('/', 1) b['binding']['config']['tagPath'] = f"{base}/{subpath}/{suffix}" return b # Special FL_1CH_PE1 elements (starts with FL and ends with _1CH_PE1) # Check both ID and name is_fl_1ch_pe1 = (name_or_id_startswith('FL') and name_or_id_endswith('_1CH_PE1')) if is_fl_1ch_pe1: # Generate name2 by adding "J" before "PE1" in the name name2 = name.replace('_PE1', '_JPE1') if name else "" el['name2'] = name2 el['full'] = False el['jammed'] = False # Full binding (uses name) prop_config[f"{element_path}.full"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Full", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},false)", "type": "expression" } ] } } # Jammed binding (uses name2) prop_config[f"{element_path}.jammed"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Jammed", "references": { "0": f"{{this.{element_path}.name2}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},false)", "type": "expression" } ] } } # Color tag binding with /Full path and expression transform color_expr = f'''if( !isGood({{value}}), "#000000", if({{this.{element_path}.full}} && {{this.{element_path}.jammed}}, "#FFA500", if({{this.{element_path}.jammed}}, "#FFA500", if({{this.{element_path}.full}} , "#0008FF", "#FFFFFF" ) ) ) )''' prop_config[f"{element_path}.color"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Full", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": color_expr, "type": "expression" } ] } } # Priority tag binding with /Full path and expression transform priority_expr = f'''if( !isGood({{value}}), "No Active Alarms", if({{this.{element_path}.full}} && {{this.{element_path}.jammed}}, "High", if({{this.{element_path}.jammed}}, "High", if({{this.{element_path}.full}} , "Low", "No Active Alarms" ) ) ) )''' prop_config[f"{element_path}.priority"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Full", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": priority_expr, "type": "expression" } ] } } # State tag binding with /Full path and expression transform state_expr = f'''if( !isGood({{value}}), "Unknown", if({{this.{element_path}.full}} && {{this.{element_path}.jammed}}, "Jammed and Full", if({{this.{element_path}.jammed}}, "Jammed", if({{this.{element_path}.full}} , "Full", "Normal" ) ) ) )''' prop_config[f"{element_path}.state"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Full", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": state_expr, "type": "expression" } ] } } el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_SS\d+$'): # Special SS Buttons: Start/Stop properties # Start prop_config[f"{element_path}.start_state"] = create_binding(STATE_BINDING_TEMPLATE, "Start") prop_config[f"{element_path}.start_priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "Start") prop_config[f"{element_path}.start_color"] = create_binding(COLOR_BINDING_TEMPLATE, "Start") # Stop prop_config[f"{element_path}.stop_state"] = create_binding(STATE_BINDING_TEMPLATE, "Stop") prop_config[f"{element_path}.stop_priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "Stop") prop_config[f"{element_path}.stop_color"] = create_binding(COLOR_BINDING_TEMPLATE, "Stop") el['start_state'] = "Normal" el['start_priority'] = "No Active Alarms" el['start_color'] = "#006400" el['stop_state'] = "Normal" el['stop_priority'] = "No Active Alarms" el['stop_color'] = "#B43434" elif matches_pattern(r'_S\d+$') and name_contains('/DIV/'): # S buttons with DIV in name -> /Enable/ tagPath prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE, "Enable") prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "Enable") prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE, "Enable") el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_S\d+$'): # S buttons -> /Start/ tagPath prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE, "Start") prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "Start") prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE, "Start") el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_JR\d+$') and name_contains('Chute_JR'): # JR buttons with Chute_JR in name -> /Chute_JR/ tagPath prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE, "Chute_JR") prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "Chute_JR") prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE, "Chute_JR") el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_JR\d+$'): # JR buttons -> /JR/ tagPath prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE, "JR") prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "JR") prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE, "JR") el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_GS\d+$'): # GS buttons -> /Commands/bBlockHost1 tagPath with custom mappings # Color binding prop_config[f"{element_path}.color"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Commands/bBlockHost1", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},\"err\")", "type": "expression" }, { "inputType": "scalar", "outputType": "color", "mappings": [ {"input": True, "output": "#00FF00"}, {"input": False, "output": "#00A700"} ], "fallback": "#000000", "type": "map" } ] } } # Priority binding prop_config[f"{element_path}.priority"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Commands/bBlockHost1", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},\"err\")", "type": "expression" }, { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": True, "output": "Low"}, {"input": False, "output": "No Active Alarms"} ], "fallback": "No Active Alarms", "type": "map" } ] } } # State binding prop_config[f"{element_path}.state"] = { "binding": { "type": "tag", "config": { "mode": "indirect", "tagPath": "[{fc}_SCADA_TAG_PROVIDER]{0}/Commands/bBlockHost1", "references": { "0": f"{{this.{element_path}.name}}", "fc": "{session.custom.fc}" }, "fallbackDelay": 2.5 }, "transforms": [ { "expression": "coalesce({value},\"err\")", "type": "expression" }, { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": True, "output": "Enable Pressed"}, {"input": False, "output": "Enable Not Pressed"} ], "fallback": "Unknown", "type": "map" } ] } } el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_PR\d+$'): # PR buttons -> /PR/ tagPath prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE, "PR") prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE, "PR") prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE, "PR") el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'Induction_\d+$'): # Induction logic: Script Transforms on Base Tag Path # Color c_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(c_bind['binding']['config']['references'].keys())[0] c_bind['binding']['config']['references'][ref_key] = c_bind['binding']['config']['references'][ref_key] % element_path c_bind['binding']['transforms'][0]['code'] = INDUCTION_COLOR_SCRIPT prop_config[f"{element_path}.color"] = c_bind # Priority p_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(p_bind['binding']['config']['references'].keys())[0] p_bind['binding']['config']['references'][ref_key] = p_bind['binding']['config']['references'][ref_key] % element_path p_bind['binding']['transforms'][0]['code'] = INDUCTION_PRIORITY_SCRIPT prop_config[f"{element_path}.priority"] = p_bind # State s_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(s_bind['binding']['config']['references'].keys())[0] s_bind['binding']['config']['references'][ref_key] = s_bind['binding']['config']['references'][ref_key] % element_path s_bind['binding']['transforms'][0]['code'] = INDUCTION_STATE_SCRIPT prop_config[f"{element_path}.state"] = s_bind el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif (name_contains('/CMC/Conveyors/') and not matches_pattern(r'_(VFD|EX)\d*$')): # Special Conveyor logic (System/CMC/Conveyors/...) # Color c_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(c_bind['binding']['config']['references'].keys())[0] c_bind['binding']['config']['references'][ref_key] = c_bind['binding']['config']['references'][ref_key] % element_path c_bind['binding']['transforms'][0]['code'] = CONVEYOR_COLOR_SCRIPT prop_config[f"{element_path}.color"] = c_bind # Priority p_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(p_bind['binding']['config']['references'].keys())[0] p_bind['binding']['config']['references'][ref_key] = p_bind['binding']['config']['references'][ref_key] % element_path p_bind['binding']['transforms'][0]['code'] = CONVEYOR_PRIORITY_SCRIPT prop_config[f"{element_path}.priority"] = p_bind # State s_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(s_bind['binding']['config']['references'].keys())[0] s_bind['binding']['config']['references'][ref_key] = s_bind['binding']['config']['references'][ref_key] % element_path s_bind['binding']['transforms'][0]['code'] = CONVEYOR_STATE_SCRIPT prop_config[f"{element_path}.state"] = s_bind el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif el_id and 'CH' in el_id.upper() and name and 'CH' in name.upper() and not re.search(r'_(FPE|JPE|TPE|PE)\d+', el_id) and not re.search(r'^Chute_\d+$', el_id): # CH elements where both ID and name contain "CH" (but ID has no PE pattern): chute bindings # Excludes Chute_{nnn} elements which have their own handling # Uses standard color expression coalesce({value},999) prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE) prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE) prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE) el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif el_id and 'CH' in el_id.upper() and name and (re.search(r'_(FPE|JPE|TPE|PE)\d+', name) or '/PE/' in name) and not re.search(r'^Chute_\d+$', el_id): # CH (Chute) elements with PE-related names prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE) prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE) # If ID also has PE pattern, use coalesce({value},0); otherwise use standard coalesce({value},999) if re.search(r'_(FPE|JPE|TPE|PE)\d+', el_id): c_bind = create_binding(COLOR_BINDING_TEMPLATE) c_bind['binding']['transforms'][0]['expression'] = "coalesce({value},0)" prop_config[f"{element_path}.color"] = c_bind else: prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE) el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'_(TPE|JPE|FPE|PE)\d+$') or matches_pattern(r'_DPM\d+$'): # PE and DPM: Color expression coalesce({value},0) prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE) prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE) c_bind = create_binding(COLOR_BINDING_TEMPLATE) c_bind['binding']['transforms'][0]['expression'] = "coalesce({value},0)" prop_config[f"{element_path}.color"] = c_bind el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif matches_pattern(r'Chute_\d+'): # Chute elements: Script-based bindings with base tag path (no sub-property) # Uses INDUCTION_BINDING_TEMPLATE which has tagPath without /State, /Priority, /Color # Color c_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(c_bind['binding']['config']['references'].keys())[0] c_bind['binding']['config']['references'][ref_key] = c_bind['binding']['config']['references'][ref_key] % element_path c_bind['binding']['transforms'][0]['code'] = CHUTE_COLOR_SCRIPT prop_config[f"{element_path}.color"] = c_bind # Priority p_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(p_bind['binding']['config']['references'].keys())[0] p_bind['binding']['config']['references'][ref_key] = p_bind['binding']['config']['references'][ref_key] % element_path p_bind['binding']['transforms'][0]['code'] = CHUTE_PRIORITY_SCRIPT prop_config[f"{element_path}.priority"] = p_bind # State s_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(s_bind['binding']['config']['references'].keys())[0] s_bind['binding']['config']['references'][ref_key] = s_bind['binding']['config']['references'][ref_key] % element_path s_bind['binding']['transforms'][0]['code'] = CHUTE_STATE_SCRIPT prop_config[f"{element_path}.state"] = s_bind el['state'] = "value" el['priority'] = "value" el['color'] = "value" elif check_name and 'sorter' in check_name.lower(): # Sorter elements: Script-based bindings with base tag path (no sub-property) # Color c_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(c_bind['binding']['config']['references'].keys())[0] c_bind['binding']['config']['references'][ref_key] = c_bind['binding']['config']['references'][ref_key] % element_path c_bind['binding']['transforms'][0]['code'] = SORTER_COLOR_SCRIPT prop_config[f"{element_path}.color"] = c_bind # Priority p_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(p_bind['binding']['config']['references'].keys())[0] p_bind['binding']['config']['references'][ref_key] = p_bind['binding']['config']['references'][ref_key] % element_path p_bind['binding']['transforms'][0]['code'] = SORTER_PRIORITY_SCRIPT prop_config[f"{element_path}.priority"] = p_bind # State s_bind = json.loads(json.dumps(INDUCTION_BINDING_TEMPLATE)) ref_key = list(s_bind['binding']['config']['references'].keys())[0] s_bind['binding']['config']['references'][ref_key] = s_bind['binding']['config']['references'][ref_key] % element_path s_bind['binding']['transforms'][0]['code'] = SORTER_STATE_SCRIPT prop_config[f"{element_path}.state"] = s_bind el['state'] = "value" el['priority'] = "value" el['color'] = "value" else: # Generic Devices prop_config[f"{element_path}.state"] = create_binding(STATE_BINDING_TEMPLATE) prop_config[f"{element_path}.priority"] = create_binding(PRIORITY_BINDING_TEMPLATE) prop_config[f"{element_path}.color"] = create_binding(COLOR_BINDING_TEMPLATE) el['state'] = "value" el['priority'] = "value" el['color'] = "value" # 3. Add Specific Inner Bindings (Style/Paint bindings) AND Display Bindings # Helper to check if a property path exists in the element def property_exists(target_subpath): """Check if the property path exists in the element structure""" parts = target_subpath.split('.') current = el for part in parts: # Handle array access like elements[0] if '[' in part: key, index_str = part.split('[') index = int(index_str.rstrip(']')) if key not in current or not isinstance(current[key], list): return False if index >= len(current[key]): return False current = current[key][index] else: if part not in current: return False current = current[part] # Check if the final property exists and has a value return current is not None # Define helper to add property binding (only if property exists) def add_prop_binding(target_subpath, source_prop="color"): if not property_exists(target_subpath): return # Skip if property doesn't exist binding_key = f"{element_path}.{target_subpath}" prop_config[binding_key] = { "binding": { "type": "property", "config": { "path": f"this.{element_path}.{source_prop}" } } } DISABLED_DISPLAY_FLAGS = { "show_chutes", "show_running", } # Helper to add display binding def add_display_binding(flag_name, priority_prop="priority", is_ss=False): # Skip disabled display flags if flag_name in DISABLED_DISPLAY_FLAGS: return d_bind = json.loads(json.dumps(DISPLAY_BINDING_TEMPLATE)) path_to_elements = f"this.{element_path}" path_to_priority = f"{{{path_to_elements}.{priority_prop}}}" path_to_flag = f"{{session.custom.devices_filter.{flag_name}}}" if is_ss: path_to_start = f"{{{path_to_elements}.start_priority}}" path_to_stop = f"{{{path_to_elements}.stop_priority}}" expression = f"({path_to_start} = 'High' || {path_to_stop} = 'High') || {path_to_flag}" else: expression = f"({path_to_priority} = 'High') || {path_to_flag}" d_bind['binding']['config']['expression'] = expression prop_config[f"{element_path}.style.display"] = d_bind if 'style' not in el: el['style'] = {"classes": ""} el['style']['display'] = "block" # Special FL_1CH_PE1 elements if is_fl_1ch_pe1: # fill.paint bound to color add_prop_binding("fill.paint") # Custom style.display with show_chutes display_expr = f"({{this.{element_path}.priority}} = 'High') || {{session.custom.alarm_filter.show_chutes}}" prop_config[f"{element_path}.style.display"] = { "binding": { "type": "expr", "config": { "expression": display_expr }, "transforms": [ { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": False, "output": "none"} ], "fallback": "block", "type": "map" } ] } } if 'style' not in el: el['style'] = {"classes": ""} el['style']['display'] = "block" elif matches_pattern(r'_EPC\d+$'): add_prop_binding("elements[0].stroke.paint") add_prop_binding("elements[3].fill.paint") add_display_binding("show_safety") elif name and 'Diverter' in name: # Diverter elements - bind elements[1], [2], [3] fill.paint and custom style.display add_prop_binding("elements[1].fill.paint") add_prop_binding("elements[2].fill.paint") add_prop_binding("elements[3].fill.paint") # Custom style.display for element[1]: hide when state is "Diverted", show otherwise prop_config[f"{element_path}.elements[1].style.display"] = { "binding": { "type": "property", "config": { "path": f"this.{element_path}.state" }, "transforms": [ { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": "Diverted", "output": "none"} ], "fallback": "block", "type": "map" } ] } } # Custom style.display for element[2]: show when state is "Diverted", hide otherwise prop_config[f"{element_path}.elements[2].style.display"] = { "binding": { "type": "property", "config": { "path": f"this.{element_path}.state" }, "transforms": [ { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": "Diverted", "output": "block"} ], "fallback": "none", "type": "map" } ] } } # Custom style.display for element[3]: show when state is "Diverted", hide otherwise prop_config[f"{element_path}.elements[3].style.display"] = { "binding": { "type": "property", "config": { "path": f"this.{element_path}.state" }, "transforms": [ { "inputType": "scalar", "outputType": "scalar", "mappings": [ {"input": "Diverted", "output": "block"} ], "fallback": "none", "type": "map" } ] } } # Main element display binding add_display_binding("show_gateways") # Set default stroke and fill for elements[0] if 'elements' in el and isinstance(el['elements'], list) and len(el['elements']) > 0: if 'stroke' not in el['elements'][0]: el['elements'][0]['stroke'] = {} el['elements'][0]['stroke']['paint'] = "#000000" el['elements'][0]['stroke']['width'] = "1" if 'fill' not in el['elements'][0]: el['elements'][0]['fill'] = {} el['elements'][0]['fill']['paint'] = "#aaaaaa" elif matches_pattern(r'_S\d+$') and name_contains('/DIV/'): # S buttons with DIV in name add_prop_binding("elements[1].fill.paint") add_display_binding("show_buttons") elif matches_pattern(r'_(S|JR|GS|PR)\d+$'): add_prop_binding("elements[1].fill.paint") add_display_binding("show_buttons") elif matches_pattern(r'_SS\d+$'): add_prop_binding("elements[1].fill.paint", "start_color") add_prop_binding("elements[2].fill.paint", "stop_color") add_display_binding("show_buttons", is_ss=True) elif matches_pattern(r'_VFD\d*$'): # VFD pattern: matches _VFD, _VFD1, _VFD2, etc. add_prop_binding("fill.paint") add_display_binding("show_running") elif matches_pattern(r'_ST\d+$'): # ST pattern: matches _ST1, _ST2, etc. in ID or name # Check if fill.paint exists, add binding if it does if 'fill' in el and isinstance(el.get('fill'), dict) and 'paint' in el.get('fill', {}): add_prop_binding("fill.paint") # Always add display binding add_display_binding("show_running") elif matches_pattern(r'_DPM\d+$'): add_prop_binding("elements[0].fill.paint") add_display_binding("show_gateways") # CH elements where both ID and name contain "CH" (but ID has no PE pattern) # Excludes Chute_{nnn} elements which have their own handling elif el_id and 'CH' in el_id.upper() and name and 'CH' in name.upper() and not re.search(r'_(FPE|JPE|TPE|PE)\d+', el_id) and not re.search(r'^Chute_\d+$', el_id): # Both ID and name contain "CH" but ID has no PE pattern - apply chute bindings if 'fill' in el and isinstance(el.get('fill'), dict) and 'paint' in el.get('fill', {}): add_prop_binding("fill.paint") add_display_binding("show_chutes") # Special handling for CH (Chute) elements with PE-related names # Excludes Chute_{nnn} elements which have their own handling elif el_id and 'CH' in el_id.upper() and name and (re.search(r'_(FPE|JPE|TPE|PE)\d+', name) or '/PE/' in name) and not re.search(r'^Chute_\d+$', el_id): # ID contains "CH" and name is PE-related (has PE pattern or /PE/ path) # Check if fill.paint exists before adding binding if 'fill' in el and isinstance(el.get('fill'), dict) and 'paint' in el.get('fill', {}): add_prop_binding("fill.paint") # Determine display binding: # - If ID also has PE pattern (like FL1018_3CH_PE1), use show_pes # - If ID has only CH (like PS11_5CH), use show_chutes if re.search(r'_(FPE|JPE|TPE|PE)\d+', el_id): add_display_binding("show_pes") else: add_display_binding("show_chutes") elif matches_pattern(r'_(TPE|JPE|FPE|PE)\d+$') or name_contains('/PE/'): # PE element (by ID pattern or by name containing /PE/) add_prop_binding("fill.paint") add_display_binding("show_pes") elif matches_pattern(r'_PMM\d+$'): add_prop_binding("elements[2].fill.paint") add_display_binding("show_gateways") elif matches_pattern(r'_FIO\d+$'): add_prop_binding("fill.paint") add_display_binding("show_fio") elif matches_pattern(r'_EX\d+$'): add_prop_binding("fill.paint") add_display_binding("show_running") elif matches_pattern(r'Induction_\d+$'): add_prop_binding("fill.paint") add_display_binding("show_inductions") elif matches_pattern(r'Chute_\d+'): # Chute elements: fill.paint bound to color, display uses show_chutes add_prop_binding("fill.paint") add_display_binding("show_chutes") elif check_name and 'sorter' in check_name.lower(): # Sorter elements: fill.paint bound to color, display uses show_running add_prop_binding("fill.paint") add_display_binding("show_running") elif matches_pattern(r'_FIOH\d+$'): # FIOH (Field IO Hub) elements: fill.paint bound to color add_prop_binding("fill.paint") add_display_binding("show_gateways") # Always set stroke values if 'stroke' not in el: el['stroke'] = {} el['stroke']['paint'] = "#000000" el['stroke']['width'] = "1" elif matches_pattern(r'_SIO\d+$'): # SIO (Serial IO) elements: fill.paint bound to color add_prop_binding("fill.paint") add_display_binding("show_gateways") elif matches_pattern(r'^MCM\d+$'): add_prop_binding("elements[1].fill.paint") add_display_binding("show_gateways") elif name_contains('/CMC/Conveyors/'): add_prop_binding("fill.paint") add_display_binding("show_running") # Count this top-level element as processed once we've applied # naming + bindings logic. processed_count += 1 # Recurse into nested elements ONLY for cleanup (is_top_level=False) if 'elements' in el and isinstance(el['elements'], list): traverse_elements( el['elements'], f"{element_path}.elements", prop_config, is_top_level=False, system_name=system_name ) return processed_count def process_system(system_id, view_name=None): print(f"Searching for system: {system_id}") if view_name: print(f"Filtering for views containing: '{view_name}'") # 1. Find matching folders in Detailed-Views (ONLY SOURCE NOW) target_folders = [] if os.path.exists(BASE_NEW): for item in os.listdir(BASE_NEW): item_path = os.path.join(BASE_NEW, item) if item.startswith(system_id) and os.path.isdir(item_path): if view_name is None or view_name.lower() in item.lower(): target_folders.append(item) if not target_folders: print(f"Error: Could not find folder(s) starting with '{system_id}' in {BASE_NEW}") if view_name: print(f" (with name containing '{view_name}')") return print(f"\nFound {len(target_folders)} matching view(s):") for i, folder in enumerate(target_folders, 1): print(f" {i}. {folder}") # 2. Process each target folder success_count = 0 for target_folder_name in target_folders: print(f"\n{'='*60}") print(f"Processing folder: {target_folder_name}") print(f"{'='*60}") # Extract MCM ID from folder name mcm_id = extract_mcm_id(target_folder_name) print(f"Using System Name: {mcm_id}") scada_view_path = os.path.join(BASE_NEW, target_folder_name, "view.json") if not os.path.exists(scada_view_path): print(f"⚠ view.json not found: {scada_view_path}") continue temp_target = f"{mcm_id}_target_{target_folder_name.replace(' ', '_')}.json" try: print(f"Loading SCADA SVG View: {scada_view_path}...") scada_view_data = load_json(scada_view_path) if not scada_view_data: print(f"✗ Failed to load target view.json") continue print("Updating SCADA SVG Bindings...") updated_count = update_scada_svg(scada_view_data, mcm_id) print(f"Generated bindings for {updated_count} elements.") # Write to temp first with open(temp_target, 'w', encoding='utf-8') as f: json.dump(scada_view_data, f, indent=2) # Write back to Ignition project try: with open(scada_view_path, 'w', encoding='utf-8') as f: json.dump(scada_view_data, f, indent=2) print(f"✓ Saved directly to {scada_view_path}") success_count += 1 except PermissionError: print("Direct write failed (Permission denied). Attempting copy...") import shutil shutil.copy2(temp_target, scada_view_path) print("✓ Copy completed.") success_count += 1 except Exception as e: print(f"✗ Error processing {target_folder_name}: {e}") import traceback traceback.print_exc() finally: if os.path.exists(temp_target): os.remove(temp_target) print(f"\n{'='*60}") print(f"Summary: {success_count}/{len(target_folders)} view(s) processed successfully") print(f"{'='*60}") def main(): if len(sys.argv) < 2: print("Usage: python update_scada_names.py [view_name]") print("Example: python update_scada_names.py MCM02") print("Example: python update_scada_names.py MCM02 'Fluid Inbound Upper'") print("\nIf view_name is not specified, all views starting with MCM_ID will be processed.") return system_id = sys.argv[1] view_name = sys.argv[2] if len(sys.argv) > 2 else None if view_name: print(f"Processing specific view containing: '{view_name}'") process_system(system_id, view_name) if __name__ == '__main__': main()