470 lines
23 KiB
Python
470 lines
23 KiB
Python
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 |