Merge pull request #70 from upb-lea/pecst

Add prototype for capacitor optimization

Author: SevenOfNinePE

.github/workflows/main.yml

@@ -109,6 +109,21 @@ jobs:
             echo show folder structure:
             ls 
 
+      - name: install PE-CST developer version
+        run: |
+            pwd
+            ls
+            wget https://github.com/upb-lea/capacitor_selection_toolbox/archive/refs/heads/main.zip
+            unzip main.zip
+            rm main.zip
+            cd capacitor_selection_toolbox-main
+            pip install -e .
+            cd ..
+            echo show folder path
+            pwd
+            echo show folder structure:
+            ls 
+
       - name: install the package
         run: |
             pip install -e .

.github/workflows/sphinx_render_docs.yml

@@ -69,6 +69,21 @@ jobs:
           echo show folder structure:
           ls 
 
+    - name: install PE-CST developer version
+      run: |
+          pwd
+          ls
+          wget https://github.com/upb-lea/capacitor_selection_toolbox/archive/refs/heads/main.zip
+          unzip main.zip
+          rm main.zip
+          cd capacitor_selection_toolbox-main
+          pip install -e .
+          cd ..
+          echo show folder path
+          pwd
+          echo show folder structure:
+          ls 
+
     - name: install the package
       run: |
           pip install -e .

dct/__init__.py

@@ -24,6 +24,9 @@
 # summary
 from dct.summary_processing import *
 from dct.summary_pre_processing import *
+# capacitor selection classes
+from dct.capacitor_selection import *
+from dct.capacitor_optimization_dtos import *
 # optimization classes
 from dct.inductor_optimization import *
 from dct.transformer_optimization import *

dct/capacitor_optimization_dtos.py

@@ -0,0 +1,17 @@
+"""Data transfer objects (DTOs) for the capacitor optimization."""
+
+# python libraries
+import dataclasses
+
+# 3rd party libraries
+import pecst
+# own libraries
+from dct.server_ctl_dtos import ProgressData
+
+@dataclasses.dataclass
+class CapacitorOptimizationDto:
+    """DTO for the inductor optimization."""
+
+    circuit_filtered_point_filename: str
+    progress_data: ProgressData
+    capacitor_optimization_dto: pecst.CapacitorRequirements

dct/capacitor_selection.py

@@ -0,0 +1,225 @@
+"""Capacitor selection."""
+
+# python libraries
+import logging
+import os
+from multiprocessing import Pool, cpu_count
+import copy
+import pickle
+
+# 3rd party libraries
+import numpy as np
+import pandas as pd
+import tqdm
+
+# own libraries
+import pecst
+from dct.capacitor_optimization_dtos import CapacitorOptimizationDto
+from dct.toml_checker import TomlCapacitorSelection, Debug
+from dct.datasets import HandleDabDto
+from dct.datasets_dtos import StudyData, FilterData
+from dct.server_ctl_dtos import ProgressData, ProgressStatus
+from dct.functions_waveforms import full_current_waveform_from_currents, full_angle_waveform_from_angles
+from dct.datasets_dtos import CapacitorResults
+
+logger = logging.getLogger(__name__)
+
+class CapacitorSelection:
+    """Select suitable capacitors."""
+
+    c_df: pd.DataFrame
+    _optimization_config_list: list[CapacitorOptimizationDto]
+
+    def __init__(self):
+        self._optimization_config_list = []
+
+    @staticmethod
+    def verify_optimization_parameter(toml_capacitor: TomlCapacitorSelection) -> tuple[bool, str]:
+        """
+        Verify the parameters from toml file for the capacitor optimization.
+
+        Dummy method so far.
+
+        :param toml_capacitor: capacitor toml file to check
+        :type toml_capacitor: TomlCapacitorSelection
+        :return: is_consistent, issue_report
+        :rtype: tuple[bool, str]
+        """
+        if toml_capacitor:
+            pass
+        return True, ""
+
+    def initialize_capacitor_selection(self, toml_capacitor: TomlCapacitorSelection, study_data: StudyData, circuit_filter_data: FilterData) -> None:
+        """
+        Initialize the capacitor selection.
+
+        :param toml_capacitor: capacitor data
+        :type toml_capacitor: TomlCapacitorSelection
+        :param study_data: capacitor study data
+        :type study_data: StudyData
+        :param circuit_filter_data: filtered circuit data
+        :type circuit_filter_data: FilterData
+        """
+        pecst.download_esr_csv_files()
+
+        # Create the io_config_list for all trials
+        for circuit_trial_file in circuit_filter_data.filtered_list_files:
+            circuit_filepath = os.path.join(circuit_filter_data.filtered_list_pathname, f"{circuit_trial_file}.pkl")
+            # Check filename
+            if os.path.isfile(circuit_filepath):
+                # Read results from circuit optimization
+                circuit_dto = HandleDabDto.load_from_file(circuit_filepath)
+
+                optimization_directory = os.path.join(study_data.optimization_directory, circuit_trial_file, study_data.study_name)
+
+                # figure out worst case working point for the capacitor per circuit design
+                sorted_max_rms_angles, i_c1_max_rms_current_waveform = HandleDabDto.get_max_rms_waveform_capacitor(circuit_dto, plot=False)
+                time = sorted_max_rms_angles / (2 * np.pi * circuit_dto.input_config.fs)
+                v_max = np.max(circuit_dto.input_config.mesh_v1)
+
+                # generate capacitor requirements from circuit simulation data
+                capacitor_requirements_dto = pecst.CapacitorRequirements(
+                    maximum_peak_to_peak_voltage_ripple=toml_capacitor.maximum_peak_to_peak_voltage_ripple,
+                    current_waveform_for_op_max_current=np.array([time, i_c1_max_rms_current_waveform]),
+                    v_dc_for_op_max_voltage=v_max,
+                    temperature_ambient=toml_capacitor.temperature_ambient,
+                    voltage_safety_margin_percentage=toml_capacitor.voltage_safety_margin_percentage,
+                    capacitor_type_list=[pecst.CapacitorType.FilmCapacitor],
+                    maximum_number_series_capacitors=toml_capacitor.maximum_number_series_capacitors,
+                    capacitor_tolerance_percent=pecst.CapacitanceTolerance.TenPercent,
+                    lifetime_h=toml_capacitor.lifetime_h,
+                    results_directory=optimization_directory
+                )
+
+                # Initialize the statistical data
+                stat_data_init: ProgressData = ProgressData(run_time=0, number_of_filtered_points=0,
+                                                            progress_status=ProgressStatus.Idle)
+
+                capacitor_dto = CapacitorOptimizationDto(
+                    circuit_filtered_point_filename=circuit_trial_file,
+                    progress_data=copy.deepcopy(stat_data_init),
+                    capacitor_optimization_dto=capacitor_requirements_dto)
+
+                self._optimization_config_list.append(capacitor_dto)
+            else:
+                logger.info(f"Wrong path or file {circuit_filepath} does not exists!")
+
+    @staticmethod
+    def _start_optimization(circuit_filtered_point_file: str, act_cst_config: pecst.CapacitorRequirements, filter_data: FilterData,
+                            debug: Debug) -> int:
+        # capacitor requirements
+        _, c_db_df_list = pecst.select_capacitors(act_cst_config)
+
+        c_db_df = pd.concat(c_db_df_list)
+
+        if not os.path.exists(act_cst_config.results_directory):
+            os.makedirs(act_cst_config.results_directory)
+        c_db_df.to_csv(f"{act_cst_config.results_directory}/results.csv")
+
+        df_filtered = pecst.filter_df(c_db_df)
+        if debug.general.is_debug:
+            # reduce dataset to the given number from the debug configuration
+            df_filtered = df_filtered.iloc[:debug.capacitor_1.number_working_point_max]
+        # save Pareto front designs (reduced in case of active debugging)
+        df_filtered.to_csv(f"{act_cst_config.results_directory}/results_filtered.csv")
+
+        # Load configuration
+        circuit_dto = HandleDabDto.load_from_file(os.path.join(filter_data.filtered_list_pathname, f"{circuit_filtered_point_file}.pkl"))
+
+        # sweep through all current waveforms
+        i_l1_sorted = np.transpose(circuit_dto.calc_currents.i_l_1_sorted, (1, 2, 3, 0))
+        angles_rad_sorted = np.transpose(circuit_dto.calc_currents.angles_rad_sorted, (1, 2, 3, 0))
+
+        all_operation_point_ordering_codes_list = df_filtered["ordering code"].to_numpy()
+        all_operation_point_volume_list = df_filtered["volume_total"].to_numpy()
+        all_operation_point_area_list = df_filtered["area_total"].to_numpy()
+        all_operation_point_n_series_list = df_filtered["in_series_needed"].to_numpy()
+        all_operation_point_n_parallel_list = df_filtered["in_parallel_needed"].to_numpy()
+
+        # Overtake the filtered operation points
+        number_of_filtered_points = len(all_operation_point_ordering_codes_list)
+
+        logger.info(f"Full-operating point simulation list: {all_operation_point_ordering_codes_list}")
+
+        # simulate all operating points
+        for count, ordering_code in enumerate(tqdm.tqdm(all_operation_point_ordering_codes_list)):
+
+            volume_total = all_operation_point_volume_list[count]
+            area_total = all_operation_point_area_list[count]
+            df_geometry_re_simulation_number = df_filtered[df_filtered["ordering code"] == ordering_code]
+            n_series = all_operation_point_n_series_list[count]
+            n_parallel = all_operation_point_n_parallel_list[count]
+
+            logger.debug(f"ordering_code: \n"
+                         f"    {df_geometry_re_simulation_number.head()}")
+
+            loss_total_array = np.full_like(circuit_dto.calc_modulation.phi, np.nan)
+
+            new_circuit_dto_directory = os.path.join(act_cst_config.results_directory, "01_circuit_dtos_incl_capacitor_1_loss")
+            if not os.path.exists(new_circuit_dto_directory):
+                os.makedirs(new_circuit_dto_directory)
+
+            if os.path.exists(os.path.join(new_circuit_dto_directory, f"{ordering_code}.pkl")):
+                logger.info(f"Re-simulation of {circuit_dto.name} already exists. Skip.")
+            else:
+                for vec_vvp in np.ndindex(circuit_dto.calc_modulation.phi.shape):
+                    time, unique_indices = np.unique(full_angle_waveform_from_angles(
+                                                     angles_rad_sorted[vec_vvp]) / 2 / np.pi / circuit_dto.input_config.fs, return_index=True)
+                    current = full_current_waveform_from_currents(i_l1_sorted[vec_vvp])[unique_indices]
+
+                    current_waveform = np.array([time, current])
+                    logger.debug(f"{current_waveform=}")
+                    logger.debug("All operating point simulation of:")
+                    logger.debug(f"   * Circuit study: {filter_data.circuit_study_name}")
+                    logger.debug(f"   * Circuit trial: {circuit_filtered_point_file}")
+                    logger.debug(f"   * Inductor re-simulation trial: {ordering_code}")
+
+                    [frequency_list, current_amplitude_list, _] = pecst.fft(current_waveform, plot='no', mode='time', title='fft input current')
+
+                    loss_per_capacitor = pecst.power_loss_film_capacitor(ordering_code, frequency_list, current_amplitude_list,
+                                                                         number_parallel_capacitors=n_parallel)
+
+                    loss_total_array[vec_vvp] = loss_per_capacitor * n_series * n_parallel
+
+                capacitor_losses = CapacitorResults(
+                    p_combined_losses=loss_total_array,
+                    volume=volume_total,
+                    pcb_area=area_total,
+                    circuit_trial_file=circuit_filtered_point_file,
+                    capacitor_order_number=df_geometry_re_simulation_number,
+                )
+
+                pickle_file = os.path.join(new_circuit_dto_directory, f"{ordering_code}.pkl")
+                with open(pickle_file, 'wb') as output:
+                    pickle.dump(capacitor_losses, output, pickle.HIGHEST_PROTOCOL)
+
+        # returns the number of filtered results
+        return number_of_filtered_points
+
+    def optimization_handler(self, filter_data: FilterData, debug: Debug) -> None:
+        """
+        Control the multi simulation processes.
+
+        :param filter_data: Information about the filtered designs
+        :type  filter_data: dct.FilterData
+        :param debug: True to use debug mode which stops earlier
+        :type debug: bool
+        """
+        number_cpus = cpu_count()
+
+        with Pool(processes=number_cpus) as pool:
+            parameters = []
+            for count, act_optimization_configuration in enumerate(self._optimization_config_list):
+                if debug.general.is_debug:
+                    # in debug mode, stop when number of configuration parameters has reached the same as parallel cores are used
+                    if count == number_cpus:
+                        break
+
+                parameters.append((
+                    act_optimization_configuration.circuit_filtered_point_filename,
+                    act_optimization_configuration.capacitor_optimization_dto,
+                    filter_data,
+                    debug
+                ))
+
+            pool.starmap(func=CapacitorSelection._start_optimization, iterable=parameters)

dct/datasets.py

@@ -117,8 +117,10 @@ def init_config(name: str, mesh_v1: np.ndarray, mesh_v2: np.ndarray, mesh_p: np.
             gecko_additional_params=gecko_additional_params,
             gecko_results=None,
             gecko_waveforms=None,
+            capacitor_1_results=None,
+            capacitor_2_results=None,
             inductor_results=None,
-            stacked_transformer_results=None
+            stacked_transformer_results=None,
         )
         return dab_dto
 
@@ -405,6 +407,44 @@ def get_max_peak_waveform_inductor(dab_dto: d_dtos.CircuitDabDTO, plot: bool = F
 
         return sorted_max_angles, i_l1_max_current_waveform
 
+    @staticmethod
+    def get_max_rms_waveform_capacitor(dab_dto: d_dtos.CircuitDabDTO, plot: bool = False) -> tuple[np.ndarray, np.ndarray]:
+        """
+        Get the capacitor waveform with the maximum RMS current out of the three-dimensional simulation array (v_1, v_2, P).
+
+        :param dab_dto: DAB data transfer object (DTO)
+        :type dab_dto: d_dtos.CircuitDabDTO
+        :param plot: True to plot the results, mostly for understanding and debugging
+        :type plot: bool
+        :return: sorted_max_rms_angles, i_hf1_max_rms_current_waveform, All as a numpy array.
+        :rtype: List[np.ndarray]
+        """
+        i_c1_sorted = np.transpose(dab_dto.calc_currents.i_hf_1_sorted, (1, 2, 3, 0))
+        i_c1_rms = dab_dto.calc_currents.i_hf_1_rms
+        angles_rad_sorted = np.transpose(dab_dto.calc_currents.angles_rad_sorted, (1, 2, 3, 0))
+
+        max_index = (0, 0, 0)
+        for vec_vvp in np.ndindex(dab_dto.calc_modulation.phi.shape):
+            max_index = vec_vvp if np.max(i_c1_rms[vec_vvp]) > np.max(i_c1_rms[max_index]) else max_index  # type: ignore
+            if plot:
+                plt.plot(d_waveforms.full_angle_waveform_from_angles(angles_rad_sorted[vec_vvp]),
+                         d_waveforms.full_current_waveform_from_currents(i_c1_sorted[vec_vvp]), color='gray')
+
+        i_c1_max_rms_current_waveform = i_c1_sorted[max_index]
+
+        sorted_max_rms_angles, unique_indices = np.unique(d_waveforms.full_angle_waveform_from_angles(angles_rad_sorted[max_index]), return_index=True)
+        i_c1_max_rms_current_waveform = d_waveforms.full_current_waveform_from_currents(i_c1_max_rms_current_waveform)[unique_indices]
+
+        if plot:
+            plt.plot(sorted_max_rms_angles, i_c1_max_rms_current_waveform, color='red', label='peak current full waveform')
+            plt.grid()
+            plt.xlabel('Angle in rad')
+            plt.ylabel('Current in A')
+            plt.legend()
+            plt.show()
+
+        return sorted_max_rms_angles, i_c1_max_rms_current_waveform
+
     @staticmethod
     def export_transformer_target_parameters_dto(dab_dto: d_dtos.CircuitDabDTO) -> d_dtos.TransformerTargetParameters:
         """

dct/datasets_dtos.py

@@ -171,6 +171,23 @@ def __init__(self, **kwargs):
             if k in names:
                 setattr(self, k, v)
 
+
+@dataclasses.dataclass(init=False)
+class CapacitorResults:
+    """DTO contains the inductor losses."""
+
+    loss_total_array: np.ndarray
+    volume_total: float
+    area_total: float
+    circuit_trial_file: str
+    capacitor_order_number: int
+
+    def __init__(self, **kwargs):
+        names = set([f.name for f in dataclasses.fields(self)])
+        for k, v in kwargs.items():
+            if k in names:
+                setattr(self, k, v)
+
 @dataclasses.dataclass(init=False)
 class InductorResults:
     """DTO contains the inductor losses."""
@@ -292,6 +309,8 @@ class CircuitDabDTO:
     gecko_additional_params: GeckoAdditionalParameters
     gecko_results: GeckoResults | None
     gecko_waveforms: GeckoWaveforms | None
+    capacitor_1_results: CapacitorResults | None
+    capacitor_2_results: CapacitorResults | None
     inductor_results: InductorResults | None
     stacked_transformer_results: StackedTransformerResults | None