import os import json import re import zipfile # Restored zip dependency import shutil # For directory operations from datetime import datetime from PIL import Image # For thumbnail creation class ScadaCreatorError(Exception): # Renamed Error class """Custom exception for SCADA view creation errors.""" pass class ScadaViewCreator: # Renamed class """ Handles the direct creation of Ignition SCADA view files (view.json, resource.json, thumbnail.png) within the project structure. """ def __init__(self, view_data, ignition_base_dir, custom_export_dir=None, log_callback=print, status_callback=print, create_zip=True): """ Initialize the view creator. Args: view_data (dict): A dictionary containing all necessary data: - project_title (str): Ignition project title. # - parent_project (str): Removed, not needed for direct creation. - view_path (str): Path within the project's views folder (e.g., 'Group/SubGroup'). Optional. - view_name (str): Name for the view file/directory (e.g., 'MyView'). - svg_url (str): URL or path for the background SVG image (optional). - image_width (int): Width of the background image. - image_height (int): Height of the background image. - default_width (int): Default width for the view. - default_height (int): Default height for the view. - elements (list): List of processed element dictionaries. ignition_base_dir (str): The base directory for Ignition projects (e.g., 'C:/Program Files/Inductive Automation/Ignition/data/projects'). custom_export_dir (str, optional): Custom directory to export files to. If provided, ignores the ignition_base_dir and exports files directly to this location. log_callback (callable): Function to call for logging messages. status_callback (callable): Function to call for status updates. create_zip (bool): Whether to create a zip file containing the export (default: True). """ self.data = view_data self.ignition_base_dir = ignition_base_dir self.custom_export_dir = custom_export_dir self.create_zip = create_zip self._log = lambda msg: log_callback(f"[ViewCreator] {msg}") # Updated log prefix self._status = status_callback # Validate required data required_keys = [ 'project_title', 'view_name', 'image_width', 'image_height', 'default_width', 'default_height', 'elements' ] missing_keys = [k for k in required_keys if k not in self.data] if missing_keys: raise ValueError(f"Missing required view data keys: {', '.join(missing_keys)}") # Only validate base directory if no custom export directory is provided if not self.custom_export_dir and (not self.ignition_base_dir or not os.path.isdir(self.ignition_base_dir)): raise ValueError(f"Invalid ignition_base_dir provided: {self.ignition_base_dir}") def create_view_files(self): # Renamed method, removed zip_file_path """ Creates the SCADA view files (view.json, resource.json, thumbnail.png) directly in the project's file structure or in a custom export location. """ project_title_cleaned = self._clean_path_part(self.data["project_title"]) if not project_title_cleaned: raise ValueError("Project title cannot be empty after cleaning.") self._log(f"Starting SCADA view file creation for project: {project_title_cleaned}") try: # --- Create temporary working directory if creating a zip file --- temp_work_dir = None if self.create_zip and self.custom_export_dir: import tempfile temp_work_dir = tempfile.mkdtemp(prefix="scada_export_") self._log(f"Created temporary working directory: {temp_work_dir}") # --- Determine target directory --- if self.custom_export_dir: if self.create_zip: # If creating a zip, use the temporary directory to build the structure target_root_dir = temp_work_dir if temp_work_dir else self.custom_export_dir # Place perspective folder directly at root for zip perspective_rel_path = "com.inductiveautomation.perspective" # Save the root dir for later project.json creation self.zip_root_dir = target_root_dir else: # Otherwise use custom export directory directly self._log(f"Using custom export directory: {self.custom_export_dir}") target_view_dir = self.custom_export_dir target_root_dir = None else: # Use standard Ignition structure target_root_dir = self.ignition_base_dir perspective_rel_path = os.path.join("com.inductiveautomation.perspective", "views") # --- Clean and Combine View Path and Name --- view_path_raw = self.data.get("view_path", "").strip(' /\\') view_name_raw = self.data["view_name"].strip(' /\\') cleaned_view_path_parts = [self._clean_path_part(part) for part in re.split(r'[\\/]+', view_path_raw) if part] cleaned_view_name = self._clean_path_part(view_name_raw) if not cleaned_view_name: self._log("Warning: Cleaned view name is empty. Using 'DefaultView'.") cleaned_view_name = "DefaultView" # For later creation of zip file self.view_name = cleaned_view_name final_view_rel_parts = cleaned_view_path_parts + [cleaned_view_name] view_dir_rel = os.path.join(*final_view_rel_parts) # Construct the absolute path for the view directory if target_root_dir: if self.create_zip: # For zip structure, views folder is inside perspective folder target_view_dir = os.path.join( target_root_dir, perspective_rel_path, "views", view_dir_rel ) else: target_view_dir = os.path.join( target_root_dir, project_title_cleaned, perspective_rel_path, view_dir_rel ) # Save project directory for later zip creation if self.create_zip: self.project_dir = os.path.join( target_root_dir, project_title_cleaned ) self._log(f"Calculated target view directory: {target_view_dir}") # Create directories if they don't exist self._status(f"Ensuring directory structure exists: {target_view_dir}") os.makedirs(target_view_dir, exist_ok=True) self._log(f"View directory created/ensured: {target_view_dir}") # --- Create View Files --- self._status("Creating view files...") self._create_view_json(target_view_dir) self._create_resource_json(target_view_dir) self._create_thumbnail(target_view_dir) # Create project.json at top level for zip structure if self.create_zip and hasattr(self, 'zip_root_dir'): self._create_project_json(self.zip_root_dir) self._log(f"View files created in: {target_view_dir}") # --- Create Zip File If Requested --- zip_file_path = None if self.create_zip and self.custom_export_dir: self._status("Creating zip file...") zip_file_path = self._create_zip_file( project_title_cleaned, self.zip_root_dir if hasattr(self, 'zip_root_dir') else target_view_dir ) self._log(f"Zip file created at: {zip_file_path}") # Clean up temporary directory if temp_work_dir: self._log(f"Cleaning up temporary directory: {temp_work_dir}") shutil.rmtree(temp_work_dir) self._log(f"View files successfully created.") if zip_file_path: self._log(f"Zip file created at: {zip_file_path}") else: self._log(f"Files created in directory: '{target_view_dir}'.") self._status("View creation complete.") except Exception as e: # Clean up temp directory if it exists if 'temp_work_dir' in locals() and temp_work_dir and os.path.exists(temp_work_dir): try: shutil.rmtree(temp_work_dir) except: pass # Log the error and re-raise as a specific ScadaCreatorError import traceback error_details = f"Error during view file creation:\n{e}\nTraceback:\n{traceback.format_exc()}" self._log(f"View Creation Failed! Details:\n{error_details}") self._status(f"View creation failed: {e}") raise ScadaCreatorError(f"Failed to create view files: {e}") from e def _create_zip_file(self, project_title, source_dir): """Create a zip file containing the project structure.""" timestamp = datetime.now().strftime("%Y-%m-%d_%H%M") zip_filename = f"{project_title}_SCADA_{timestamp}.zip" zip_path = os.path.join(self.custom_export_dir, zip_filename) self._log(f"Creating zip file at: {zip_path}") self._status("Creating zip archive...") try: with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: # Add project.json directly to the root project_json_path = os.path.join(source_dir, "project.json") if os.path.exists(project_json_path): zipf.write(project_json_path, "project.json") self._log(f" Added to zip: project.json") # Add com.inductiveautomation.perspective folder and its contents perspective_dir = os.path.join(source_dir, "com.inductiveautomation.perspective") if os.path.exists(perspective_dir): # Walk through the perspective directory for root, dirs, files in os.walk(perspective_dir): for file in files: file_path = os.path.join(root, file) # Create relative path within the zip arcname = os.path.join("com.inductiveautomation.perspective", os.path.relpath(file_path, perspective_dir)) zipf.write(file_path, arcname) self._log(f" Added to zip: {arcname}") self._status("Zip file created successfully.") return zip_path except Exception as e: self._log(f"Error creating zip file: {e}") raise ScadaCreatorError(f"Failed to create zip file: {e}") from e # --- Helper for cleaning path parts --- def _clean_path_part(self, part): """Cleans a path component for safe file/directory naming.""" # Replace invalid filename characters and multiple spaces/underscores cleaned = re.sub(r'[<>:"/\\|?*\s]+', '_', part) cleaned = re.sub(r'_+', '_', cleaned) # Consolidate underscores return cleaned.strip('_') # --- Helper methods for creating files --- def _create_project_json(self, base_dir): """Create a project.json file at the root of the project directory.""" project_config = { "title": self.data["project_title"], "description": "JP", "parent": "SCADA_PERSPECTIVE_PARENT_PROJECT", "enabled": True, "inheritable": False } # Create/ensure project.json at the root level project_file = os.path.join(base_dir, "project.json") try: with open(project_file, 'w', encoding='utf-8') as f: json.dump(project_config, f, indent=2) self._log(" - project.json created.") except IOError as e: self._log(f"ERROR: Could not write project.json to {project_file}: {e}") raise ScadaCreatorError(f"Could not write project.json: {e}") from e def _create_resource_json(self, view_dir): """Create the resource.json file with specified structure.""" resource_config = { "scope": "G", "version": 1, "restricted": False, "overridable": True, "files": [ "view.json", "thumbnail.png" ], "attributes": { "lastModification": { # Using fixed values as requested "actor": "ilia-gu-autstand", "timestamp": "2025-04-10T15:25:02Z" }, # Using fixed signature as requested "lastModificationSignature": "578829ca084d3cc740d02206de92730e6cb6ee3ea927e5e28f4f8dfe44b95ade" } } resource_file = os.path.join(view_dir, "resource.json") try: with open(resource_file, 'w', encoding='utf-8') as f: json.dump(resource_config, f, indent=2) self._log(" - resource.json created.") except IOError as e: self._log(f"ERROR: Could not write resource.json to {resource_file}: {e}") raise ScadaCreatorError(f"Could not write resource.json: {e}") from e def _create_thumbnail(self, view_dir): """Create a minimal placeholder thumbnail.png.""" thumbnail_file = os.path.join(view_dir, "thumbnail.png") try: # Check if Pillow is available try: from PIL import Image img = Image.new('RGBA', (1, 1), (0, 0, 0, 0)) # 1x1 transparent pixel img.save(thumbnail_file, "PNG") self._log(" - thumbnail.png created (placeholder).") except ImportError: self._log("Warning: Pillow (PIL) not installed. Cannot create thumbnail PNG.") # Create empty file as fallback if Pillow is not installed with open(thumbnail_file, 'w') as f: pass self._log(" - thumbnail.png created (empty file fallback - Pillow missing).") except Exception as e: self._log(f"Error creating thumbnail PNG: {e}. Creating empty file.") # Create empty file as fallback on other errors try: with open(thumbnail_file, 'w') as f: pass self._log(" - thumbnail.png created (empty file fallback - error during creation).") except IOError as io_e: self._log(f"ERROR: Could not write empty thumbnail.png to {thumbnail_file}: {io_e}") raise ScadaCreatorError(f"Could not write thumbnail.png: {io_e}") from io_e def _create_view_json(self, view_dir): """Create the main view.json file.""" view_config = { "custom": {}, "params": {}, "props": { "defaultSize": { # Using default_width/height from data for consistency "height": self.data["default_height"], "width": self.data["default_width"] } }, "root": { "type": "ia.container.coord", "meta": {"name": "root"}, "props": {"style": {"overflow": "visible"}}, "children": [] } } # Add background image if URL is provided svg_url = self.data.get("svg_url") if svg_url: background_image = { "type": "ia.display.image", "version": 0, "meta": {"name": "BackgroundImage"}, "position": { "x": 0, "y": 0, "width": self.data["image_width"], "height": self.data["image_height"] }, "props": {"source": svg_url, "fit": {"mode": "contain"}}, "custom": {} } view_config["root"]["children"].append(background_image) # Convert processed elements scada_elements = [] processed_elements = self.data.get("elements", []) self._log(f"Processing {len(processed_elements)} elements for view.json...") for i, element_data in enumerate(processed_elements): try: element_type = element_data.get('type', 'ia.display.label') # Use cleaned element name for meta name if possible element_name_raw = element_data.get('meta', {}).get('name', f'element_{i+1}') element_name = self._clean_path_part(element_name_raw) or f'element_{i+1}' # Fallback if cleaning results in empty position = element_data.get('position', {}) props = element_data.get('props', {}) custom = element_data.get('custom', {}) # Position details x = position.get('x', 0) y = position.get('y', 0) width = position.get('width', 10) height = position.get('height', 10) rotation_info = position.get('rotate', None) # Basic structure scada_component = { "type": element_type, "version": 0, "meta": {"name": element_name}, # Use cleaned name "position": { "x": x, "y": y, "width": width, "height": height }, "props": props, "custom": custom } # Add rotation if rotation_info and isinstance(rotation_info, dict): angle_str = str(rotation_info.get('angle', '0')).lower().replace('deg', '').strip() try: angle = float(angle_str) scada_component["position"]["rotate"] = {"angle": angle} anchor = rotation_info.get('anchor') if anchor: scada_component["position"]["rotate"]["anchor"] = anchor except ValueError: self._log(f"Warning: Invalid rotation angle '{angle_str}' for '{element_name}'.") # --- Adjustments based on type (especially for ia.display.view) --- if element_type == 'ia.display.view': # Ensure props exists if 'props' not in scada_component: scada_component['props'] = {} # Ensure path exists in props, use default from mapping if available if 'path' not in scada_component['props']: # Attempt to get path from original props if it was somehow separated original_path = props.get('path', 'Error/MissingPath') scada_component['props']['path'] = original_path if original_path == 'Error/MissingPath': self._log(f"Warning: Embedded view '{element_name}' missing 'path'. Setting default 'Error/MissingPath'. Check mappings.") # Ensure params exists and construct tagProps based on example if 'params' not in scada_component['props']: scada_component['props']['params'] = {} # Construct tagProps using element name as placeholder tag path # TODO: Make tag path generation more configurable if needed placeholder_tag_path = f"System/Path/Placeholder/{element_name}" scada_component['props']['params']['tagProps'] = [ placeholder_tag_path, "value", "value", "value", "value", "value", "value", "value", "value", "value" ] # Include other potential params from original data if needed # This merge might be complex depending on desired behavior # original_params = props.get('params', {}) # scada_component['props']['params'].update(original_params) # Careful merge needed elif element_type == 'ia.display.label': # Ensure text prop exists for labels if 'props' not in scada_component: scada_component['props'] = {} if 'text' not in scada_component['props']: # Use raw name for text, cleaned name for meta scada_component['props']['text'] = element_name_raw # Add other type-specific adjustments here if necessary scada_elements.append(scada_component) except Exception as el_err: element_name_raw = element_data.get('meta',{}).get('name', f'unknown_{i+1}') self._log(f"ERROR converting element {i} ('{element_name_raw}') to SCADA format: {el_err}") # Add placeholder on error scada_elements.append({ "type": "ia.display.label", "meta": {"name": f"ERROR_{self._clean_path_part(element_name_raw)}_{i+1}"}, "position": {"x": 0, "y": i*15, "width": 100, "height": 15}, "props": {"text": f"Export Error: {el_err}"} }) view_config["root"]["children"].extend(scada_elements) # Write file view_file = os.path.join(view_dir, "view.json") try: with open(view_file, 'w', encoding='utf-8') as f: json.dump(view_config, f, indent=2) self._log(f" - view.json created with {len(scada_elements)} elements.") except IOError as e: self._log(f"ERROR: Could not write view.json to {view_file}: {e}") raise ScadaCreatorError(f"Could not write view.json: {e}") from e