Skip to content

postprocess

NexusePlugin

NexusePlugin(parameters: dict, scenario: Scenario)

Bases: Plugin

Source code in src/plugins/postprocess/nexus_e_plugin.py
def __init__(self, parameters: dict, scenario: Scenario):
    self.__data_context = scenario.get_data_context()
    if self.__data_context.type != "mysql":
        raise ValueError("postprocess only works with a MySQL database")
    if "output_host" not in parameters:
        parameters["output_host"] = self.__data_context.host
    if "output_port" not in parameters:
        logging.warning("Default output database port is set to 3307 for legacy reasons.")
    if "output_user" not in parameters:
        parameters["output_user"] = self.__data_context.user
    if "output_password" not in parameters:
        parameters["output_password"] = self.__data_context.password
    if "output_name" not in parameters:
        parameters["output_name"] = self.__data_context.name
        logging.warning(
            (
                "No output_name given, defaulting to data_context name: "
                + f"{parameters['output_name']}"
            )
        )
    self.__parameters = Parameters(**parameters)
    self.__simulation_results = SimulationResults(
        results_path = self.__parameters.results_path,
    )

get_default_parameters classmethod

get_default_parameters() -> dict
Source code in src/plugins/postprocess/nexus_e_plugin.py
@classmethod
def get_default_parameters(cls) -> dict:
    return asdict(Parameters())

move_to_mysql

move_to_mysql()

Here should be re-implemented the following logic taken from postProcess.m:

disp('---------------- MySQL -----------------')
    try
        cd(workingDir);
        command = ['python'...
            ' moveToMysql.py'...
            ' --simu-name ' simuName...
            ' --scen-name "' scenShortName...
            '" --version-wv ' wspace.v_wv];
        disp(command)
        system(ACTIVATE_ENVIRONMENT + command);
    catch
        disp("FAILED to upload results to the output database.")
    end
Source code in src/plugins/postprocess/nexus_e_plugin.py
def move_to_mysql(self):
    """ Here should be re-implemented the following logic taken from 
    postProcess.m:

        disp('---------------- MySQL -----------------')
            try
                cd(workingDir);
                command = ['python'...
                    ' moveToMysql.py'...
                    ' --simu-name ' simuName...
                    ' --scen-name "' scenShortName...
                    '" --version-wv ' wspace.v_wv];
                disp(command)
                system(ACTIVATE_ENVIRONMENT + command);
            catch
                disp("FAILED to upload results to the output database.")
            end
    """
    logging.info("Moving results to the output database")
    moveToMysql.main(
        simulation_postprocess_path = self.__simulation_results.postprocess_path,
        simulation_execution_date = self.__parameters.execution_date,
        scenario = self.__parameters.output_name,
        webviewer_version = "results",
        host=self.__parameters.output_host,
        port=self.__parameters.output_port,
        user=self.__parameters.output_user,
        password=self.__parameters.output_password
    )
    logging.info("DONE")

run

run() -> None
Source code in src/plugins/postprocess/nexus_e_plugin.py
def run(self) -> None:
    self.write_metadata()
    if os.path.exists(self.__parameters.plot_config_file_path):
        self.__simulation_results.copy_plot_config(
            self.__parameters.plot_config_file_path
        )
    if self.__parameters.centiv:
        centiv_postprocess = CentivPostprocess(
            results_path=self.__parameters.results_path,
            input_data_name=self.__data_context.name,
            input_host=self.__data_context.host,
            input_user=self.__data_context.user,
            input_password=self.__data_context.password,
            single_electric_node=self.__parameters.single_electric_node,
        )
        centiv_postprocess.run()
    if self.__parameters.cascades:
        cascades_postprocess = CascadesPostprocess(
            self.__simulation_results.postprocess_path
        )
        cascades_postprocess.run()
    if self.__parameters.move_to_mysql:
        self.move_to_mysql()
    logging.info("Postprocess finished")

write_metadata

write_metadata()
Source code in src/plugins/postprocess/nexus_e_plugin.py
def write_metadata(self):
    metadata = self.__simulation_results.get_metadata(
        scenario_description = self.__parameters.scenario_description,
    )
    self.__simulation_results.write_csv_in_postprocess(
        data_to_write = metadata,
        file_name = "metadata.csv"
    )

Parameters dataclass

Parameters(
    results_path: str = "",
    plot_config_file_path: str = "",
    centiv: bool = False,
    cascades: bool = False,
    move_to_mysql: bool = False,
    scenario_description: str = "",
    execution_date: str = "",
    output_name: str = "",
    output_host: str = "",
    output_port: str = "3307",
    output_user: str = "",
    output_password: str = "",
    single_electric_node: bool = False,
)

cascades class-attribute instance-attribute

cascades: bool = False

centiv class-attribute instance-attribute

centiv: bool = False

execution_date class-attribute instance-attribute

execution_date: str = ''

move_to_mysql class-attribute instance-attribute

move_to_mysql: bool = False

output_host class-attribute instance-attribute

output_host: str = ''

output_name class-attribute instance-attribute

output_name: str = ''

output_password class-attribute instance-attribute

output_password: str = ''

output_port class-attribute instance-attribute

output_port: str = '3307'

output_user class-attribute instance-attribute

output_user: str = ''

plot_config_file_path class-attribute instance-attribute

plot_config_file_path: str = ''

results_path class-attribute instance-attribute

results_path: str = ''

scenario_description class-attribute instance-attribute

scenario_description: str = ''

single_electric_node class-attribute instance-attribute

single_electric_node: bool = False