Skip to content

Commit

Permalink
Add parser and update UI
Browse files Browse the repository at this point in the history
  • Loading branch information
proy30 committed Jan 23, 2025
1 parent d7667fe commit e2de344
Show file tree
Hide file tree
Showing 4 changed files with 375 additions and 8 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,9 @@ def distribution_parameters():

@state.change("distribution")
def on_distribution_name_change(distribution, **kwargs):
if state.importing_file:
return

if distribution == "Thermal" or distribution == "Empty":
state.distribution_type = ""
state.distribution_type_disable = True
Expand All @@ -140,6 +143,8 @@ def on_distribution_name_change(distribution, **kwargs):

@state.change("distribution_type")
def on_distribution_type_change(**kwargs):
if state.importing_file:
return
populate_distribution_parameters()


Expand Down
131 changes: 131 additions & 0 deletions src/python/impactx/dashboard/Toolbar/importParser.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
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.
"""

@staticmethod
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"]
size_str = ""

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']}"

@staticmethod
def parse_impactx_simulation_file(file) -> None:
"""
Parses ImpactX simulation file contents.
:param file: ImpactX simulation file uploaded by the user.
"""

file_content = DashboardParserHelper.import_file_content(file, state)

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

@staticmethod
def populate_impactx_simulation_file_to_ui(file) -> None:
"""
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.distribution = imported_data["distribution"]["name"]
state.distribution_type = imported_data["distribution"]["type"]
state.flush()
populate_distribution_parameters()

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.selected_lattice_list = []

for lattice_element_index, element in enumerate(imported_lattice_data):
parsed_element = element["element"]
parsed_parameters = element["parameters"]

state.selected_lattice = parsed_element
add_lattice_element()

lattice_list_parameters = state.selected_lattice_list[
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,
)
151 changes: 151 additions & 0 deletions src/python/impactx/dashboard/Toolbar/importParserHelper.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
import ast
import re

from ..Input.defaults import DashboardDefaults


class DashboardParserHelper:
"""
Helper functions for building dashboard parser.
"""

@staticmethod
def import_file_content(file: str, state) -> 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_on_ui"] = 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
Loading

0 comments on commit e2de344

Please sign in to comment.