from typing import List
import simplnx as nx
import numpy as np
import os
import csv
class EulerAngleToCSV:
# -----------------------------------------------------------------------------
# These methods should not be edited
# -----------------------------------------------------------------------------
def uuid(self) -> nx.Uuid:
"""This returns the UUID of the filter. Each filter has a unique UUID value
:return: The Filter's Uuid value
:rtype: string
"""
return nx.Uuid('c68801c3-a29e-465e-a447-fb13b739fe4b')
def class_name(self) -> str:
"""The returns the name of the class that implements the filter
:return: The name of the implementation class
:rtype: string
"""
return 'EulerAngleToCSV'
def name(self) -> str:
"""The returns the name of filter
:return: The name of the filter
:rtype: string
"""
return 'EulerAngleToCSV'
def clone(self):
"""Clones the filter
:return: A new instance of the filter
:rtype: EulerAngleToCSV
"""
return EulerAngleToCSV()
# -----------------------------------------------------------------------------
# These methods CAN (and probably should) be updated. For instance, the
# human_name() is what users of the filter will see in the DREAM3D-NX GUI. You
# might want to consider putting spaces between workd, using proper capitalization
# and putting "(Python)" at the end of the name (or beginning if you want the
# filter list to group your filters togther)
# -----------------------------------------------------------------------------
def human_name(self) -> str:
"""This returns the name of the filter as a user of DREAM3DNX would see it
:return: The filter's human name
:rtype: string
"""
return 'EulerAngleToCSV (Python)'
def default_tags(self) -> List[str]:
"""This returns the default tags for this filter
:return: The default tags for the filter
:rtype: list
"""
return ['python', 'EulerAngleToCSV']
"""
This section should contain the 'keys' that store each parameter. The value of the key should be snake_case. The name
of the value should be ALL_CAPITOL_KEY
"""
FEATURE_IDS_ARRAY_PATH_KEY = 'feature_ids_array_path'
EULER_ANGLES_ARRAY_PATH_KEY = 'euler_angles_array_path'
OUTPUT_DIRECTORY_KEY = 'output_directory'
def parameters(self) -> nx.Parameters:
"""This function defines the parameters that are needed by the filter. Parameters collect the values from the user interface
and pack them up into a dictionary for use in the preflight and execute methods.
"""
params = nx.Parameters()
params.insert(nx.ArraySelectionParameter(EulerAngleToCSV.FEATURE_IDS_ARRAY_PATH_KEY, 'Feature IDs Array', 'Array containing the Feature IDs', nx.DataPath(), {nx.DataType.int32}))
params.insert(nx.ArraySelectionParameter(EulerAngleToCSV.EULER_ANGLES_ARRAY_PATH_KEY, 'Euler Angles Array', 'Array containing the Euler Angles (phi1, PHI, phi2)', nx.DataPath(), {nx.DataType.float32}))
params.insert(nx.FileSystemPathParameter(EulerAngleToCSV.OUTPUT_DIRECTORY_KEY, 'Output Directory', 'Directory path where CSV files will be saved', '/tmp', set(), nx.FileSystemPathParameter.PathType.OutputDir, False))
return params
def parameters_version(self) -> int:
return 1
def preflight_impl(self, data_structure: nx.DataStructure, args: dict, message_handler: nx.IFilter.MessageHandler, should_cancel: nx.AtomicBoolProxy) -> nx.IFilter.PreflightResult:
"""This method preflights the filter and should ensure that all inputs are sanity checked as best as possible. Array
sizes can be checked if the array sizes are actually known at preflight time. Some filters will not be able to report output
array sizes during preflight (segmentation filters for example). If in doubt, set the tuple dimensions of an array to [1].
:returns:
:rtype: nx.IFilter.PreflightResult
"""
# Extract the values from the user interface from the 'args'
feature_ids_array_path: nx.DataPath = args[EulerAngleToCSV.FEATURE_IDS_ARRAY_PATH_KEY]
euler_angles_array_path: nx.DataPath = args[EulerAngleToCSV.EULER_ANGLES_ARRAY_PATH_KEY]
output_directory: str = args[EulerAngleToCSV.OUTPUT_DIRECTORY_KEY]
# Create an OutputActions object to hold any DataStructure modifications that we are going to make
output_actions = nx.OutputActions()
# Create the Errors and Warnings Lists to commuicate back to the user if anything has gone wrong
errors = []
warnings = []
# Validate that the output directory is provided
if not output_directory or output_directory.strip() == "":
errors.append(nx.Error(-65020, "Output directory must be specified."))
# Validate that the feature IDs array exists in the data structure
if not data_structure.does_path_exist(feature_ids_array_path):
errors.append(nx.Error(-65021, f"Feature IDs array path '{feature_ids_array_path.to_string('/')}' does not exist in the data structure."))
# Validate that the euler angles array exists in the data structure
if not data_structure.does_path_exist(euler_angles_array_path):
errors.append(nx.Error(-65022, f"Euler Angles array path '{euler_angles_array_path.to_string('/')}' does not exist in the data structure."))
# If we have both arrays, validate their dimensions
if data_structure.does_path_exist(feature_ids_array_path) and data_structure.does_path_exist(euler_angles_array_path):
feature_ids_array = data_structure[feature_ids_array_path]
euler_angles_array = data_structure[euler_angles_array_path]
# Check that both arrays have the same number of tuples
if feature_ids_array.shape[0] != euler_angles_array.shape[0]:
errors.append(nx.Error(-65023, f"Feature IDs array and Euler Angles array must have the same number of tuples. Feature IDs: {feature_ids_array.shape[0]}, Euler Angles: {euler_angles_array.shape[0]}"))
# Check that Euler Angles array has 3 components (phi1, PHI, phi2)
if len(euler_angles_array.shape) != 2 or euler_angles_array.shape[1] != 3:
errors.append(nx.Error(-65024, f"Euler Angles array must have 3 components (phi1, PHI, phi2). Current shape: {euler_angles_array.shape}"))
# If there are errors, return them
if errors:
return nx.IFilter.PreflightResult(errors=errors)
# Send back any messages that will appear in the "Output" widget in the UI. This is optional.
message_handler(nx.IFilter.Message(nx.IFilter.Message.Type.Info, f"Will export Euler angles to CSV files in directory: '{output_directory}'"))
# Return the output_actions so the changes are reflected in the User Interface.
return nx.IFilter.PreflightResult(output_actions=output_actions, errors=None, warnings=warnings, preflight_values=None)
def execute_impl(self, data_structure: nx.DataStructure, args: dict, message_handler: nx.IFilter.MessageHandler, should_cancel: nx.AtomicBoolProxy) -> nx.IFilter.ExecuteResult:
""" This method actually executes the filter algorithm and reports results.
:returns:
:rtype: nx.IFilter.ExecuteResult
"""
# Extract the values from the user interface from the 'args'
# This is basically repeated from the preflight because the variables are scoped to the method()
feature_ids_array_path: nx.DataPath = args[EulerAngleToCSV.FEATURE_IDS_ARRAY_PATH_KEY]
euler_angles_array_path: nx.DataPath = args[EulerAngleToCSV.EULER_ANGLES_ARRAY_PATH_KEY]
output_directory: str = args[EulerAngleToCSV.OUTPUT_DIRECTORY_KEY]
# Get numpy views of the arrays
feature_ids_array = data_structure[feature_ids_array_path].npview()
euler_angles_array = data_structure[euler_angles_array_path].npview()
# Create output directory if it doesn't exist
try:
os.makedirs(output_directory, exist_ok=True)
except Exception as e:
return nx.IFilter.ExecuteResult(errors=[nx.Error(-65030, f"Failed to create output directory '{output_directory}': {str(e)}")])
# Get unique feature IDs
unique_feature_ids = np.unique(feature_ids_array)
message_handler(nx.IFilter.Message(nx.IFilter.Message.Type.Info, f'Processing {len(unique_feature_ids)} unique feature IDs'))
# Process each unique feature ID
for i, feature_id in enumerate(unique_feature_ids):
# Check if user cancelled the filter
if should_cancel:
return nx.IFilter.ExecuteResult(errors=[nx.Error(-65031, "Filter was cancelled by user")])
# Find all indices where this feature ID appears
feature_indices = np.where(feature_ids_array == feature_id)[0]
# Get the Euler angles for this feature ID
feature_euler_angles = euler_angles_array[feature_indices]
# Create the output filename
output_filename = f"Feature_{feature_id}_EulerAngles.csv"
output_filepath = os.path.join(output_directory, output_filename)
try:
# Write the CSV file
with open(output_filepath, 'w', newline='') as csvfile:
csv_writer = csv.writer(csvfile)
# Write header
csv_writer.writerow(['phi1', 'PHI', 'phi2'])
# Write data rows
for euler_angle in feature_euler_angles:
csv_writer.writerow([euler_angle[0], euler_angle[1], euler_angle[2]])
message_handler(nx.IFilter.Message(nx.IFilter.Message.Type.Info, f'Created {output_filename} with {len(feature_euler_angles)} Euler angle entries'))
except Exception as e:
return nx.IFilter.ExecuteResult(errors=[nx.Error(-65032, f"Failed to write CSV file '{output_filepath}': {str(e)}")])
# Update progress
progress = (i + 1) / len(unique_feature_ids) * 100
message_handler(nx.IFilter.Message(nx.IFilter.Message.Type.Info, f'Progress: {progress:.1f}% ({i + 1}/{len(unique_feature_ids)} features processed)'))
message_handler(nx.IFilter.Message(nx.IFilter.Message.Type.Info, f'Successfully exported Euler angles for {len(unique_feature_ids)} features to {output_directory}'))
return nx.Result()