Skip to content

upload_res_data

NexusePlugin 

NexusePlugin(parameters: dict, scenario: Scenario)

Bases: Plugin

Source code in src/plugins/upload_res_data/nexus_e_plugin.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, parameters: dict, scenario: Scenario):
    self.__settings = Parameters(**parameters)
    self.__data_context = scenario.get_data_context()
    if self.__data_context.type != "mysql":
        raise ValueError("upload_res_data only works with a MySQL database")

    # Copy database if requested
    if self.__settings.copy_database:
        if not self.__settings.new_dbName:
            raise ValueError("new_dbName must be specified when copy_database is True")
        self.__copy_database()
        target_db = self.__settings.new_dbName
    else:
        target_db = self.__data_context.name

    logging.info(f"Connecting to database: {target_db} on host: {self.__data_context.host}")
    self.__engine = create_engine((
        f"mysql+pymysql://{self.__data_context.user}:{self.__data_context.password}"
        f"@{self.__data_context.host}/{target_db}"))

    # Dictionary to store GenName -> idProfile mapping for current upload session
    self.__genname_to_idprofile = {}

get_default_parameters classmethod 

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

run 

run() -> None
Source code in src/plugins/upload_res_data/nexus_e_plugin.py 
59
60
61
62
63
64
65
66
67
68
69
def run(self) -> None:
    SQLModel.metadata.create_all(self.__engine)
    self.__ensure_gendata_columns()
    if self.__settings.upload_wind:
        self.__upload_wind()
    if self.__settings.upload_alpine_PV:
        self.__upload_alpine_PV()
    if self.__settings.upload_pv:
        self.__upload_pv()
    if self.__settings.upload_ror:
        self.__upload_ror()

Parameters dataclass 

Parameters(
    upload_pv: bool = False,
    pv_data_year: str = "weather year",
    pv_data_scenario: str = "aggregation level",
    upload_ror: bool = False,
    ror_data_scenario: str = "historical",
    upload_wind: bool = False,
    upload_alpine_PV: bool = False,
    enable_pathway_runs: bool = False,
    biodiversity_scenario: str = "current",
    public_acceptance_scenario: str = "00",
    wind_data_scenario: str = "cat:(1-1)",
    wind_rcp_scenario: str = "rcp85",
    copy_database: bool = False,
    new_dbName: str | None = None,
)

biodiversity_scenario class-attribute instance-attribute 

biodiversity_scenario: str = 'current'

copy_database class-attribute instance-attribute 

copy_database: bool = False

enable_pathway_runs class-attribute instance-attribute 

enable_pathway_runs: bool = False

new_dbName class-attribute instance-attribute 

new_dbName: str | None = None

public_acceptance_scenario class-attribute instance-attribute 

public_acceptance_scenario: str = '00'

pv_data_scenario class-attribute instance-attribute 

pv_data_scenario: str = 'aggregation level'

pv_data_year class-attribute instance-attribute 

pv_data_year: str = 'weather year'

ror_data_scenario class-attribute instance-attribute 

ror_data_scenario: str = 'historical'

upload_alpine_PV class-attribute instance-attribute 

upload_alpine_PV: bool = False

upload_pv class-attribute instance-attribute 

upload_pv: bool = False

upload_ror class-attribute instance-attribute 

upload_ror: bool = False

upload_wind class-attribute instance-attribute 

upload_wind: bool = False

wind_data_scenario class-attribute instance-attribute 

wind_data_scenario: str = 'cat:(1-1)'

wind_rcp_scenario class-attribute instance-attribute 

wind_rcp_scenario: str = 'rcp85'