Skip to content

update_inv_costs

NexusePlugin 

NexusePlugin(parameters: dict, scenario: Scenario)

Bases: Plugin

Updates investment cost data in the database using cost data from CSV files.

Implements: simulation.Module Protocol

Features: - Supports Low, Reference, High cost scenarios - Automatically annualizes investment costs using WACC and lifetime parameters - Converts currencies using configurable conversion factors from CSV - Maps generator types to cost data technology categories

Source code in src/plugins/update_inv_costs/nexus_e_plugin.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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("update_inv_costs only works with a MySQL database")

    # Suppress verbose MySQL connector logs
    logging.getLogger('mysql.connector').setLevel(logging.WARNING)

    # Add connection timeout to fail faster - increased for slow networks
    # Disable SSL/TLS to avoid handshake hangs
    connection_string = (
        f"mysql+mysqlconnector://{self.__data_context.user}:"
        f"{self.__data_context.password}@{self.__data_context.host}/"
        f"{self.__data_context.name}?connect_timeout=10&"
        f"ssl_disabled=true&use_pure=true"
    )
    self.__engine = create_engine(
        connection_string,
        pool_timeout=10,
        pool_recycle=300,
        pool_pre_ping=False
    )

get_default_parameters classmethod 

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

run 

run() -> None
Source code in src/plugins/update_inv_costs/nexus_e_plugin.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def run(self) -> None:
    logging.debug("Starting Investment cost data update process...")

    # Test database connection first
    self.database_available = False
    try:
        with self.__engine.connect() as conn:
            conn.execute(text("SELECT 1"))
            self.database_available = True
    except Exception as e:
        logging.error(f"Failed to connect to database: {e}")
        self.database_available = False

    self.__update_cost()

Parameters dataclass 

Parameters(cost_scenario: Literal['Low', 'Reference', 'High'] = 'Reference')

cost_scenario class-attribute instance-attribute 

cost_scenario: Literal['Low', 'Reference', 'High'] = 'Reference'