-
Notifications
You must be signed in to change notification settings - Fork 21
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Draft] Dashboard File Importing #785
base: development
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,126 @@ | ||
from ..Input.distributionParameters.distributionMain import ( | ||
on_distribution_parameter_change, | ||
populate_distribution_parameters, | ||
) | ||
from ..Input.latticeConfiguration.latticeMain import ( | ||
add_lattice_element, | ||
on_lattice_element_parameter_change, | ||
) | ||
from ..trame_setup import setup_server | ||
from .importParserHelper import DashboardParserHelper | ||
|
||
server, state, ctrl = setup_server() | ||
|
||
|
||
class DashboardParser: | ||
""" | ||
Provides functionality to import ImpactX simulation files | ||
to the dashboard and auto-populate the UI with their configurations. | ||
""" | ||
|
||
def file_details(file) -> None: | ||
""" | ||
Displays the size of the imported simulation file. | ||
|
||
:param file: ImpactX simulation file uploaded by the user. | ||
""" | ||
|
||
file_size_in_bytes = file["size"] | ||
if file_size_in_bytes < 1024: | ||
size_str = f"{file_size_in_bytes} B" | ||
elif file_size_in_bytes < 1024 * 1024: | ||
size_str = f"{file_size_in_bytes/1024:.1f} KB" | ||
|
||
state.import_file_details = f"({size_str}) {file['name']}" | ||
Check failure Code scanning / CodeQL Potentially uninitialized local variable Error
Local variable 'size_str' may be used before it is initialized.
|
||
|
||
def parse_impactx_simulation_file(file) -> None: | ||
Check notice Code scanning / CodeQL First parameter of a method is not named 'self' Note
Normal methods should have 'self', rather than 'file', as their first parameter.
|
||
""" | ||
Parses ImpactX simulation file contents. | ||
|
||
:param file: ImpactX simulation file uploaded by the user. | ||
""" | ||
|
||
file_content = DashboardParserHelper.import_file_content(file) | ||
|
||
single_input_contents = DashboardParserHelper.parse_single_inputs(file_content) | ||
list_input_contents = DashboardParserHelper.parse_list_inputs(file_content) | ||
distribution_contents = DashboardParserHelper.parse_distribution(file_content) | ||
lattice_element_contents = DashboardParserHelper.parse_lattice_elements( | ||
file_content | ||
) | ||
|
||
parsed_values_dictionary = { | ||
**single_input_contents, | ||
**list_input_contents, | ||
**distribution_contents, | ||
**lattice_element_contents, | ||
} | ||
|
||
return parsed_values_dictionary | ||
|
||
def populate_impactx_simulation_file_to_ui(file) -> None: | ||
Check notice Code scanning / CodeQL First parameter of a method is not named 'self' Note
Normal methods should have 'self', rather than 'file', as their first parameter.
|
||
""" | ||
Auto fills the dashboard with parsed inputs. | ||
|
||
:param file: ImpactX simulation file uploaded by the user. | ||
""" | ||
|
||
imported_data = DashboardParser.parse_impactx_simulation_file(file) | ||
|
||
imported_distribution_data = imported_data["distribution"]["parameters"].items() | ||
imported_lattice_data = imported_data["lattice_elements"] | ||
non_state_inputs = ["distribution", "lattice_elements"] | ||
|
||
# Update state inputs (inputParameters, Space Charge, CSR) | ||
for input_name, input_value in imported_data.items(): | ||
if hasattr(state, input_name) and input_name not in non_state_inputs: | ||
setattr(state, input_name, input_value) | ||
|
||
# Update distribution inputs | ||
if imported_distribution_data: | ||
state.selectedDistribution = imported_data["distribution"]["name"] | ||
state.selectedDistributionType = imported_data["distribution"]["type"] | ||
state.flush() | ||
populate_distribution_parameters(state.selectedDistribution) | ||
|
||
for ( | ||
distr_parameter_name, | ||
distr_parameter_value, | ||
) in imported_distribution_data: | ||
on_distribution_parameter_change( | ||
distr_parameter_name, distr_parameter_value, "float" | ||
) | ||
|
||
# Update lattice elements | ||
state.selectedLatticeList = [] | ||
|
||
for lattice_element_index, element in enumerate(imported_lattice_data): | ||
parsed_element = element["element"] | ||
parsed_parameters = element["parameters"] | ||
|
||
state.selectedLattice = parsed_element | ||
add_lattice_element() | ||
|
||
lattice_list_parameters = state.selectedLatticeList[lattice_element_index][ | ||
"parameters" | ||
] | ||
|
||
for ( | ||
parsed_parameter_name, | ||
parsed_parameter_value, | ||
) in parsed_parameters.items(): | ||
parameter_type = None | ||
|
||
for parameter_info in lattice_list_parameters: | ||
parameter_info_name = parameter_info["parameter_name"] | ||
if parameter_info_name == parsed_parameter_name: | ||
parameter_type = parameter_info["parameter_type"] | ||
break | ||
|
||
if parameter_type: | ||
on_lattice_element_parameter_change( | ||
lattice_element_index, | ||
parsed_parameter_name, | ||
parsed_parameter_value, | ||
parameter_type, | ||
) |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,154 @@ | ||
import ast | ||
import re | ||
|
||
from ..Input.defaults import DashboardDefaults | ||
from ..trame_setup import setup_server | ||
|
||
server, state, ctrl = setup_server() | ||
|
||
|
||
class DashboardParserHelper: | ||
""" | ||
Helper functions for building dashboard parser. | ||
""" | ||
|
||
@staticmethod | ||
def import_file_content(file: str) -> dict: | ||
""" | ||
Retrieves and prints the content of the uploaded simulation file. | ||
|
||
:param content: The content of the ImpactX simulation file. | ||
""" | ||
if file: | ||
content = file["content"].decode("utf-8") | ||
return content | ||
else: | ||
state.file_content = "" | ||
return "" | ||
|
||
@staticmethod | ||
def parse_single_inputs(content: str) -> dict: | ||
""" | ||
Parses individual input parameters from the simulation file content. | ||
|
||
:param content: The content of the ImpactX simulation file. | ||
""" | ||
reference_dictionary = DashboardDefaults.DEFAULT_VALUES.copy() | ||
|
||
parsing_patterns = [ | ||
r"\b{}\s*=\s*([^#\n]+)", # (param = value) | ||
r"set_{}\(([^)]+)\)", # (set_param(value)) | ||
] | ||
|
||
for parameter_name in reference_dictionary.keys(): | ||
if parameter_name.endswith("_list"): | ||
continue | ||
|
||
for pattern in parsing_patterns: | ||
pattern_match = re.search(pattern.format(parameter_name), content) | ||
if pattern_match: | ||
value = ast.literal_eval(pattern_match.group(1)) | ||
reference_dictionary[parameter_name] = value | ||
break | ||
|
||
# Handling for kin_energy | ||
kin_energy_pattern_match = re.search( | ||
r"\bkin_energy_MeV\s*=\s*([^#\n]+)", content | ||
) | ||
if kin_energy_pattern_match: | ||
kin_energy_value = kin_energy_pattern_match.group(1) | ||
reference_dictionary["kin_energy"] = kin_energy_value | ||
|
||
return reference_dictionary | ||
|
||
@staticmethod | ||
def parse_list_inputs(content: str) -> dict: | ||
""" | ||
Parses list-based input parameters from the simulation file content. | ||
|
||
:param content: The content of the ImpactX simulation file. | ||
""" | ||
dictionary = {} | ||
list_inputs = ["n_cell", "prob_relative"] | ||
list_parsing = "{} = (\\[.*?\\])" | ||
|
||
for input_name in list_inputs: | ||
match = re.search(list_parsing.format(input_name), content) | ||
if match: | ||
values = ast.literal_eval(match.group(1).strip()) | ||
|
||
if input_name == "n_cell": | ||
for i, dim in enumerate(["x", "y", "z"]): | ||
dictionary[f"n_cell_{dim}"] = values[i] | ||
|
||
if input_name == "prob_relative": | ||
dictionary["prob_relative"] = values | ||
|
||
return dictionary | ||
|
||
@staticmethod | ||
def parse_distribution(content: str) -> dict: | ||
""" | ||
Parses distribution section from the simulation file content. | ||
|
||
:param content: The content of the ImpactX simulation file. | ||
""" | ||
|
||
dictionary = {"distribution": {"name": "", "type": "", "parameters": {}}} | ||
|
||
distribution_name = re.search(r"distribution\.(\w+)\(", content) | ||
distribution_type_twiss = re.search(r"twiss\((.*?)\)", content, re.DOTALL) | ||
distribution_type_quadratic = re.search( | ||
r"distribution\.\w+\((.*?)\)", content, re.DOTALL | ||
) | ||
parameters = {} | ||
|
||
def extract_parameters(distribution_type, parsing_pattern): | ||
parameter_pairs = re.findall(parsing_pattern, distribution_type.group(1)) | ||
parsed_parameters = {} | ||
|
||
for param_name, param_value in parameter_pairs: | ||
parsed_parameters[param_name] = param_value | ||
return parsed_parameters | ||
|
||
if distribution_name: | ||
dictionary["distribution"]["name"] = distribution_name.group(1) | ||
|
||
if distribution_type_twiss: | ||
dictionary["distribution"]["type"] = "Twiss" | ||
parameters = extract_parameters( | ||
distribution_type_twiss, r"(\w+)=(\d+\.?\d*)" | ||
) | ||
elif distribution_type_quadratic: | ||
dictionary["distribution"]["type"] = "Quadratic" | ||
parameters = extract_parameters( | ||
distribution_type_quadratic, r"(\w+)=([^,\)]+)" | ||
) | ||
|
||
dictionary["distribution"]["parameters"] = parameters | ||
|
||
return dictionary | ||
|
||
@staticmethod | ||
def parse_lattice_elements(content: str) -> dict: | ||
""" | ||
Parses lattice elements from the simulation file content. | ||
|
||
:param content: The content of the ImpactX simulation file. | ||
""" | ||
|
||
dictionary = {"lattice_elements": []} | ||
|
||
lattice_elements = re.findall(r"elements\.(\w+)\((.*?)\)", content) | ||
|
||
for element_name, element_parameter in lattice_elements: | ||
element = {"element": element_name, "parameters": {}} | ||
|
||
parameter_pairs = re.findall(r"(\w+)=([^,\)]+)", element_parameter) | ||
for parameter_name, parameter_value in parameter_pairs: | ||
parameter_value_cleaned = parameter_value.strip("'\"") | ||
element["parameters"][parameter_name] = parameter_value_cleaned | ||
|
||
dictionary["lattice_elements"].append(element) | ||
|
||
return dictionary |
Check notice
Code scanning / CodeQL
First parameter of a method is not named 'self' Note