Code Reference

framjules

JulES

JulES

Bases: Solver

Solver class for JulES - a fundamental energy market simulation model for operational planning.

The JulES solver takes a populated Model, and uses the configuration set by the user in JulESConfig to run JulES and read results back into the Model. The main steps of the JulES solver class are described below: 1. Transform Model into JulES compatible Components (SolveHandler) 2. Build JulES input files (BuildHandler) 3. Configure JulES according to the JulESConfig (ConfigHandler) 4. Run JulES (and setup Julia environment if needed) (RunHandler) 5. Read results from JulES back into the Model (ResultsHandler)

SolveHandler - Initialize - Transform the Model Components into JulES compatible Components. JulES uses the main Components Node and Flow to represent the energy system. - If Aggregators are set in JulEConfig.short_term_aggregations, make an aggregated version of Model. The JulES, algorithm has different problems that will use different Models. The clearing, stochastic subsystem and end value problems will use the full detailed Model, while the price prognosis problem will use the aggregated Model. SolveHandler also makes the mappings between the detailed and aggregated Models to couple the problems. - JulES only support the commodities Power, Hydro and Battery at the moment. All commodities in Model will be mapped to these. The main property of the commodity in JulES is the horizon (with type, duration and resolution). Power is the default commodity, and all commodities with no storage will be mapped to Power. Battery represents short-term storage commodities with a detailed time resolution, while Hydro represents long-term storage commodities with a coarser time resolution. - Identify storagesystems (e.g. watersheds or batteries), and identify if they are long-term or short-term storage systems. Storage systems are short-term if all storages in the subsystem have lower storage duration than JulESConfig.get_short_term_storage_cutoff_hours(). All storage subsystems in the same category (short-term or long-term) will get the same storage commodities and horizons, problem structure and end-condition type. - Short-term: StochSubsystem, startequalstop, no skipmed, Battery commodity, short horizon duration. - Long-term: EVP and/or StochSubsystem, endvalues from ppp, skipmed, Hydro commodity, long horizon duration. - TODO: This implementation is built around the first JulES version and will be improved in the future. We would like to add more tailored configurations for each storage system. Also make models for each problem and subsystem in JulES, not just a detailed and aggregated version of Model that JulES has to derive all problems from.

BuildHandler - Build JulES input files. - Write JulES input files for the detailed and aggregated elements, together with their timevectors. - Write JulES input files for detailed and aggregated start storages. - Write JulES input file for the mapping between detailed and aggregated storages.

ConfigHandler - Configure JulES according to the JulESConfig set by the user and the Model. - Simulation mode, simulation periods, weather years and scenario generation. - Number of CPU cores to use, parallelization settings and optimization solvers. - Problem structure and horizons for each problem and commodity. Horizon type, duration and resolution. - The problem structure will in most cases consist of the following, which are run for each simulation step: - Deterministic price prognosis problems for each weather scenario - Deterministic end value problems for each storage subsystem and weather scenario - Stochastic (two-stage) subsystems problem for each storage subsystem - Market clearing problem - Exception 1: If there are no storages in the system, only the market clearing problem will be run. TODO: Should also check if there are other constraints coupling time periods. Then we need the price prognosis problems. - Exception 2: If there is only exogenous market nodes only stochastic subsystem problems will not be run. - TODO: Improve configuration possibilities for the different problem structure cases. - Result settings. - Turn on or off various JulES features.

See JulES documentation at https://nve.github.io/JulES/ for more.

Methods:

Name	Description
__init__	Initializes the solver with default configuration.
get_config	Returns the internal configuration object for customization.
solve	Model): Solves the given model using JulES. Parent class method (Solver).

Source code in framjules/JulES.py

class JulES(Solver):
    """Solver class for JulES - a fundamental energy market simulation model for operational planning.

    The JulES solver takes a populated Model, and uses the configuration set by the user in JulESConfig to run
    JulES and read results back into the Model. The main steps of the JulES solver class are described below:
    1. Transform Model into JulES compatible Components (SolveHandler)
    2. Build JulES input files (BuildHandler)
    3. Configure JulES according to the JulESConfig (ConfigHandler)
    4. Run JulES (and setup Julia environment if needed) (RunHandler)
    5. Read results from JulES back into the Model (ResultsHandler)

    SolveHandler - Initialize
    - Transform the Model Components into JulES compatible Components. JulES uses the main Components
    Node and Flow to represent the energy system.
    - If Aggregators are set in JulEConfig.short_term_aggregations, make an aggregated version of Model. The JulES,
    algorithm has different problems that will use different Models. The clearing, stochastic subsystem and end
    value problems will use the full detailed Model, while the price prognosis problem will use the aggregated Model.
    SolveHandler also makes the mappings between the detailed and aggregated Models to couple the problems.
    - JulES only support the commodities Power, Hydro and Battery at the moment. All commodities in Model will be mapped
    to these. The main property of the commodity in JulES is the horizon (with type, duration and resolution). Power is
    the default commodity, and all commodities with no storage will be mapped to Power. Battery represents short-term
    storage commodities with a detailed time resolution, while Hydro represents long-term storage commodities with a
    coarser time resolution.
    - Identify storagesystems (e.g. watersheds or batteries), and identify if they are long-term or short-term
    storage systems. Storage systems are short-term if all storages in the subsystem have lower storage duration than
    JulESConfig.get_short_term_storage_cutoff_hours(). All storage subsystems in the same category
    (short-term or long-term) will get the same storage commodities and horizons, problem structure and end-condition
    type.
        - Short-term: StochSubsystem, startequalstop, no skipmed, Battery commodity, short horizon duration.
        - Long-term: EVP and/or StochSubsystem, endvalues from ppp, skipmed, Hydro commodity, long horizon duration.
    - TODO: This implementation is built around the first JulES version and will be improved in the future.
    We would like to add more tailored configurations for each storage system. Also make models for each problem and
    subsystem in JulES, not just a detailed and aggregated version of Model that JulES has to derive all problems from.

    BuildHandler - Build JulES input files.
    - Write JulES input files for the detailed and aggregated elements, together with their timevectors.
    - Write JulES input files for detailed and aggregated start storages.
    - Write JulES input file for the mapping between detailed and aggregated storages.

    ConfigHandler - Configure JulES according to the JulESConfig set by the user and the Model.
    - Simulation mode, simulation periods, weather years and scenario generation.
    - Number of CPU cores to use, parallelization settings and optimization solvers.
    - Problem structure and horizons for each problem and commodity. Horizon type, duration and resolution.
        - The problem structure will in most cases consist of the following, which are run for each simulation step:
            - Deterministic price prognosis problems for each weather scenario
            - Deterministic end value problems for each storage subsystem and weather scenario
            - Stochastic (two-stage) subsystems problem for each storage subsystem
            - Market clearing problem
        - Exception 1: If there are no storages in the system, only the market clearing problem will be run.
            TODO: Should also check if there are other constraints coupling time periods.
            Then we need the price prognosis problems.
        - Exception 2: If there is only exogenous market nodes only stochastic subsystem problems will not be run.
        - TODO: Improve configuration possibilities for the different problem structure cases.
    - Result settings.
    - Turn on or off various JulES features.

    See JulES documentation at https://nve.github.io/JulES/ for more.

    Methods:
        __init__(): Initializes the solver with default configuration.
        get_config(): Returns the internal configuration object for customization.
        solve(model: Model): Solves the given model using JulES. Parent class method (Solver).

    """

    def __init__(self) -> None:
        """Create new JulES solver with default config set."""
        super().__init__()
        self._config = JulESConfig()

    def get_config(self) -> JulESConfig:
        """Get internal config object. Modify this to configure JulES."""
        return self._config

    def _solve(
        self,
        folder: Path,
        model: Model,
    ) -> None:
        t0 = time()
        if _PROFILE_PERFORMANCE:
            profiler = cProfile.Profile()
            profiler.enable()
        handler = SolveHandler(folder, model, self.get_config())
        self.send_debug_event(f"SolveHandler time: {round(time() - t0, 2)} seconds")
        if _PROFILE_PERFORMANCE:
            profiler.disable()  # Stop profiling
            profiler.dump_stats("profile_solvehandler_init.prof")

        t = time()
        if _PROFILE_PERFORMANCE:
            profiler = cProfile.Profile()
            profiler.enable()
        handler.build()
        if _PROFILE_PERFORMANCE:
            profiler.disable()  # Stop profiling
            profiler.dump_stats("profile_solvehandler_build.prof")
        self.send_debug_event(f"build time: {round(time() - t, 2)} seconds")

        t = time()
        handler.configure()
        self.send_debug_event(f"configure time: {round(time() - t, 2)} seconds")

        t = time()
        loaders: set[Loader] = set()
        add_loaders(loaders, model)
        for loader in loaders:
            loader.clear_cache()
        gc.collect()
        self.send_debug_event(f"clear_cache time: {round(time() - t, 2)} seconds")

        t = time()
        handler.run()
        self.send_debug_event(f"run time: {round(time() - t, 2)} seconds")

        t = time()
        if _PROFILE_PERFORMANCE:
            profiler = cProfile.Profile()
            profiler.enable()
        handler.set_results()
        if _PROFILE_PERFORMANCE:
            profiler.disable()  # Stop profiling
            profiler.dump_stats("profile_solvehandler_results.prof")
        self.send_debug_event(f"set_results time: {round(time() - t, 2)} seconds")

        self.send_debug_event(f"JulES._solve time: {round(time() - t0, 2)} seconds")

__init__() -> None

Create new JulES solver with default config set.

Source code in framjules/JulES.py

def __init__(self) -> None:
    """Create new JulES solver with default config set."""
    super().__init__()
    self._config = JulESConfig()

get_config() -> JulESConfig

Get internal config object. Modify this to configure JulES.

Source code in framjules/JulES.py

def get_config(self) -> JulESConfig:
    """Get internal config object. Modify this to configure JulES."""
    return self._config

JulESConfig

JulESConfig

Bases: SolverConfig

Class containing all config for JulES. Subclass of SolverConfig.

Source code in framjules/JulESConfig.py

class JulESConfig(SolverConfig):
    """Class containing all config for JulES. Subclass of SolverConfig."""

    def __init__(self) -> None:
        """Create new JulESConfig object."""
        super().__init__()

        self._julia_exe_path: Path | str | None = None
        self._julia_env_path: Path | str | None = None
        self._julia_depot_path: Path | str | None = None
        self._force_julia_install: bool = True

        self._branch_jules: str | None = None
        self._branch_tulipa: str | None = None

        self._short_term_aggregations: list[Aggregator] = []

        self._time_resolution = JulESTimeResolution()

        self._duration_clearing = timedelta(days=1)
        self._duration_short_term = timedelta(days=6)
        self._duration_medium_term = timedelta(days=364)
        self._duration_long_term = timedelta(days=364 * 5)

        self._market_resolution_clearing = timedelta(hours=1)
        self._market_resolution_short_term = timedelta(hours=2)

        self._market_num_blocks_medium_term = 5
        self._market_num_blocks_long_term = 4

        self._storage_resolution_clearing = timedelta(days=1)
        self._storage_resolution_short_term = timedelta(days=2)
        self._storage_resolution_medium_term = timedelta(days=7)
        self._storage_resolution_long_term = timedelta(days=28)

        self._short_term_storage_cutoff_hours = 24 * 7

        self._debug_short_opt_solver = False
        self._debug_med_opt_solver = False
        self._debug_long_opt_solver = False
        self._debug_end_value_opt_solver = False
        self._debug_subsystem_master_opt_solver = False
        self._debug_subsystem_sub_opt_solver = False
        self._debug_clearing_opt_solver = False
        self._is_cache_db = True
        self._skip_install_dependencies = False

        self._clearing_days = 2
        self._market_duration_minutes = 6 * 60
        self._storage_duration_minutes = 2 * 24 * 60
        self._lookahead_days = 5 * 365
        self._detail_level = "fast"

        self._skipmax_days = 6
        self._warn_skipmax_days = 32

    def set_skipmax_days(self, days: int) -> None:
        """Set number of days between calculation of medium and long term storage values.

        This can speed up a simulation as medium and long term price prognosis problems, and long term
        storage value problems are solved less often. The cost is less good storage values. The longer between
        re-calculation of storage values, the bigger negative impact on simulation result quality.

        If skipmax_days = 6 and clearing_days = 2, JulES will calculate medium and long
        term storage values every 3rd simulation step.

        Short term price prognosis problems and storage values problems (i.e. for batteries) are not affected by this
        setting, and are calculated every simulation step.
        """
        self._check_type(days, int)
        self._check_int(days, lower_bound=0, upper_bound=None)
        if days > self._warn_skipmax_days:
            message = (
                "Unusually high value for skipmax_days: "
                f"Medium and long term storage values updated only {days}th day. "
                "This can give poor simulation results due to poor storage utilization."
            )
            self.send_warning_event(message)
        self._skipmax_days = days

    def get_skipmax_days(self) -> int:
        """Get number of days between calculation of medium and long term storage values."""
        return self._skipmax_days

    def is_skip_install_dependencies(self) -> bool:
        """Return True if install julia dependencies will be skipped during JulES.solve."""
        return self._skip_install_dependencies

    def activate_skip_install_dependencies(self) -> None:
        """Tell JulES to not install julia dependencies, assuming they are already installed.

        Default is to install.
        """
        self._skip_install_dependencies = True

    def deactivate_skip_install_dependencies(self) -> None:
        """Tell JulES to install julia dependencies. (This is the default)."""
        self._skip_install_dependencies = False

    def is_cache_db(self) -> bool:
        """Return True if JulES is allowed to use a cache to store precomputed values while building."""
        return self._is_cache_db

    def activate_cache_db(self) -> None:
        """Activates use of cache db."""
        self._is_cache_db = True

    def deactivate_cache_db(self) -> None:
        """Activates use of db without cache."""
        self._is_cache_db = False

    def get_time_resolution(self) -> JulESTimeResolution:
        """Get time resolution object. Modify this to modify time resolution of JulES."""
        return self._time_resolution

    def get_short_term_storage_cutoff_hours(self) -> int:
        """Return number of hours.

        JulES will classify all storage subsystems with max storage duration less than cutoff as short term subsystems.
        """
        return self._short_term_storage_cutoff_hours

    def set_jules_version(self, jules_branch: str | None = None, tulipa_branch: str | None = None) -> None:
        """Set which version of JulES and/or TuLiPa to use.

        Can be a git branch name, or a local path to a git repo which will activate development mode.
        """
        self._check_type(jules_branch, (str, type(None)))
        self._check_type(tulipa_branch, (str, type(None)))
        if jules_branch is not None:
            self._branch_jules = jules_branch
        if tulipa_branch is not None:
            self._branch_tulipa = tulipa_branch
        if self._branch_tulipa is None and self._branch_jules is not None:
            self._branch_tulipa = self._branch_jules

    def get_jules_version(self) -> str | None:
        """Get JulES version.

        Can be a git branch name, or a local path to a git repo for use in development mode.
        """
        return self._branch_jules

    def get_tulipa_version(self) -> str | None:
        """Get TuLiPa git branch.

        Can be a git branch name, or a local path to a git repo for use in development mode.
        """
        return self._branch_tulipa

    def set_julia_depot_path(self, path: Path) -> None:
        """Set folder where Julia installs new packages."""
        self._check_type(path, Path)
        self._julia_depot_path = path

    def get_julia_depot_path(self) -> Path | None:
        """Get folder where Julia installs new packages."""
        return self._julia_depot_path

    def set_julia_env_path(self, path: Path) -> None:
        """Set which Julia environment to use."""
        self._check_type(path, Path)
        self._julia_env_path = path

    def get_julia_env_path(self) -> Path | None:
        """Get Julia environment being used."""
        return self._julia_env_path

    def set_julia_exe_path(self, path: Path) -> None:
        """Set which Julia installation to use."""
        self._check_type(path, Path)
        self._julia_exe_path = path

    def get_julia_exe_path(self) -> Path | None:
        """Get Julia installation being used."""
        return self._julia_exe_path

    def set_force_julia_install(self, flag: bool) -> bool:
        """Set bool for force new julia install."""
        self._force_julia_install = flag

    def get_force_julia_install(self) -> bool:
        """Get bool for force new julia install."""
        return self._force_julia_install

    def _check_supported_aggregators(self, aggregators: list[Aggregator]) -> None:
        for aggr in aggregators:
            if not isinstance(aggr, tuple(_SUPPORTED_AGGREGATORS)):
                message = (
                    f"Aggregator of type {type(aggr)} is not supported in JulES.",
                    f"Supported types are: {_SUPPORTED_AGGREGATORS}",
                )
                raise TypeError(message)

    def set_short_term_aggregations(self, aggregators: list[Aggregator]) -> None:
        """Set aggregations to create the short term model used in the price prognosis problems."""
        self._check_supported_aggregators(aggregators)
        self._short_term_aggregations = aggregators

    def get_short_term_aggregations(self) -> list[Aggregator]:
        """Get aggregations to create the short term model used in the price prognosis problems."""
        return self._short_term_aggregations

    """
    Debug optimization solvers of the different JulES problems.
    Helpful if problems are encountered during solving, most commonly infeasibility issues.
    Replaces TuLiPa.HiGHS_Prob with TuLiPa.JuMP_Prob which has better debugging features:
    - More checks while building the optimization problem
    - If infeasible, solve the problem again with relaxed constraints (with penalties) and return the broken constraints
    - Outputs the optimization problem with variable and constraint names from FRAM
    At the cost of performance, as JuMP_Prob is slower than HiGHS_Prob.

    JulES will now automatically switch to JuMP_Prob if HiGHS_Prob fails, but these can still be used for testing
    and debugging.
    """

    def set_debug_all_opt_solver(self, debug: bool) -> None:
        """Set whether to debug all optimization solvers."""
        self._debug_short_opt_solver = debug
        self._debug_med_opt_solver = debug
        self._debug_long_opt_solver = debug
        self._debug_end_value_opt_solver = debug
        self._debug_subsystem_master_opt_solver = debug
        self._debug_subsystem_sub_opt_solver = debug
        self._debug_clearing_opt_solver = debug

    def set_debug_short_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the short-term optimization solver."""
        self._debug_short_opt_solver = debug

    def get_debug_short_opt_solver(self) -> bool:
        """Get whether to debug the short-term optimization solver."""
        return self._debug_short_opt_solver

    def set_debug_med_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the medium-term optimization solver."""
        self._debug_med_opt_solver = debug

    def get_debug_med_opt_solver(self) -> bool:
        """Get whether to debug the medium-term optimization solver."""
        return self._debug_med_opt_solver

    def set_debug_long_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the long-term optimization solver."""
        self._debug_long_opt_solver = debug

    def get_debug_long_opt_solver(self) -> bool:
        """Get whether to debug the long-term optimization solver."""
        return self._debug_long_opt_solver

    def set_debug_end_value_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the end value optimization solver."""
        self._debug_end_value_opt_solver = debug

    def get_debug_end_value_opt_solver(self) -> bool:
        """Get whether to debug the end value optimization solver."""
        return self._debug_end_value_opt_solver

    def set_debug_subsystem_master_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the subsystem master optimization solver."""
        self._debug_subsystem_master_opt_solver = debug

    def get_debug_subsystem_master_opt_solver(self) -> bool:
        """Get whether to debug the subsystem master optimization solver."""
        return self._debug_subsystem_master_opt_solver

    def set_debug_subsystem_sub_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the subsystem sub optimization solver."""
        self._debug_subsystem_sub_opt_solver = debug

    def get_debug_subsystem_sub_opt_solver(self) -> bool:
        """Get whether to debug the subsystem sub optimization solver."""
        return self._debug_subsystem_sub_opt_solver

    def set_debug_clearing_opt_solver(self, debug: bool) -> None:
        """Set whether to debug the clearing optimization solver."""
        self._debug_clearing_opt_solver = debug

    def get_debug_clearing_opt_solver(self) -> bool:
        """Get whether to debug the clearing optimization solver."""
        return self._debug_clearing_opt_solver

__init__() -> None

Create new JulESConfig object.

Source code in framjules/JulESConfig.py

def __init__(self) -> None:
    """Create new JulESConfig object."""
    super().__init__()

    self._julia_exe_path: Path | str | None = None
    self._julia_env_path: Path | str | None = None
    self._julia_depot_path: Path | str | None = None
    self._force_julia_install: bool = True

    self._branch_jules: str | None = None
    self._branch_tulipa: str | None = None

    self._short_term_aggregations: list[Aggregator] = []

    self._time_resolution = JulESTimeResolution()

    self._duration_clearing = timedelta(days=1)
    self._duration_short_term = timedelta(days=6)
    self._duration_medium_term = timedelta(days=364)
    self._duration_long_term = timedelta(days=364 * 5)

    self._market_resolution_clearing = timedelta(hours=1)
    self._market_resolution_short_term = timedelta(hours=2)

    self._market_num_blocks_medium_term = 5
    self._market_num_blocks_long_term = 4

    self._storage_resolution_clearing = timedelta(days=1)
    self._storage_resolution_short_term = timedelta(days=2)
    self._storage_resolution_medium_term = timedelta(days=7)
    self._storage_resolution_long_term = timedelta(days=28)

    self._short_term_storage_cutoff_hours = 24 * 7

    self._debug_short_opt_solver = False
    self._debug_med_opt_solver = False
    self._debug_long_opt_solver = False
    self._debug_end_value_opt_solver = False
    self._debug_subsystem_master_opt_solver = False
    self._debug_subsystem_sub_opt_solver = False
    self._debug_clearing_opt_solver = False
    self._is_cache_db = True
    self._skip_install_dependencies = False

    self._clearing_days = 2
    self._market_duration_minutes = 6 * 60
    self._storage_duration_minutes = 2 * 24 * 60
    self._lookahead_days = 5 * 365
    self._detail_level = "fast"

    self._skipmax_days = 6
    self._warn_skipmax_days = 32

activate_cache_db() -> None

Activates use of cache db.

Source code in framjules/JulESConfig.py

def activate_cache_db(self) -> None:
    """Activates use of cache db."""
    self._is_cache_db = True

activate_skip_install_dependencies() -> None

Tell JulES to not install julia dependencies, assuming they are already installed.

Default is to install.

Source code in framjules/JulESConfig.py

def activate_skip_install_dependencies(self) -> None:
    """Tell JulES to not install julia dependencies, assuming they are already installed.

    Default is to install.
    """
    self._skip_install_dependencies = True

deactivate_cache_db() -> None

Activates use of db without cache.

Source code in framjules/JulESConfig.py

def deactivate_cache_db(self) -> None:
    """Activates use of db without cache."""
    self._is_cache_db = False

deactivate_skip_install_dependencies() -> None

Tell JulES to install julia dependencies. (This is the default).

Source code in framjules/JulESConfig.py

def deactivate_skip_install_dependencies(self) -> None:
    """Tell JulES to install julia dependencies. (This is the default)."""
    self._skip_install_dependencies = False

get_debug_clearing_opt_solver() -> bool

Get whether to debug the clearing optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_clearing_opt_solver(self) -> bool:
    """Get whether to debug the clearing optimization solver."""
    return self._debug_clearing_opt_solver

get_debug_end_value_opt_solver() -> bool

Get whether to debug the end value optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_end_value_opt_solver(self) -> bool:
    """Get whether to debug the end value optimization solver."""
    return self._debug_end_value_opt_solver

get_debug_long_opt_solver() -> bool

Get whether to debug the long-term optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_long_opt_solver(self) -> bool:
    """Get whether to debug the long-term optimization solver."""
    return self._debug_long_opt_solver

get_debug_med_opt_solver() -> bool

Get whether to debug the medium-term optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_med_opt_solver(self) -> bool:
    """Get whether to debug the medium-term optimization solver."""
    return self._debug_med_opt_solver

get_debug_short_opt_solver() -> bool

Get whether to debug the short-term optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_short_opt_solver(self) -> bool:
    """Get whether to debug the short-term optimization solver."""
    return self._debug_short_opt_solver

get_debug_subsystem_master_opt_solver() -> bool

Get whether to debug the subsystem master optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_subsystem_master_opt_solver(self) -> bool:
    """Get whether to debug the subsystem master optimization solver."""
    return self._debug_subsystem_master_opt_solver

get_debug_subsystem_sub_opt_solver() -> bool

Get whether to debug the subsystem sub optimization solver.

Source code in framjules/JulESConfig.py

def get_debug_subsystem_sub_opt_solver(self) -> bool:
    """Get whether to debug the subsystem sub optimization solver."""
    return self._debug_subsystem_sub_opt_solver

get_force_julia_install() -> bool

Get bool for force new julia install.

Source code in framjules/JulESConfig.py

def get_force_julia_install(self) -> bool:
    """Get bool for force new julia install."""
    return self._force_julia_install

get_jules_version() -> str | None

Get JulES version.

Can be a git branch name, or a local path to a git repo for use in development mode.

Source code in framjules/JulESConfig.py

def get_jules_version(self) -> str | None:
    """Get JulES version.

    Can be a git branch name, or a local path to a git repo for use in development mode.
    """
    return self._branch_jules

get_julia_depot_path() -> Path | None

Get folder where Julia installs new packages.

Source code in framjules/JulESConfig.py

def get_julia_depot_path(self) -> Path | None:
    """Get folder where Julia installs new packages."""
    return self._julia_depot_path

get_julia_env_path() -> Path | None

Get Julia environment being used.

Source code in framjules/JulESConfig.py

def get_julia_env_path(self) -> Path | None:
    """Get Julia environment being used."""
    return self._julia_env_path

get_julia_exe_path() -> Path | None

Get Julia installation being used.

Source code in framjules/JulESConfig.py

def get_julia_exe_path(self) -> Path | None:
    """Get Julia installation being used."""
    return self._julia_exe_path

get_short_term_aggregations() -> list[Aggregator]

Get aggregations to create the short term model used in the price prognosis problems.

Source code in framjules/JulESConfig.py

def get_short_term_aggregations(self) -> list[Aggregator]:
    """Get aggregations to create the short term model used in the price prognosis problems."""
    return self._short_term_aggregations

get_short_term_storage_cutoff_hours() -> int

Return number of hours.

JulES will classify all storage subsystems with max storage duration less than cutoff as short term subsystems.

Source code in framjules/JulESConfig.py

def get_short_term_storage_cutoff_hours(self) -> int:
    """Return number of hours.

    JulES will classify all storage subsystems with max storage duration less than cutoff as short term subsystems.
    """
    return self._short_term_storage_cutoff_hours

get_skipmax_days() -> int

Get number of days between calculation of medium and long term storage values.

Source code in framjules/JulESConfig.py

def get_skipmax_days(self) -> int:
    """Get number of days between calculation of medium and long term storage values."""
    return self._skipmax_days

get_time_resolution() -> JulESTimeResolution

Get time resolution object. Modify this to modify time resolution of JulES.

Source code in framjules/JulESConfig.py

def get_time_resolution(self) -> JulESTimeResolution:
    """Get time resolution object. Modify this to modify time resolution of JulES."""
    return self._time_resolution

get_tulipa_version() -> str | None

Get TuLiPa git branch.

Can be a git branch name, or a local path to a git repo for use in development mode.

Source code in framjules/JulESConfig.py

def get_tulipa_version(self) -> str | None:
    """Get TuLiPa git branch.

    Can be a git branch name, or a local path to a git repo for use in development mode.
    """
    return self._branch_tulipa

is_cache_db() -> bool

Return True if JulES is allowed to use a cache to store precomputed values while building.

Source code in framjules/JulESConfig.py

def is_cache_db(self) -> bool:
    """Return True if JulES is allowed to use a cache to store precomputed values while building."""
    return self._is_cache_db

is_skip_install_dependencies() -> bool

Return True if install julia dependencies will be skipped during JulES.solve.

Source code in framjules/JulESConfig.py

def is_skip_install_dependencies(self) -> bool:
    """Return True if install julia dependencies will be skipped during JulES.solve."""
    return self._skip_install_dependencies

set_debug_all_opt_solver(debug: bool) -> None

Set whether to debug all optimization solvers.

Source code in framjules/JulESConfig.py

def set_debug_all_opt_solver(self, debug: bool) -> None:
    """Set whether to debug all optimization solvers."""
    self._debug_short_opt_solver = debug
    self._debug_med_opt_solver = debug
    self._debug_long_opt_solver = debug
    self._debug_end_value_opt_solver = debug
    self._debug_subsystem_master_opt_solver = debug
    self._debug_subsystem_sub_opt_solver = debug
    self._debug_clearing_opt_solver = debug

set_debug_clearing_opt_solver(debug: bool) -> None

Set whether to debug the clearing optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_clearing_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the clearing optimization solver."""
    self._debug_clearing_opt_solver = debug

set_debug_end_value_opt_solver(debug: bool) -> None

Set whether to debug the end value optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_end_value_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the end value optimization solver."""
    self._debug_end_value_opt_solver = debug

set_debug_long_opt_solver(debug: bool) -> None

Set whether to debug the long-term optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_long_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the long-term optimization solver."""
    self._debug_long_opt_solver = debug

set_debug_med_opt_solver(debug: bool) -> None

Set whether to debug the medium-term optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_med_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the medium-term optimization solver."""
    self._debug_med_opt_solver = debug

set_debug_short_opt_solver(debug: bool) -> None

Set whether to debug the short-term optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_short_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the short-term optimization solver."""
    self._debug_short_opt_solver = debug

set_debug_subsystem_master_opt_solver(debug: bool) -> None

Set whether to debug the subsystem master optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_subsystem_master_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the subsystem master optimization solver."""
    self._debug_subsystem_master_opt_solver = debug

set_debug_subsystem_sub_opt_solver(debug: bool) -> None

Set whether to debug the subsystem sub optimization solver.

Source code in framjules/JulESConfig.py

def set_debug_subsystem_sub_opt_solver(self, debug: bool) -> None:
    """Set whether to debug the subsystem sub optimization solver."""
    self._debug_subsystem_sub_opt_solver = debug

set_force_julia_install(flag: bool) -> bool

Set bool for force new julia install.

Source code in framjules/JulESConfig.py

def set_force_julia_install(self, flag: bool) -> bool:
    """Set bool for force new julia install."""
    self._force_julia_install = flag

set_jules_version(jules_branch: str | None = None, tulipa_branch: str | None = None) -> None

Set which version of JulES and/or TuLiPa to use.

Can be a git branch name, or a local path to a git repo which will activate development mode.

Source code in framjules/JulESConfig.py

def set_jules_version(self, jules_branch: str | None = None, tulipa_branch: str | None = None) -> None:
    """Set which version of JulES and/or TuLiPa to use.

    Can be a git branch name, or a local path to a git repo which will activate development mode.
    """
    self._check_type(jules_branch, (str, type(None)))
    self._check_type(tulipa_branch, (str, type(None)))
    if jules_branch is not None:
        self._branch_jules = jules_branch
    if tulipa_branch is not None:
        self._branch_tulipa = tulipa_branch
    if self._branch_tulipa is None and self._branch_jules is not None:
        self._branch_tulipa = self._branch_jules

set_julia_depot_path(path: Path) -> None

Set folder where Julia installs new packages.

Source code in framjules/JulESConfig.py

def set_julia_depot_path(self, path: Path) -> None:
    """Set folder where Julia installs new packages."""
    self._check_type(path, Path)
    self._julia_depot_path = path

set_julia_env_path(path: Path) -> None

Set which Julia environment to use.

Source code in framjules/JulESConfig.py

def set_julia_env_path(self, path: Path) -> None:
    """Set which Julia environment to use."""
    self._check_type(path, Path)
    self._julia_env_path = path

set_julia_exe_path(path: Path) -> None

Set which Julia installation to use.

Source code in framjules/JulESConfig.py

def set_julia_exe_path(self, path: Path) -> None:
    """Set which Julia installation to use."""
    self._check_type(path, Path)
    self._julia_exe_path = path

set_short_term_aggregations(aggregators: list[Aggregator]) -> None

Set aggregations to create the short term model used in the price prognosis problems.

Source code in framjules/JulESConfig.py

def set_short_term_aggregations(self, aggregators: list[Aggregator]) -> None:
    """Set aggregations to create the short term model used in the price prognosis problems."""
    self._check_supported_aggregators(aggregators)
    self._short_term_aggregations = aggregators

set_skipmax_days(days: int) -> None

Set number of days between calculation of medium and long term storage values.

This can speed up a simulation as medium and long term price prognosis problems, and long term storage value problems are solved less often. The cost is less good storage values. The longer between re-calculation of storage values, the bigger negative impact on simulation result quality.

If skipmax_days = 6 and clearing_days = 2, JulES will calculate medium and long term storage values every 3rd simulation step.

Short term price prognosis problems and storage values problems (i.e. for batteries) are not affected by this setting, and are calculated every simulation step.

Source code in framjules/JulESConfig.py

def set_skipmax_days(self, days: int) -> None:
    """Set number of days between calculation of medium and long term storage values.

    This can speed up a simulation as medium and long term price prognosis problems, and long term
    storage value problems are solved less often. The cost is less good storage values. The longer between
    re-calculation of storage values, the bigger negative impact on simulation result quality.

    If skipmax_days = 6 and clearing_days = 2, JulES will calculate medium and long
    term storage values every 3rd simulation step.

    Short term price prognosis problems and storage values problems (i.e. for batteries) are not affected by this
    setting, and are calculated every simulation step.
    """
    self._check_type(days, int)
    self._check_int(days, lower_bound=0, upper_bound=None)
    if days > self._warn_skipmax_days:
        message = (
            "Unusually high value for skipmax_days: "
            f"Medium and long term storage values updated only {days}th day. "
            "This can give poor simulation results due to poor storage utilization."
        )
        self.send_warning_event(message)
    self._skipmax_days = days

JulESTimeResolution

JulESTimeResolution

Bases: Base

Time resolution settings for JulES (only some are modifiable).

Source code in framjules/JulESTimeResolution.py

class JulESTimeResolution(Base):
    """Time resolution settings for JulES (only some are modifiable)."""

    def __init__(self) -> None:
        """Create instance with default values."""
        self._target_med_days = 12 * 7
        self._target_long_storage_days = 6 * 7
        self._target_lookahead_days = 365 * 5
        self._target_ev_days = 365 * 3

        self._clearing_days = 2
        self._short_days = 5

        self._clearing_market_minutes = 60 * 3
        self._clearing_storage_minutes = 60 * 24

        self._short_market_minutes = self._get_incremented_divisor(
            n=self._clearing_days * 24 * 60,
            divisor=self._clearing_market_minutes,
            num_increments=1,
        )
        self._short_storage_minutes = 60 * 24 * self._short_days

        self._long_adaptive_blocks = 4
        self._long_adaptive_hours = 6

        self._med_adaptive_blocks = 4
        self._med_adaptive_hours = 6

        med_days, long_days, med_storage_days, long_storage_days = self._get_med_long_days_and_storage_days(
            self._target_lookahead_days,
            self._clearing_days,
            self._short_days,
        )

        self._med_days = med_days
        self._long_days = long_days
        self._med_storage_days = med_storage_days
        self._long_storage_days = long_storage_days

    def set_target_ev_days(self, x: int) -> None:
        """Set prefered value for length in days of end value problems.

        Will choose a close valid value.
        """
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
        self._target_ev_days = x

    def set_target_long_storage_days(self, x: int) -> None:
        """Set prefered value for long_storage_days.

        Will choose a close valid value.
        """
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
        self._target_long_storage_days = x

    def set_target_med_days(self, x: int) -> None:
        """Set prefered value for horizon length in days in medium prognosis problem.

        Will choose a close valid value.
        """
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
        self._target_med_num_period = x

    def set_clearing_days(self, x: int) -> None:
        """Set length of clearing problem in days."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)

        self._clearing_days = x

        clearing_minutes = self._clearing_days * 24 * 60

        self._clearing_market_minutes = min(self._clearing_market_minutes, clearing_minutes)
        self._clearing_storage_minutes = min(self._clearing_storage_minutes, clearing_minutes)

        if clearing_minutes % self._clearing_market_minutes != 0:
            message = (
                f"clearing_market_minutes ({self._clearing_market_minutes}) does not go up"
                f" in clearing_days in minutes ({clearing_minutes})."
            )
            raise ValueError(message)

        if clearing_minutes % self._clearing_storage_minutes != 0:
            message = (
                f"clearing_storage_minutes ({self._clearing_storage_minutes}) does not go up"
                f" in clearing_days in minutes ({clearing_minutes})."
            )
            raise ValueError(message)

        self.set_target_lookahead_days(self._target_lookahead_days)

    def set_short_days(self, x: int) -> None:
        """Set length of short term prognosis problem in days."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)

        self._short_days = x

        short_minutes = self._short_days * 24 * 60

        self._short_market_minutes = min(self._short_market_minutes, short_minutes)
        self._short_storage_minutes = min(self._short_storage_minutes, short_minutes)

        if short_minutes % self._short_market_minutes != 0:
            message = (
                f"short_market_minutes ({self._short_market_minutes}) does not go up"
                f" in short_days in minutes ({short_minutes})."
            )
            raise ValueError(message)

        if short_minutes % self._short_storage_minutes != 0:
            message = (
                f"short_storage_minutes ({self._short_storage_minutes}) does not go up"
                f" in short_days in minutes ({short_minutes})."
            )
            raise ValueError(message)

        self.set_target_lookahead_days(self._target_lookahead_days)

    def set_target_lookahead_days(self, x: int) -> None:
        """Set target length of prognosis problems in days.

        Will set med_days and long_days and make sure their sum is minimum this length.

        Will set short_days if target_lookahead_days < short_days.
        """
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)

        self._target_lookahead_days = x

        if x < self._short_days:
            self.set_short_days(x)

        med_days, long_days, med_storage_days, long_storage_days = self._get_med_long_days_and_storage_days(
            self._target_lookahead_days,
            self._clearing_days,
            self._short_days,
        )

        self._med_days = med_days
        self._long_days = long_days
        self._med_storage_days = med_storage_days
        self._long_storage_days = long_storage_days

    def set_clearing_market_minutes(self, x: int) -> None:
        """Set market period length in clearing problem in minutes. Currently only support whole hours."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)
        self._check_hourly(x)
        self._clearing_market_minutes = x

    def set_clearing_storage_minutes(self, x: int) -> None:
        """Set storage period length in clearing problem in minutes. Currently only support whole hours."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)
        self._clearing_storage_minutes = x

    def set_short_market_minutes(self, x: int) -> None:
        """Set market period length in short prognosis problem in minutes. Currently only support whole hours."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)
        self._short_market_minutes = x

    def set_short_storage_minutes(self, x: int) -> None:
        """Set storage period length in short prognosis problem in minutes. Currently only support whole hours."""
        self._check_type(x, int)
        self._check_int(value=x, lower_bound=1, upper_bound=None)
        self._short_storage_minutes = x

    def get_clearing_days(self) -> int:
        """Get length of clearing problem in days."""
        return self._clearing_days

    def get_short_days(self) -> int:
        """Get length of short prognosis problem in days."""
        return self._short_days

    def get_target_lookahead_days(self) -> int:
        """Get target (minimum) length of prognosis problems in days."""
        return self._target_lookahead_days

    def get_med_days(self) -> int:
        """Get length of medium prognosis problem in days."""
        return self._med_days

    def get_long_days(self) -> int:
        """Get length of long prognosis problem in days."""
        return self._long_days

    def get_med_storage_days(self) -> int:
        """Get storage period length in medium prognosis problem in days."""
        return self._med_storage_days

    def get_long_storage_days(self) -> int:
        """Get storage period length in long prognosis problem in days."""
        return self._long_storage_days

    def get_clearing_market_minutes(self) -> int:
        """Get market period length in clearing problem in minutes. Currently only support whole hours."""
        return self._clearing_market_minutes

    def get_clearing_storage_minutes(self) -> int:
        """Get storage period length in clearing problem in minutes. Currently only support whole hours."""
        return self._clearing_storage_minutes

    def get_short_market_minutes(self) -> int:
        """Get market period length in short prognosis problem in minutes. Currently only support whole hours."""
        return self._short_market_minutes

    def get_short_storage_minutes(self) -> int:
        """Get storage period length in short prognosis problem in minutes. Currently only support whole hours."""
        return self._short_storage_minutes

    def get_long_adaptive_blocks(self) -> int:
        """Get number of market periods long prognosis problem."""
        return self._long_adaptive_blocks

    def get_long_adaptive_hours(self) -> int:
        """Get resolution in hours used in clustering of market period blocks in medium prognosis problem."""
        return self._long_adaptive_hours

    def get_med_adaptive_blocks(self) -> int:
        """Get number of market periods long prognosis problem."""
        return self._med_adaptive_blocks

    def get_med_adaptive_hours(self) -> int:
        """Get resolution in hours used in clustering of market period blocks in medium prognosis problem."""
        return self._med_adaptive_hours

    def get_target_long_storage_days(self) -> int:
        """Get prefered value for long_storage_days."""
        return self._target_long_storage_days

    def get_target_med_days(self) -> int:
        """Get prefered value for horizon length of medium prognosis problem."""
        return self._target_med_days

    def get_target_ev_days(self) -> int:
        """Get prefered value for horizon length of end value problem."""
        return self._target_ev_days

    def get_ev_days(self) -> int:
        """Get number of days in horizon of end value problems."""
        target_horizon_days = self.get_target_ev_days()
        long_period_days = self.get_long_storage_days()
        return math.ceil(target_horizon_days / long_period_days) * long_period_days

    def get_content_dict(self) -> dict[str, int]:
        """Return dict of all settings. Useful to get an overview."""
        return {
            "clearing_days": self.get_clearing_days(),
            "short_days": self.get_short_days(),
            "med_days": self.get_med_days(),
            "long_days": self.get_long_days(),
            "long_storage_days": self.get_long_storage_days(),
            "med_storage_days": self.get_med_storage_days(),
            "clearing_market_minutes": self.get_clearing_market_minutes(),
            "clearing_storage_minutes": self.get_clearing_storage_minutes(),
            "short_market_minutes": self.get_short_market_minutes(),
            "short_storage_minutes": self.get_short_storage_minutes(),
            "long_adaptive_blocks": self.get_long_adaptive_blocks(),
            "long_adaptive_hours": self.get_long_adaptive_hours(),
            "med_adaptive_blocks": self.get_med_adaptive_blocks(),
            "med_adaptive_hours": self.get_med_adaptive_hours(),
            "target_lookahead_days": self.get_target_lookahead_days(),
            "target_long_storage_days": self.get_target_long_storage_days(),
            "target_med_days": self.get_target_med_days(),
            "target_ev_days": self.get_target_ev_days(),
            "ev_days": self.get_ev_days(),
        }

    def _get_med_long_days_and_storage_days(
        self,
        target_lookahead_days: int,
        clearing_days: int,
        short_days: int,
    ) -> tuple[int, int, int, int]:
        """Find the valid configuration that is closest to the user supplied targets."""
        med_period = clearing_days + short_days

        possible_med_pairs = self._get_possible_med_pairs(med_period, target_lookahead_days, short_days)

        if not possible_med_pairs:
            return tuple([clearing_days + short_days] * 4)

        candidates = set()
        for med_period, med_num_periods in possible_med_pairs:
            possible_long_pairs = self._get_possible_long_pairs(
                med_period,
                med_num_periods,
                target_lookahead_days,
                short_days,
            )
            for long_period, long_num_periods in possible_long_pairs:
                med_days = med_period * med_num_periods
                long_days = long_period * long_num_periods
                candidate = (med_days, long_days, med_period, long_period)
                candidates.add(candidate)

        target_med_days = self.get_target_med_days()
        target_long_storage_days = self.get_target_long_storage_days()

        def distance_from_targets(candidate: tuple[int, int, int, int]) -> float | int:
            med_days = candidate[0]
            long_storage_days = candidate[3]
            med_square_diff = (target_med_days - med_days) ** 2
            long_square_diff = (target_long_storage_days - long_storage_days) ** 2
            return med_square_diff + long_square_diff

        return min(candidates, key=distance_from_targets)

    def _get_possible_med_pairs(
        self,
        med_period: int,
        target_lookahead_days: int,
        short_days: int,
    ) -> list[tuple[int, int]]:
        """Fuzz number of med storage periods to get more candidates."""
        target_med_days = self.get_target_med_days()
        out = []
        n = math.ceil(target_med_days / med_period)
        for m in [n - 1, n, n + 1]:
            implied_target_lookahead_days = med_period * (m + 1) + short_days
            if m > 1 and implied_target_lookahead_days <= target_lookahead_days:
                out.append((med_period, m))
        return out

    def _get_possible_long_pairs(
        self,
        med_period: int,
        med_num_periods: int,
        target_lookahead_days: int,
        short_days: int,
    ) -> list[tuple[int, int]]:
        """Find valid long pairs. Valid if long_period is divisor of med_days."""
        target_long_storage_days = self.get_target_long_storage_days()
        med_days = med_period * med_num_periods
        divisors = self._get_divisors(med_days)
        divisors = sorted(divisors, key=lambda x: abs(target_long_storage_days - x))
        divisors = divisors[:4]
        out = []
        for long_period in divisors:
            long_num_periods = math.ceil(max(1, target_lookahead_days - med_days - short_days) / long_period)
            out.append((long_period, long_num_periods))
        return out

    def _get_divisors(self, n: int) -> list[int]:
        """Return sorted list of divisors of n.

        Inspiration from: https://stackoverflow.com/questions/171765/what-is-the-best-way-to-get-all-the-divisors-of-a-number
        """
        divs = [1]
        for i in range(2, int(math.sqrt(n)) + 1):
            if n % i == 0:
                divs.extend([i, int(n / i)])
        divs.extend([n])
        return sorted(list(set(divs)))

    def _get_incremented_divisor(self, n: int, divisor: int, num_increments: int) -> int:
        """Get divisor of n num_increments greater than k if <= n else n."""
        divs = self._get_divisors(n)
        try:
            i = divs.index(divisor)
        except ValueError:
            message = f"{divisor} is not a divisor of {n}."
            raise ValueError(message) from None
        return divs[min(i + num_increments, len(divs) - 1)]

    def _check_hourly(self, x: int) -> None:
        if not (x / 60).is_integer():
            message = "Currently, JulES only support hourly resolutions."
            raise ValueError(message)

__init__() -> None

Create instance with default values.

Source code in framjules/JulESTimeResolution.py

def __init__(self) -> None:
    """Create instance with default values."""
    self._target_med_days = 12 * 7
    self._target_long_storage_days = 6 * 7
    self._target_lookahead_days = 365 * 5
    self._target_ev_days = 365 * 3

    self._clearing_days = 2
    self._short_days = 5

    self._clearing_market_minutes = 60 * 3
    self._clearing_storage_minutes = 60 * 24

    self._short_market_minutes = self._get_incremented_divisor(
        n=self._clearing_days * 24 * 60,
        divisor=self._clearing_market_minutes,
        num_increments=1,
    )
    self._short_storage_minutes = 60 * 24 * self._short_days

    self._long_adaptive_blocks = 4
    self._long_adaptive_hours = 6

    self._med_adaptive_blocks = 4
    self._med_adaptive_hours = 6

    med_days, long_days, med_storage_days, long_storage_days = self._get_med_long_days_and_storage_days(
        self._target_lookahead_days,
        self._clearing_days,
        self._short_days,
    )

    self._med_days = med_days
    self._long_days = long_days
    self._med_storage_days = med_storage_days
    self._long_storage_days = long_storage_days

get_clearing_days() -> int

Get length of clearing problem in days.

Source code in framjules/JulESTimeResolution.py

def get_clearing_days(self) -> int:
    """Get length of clearing problem in days."""
    return self._clearing_days

get_clearing_market_minutes() -> int

Get market period length in clearing problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def get_clearing_market_minutes(self) -> int:
    """Get market period length in clearing problem in minutes. Currently only support whole hours."""
    return self._clearing_market_minutes

get_clearing_storage_minutes() -> int

Get storage period length in clearing problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def get_clearing_storage_minutes(self) -> int:
    """Get storage period length in clearing problem in minutes. Currently only support whole hours."""
    return self._clearing_storage_minutes

get_content_dict() -> dict[str, int]

Return dict of all settings. Useful to get an overview.

Source code in framjules/JulESTimeResolution.py

def get_content_dict(self) -> dict[str, int]:
    """Return dict of all settings. Useful to get an overview."""
    return {
        "clearing_days": self.get_clearing_days(),
        "short_days": self.get_short_days(),
        "med_days": self.get_med_days(),
        "long_days": self.get_long_days(),
        "long_storage_days": self.get_long_storage_days(),
        "med_storage_days": self.get_med_storage_days(),
        "clearing_market_minutes": self.get_clearing_market_minutes(),
        "clearing_storage_minutes": self.get_clearing_storage_minutes(),
        "short_market_minutes": self.get_short_market_minutes(),
        "short_storage_minutes": self.get_short_storage_minutes(),
        "long_adaptive_blocks": self.get_long_adaptive_blocks(),
        "long_adaptive_hours": self.get_long_adaptive_hours(),
        "med_adaptive_blocks": self.get_med_adaptive_blocks(),
        "med_adaptive_hours": self.get_med_adaptive_hours(),
        "target_lookahead_days": self.get_target_lookahead_days(),
        "target_long_storage_days": self.get_target_long_storage_days(),
        "target_med_days": self.get_target_med_days(),
        "target_ev_days": self.get_target_ev_days(),
        "ev_days": self.get_ev_days(),
    }

get_ev_days() -> int

Get number of days in horizon of end value problems.

Source code in framjules/JulESTimeResolution.py

def get_ev_days(self) -> int:
    """Get number of days in horizon of end value problems."""
    target_horizon_days = self.get_target_ev_days()
    long_period_days = self.get_long_storage_days()
    return math.ceil(target_horizon_days / long_period_days) * long_period_days

get_long_adaptive_blocks() -> int

Get number of market periods long prognosis problem.

Source code in framjules/JulESTimeResolution.py

def get_long_adaptive_blocks(self) -> int:
    """Get number of market periods long prognosis problem."""
    return self._long_adaptive_blocks

get_long_adaptive_hours() -> int

Get resolution in hours used in clustering of market period blocks in medium prognosis problem.

Source code in framjules/JulESTimeResolution.py

def get_long_adaptive_hours(self) -> int:
    """Get resolution in hours used in clustering of market period blocks in medium prognosis problem."""
    return self._long_adaptive_hours

get_long_days() -> int

Get length of long prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def get_long_days(self) -> int:
    """Get length of long prognosis problem in days."""
    return self._long_days

get_long_storage_days() -> int

Get storage period length in long prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def get_long_storage_days(self) -> int:
    """Get storage period length in long prognosis problem in days."""
    return self._long_storage_days

get_med_adaptive_blocks() -> int

Get number of market periods long prognosis problem.

Source code in framjules/JulESTimeResolution.py

def get_med_adaptive_blocks(self) -> int:
    """Get number of market periods long prognosis problem."""
    return self._med_adaptive_blocks

get_med_adaptive_hours() -> int

Get resolution in hours used in clustering of market period blocks in medium prognosis problem.

Source code in framjules/JulESTimeResolution.py

def get_med_adaptive_hours(self) -> int:
    """Get resolution in hours used in clustering of market period blocks in medium prognosis problem."""
    return self._med_adaptive_hours

get_med_days() -> int

Get length of medium prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def get_med_days(self) -> int:
    """Get length of medium prognosis problem in days."""
    return self._med_days

get_med_storage_days() -> int

Get storage period length in medium prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def get_med_storage_days(self) -> int:
    """Get storage period length in medium prognosis problem in days."""
    return self._med_storage_days

get_short_days() -> int

Get length of short prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def get_short_days(self) -> int:
    """Get length of short prognosis problem in days."""
    return self._short_days

get_short_market_minutes() -> int

Get market period length in short prognosis problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def get_short_market_minutes(self) -> int:
    """Get market period length in short prognosis problem in minutes. Currently only support whole hours."""
    return self._short_market_minutes

get_short_storage_minutes() -> int

Get storage period length in short prognosis problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def get_short_storage_minutes(self) -> int:
    """Get storage period length in short prognosis problem in minutes. Currently only support whole hours."""
    return self._short_storage_minutes

get_target_ev_days() -> int

Get prefered value for horizon length of end value problem.

Source code in framjules/JulESTimeResolution.py

def get_target_ev_days(self) -> int:
    """Get prefered value for horizon length of end value problem."""
    return self._target_ev_days

get_target_long_storage_days() -> int

Get prefered value for long_storage_days.

Source code in framjules/JulESTimeResolution.py

def get_target_long_storage_days(self) -> int:
    """Get prefered value for long_storage_days."""
    return self._target_long_storage_days

get_target_lookahead_days() -> int

Get target (minimum) length of prognosis problems in days.

Source code in framjules/JulESTimeResolution.py

def get_target_lookahead_days(self) -> int:
    """Get target (minimum) length of prognosis problems in days."""
    return self._target_lookahead_days

get_target_med_days() -> int

Get prefered value for horizon length of medium prognosis problem.

Source code in framjules/JulESTimeResolution.py

def get_target_med_days(self) -> int:
    """Get prefered value for horizon length of medium prognosis problem."""
    return self._target_med_days

set_clearing_days(x: int) -> None

Set length of clearing problem in days.

Source code in framjules/JulESTimeResolution.py

def set_clearing_days(self, x: int) -> None:
    """Set length of clearing problem in days."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)

    self._clearing_days = x

    clearing_minutes = self._clearing_days * 24 * 60

    self._clearing_market_minutes = min(self._clearing_market_minutes, clearing_minutes)
    self._clearing_storage_minutes = min(self._clearing_storage_minutes, clearing_minutes)

    if clearing_minutes % self._clearing_market_minutes != 0:
        message = (
            f"clearing_market_minutes ({self._clearing_market_minutes}) does not go up"
            f" in clearing_days in minutes ({clearing_minutes})."
        )
        raise ValueError(message)

    if clearing_minutes % self._clearing_storage_minutes != 0:
        message = (
            f"clearing_storage_minutes ({self._clearing_storage_minutes}) does not go up"
            f" in clearing_days in minutes ({clearing_minutes})."
        )
        raise ValueError(message)

    self.set_target_lookahead_days(self._target_lookahead_days)

set_clearing_market_minutes(x: int) -> None

Set market period length in clearing problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def set_clearing_market_minutes(self, x: int) -> None:
    """Set market period length in clearing problem in minutes. Currently only support whole hours."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)
    self._check_hourly(x)
    self._clearing_market_minutes = x

set_clearing_storage_minutes(x: int) -> None

Set storage period length in clearing problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def set_clearing_storage_minutes(self, x: int) -> None:
    """Set storage period length in clearing problem in minutes. Currently only support whole hours."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)
    self._clearing_storage_minutes = x

set_short_days(x: int) -> None

Set length of short term prognosis problem in days.

Source code in framjules/JulESTimeResolution.py

def set_short_days(self, x: int) -> None:
    """Set length of short term prognosis problem in days."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)

    self._short_days = x

    short_minutes = self._short_days * 24 * 60

    self._short_market_minutes = min(self._short_market_minutes, short_minutes)
    self._short_storage_minutes = min(self._short_storage_minutes, short_minutes)

    if short_minutes % self._short_market_minutes != 0:
        message = (
            f"short_market_minutes ({self._short_market_minutes}) does not go up"
            f" in short_days in minutes ({short_minutes})."
        )
        raise ValueError(message)

    if short_minutes % self._short_storage_minutes != 0:
        message = (
            f"short_storage_minutes ({self._short_storage_minutes}) does not go up"
            f" in short_days in minutes ({short_minutes})."
        )
        raise ValueError(message)

    self.set_target_lookahead_days(self._target_lookahead_days)

set_short_market_minutes(x: int) -> None

Set market period length in short prognosis problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def set_short_market_minutes(self, x: int) -> None:
    """Set market period length in short prognosis problem in minutes. Currently only support whole hours."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)
    self._short_market_minutes = x

set_short_storage_minutes(x: int) -> None

Set storage period length in short prognosis problem in minutes. Currently only support whole hours.

Source code in framjules/JulESTimeResolution.py

def set_short_storage_minutes(self, x: int) -> None:
    """Set storage period length in short prognosis problem in minutes. Currently only support whole hours."""
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)
    self._short_storage_minutes = x

set_target_ev_days(x: int) -> None

Set prefered value for length in days of end value problems.

Will choose a close valid value.

Source code in framjules/JulESTimeResolution.py

def set_target_ev_days(self, x: int) -> None:
    """Set prefered value for length in days of end value problems.

    Will choose a close valid value.
    """
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
    self._target_ev_days = x

set_target_long_storage_days(x: int) -> None

Set prefered value for long_storage_days.

Will choose a close valid value.

Source code in framjules/JulESTimeResolution.py

def set_target_long_storage_days(self, x: int) -> None:
    """Set prefered value for long_storage_days.

    Will choose a close valid value.
    """
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
    self._target_long_storage_days = x

set_target_lookahead_days(x: int) -> None

Set target length of prognosis problems in days.

Will set med_days and long_days and make sure their sum is minimum this length.

Will set short_days if target_lookahead_days < short_days.

Source code in framjules/JulESTimeResolution.py

def set_target_lookahead_days(self, x: int) -> None:
    """Set target length of prognosis problems in days.

    Will set med_days and long_days and make sure their sum is minimum this length.

    Will set short_days if target_lookahead_days < short_days.
    """
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=1, upper_bound=None)

    self._target_lookahead_days = x

    if x < self._short_days:
        self.set_short_days(x)

    med_days, long_days, med_storage_days, long_storage_days = self._get_med_long_days_and_storage_days(
        self._target_lookahead_days,
        self._clearing_days,
        self._short_days,
    )

    self._med_days = med_days
    self._long_days = long_days
    self._med_storage_days = med_storage_days
    self._long_storage_days = long_storage_days

set_target_med_days(x: int) -> None

Set prefered value for horizon length in days in medium prognosis problem.

Will choose a close valid value.

Source code in framjules/JulESTimeResolution.py

def set_target_med_days(self, x: int) -> None:
    """Set prefered value for horizon length in days in medium prognosis problem.

    Will choose a close valid value.
    """
    self._check_type(x, int)
    self._check_int(value=x, lower_bound=self._clearing_days + self._short_days, upper_bound=None)
    self._target_med_num_period = x

loaders

time_vector_loaders

DemandJulESH5TimeVectorLoader

Bases: JulESH5TimeVectorLoader

Workaround to get demand results and at the same time avoid name conflicts.

Source code in framjules/loaders/time_vector_loaders.py

class DemandJulESH5TimeVectorLoader(JulESH5TimeVectorLoader):
    """Workaround to get demand results and at the same time avoid name conflicts."""

    _SUPPORTED_SUFFIXES: ClassVar[list] = [".h5"]
    _SEARCH_FIELDS: ClassVar[list] = [
        ("demandnames", "priceindex", "demandvalues", True),
    ]
    _DEFAULT_INDEX = "priceindex"
    pass

JulESH5TimeVectorLoader

Bases: FileLoader, TimeVectorLoader

Loader for JulES H5 files containing time vectors.

Source code in framjules/loaders/time_vector_loaders.py

class JulESH5TimeVectorLoader(FileLoader, TimeVectorLoader):
    """Loader for JulES H5 files containing time vectors."""

    _SUPPORTED_SUFFIXES: ClassVar[list] = [".h5"]
    # find time vector id in one of the names, then use the corresponding names to get the data
    _SEARCH_FIELDS: ClassVar[list] = [
        ("areanames", "priceindex", "pricematrix", True),
        ("batnames", "batindex", "batmatrix", True),
        ("resnames", "resindex", "resmatrix_water", True),
        ("othernames_Vars_Power", "priceindex", "othervalues_Vars_Power", True),
        ("othernames_Vars_Hydro", "resindex", "othervalues_Vars_Hydro", True),
        (
            "storagenames",
            "stateindex",
            "storagevalues_main",
            True,
        ),
    ]
    _DEFAULT_INDEX = "priceindex"

    def __init__(
        self,
        source: Path | str,
        units: dict[str, str],
        relative_loc: Path | str | None = None,
        is_whole_years: bool = False,
    ) -> None:
        """Initialize the NVEH5TimeVectorLoader."""
        super().__init__(source, relative_loc)
        self._data: dict[str, NDArray] = {}
        self._index: dict[str, TimeIndex] = {}
        self._units: dict[str, str] = units
        self._is_whole_years = is_whole_years

        self._id_fields_map: dict[str, list[set]] = {}

    def clear_cache(self) -> None:
        """Clear cached data."""
        self._data = {}
        self._index = {}
        self._id_fields_map = {}

    def get_reference_period(self, vector_id: str) -> None:
        """Return None."""
        return

    def is_max_level(self, vector_id: str) -> None:
        """Return None."""
        return

    def is_zero_one_profile(self, vector_id: str) -> None:
        """Return None."""
        return True

    def get_values(self, vector_id: str) -> NDArray:
        """Find the values for a given vector id.

        Args:
            vector_id: (str)

        Returns:
            NDArray: Values for the vector id.

        """
        self._id_exsists(vector_id)
        self._check_multiple_fields(vector_id)
        if vector_id not in self._data:
            id_field, values_field = self._get_id_values_field(vector_id)
            with h5py.File(self.get_source(), "r") as f:
                ids = np.array([item.decode() for item in f[id_field]])
                values_matrix = np.array(f[values_field])
                new_data = {id_: values_matrix[i, :] for i, id_ in enumerate(ids)}
                self._data.update(new_data)
        # index may not use all values
        index = self.get_index(vector_id)
        vector = self._data[vector_id]
        n = index.get_num_periods()
        return vector[:n]

    def get_index(self, vector_id: str) -> TimeIndex:
        """Find the time index for a given vector id.

        Args:
            vector_id: (str)

        Returns:
            TimeIndex

        """
        self._id_exsists(vector_id)  # calls get_ids which calls _get_ids which sets _id_fields_map
        self._check_multiple_fields(vector_id)  # therefore we can use this afterwards
        with h5py.File(self.get_source(), "r") as f:
            index_field = self._id_fields_map.get(vector_id)[0][1]
            fmt = "%Y-%m-%dT%H:%M:%S"
            if self._index is None or (index_field not in self._index):
                t0 = datetime.strptime(f[index_field][0].decode(), fmt)
                t1 = datetime.strptime(f[index_field][1].decode(), fmt)
                index = FixedFrequencyTimeIndex(
                    start_time=t0,
                    period_duration=t1 - t0,
                    num_periods=len(f[index_field]),
                    is_52_week_years=False,
                    extrapolate_first_point=False,
                    extrapolate_last_point=False,
                )
                if not index.is_whole_years() and self._is_whole_years:
                    # start time for each period
                    datetime_list = [datetime.strptime(x.decode(), fmt) for x in f[index_field]]
                    period_duration = datetime_list[1] - datetime_list[0]

                    # add end index since JulES index represents periods
                    datetime_list.append(datetime_list[-1] + period_duration)
                    num_periods = len(datetime_list)

                    # find last index before new iso year
                    last_in_year_ix = num_periods - 1
                    while last_in_year_ix >= 0:
                        last_in_year_ix -= 1
                        this_period = datetime_list[last_in_year_ix]
                        next_period = this_period + period_duration
                        this_year = this_period.isocalendar().year
                        next_year = next_period.isocalendar().year
                        if next_year > this_year:
                            break

                    last_in_year = datetime_list[last_in_year_ix]

                    first_next_year = last_in_year.fromisocalendar(last_in_year.isocalendar().year + 1, 1, 1)

                    if last_in_year + period_duration == first_next_year:
                        index = FixedFrequencyTimeIndex(
                            start_time=datetime_list[0],
                            period_duration=period_duration,
                            num_periods=last_in_year_ix + 1,
                            is_52_week_years=False,
                            extrapolate_first_point=False,
                            extrapolate_last_point=False,
                        )
                    elif last_in_year + period_duration > first_next_year:
                        # TODO: test (only reachable without profiles as ProfileTimeIndex enforces whole years)
                        datetime_list[last_in_year_ix + 1] = first_next_year
                        del datetime_list[last_in_year_ix + 2 :]  # (slice delete does not error when out-of-bounds)
                        index = ListTimeIndex(
                            datetime_list=datetime_list,
                            is_52_week_years=False,
                            extrapolate_first_point=False,
                            extrapolate_last_point=False,
                        )
                    else:
                        n = last_in_year_ix
                        message = (
                            f"Unexpected last_in_year + period_duration < first_next_year.\n"
                            f"vector_id = {vector_id}\n"
                            f"last_in_year = {last_in_year}\n"
                            f"period_duration = {period_duration}\n"
                            f"first_next_year = {first_next_year}\n"
                            f"datetime_list around last_in_year_ix = {datetime_list[n - 10 : n + 10]}"
                        )
                        raise RuntimeError(message)

                self._index[index_field] = index
        return self._index[index_field]

    def get_unit(self, vector_id: str) -> str:
        """Get the unit of the time vector."""
        return self._units[vector_id]

    def get_metadata(self) -> str:
        """Get metadata from the file."""
        return ""

    def _get_id_values_field(self, vector_id: str) -> str:
        search_fields = self._id_fields_map.get(vector_id)
        return search_fields[0][0], search_fields[0][2]

    def _get_ids(self) -> list[str]:
        if not self._id_fields_map:
            self._create_id_fields_map()
        return list(self._id_fields_map.keys())

    def _create_id_fields_map(self) -> dict[str, list[str]]:
        if not self._id_fields_map:
            self._id_fields_map: dict[str, list[str]] = dict()
            with h5py.File(self.get_source(), "r") as f:
                for search_name in self._SEARCH_FIELDS:
                    if search_name[0] in f:
                        new_ids = [item.decode() for item in f[search_name[0]]]
                        for vector_id in new_ids:
                            if vector_id not in self._id_fields_map:
                                self._id_fields_map[vector_id] = [search_name]
                            else:
                                self._id_fields_map[vector_id].append(search_name)

    def _check_multiple_fields(self, vector_id: str) -> None:
        self._create_id_fields_map()
        # check if the vector id is found in multiple fields.
        if len(self._id_fields_map[vector_id]) > 1:
            msg = (
                f"Vector ID {vector_id} found in multiple fields: {self._id_fields_map[vector_id]}. "
                "Could not determine which field to use."
            )
            raise NotImplementedError(msg)

    def get_fingerprint(self) -> Fingerprint:
        """Get the fingerprint of the NVEH5TimeVectorLoader."""
        return None

    def __eq__(self, other: object) -> bool:
        """Check if self and other are equal."""
        if not isinstance(other, type(self)):
            return False
        return self.get_source() == other.get_source() and self._SEARCH_FIELDS == other._SEARCH_FIELDS

    def __hash__(self) -> int:
        """Return hash of NVEH5TimeVectorLoader object."""
        return hash(
            (
                self.get_source(),
                frozenset(self._SEARCH_FIELDS),
            ),
        )

__eq__(other: object) -> bool

Check if self and other are equal.

Source code in framjules/loaders/time_vector_loaders.py

def __eq__(self, other: object) -> bool:
    """Check if self and other are equal."""
    if not isinstance(other, type(self)):
        return False
    return self.get_source() == other.get_source() and self._SEARCH_FIELDS == other._SEARCH_FIELDS

__hash__() -> int

Return hash of NVEH5TimeVectorLoader object.

Source code in framjules/loaders/time_vector_loaders.py

def __hash__(self) -> int:
    """Return hash of NVEH5TimeVectorLoader object."""
    return hash(
        (
            self.get_source(),
            frozenset(self._SEARCH_FIELDS),
        ),
    )

__init__(source: Path | str, units: dict[str, str], relative_loc: Path | str | None = None, is_whole_years: bool = False) -> None

Initialize the NVEH5TimeVectorLoader.

Source code in framjules/loaders/time_vector_loaders.py

def __init__(
    self,
    source: Path | str,
    units: dict[str, str],
    relative_loc: Path | str | None = None,
    is_whole_years: bool = False,
) -> None:
    """Initialize the NVEH5TimeVectorLoader."""
    super().__init__(source, relative_loc)
    self._data: dict[str, NDArray] = {}
    self._index: dict[str, TimeIndex] = {}
    self._units: dict[str, str] = units
    self._is_whole_years = is_whole_years

    self._id_fields_map: dict[str, list[set]] = {}

clear_cache() -> None

Clear cached data.

Source code in framjules/loaders/time_vector_loaders.py

def clear_cache(self) -> None:
    """Clear cached data."""
    self._data = {}
    self._index = {}
    self._id_fields_map = {}

get_fingerprint() -> Fingerprint

Get the fingerprint of the NVEH5TimeVectorLoader.

Source code in framjules/loaders/time_vector_loaders.py

def get_fingerprint(self) -> Fingerprint:
    """Get the fingerprint of the NVEH5TimeVectorLoader."""
    return None

get_index(vector_id: str) -> TimeIndex

Find the time index for a given vector id.

Parameters:

Name	Type	Description	Default
vector_id	str	(str)	required

Returns:

Type	Description
TimeIndex	TimeIndex

Source code in framjules/loaders/time_vector_loaders.py

def get_index(self, vector_id: str) -> TimeIndex:
    """Find the time index for a given vector id.

    Args:
        vector_id: (str)

    Returns:
        TimeIndex

    """
    self._id_exsists(vector_id)  # calls get_ids which calls _get_ids which sets _id_fields_map
    self._check_multiple_fields(vector_id)  # therefore we can use this afterwards
    with h5py.File(self.get_source(), "r") as f:
        index_field = self._id_fields_map.get(vector_id)[0][1]
        fmt = "%Y-%m-%dT%H:%M:%S"
        if self._index is None or (index_field not in self._index):
            t0 = datetime.strptime(f[index_field][0].decode(), fmt)
            t1 = datetime.strptime(f[index_field][1].decode(), fmt)
            index = FixedFrequencyTimeIndex(
                start_time=t0,
                period_duration=t1 - t0,
                num_periods=len(f[index_field]),
                is_52_week_years=False,
                extrapolate_first_point=False,
                extrapolate_last_point=False,
            )
            if not index.is_whole_years() and self._is_whole_years:
                # start time for each period
                datetime_list = [datetime.strptime(x.decode(), fmt) for x in f[index_field]]
                period_duration = datetime_list[1] - datetime_list[0]

                # add end index since JulES index represents periods
                datetime_list.append(datetime_list[-1] + period_duration)
                num_periods = len(datetime_list)

                # find last index before new iso year
                last_in_year_ix = num_periods - 1
                while last_in_year_ix >= 0:
                    last_in_year_ix -= 1
                    this_period = datetime_list[last_in_year_ix]
                    next_period = this_period + period_duration
                    this_year = this_period.isocalendar().year
                    next_year = next_period.isocalendar().year
                    if next_year > this_year:
                        break

                last_in_year = datetime_list[last_in_year_ix]

                first_next_year = last_in_year.fromisocalendar(last_in_year.isocalendar().year + 1, 1, 1)

                if last_in_year + period_duration == first_next_year:
                    index = FixedFrequencyTimeIndex(
                        start_time=datetime_list[0],
                        period_duration=period_duration,
                        num_periods=last_in_year_ix + 1,
                        is_52_week_years=False,
                        extrapolate_first_point=False,
                        extrapolate_last_point=False,
                    )
                elif last_in_year + period_duration > first_next_year:
                    # TODO: test (only reachable without profiles as ProfileTimeIndex enforces whole years)
                    datetime_list[last_in_year_ix + 1] = first_next_year
                    del datetime_list[last_in_year_ix + 2 :]  # (slice delete does not error when out-of-bounds)
                    index = ListTimeIndex(
                        datetime_list=datetime_list,
                        is_52_week_years=False,
                        extrapolate_first_point=False,
                        extrapolate_last_point=False,
                    )
                else:
                    n = last_in_year_ix
                    message = (
                        f"Unexpected last_in_year + period_duration < first_next_year.\n"
                        f"vector_id = {vector_id}\n"
                        f"last_in_year = {last_in_year}\n"
                        f"period_duration = {period_duration}\n"
                        f"first_next_year = {first_next_year}\n"
                        f"datetime_list around last_in_year_ix = {datetime_list[n - 10 : n + 10]}"
                    )
                    raise RuntimeError(message)

            self._index[index_field] = index
    return self._index[index_field]

get_metadata() -> str

Get metadata from the file.

Source code in framjules/loaders/time_vector_loaders.py

def get_metadata(self) -> str:
    """Get metadata from the file."""
    return ""

get_reference_period(vector_id: str) -> None

Return None.

Source code in framjules/loaders/time_vector_loaders.py

def get_reference_period(self, vector_id: str) -> None:
    """Return None."""
    return

get_unit(vector_id: str) -> str

Get the unit of the time vector.

Source code in framjules/loaders/time_vector_loaders.py

def get_unit(self, vector_id: str) -> str:
    """Get the unit of the time vector."""
    return self._units[vector_id]

get_values(vector_id: str) -> NDArray

Find the values for a given vector id.

Parameters:

Name	Type	Description	Default
vector_id	str	(str)	required

Returns:

Name	Type	Description
NDArray	NDArray	Values for the vector id.

Source code in framjules/loaders/time_vector_loaders.py

def get_values(self, vector_id: str) -> NDArray:
    """Find the values for a given vector id.

    Args:
        vector_id: (str)

    Returns:
        NDArray: Values for the vector id.

    """
    self._id_exsists(vector_id)
    self._check_multiple_fields(vector_id)
    if vector_id not in self._data:
        id_field, values_field = self._get_id_values_field(vector_id)
        with h5py.File(self.get_source(), "r") as f:
            ids = np.array([item.decode() for item in f[id_field]])
            values_matrix = np.array(f[values_field])
            new_data = {id_: values_matrix[i, :] for i, id_ in enumerate(ids)}
            self._data.update(new_data)
    # index may not use all values
    index = self.get_index(vector_id)
    vector = self._data[vector_id]
    n = index.get_num_periods()
    return vector[:n]

is_max_level(vector_id: str) -> None

Return None.

Source code in framjules/loaders/time_vector_loaders.py

def is_max_level(self, vector_id: str) -> None:
    """Return None."""
    return

is_zero_one_profile(vector_id: str) -> None

Return None.

Source code in framjules/loaders/time_vector_loaders.py

def is_zero_one_profile(self, vector_id: str) -> None:
    """Return None."""
    return True

SupplyJulESH5TimeVectorLoader

Bases: JulESH5TimeVectorLoader

Workaround to get supply results and at the same time avoid name conflicts.

Source code in framjules/loaders/time_vector_loaders.py

class SupplyJulESH5TimeVectorLoader(JulESH5TimeVectorLoader):
    """Workaround to get supply results and at the same time avoid name conflicts."""

    _SUPPORTED_SUFFIXES: ClassVar[list] = [".h5"]
    _SEARCH_FIELDS: ClassVar[list] = [
        ("supplynames", "priceindex", "supplyvalues", True),
    ]
    _DEFAULT_INDEX = "priceindex"
    pass

solve_handler

JulESAggregator

JulESAggregator

Bases: Base

Class for defining and calculating aggregated Model instances based on a clearing Model.

In JulES, the clearing Model is the main Model we simulate and collect results from. JulES also has short, medium and long term price prognosis problems, which needs aggregated Models to be solved efficiently. This class helps create and manage these aggregated Models.

Note

• Short term price prognosis Model is aggregated from a Clearing Model and a list of Aggregators.
• Medium term price prognosis Model is aggregated from the Short term Model and a list of Aggregators.
• Long term price prognosis Model is aggregated from the medium term Model and a list of Aggregators.
• Storages must be the same in all aggregations
• At the moment the short, medium, and long term Models are the same, du to limitations in JulES (TODO)

Source code in framjules/solve_handler/JulESAggregator.py

class JulESAggregator(Base):
    """Class for defining and calculating aggregated Model instances based on a clearing Model.

    In JulES, the clearing Model is the main Model we simulate and collect results from. JulES also has short,
    medium and long term price prognosis problems, which needs aggregated Models to be solved efficiently.
    This class helps create and manage these aggregated Models.

    Note:
        - Short term price prognosis Model is aggregated from a Clearing Model and a list of Aggregators.
        - Medium term price prognosis Model is aggregated from the Short term Model and a list of Aggregators.
        - Long term price prognosis Model is aggregated from the medium term Model and a list of Aggregators.
        - Storages must be the same in all aggregations
        - At the moment the short, medium, and long term Models are the same, du to limitations in JulES (TODO)

    """

    def __init__(
        self,
        clearing: Model,
        short: list[Aggregator],
        medium: list[Aggregator],
        long: list[Aggregator],
    ) -> None:
        """Initialize a JulESAggregator instance.

        Args:
            clearing (Model): The clearing Model to aggregate from.
            short (list[Aggregator]): List of aggregations to create the short term price prognosis Model from clearing.
            medium (list[Aggregator]): List of aggregations to create the medium term price prognosis Model from short
            long (list[Aggregator]): List of aggregations to create the long term price prognosis Model from medium.

        """
        self._clearing = clearing
        self._short = short
        self._medium = medium
        self._long = long

        self._short_model: Model | None = None
        self._medium_model: Model | None = None
        self._long_model: Model | None = None

    def get_short_term_model(self) -> Model:
        """Apply defined aggregations for short term Model."""
        if self._short_model is None:
            self._short_model = self._aggregate(self._clearing, self._short)
        return self._short_model

    def get_medium_term_model(self) -> Model:
        """Apply defined aggregations for medium term Model."""
        if self._medium_model is None:
            self._medium_model = self._aggregate(self.get_short_term_model(), self._medium)
        return self._medium_model

    def get_long_term_model(self) -> Model:
        """Apply defined aggregations for long term Model."""
        if self._long_model is None:
            self._long_model = self._aggregate(self.get_medium_term_model(), self._long)
        return self._long_model

    def get_short_term_aggregation_map(self) -> dict[str, set[str] | None]:
        """Get the aggregation map of Components from clearing to short term Model."""
        return self._create_aggregation_map(self._clearing, self._short)

    def get_medium_term_aggregation_map(self) -> dict[str, set[str] | None]:
        """Get the aggregation map of Components from clearing to medium term Model."""
        return self._create_aggregation_map(self._clearing, self._short + self._medium)

    def get_long_term_aggregation_map(self) -> dict[str, set[str] | None]:
        """Get the aggregation map of Components from clearing to long term Model."""
        return self._create_aggregation_map(self._clearing, self._short + self._medium + self._long)

    def get_short_term_graph_map(
        self,
        graph_clearing: dict[str, Component],
        graph_short: dict[str, Component],
    ) -> dict[str, set[str] | None]:
        """Get aggregation map for version of short term Model with graph of Flows and Nodes."""
        return self._get_graph_aggregation_map(
            original_agg_map=self.get_short_term_aggregation_map(),
            clearing=self._clearing,
            graph_clearing=graph_clearing,
            aggregated=self.get_short_term_model(),
            graph_aggregated=graph_short,
        )

    def get_medium_term_graph_map(
        self,
        graph_clearing: dict[str, Component],
        graph_medium: dict[str, Component],
    ) -> dict[str, set[str] | None]:
        """Get aggregation map for version of medium term Model with graph of Flows and Nodes."""
        return self._get_graph_aggregation_map(
            original_agg_map=self.get_medium_term_aggregation_map(),
            clearing=self._clearing,
            graph_clearing=graph_clearing,
            aggregated=self.get_medium_term_model(),
            graph_aggregated=graph_medium,
        )

    def get_long_term_graph_map(
        self,
        graph_clearing: dict[str, Component],
        graph_long: dict[str, Component],
    ) -> dict[str, set[str] | None]:
        """Get aggregation map for version of long term Model with graph of Flows and Nodes."""
        return self._get_graph_aggregation_map(
            original_agg_map=self.get_long_term_aggregation_map(),
            clearing=self._clearing,
            graph_clearing=graph_clearing,
            aggregated=self.get_long_term_model(),
            graph_aggregated=graph_long,
        )

    def assert_equal_storages(
        self,
        simpler_short: dict[str, Component],
        simpler_medium: dict[str, Component],
        simpler_long: dict[str, Component],
    ) -> None:
        """Check that all Nodes with Storages are preserved between short, medium and long term Models.

        Args:
            simpler_short (dict[str, Component]): Short term Model Components.
            simpler_medium (dict[str, Component]): Medium term Model Components.
            simpler_long (dict[str, Component]): Long term Model Components.

        Raises:
            ValueError: If the Models have differing Storages.

        """
        short_storages = self._get_storages(simpler_short)
        medium_storages = self._get_storages(simpler_medium)
        long_storages = self._get_storages(simpler_long)

        if short_storages != medium_storages != long_storages:
            message = "Storages are not equal between short, medium and long term Models."
            unique_short = short_storages - (medium_storages | long_storages)
            unique_medium = medium_storages - (short_storages | long_storages)
            unique_long = long_storages - (short_storages | medium_storages)
            if unique_short:
                message += f"\n - Unique Nodes with Storages in Short Model: {unique_short}"
            if unique_medium:
                message += f"\n - Unique Nodes with Storages in Medium Model: {unique_medium}"
            if unique_long:
                message += f"\n - Unique Nodes with Storages in Long Model: {unique_long}"
            raise ValueError(message)

    def _aggregate(self, model: Model, aggs: list[Aggregator]) -> Model:
        if aggs:
            # works because aggregators should not modify the original components
            # except if disaggregate is called, but we shall only use aggregate
            agg_model = Model()
            agg_model.get_data().update(model.get_data())
        else:
            agg_model = model
        for agg in aggs:
            agg.aggregate(agg_model)
        return agg_model

    def _create_aggregation_map(self, clearing: Model, aggs: list[Aggregator]) -> dict[str, list[str]]:
        """Merge aggregation maps of a list of aggregators from clearing to final."""
        clearing_ids = [name for name, ob in clearing.get_data().items() if isinstance(ob, Component)]
        full_agg_mapping = {k: {v} for k, v in zip(clearing_ids, clearing_ids, strict=True)}

        for agg in aggs:
            agg_map = agg.get_aggregation_map()
            for detailed_id, aggregated_ids in full_agg_mapping.items():
                if not aggregated_ids:  # Component has been deleted by an aggregation.
                    continue

                new_agg_ids = set()
                for agg_id in aggregated_ids:
                    if agg_id not in agg_map:
                        new_agg_ids.add(agg_id)  # left as is
                        continue
                    if not agg_map[agg_id]:
                        # deleted. if all agg_ids are marked deleted, so is the detailed one.
                        continue
                    new_agg_ids |= agg_map[agg_id]

                full_agg_mapping[detailed_id] = new_agg_ids  # empty set signifies deleted component

        return full_agg_mapping

    def _get_graph_aggregation_map(
        self,
        original_agg_map: dict[str, set[str]],
        clearing: Model | dict[str, Component],
        graph_clearing: dict[str, Component],
        aggregated: Model | dict[str, Component],
        graph_aggregated: dict[str, Component],
    ) -> dict[str, set[str]]:
        """Create aggregation map with simpler Component IDs based on an original mapping from clearing to aggregated.

        Use get_top_parent of components to find IDs in original_agg_map then change to the Flow/Node ID.

        Args:
            original_agg_map (dict[str, set[str]]): Mapping between Components of clearing and aggregated Models.
            clearing (dict[str, Component]): Clearing Model with top parents.
            graph_clearing (dict[str, Component]): Clearing Model version with Flows and Nodes. Derived from
                                                     clearing.
            aggregated (dict[str, Component]): Aggregated Model with top parents. Aggregated from clearing.
            graph_aggregated (dict[str, Component]): Aggregated Model version with Flows and Nodes. Derived from
                                                       aggregated.

        Returns:
            dict[str, set[str]]: Mapping between components of simpler clearing and simpler aggregated Models.

        """
        if isinstance(clearing, Model):
            clearing = {k: v for k, v in clearing.get_data().items() if isinstance(v, Component)}
        if isinstance(aggregated, Model):
            aggregated = {k: v for k, v in aggregated.get_data().items() if isinstance(v, Component)}

        self._check_agg_map_compatibility(clearing, aggregated, original_agg_map)

        graph_clearing_map = self._get_top_parent_to_simple(original=clearing, simpler=graph_clearing)
        graph_aggregated_map = self._get_top_parent_to_simple(original=aggregated, simpler=graph_aggregated)
        simple_agg_map = {}

        for clearing_id, agg_ids in original_agg_map.items():
            # the two if statements are there for if we want to map only a subset of the simpler Components.
            if clearing_id in graph_clearing_map:
                if not agg_ids:
                    continue  # choose not to add deleted components. May change this later.
                simple_agg_ids = set()
                for agg_id in agg_ids:
                    if agg_id in graph_aggregated_map:  # Again to allow subset to be mapped
                        simple_agg_ids |= graph_aggregated_map[agg_id]  # add set if simple component ids
                if simple_agg_ids:
                    for graph_clearing_id in graph_clearing_map[clearing_id]:
                        simple_agg_map[graph_clearing_id] = simple_agg_ids

        self._check_agg_map_validity(graph_clearing, graph_aggregated, simple_agg_map)
        return simple_agg_map

    def _check_agg_map_compatibility(
        self,
        clearing: Model | dict[str, Component],
        aggregated: Model | dict[str, Component],
        original_agg_map: dict[str, set[str]],
    ) -> None:
        if set(clearing.keys()) != set(original_agg_map.keys()):
            missing_in_clearing = set(original_agg_map.keys()).difference(clearing.keys())
            extra_in_clearing = set(clearing.keys()).difference(original_agg_map.keys())
            message = (
                "clearing is incompatible with the aggregation mapping between clearing and aggregated Models.\n"
                f"Missing in clearing: {missing_in_clearing}\n"
                f"Extra in clearing: {extra_in_clearing}"
            )
            raise KeyError(message)

        original_agg_map_values = set().union(*(v for v in original_agg_map.values() if v))
        if set(aggregated.keys()) != original_agg_map_values:
            missing_in_aggregated = original_agg_map_values.difference(aggregated.keys())
            extra_in_aggregated = set(aggregated.keys()).difference(original_agg_map_values)
            message = (
                "aggregated is incompatible with the aggregation mapping between clearing and aggregated Models.\n"
                f"Missing in aggregated: {missing_in_aggregated}\n"
                f"Extra in aggregated: {extra_in_aggregated}"
            )
            raise KeyError(message)

    def _check_agg_map_validity(
        self,
        original_components: dict[str, Component],
        aggregated_components: dict[str, Component],
        agg_map: dict[str, set[str] | None],
    ) -> None:
        """Check Flow and Node rules for all mappings in an aggregation map."""
        errors = set()
        for original_id, aggregated_ids in agg_map.items():
            component = original_components[original_id]
            if isinstance(component, Node):
                self._check_node_rules(original_id, component, aggregated_ids, aggregated_components, errors)

            if isinstance(component, Flow) and component.get_startupcost() is not None:
                self._check_flow_rules(original_id, aggregated_ids, aggregated_components, errors)

            if not isinstance(component, (Flow, Node)):
                message = (
                    f"Invalig Model of simpler Components. Must consist of only Flows and Nodes. Found: {component}"
                )
                raise ValueError(message)

        self._report_errors(errors)

    def _check_node_rules(
        self,
        original_id: str,
        node: Node,
        aggregated_ids: set[str] | None,
        aggregated_components: dict[str, Component],
        errors: set[str],
    ) -> None:
        """Check rules for Nodes for a Component ID in an aggregation map.

        A Node on the disaggregated side (keys) must map to exactly one other Node. More keys are alowed to map to the
        same aggregated Node.

        """
        if node.get_storage() is None:
            # Check rules here?
            return

        if aggregated_ids is None:
            e = f"Node with Storage {original_id} was deleted during aggregations. This is not supported in JulES."
            errors.add(e)
            return
        aggregated_storages = set()
        for agg_id in aggregated_ids:
            agg_component = aggregated_components[agg_id]
            if isinstance(agg_component, Node) and agg_component.get_storage() is not None:
                aggregated_storages.add(agg_id)
        if len(aggregated_storages) != 1:
            errors.add(
                f"Node with Storage {original_id} must be connected to exactly one Node with Storage in the "
                f"aggregation map in JulES. Currently connected to: {aggregated_storages}.",
            )

    def _check_flow_rules(
        self,
        original_id: str,
        aggregated_ids: set[str] | None,
        aggregated_components: dict[str, Component],
        errors: set[str],
    ) -> None:
        """Check rules for Flows for a Component ID in an aggregation map.

        A Flow on the disaggregated side (keys) must map to exactly one other Flow. More keys are alowed to map to the
        same aggregated Flow.

        """
        if aggregated_ids is None:
            e = f"Flow with StartUpCost {original_id} was deleted during aggregations. This is not supported in JulES."
            errors.add(e)
            return
        aggregated_flows = set()
        for agg_id in aggregated_ids:
            agg_component = aggregated_components[agg_id]
            if isinstance(agg_component, Flow) and agg_component.get_startupcost() is not None:
                aggregated_flows.add(agg_id)
        if len(aggregated_flows) != 1:
            errors.add(
                f"Flow with StartUpCost {original_id} must be connected to exactly one Flow with StartUpCost in the "
                f"aggregation map in JulES. Currently connected to: {aggregated_flows}.",
            )

    @staticmethod
    def _get_storages(simpler: dict[str, Component]) -> set[str]:
        nodes_with_storages = set()
        for n, c in simpler.items():
            if isinstance(c, Node) and c.get_storage() is not None:
                nodes_with_storages.add(n)
        return nodes_with_storages

    @staticmethod
    def _get_top_parent_to_simple(original: dict[str, Component], simpler: dict[str, Component]) -> dict[str, set[str]]:
        """Map simpler components to their top parent."""
        inv_original = {c: n for n, c in original.items()}
        simpler_map: dict[str, set[str]] = {}

        for simple_id, component in simpler.items():
            top_parent = component.get_top_parent()
            if top_parent is None:
                message = (
                    f"Component {component} with ID {simple_id} has no parents. This means it has not been "
                    "derived from original."
                )
                raise ValueError(message)
            try:
                top_parent_id = inv_original[top_parent]
            except KeyError as e:
                message = (
                    f"Component {top_parent} does not exist in original Model. This means simpler has not been "
                    "derived from original."
                )
                raise KeyError(message) from e
            if top_parent_id in simpler_map:
                # list has been set, wo we add the simple component id to the
                simpler_map[top_parent_id].add(simple_id)
            else:
                simpler_map[top_parent_id] = {simple_id}

        return simpler_map

__init__(clearing: Model, short: list[Aggregator], medium: list[Aggregator], long: list[Aggregator]) -> None

Initialize a JulESAggregator instance.

Parameters:

Name	Type	Description	Default
clearing	Model	The clearing Model to aggregate from.	required
short	list[Aggregator]	List of aggregations to create the short term price prognosis Model from clearing.	required
medium	list[Aggregator]	List of aggregations to create the medium term price prognosis Model from short	required
long	list[Aggregator]	List of aggregations to create the long term price prognosis Model from medium.	required

Source code in framjules/solve_handler/JulESAggregator.py

def __init__(
    self,
    clearing: Model,
    short: list[Aggregator],
    medium: list[Aggregator],
    long: list[Aggregator],
) -> None:
    """Initialize a JulESAggregator instance.

    Args:
        clearing (Model): The clearing Model to aggregate from.
        short (list[Aggregator]): List of aggregations to create the short term price prognosis Model from clearing.
        medium (list[Aggregator]): List of aggregations to create the medium term price prognosis Model from short
        long (list[Aggregator]): List of aggregations to create the long term price prognosis Model from medium.

    """
    self._clearing = clearing
    self._short = short
    self._medium = medium
    self._long = long

    self._short_model: Model | None = None
    self._medium_model: Model | None = None
    self._long_model: Model | None = None

assert_equal_storages(simpler_short: dict[str, Component], simpler_medium: dict[str, Component], simpler_long: dict[str, Component]) -> None

Check that all Nodes with Storages are preserved between short, medium and long term Models.

Parameters:

Name	Type	Description	Default
simpler_short	dict[str, Component]	Short term Model Components.	required
simpler_medium	dict[str, Component]	Medium term Model Components.	required
simpler_long	dict[str, Component]	Long term Model Components.	required

Raises:

Type	Description
ValueError	If the Models have differing Storages.

Source code in framjules/solve_handler/JulESAggregator.py

def assert_equal_storages(
    self,
    simpler_short: dict[str, Component],
    simpler_medium: dict[str, Component],
    simpler_long: dict[str, Component],
) -> None:
    """Check that all Nodes with Storages are preserved between short, medium and long term Models.

    Args:
        simpler_short (dict[str, Component]): Short term Model Components.
        simpler_medium (dict[str, Component]): Medium term Model Components.
        simpler_long (dict[str, Component]): Long term Model Components.

    Raises:
        ValueError: If the Models have differing Storages.

    """
    short_storages = self._get_storages(simpler_short)
    medium_storages = self._get_storages(simpler_medium)
    long_storages = self._get_storages(simpler_long)

    if short_storages != medium_storages != long_storages:
        message = "Storages are not equal between short, medium and long term Models."
        unique_short = short_storages - (medium_storages | long_storages)
        unique_medium = medium_storages - (short_storages | long_storages)
        unique_long = long_storages - (short_storages | medium_storages)
        if unique_short:
            message += f"\n - Unique Nodes with Storages in Short Model: {unique_short}"
        if unique_medium:
            message += f"\n - Unique Nodes with Storages in Medium Model: {unique_medium}"
        if unique_long:
            message += f"\n - Unique Nodes with Storages in Long Model: {unique_long}"
        raise ValueError(message)

get_long_term_aggregation_map() -> dict[str, set[str] | None]

Get the aggregation map of Components from clearing to long term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_long_term_aggregation_map(self) -> dict[str, set[str] | None]:
    """Get the aggregation map of Components from clearing to long term Model."""
    return self._create_aggregation_map(self._clearing, self._short + self._medium + self._long)

get_long_term_graph_map(graph_clearing: dict[str, Component], graph_long: dict[str, Component]) -> dict[str, set[str] | None]

Get aggregation map for version of long term Model with graph of Flows and Nodes.

Source code in framjules/solve_handler/JulESAggregator.py

def get_long_term_graph_map(
    self,
    graph_clearing: dict[str, Component],
    graph_long: dict[str, Component],
) -> dict[str, set[str] | None]:
    """Get aggregation map for version of long term Model with graph of Flows and Nodes."""
    return self._get_graph_aggregation_map(
        original_agg_map=self.get_long_term_aggregation_map(),
        clearing=self._clearing,
        graph_clearing=graph_clearing,
        aggregated=self.get_long_term_model(),
        graph_aggregated=graph_long,
    )

get_long_term_model() -> Model

Apply defined aggregations for long term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_long_term_model(self) -> Model:
    """Apply defined aggregations for long term Model."""
    if self._long_model is None:
        self._long_model = self._aggregate(self.get_medium_term_model(), self._long)
    return self._long_model

get_medium_term_aggregation_map() -> dict[str, set[str] | None]

Get the aggregation map of Components from clearing to medium term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_medium_term_aggregation_map(self) -> dict[str, set[str] | None]:
    """Get the aggregation map of Components from clearing to medium term Model."""
    return self._create_aggregation_map(self._clearing, self._short + self._medium)

get_medium_term_graph_map(graph_clearing: dict[str, Component], graph_medium: dict[str, Component]) -> dict[str, set[str] | None]

Get aggregation map for version of medium term Model with graph of Flows and Nodes.

Source code in framjules/solve_handler/JulESAggregator.py

def get_medium_term_graph_map(
    self,
    graph_clearing: dict[str, Component],
    graph_medium: dict[str, Component],
) -> dict[str, set[str] | None]:
    """Get aggregation map for version of medium term Model with graph of Flows and Nodes."""
    return self._get_graph_aggregation_map(
        original_agg_map=self.get_medium_term_aggregation_map(),
        clearing=self._clearing,
        graph_clearing=graph_clearing,
        aggregated=self.get_medium_term_model(),
        graph_aggregated=graph_medium,
    )

get_medium_term_model() -> Model

Apply defined aggregations for medium term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_medium_term_model(self) -> Model:
    """Apply defined aggregations for medium term Model."""
    if self._medium_model is None:
        self._medium_model = self._aggregate(self.get_short_term_model(), self._medium)
    return self._medium_model

get_short_term_aggregation_map() -> dict[str, set[str] | None]

Get the aggregation map of Components from clearing to short term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_short_term_aggregation_map(self) -> dict[str, set[str] | None]:
    """Get the aggregation map of Components from clearing to short term Model."""
    return self._create_aggregation_map(self._clearing, self._short)

get_short_term_graph_map(graph_clearing: dict[str, Component], graph_short: dict[str, Component]) -> dict[str, set[str] | None]

Get aggregation map for version of short term Model with graph of Flows and Nodes.

Source code in framjules/solve_handler/JulESAggregator.py

def get_short_term_graph_map(
    self,
    graph_clearing: dict[str, Component],
    graph_short: dict[str, Component],
) -> dict[str, set[str] | None]:
    """Get aggregation map for version of short term Model with graph of Flows and Nodes."""
    return self._get_graph_aggregation_map(
        original_agg_map=self.get_short_term_aggregation_map(),
        clearing=self._clearing,
        graph_clearing=graph_clearing,
        aggregated=self.get_short_term_model(),
        graph_aggregated=graph_short,
    )

get_short_term_model() -> Model

Apply defined aggregations for short term Model.

Source code in framjules/solve_handler/JulESAggregator.py

def get_short_term_model(self) -> Model:
    """Apply defined aggregations for short term Model."""
    if self._short_model is None:
        self._short_model = self._aggregate(self._clearing, self._short)
    return self._short_model

JulESNames

JulESNames

Constants (both static and dynamic ones defined in init) used in JulES.

Source code in framjules/solve_handler/JulESNames.py

class JulESNames:
    """Constants (both static and dynamic ones defined in __init__) used in JulES."""

    JSON_INDENT = 4
    YAML_INDENT = 4

    OTHER_TERMS = "otherterms"
    VARS = "Vars"

    AGGREGATED = "aggregated"
    MACRO = "macro"
    BALANCE_RHSDATA = "balance"

    FILENAME_CONFIG = "config.yaml"
    FILENAME_H5_OUTPUT = "output.h5"
    FILENAME_STORAGE_MAPPING = "storage_mapping.json"
    FILENAME_START_STORAGES_AGGREGATED = "start_storages_aggregated.json"
    FILENAME_START_STORAGES_CLEARING = "start_storages_clearing.json"
    ROOT_FILENAME_DATAELEMENTS = "data_elements"
    FILENAME_DATAELEMENTS_TIMEVECTORS = "data_elements_timevectors.json"

    PY_JULES_SETTINGS_NAME = "python_jules_settings"
    PY_JULES_OUTPUT_NAME = "python_jules_output"

    MAIN = "main"
    MISSING_CONFIG = "missing_config"

    RESULTS = "results"
    MAINRESULTS = "mainresults"
    TIMES = "times"
    SCENARIOS = "scenarios"
    MEMORY = "memory"
    STORAGEVALUES = "storagevalues"
    STORAGEVALUES_ALL_PROBLEMS = "storagevalues_all_problems"
    ALL = "all"

    TERM_DURATION_WEEKS = "termduration_weeks"
    TERM_DURATION_DAYS = "termduration_days"
    TERM_DURATION_HOURS = "termduration_hours"
    SEQUENTIAL_HORIZON = "SequentialHorizon"
    ADAPTIVE_HORIZON = "AdaptiveHorizon"

    COMMODITIES = "commodities"
    POWER = "Power"
    HYDRO = "Hydro"
    BATTERY = "Battery"

    SHRINKAFTER_DAYS = "startafter_days"
    SHRINKATLEAST_DAYS = "shrinkatleast_days"

    TWO_STORAGE_DURATION = "twostorageduration"
    SHORT_STOCH_DURATION_HOURS = "shortstochduration_hours"
    LONG_STOCH_DURATION_DAYS = "longstochduration_days"
    LONG_EV_DURATION_DAYS = "longevduration_days"

    DISTRIBUTION_METHOD_MP = "distribution_method_mp"
    DISTRIBUTION_METHOD_SP = "distribution_method_sp"
    BYSIZE = "bysize"
    ADVANCED = "advanced"
    STORAGE = "storage"
    GREEDY = "greedy"
    WITHMP = "withmp"

    STATEDEPENDENT_PROD = "statedependentprod"
    STATEDEPENDENT_PUMP = "statedependentpump"
    HEADLOSSCOST = "headlosscost"

    SUBSYSTEMS = "subsystems"
    RESULTS = "results"
    STARTSTORAGES = "startstorages"
    ENDVALUE = "endvalue"

    OUTPUT_FORMAT = "outputformat"
    DATETIME_FORMAT = "datetimeformat"
    DATETIME_FORMAT_JULESIO = "yyyy-mm-ddTHH:MM:SS"
    HDF5 = "hdf5"
    ELASTIC = "elastic"
    TRUE = True
    FALSE = False

    JULIA = "julia"
    INPUT = "input"
    OUTPUT_PATH = "outputpath"
    NUM_CORES = "numcores"
    DATA_YEARS = "datayears"
    SCENARIO_YEARS = "weatheryears"
    WEEK_START = "weekstart"
    NUM_SIM_YEARS = "simulationyears"
    EXTRA_STEPS = "extrasteps"
    SETTINGS = "settings"
    OUTPUT_NAME = "outputname"

    OUTPUT_INDEX = "outputindex"
    WEATHER_YEAR = "weatheryear"
    DATA_YEAR = "datayear"

    TIME = "time"
    WEATHER_YEAR_START = "weatheryearstart"
    WEATHER_YEAR_STOP = "weatheryearstop"
    PROB_TIME = "probtime"
    NORMAL_TIME = "normaltime"

    FIXED_DATA_TWO_TIME = "FixedDataTwoTime"
    PHASE_IN_FIXED_DATA_TWO_TIME = "PhaseinFixedDataTwoTime"

    PHASE_IN_TIME = "phaseintime"
    PHASE_IN_DELTA_DAYS = "phaseindelta_days"
    PHASE_IN_DELTA_STEPS = "phaseinsteps"
    PROBLEMS = "problems"
    PROGNOSIS = "prognosis"
    SIMULATION = "simulation"
    SHRINKABLE = "shrinkable"
    AGGZONE = "aggzone"
    AGGSUPPLYN = "aggsupplyn"
    SHORT_TERM_STORAGE_CUTOFF_HOURS = "shorttermstoragecutoff_hours"
    SHORTER_THAN_PROGNOSIS_MED_DAYS = "shorterthanprognosismed_days"
    LONG = "long"
    MED = "med"
    SHORT = "short"
    PROB = "prob"
    SOLVER = "solver"
    FUNCTION = "function"
    AGG_STARTMAG_DICT = "aggstartmagdict"
    STARTMAG_DICT = "startmagdict"
    RESIDUAL_AREA_LIST = "residualarealist"
    END_CONDITION = "endcondition"
    START_EQUAL_STOP = "startequalstop"
    MONTHLY_PRICE = "monthly_price"

    SCENARIO_GENERATION = "scenariogeneration"
    INFLOW_CLUSTERING_METHOD = "InflowClusteringMethod"
    NUM_SCEN = "numscen"
    SCEN_DELTA_DAYS = "scendelta_days"

    PARTS = "parts"

    SKIPMAX = "skipmax"

    HIGHS_PROB = "HiGHS_Prob()"
    HIGHS_SIMPLEX = "HighsSimplexMethod()"
    HIGHS_SIMPLEX_NO_WARMSTART = "HighsSimplexMethod(warmstart=false)"
    HIGHS_SIMPLEX_SIP_NO_WARMSTART = "HighsSimplexSIPMethod(warmstart=false)"
    JUMP_HIGHS = "JuMPHiGHSMethod()"

    STOCHASTIC = "stochastic"
    MAXCUTS = "maxcuts"
    LB = "lb"
    RELTOL = "reltol"
    ONLY_AGG_HYDRO = "onlyagghydro"
    MASTER = "master"
    SUBS = "subs"
    SUB = "sub"

    HORIZONS = "horizons"
    HORIZON_DURATION_WEEKS = "horizonduration_weeks"
    HORIZON_DURATION_HOURS = "horizonduration_hours"
    PERIOD_DURATION_DAYS = "periodduration_days"
    PERIOD_DURATION_HOURS = "periodduration_hours"
    POWER_PARTS = "powerparts"

    RHSDATA = "rhsdata"
    FIND_FIRST_DYNAMIC_EXOGEN_PRICE_AH_DATA = "FindFirstDynamicExogenPriceAHData"
    DYNAMIC_EXOGEN_PRICE_AH_DATA = "DynamicExogenPriceAHData"
    DYNAMIC_RHS_AH_DATA = "DynamicRHSAHData"
    RHSMETHOD = "rhsmethod"
    KMEANS_AH_METHOD = "KMeansAHMethod()"
    CLUSTERS = "clusters"
    UNIT_DURATION_HOURS = "unitduration_hours"

    SETTINGS_SCENARIO_YEAR_START = "scenarioyearstart"

    CLEARING = "clearing"
    SHORT_TERM = "short_term"
    MEDIUM_TERM = "medium_term"
    LONG_TERM = "long_term"

    MARKET = "Power"
    STORAGE_SYSTEM = "Hydro"
    SHORT_TERM_STORAGE = "Battery"

    DFMTin = "%Y-%m-%dT%H:%M:%S"

    FLOW = "Flow"
    STORAGE = "Storage"
    BALANCE = "Balance"
    COMMODITY = "Commodity"
    PARAM = "Param"
    CAPACITY = "Capacity"
    RHSTERM = "RHSTerm"
    TIMEVECTOR = "TimeVector"
    TIMEINDEX = "TimeIndex"
    TABLE = "Table"
    TIMEDELTA = "TimeDelta"
    TIMEVALUES = "TimeValues"
    ARROW = "Arrow"
    LOSS = "Loss"
    PRICE = "Price"
    CONVERSION = "Conversion"
    COST = "Cost"
    STARTUPCOST = "StartUpCost"

    BASEFLOW = "BaseFlow"
    BASEBALANCE = "BaseBalance"
    EXOGENBALANCE = "ExogenBalance"
    BASESTORAGE = "BaseStorage"
    MWTOGWHPARAM = "MWToGWhParam"
    M3STOMM3PARAM = "M3SToMM3Param"
    MEANSERIESPARAM = "MeanSeriesParam"
    MSTIMEDELTA = "MsTimeDelta"
    INFINITETIMEVECTOR = "InfiniteTimeVector"
    ROTATINGTIMEVECTOR = "RotatingTimeVector"
    ONEYEARTIMEVECTOR = "OneYearTimeVector"
    CONSTANTTIMEVECTOR = "ConstantTimeVector"
    RANGETIMEINDEX = "RangeTimeIndex"
    VECTORTIMEINDEX = "VectorTimeIndex"
    BASETABLE = "BaseTable"
    COLUMNTIMEVALUES = "ColumnTimeValues"
    VECTORTIMEVALUES = "VectorTimeValues"
    LOWERZEROCAPACITY = "LowerZeroCapacity"
    POSITIVECAPACITY = "PositiveCapacity"
    BASERHSTERM = "BaseRHSTerm"
    BASEARROW = "BaseArrow"
    SEGMENTEDARROW = "SegmentedArrow"
    SIMPLELOSS = "SimpleLoss"
    COSTTERM = "CostTerm"
    SIMPLESTARTUPCOST = "SimpleStartUpCost"

    LOSSFACTORKEY = "LossFactor"
    UTILIZATIONKEY = "Utilization"
    FALLBACK_UTILIZATION = 0.5

    STARTCOSTKEY = "StartCost"
    MINSTABLELOADKEY = "MinStableLoad"

    WHICHCONCEPT = "WhichConcept"
    WHICHINSTANCE = "WhichInstance"

    DIRECTIONKEY = "Direction"
    DIRECTIONIN = "In"
    DIRECTIONOUT = "Out"

    BOUNDKEY = "Bound"
    BOUNDUPPER = "Upper"
    BOUNDLOWER = "Lower"

    LEVEL = "Level"
    PROFILE = "Profile"
    VALUE = "Value"
    START = "Start"
    STEPS = "Steps"
    DELTA = "Delta"
    PERIOD = "Period"
    VECTOR = "Vector"
    MATRIX = "Matrix"
    NAMES = "Names"
    NAME = "Name"

    METADATA = "Metadata"
    GLOBALENEQ = "GlobalEneq"
    RESIDUALHINT = "Residualhint"
    STORAGEHINT = "Storagehint"

    JULES_CONFIG = "config.yaml"
    OUTPUT_FOLDER = "output"
    JULIA_ENV_NAME = "JulES_julia_env"

    def __init__(self) -> None:
        """Dynamically settable names for JulES."""
        # This is set in BuildHandler when we build data elements
        # for the clearing model. It is used in ConfigHandler in
        # connection with using AdaptiveHorizon
        self.dummy_exogenous_balance_name: str | None = None
        self.dummy_exogenous_profile_id: str | None = None

__init__() -> None

Dynamically settable names for JulES.

Source code in framjules/solve_handler/JulESNames.py

def __init__(self) -> None:
    """Dynamically settable names for JulES."""
    # This is set in BuildHandler when we build data elements
    # for the clearing model. It is used in ConfigHandler in
    # connection with using AdaptiveHorizon
    self.dummy_exogenous_balance_name: str | None = None
    self.dummy_exogenous_profile_id: str | None = None

SolveHandler

SolveHandler

Bases: Base

Common data methods for different simulation modes.

Source code in framjules/solve_handler/SolveHandler.py

class SolveHandler(Base):
    """Common data methods for different simulation modes."""

    def __init__(self, folder: Path, clearing_model: Model, config: JulESConfig) -> None:
        """Hold all data and methods needed to solve JulES."""
        names = JulESNames()

        t = time()
        short_aggregations = config.get_short_term_aggregations()  # list of Aggregators
        self.send_debug_event(f"get_short_term_aggregations time: {round(time() - t, 2)} seconds")
        mid_aggregations = []  # config.get_mid_term_aggregations()  # not supported yet
        long_aggregations = []  # config.get_long_term_aggregations()  # not supported yet

        t = time()
        aggregator = JulESAggregator(
            clearing=clearing_model,
            short=short_aggregations,
            medium=mid_aggregations,
            long=long_aggregations,
        )
        self.send_debug_event(f"JulESAggregator init time: {round(time() - t, 2)} seconds")

        t = time()
        short_term_model = aggregator.get_short_term_model()
        self.send_debug_event(f"get_short_term_model init time: {round(time() - t, 2)} seconds")
        t = time()
        medium_term_model = aggregator.get_medium_term_model()
        self.send_debug_event(f"get_medium_term_model init time: {round(time() - t, 2)} seconds")
        t = time()
        long_term_model = aggregator.get_long_term_model()
        self.send_debug_event(f"get_long_term_model init time: {round(time() - t, 2)} seconds")

        t = time()
        domain_models = DomainModels(
            clearing=clearing_model,
            short_term=short_term_model,
            medium_term=medium_term_model,
            long_term=long_term_model,
        )
        self.send_debug_event(f"DomainModels init time: {round(time() - t, 2)} seconds")

        t = time()
        supported_types = (Flow, Node)
        forbidden_types = tuple()
        graphs = NodeFlowGraphs(
            clearing=get_supported_components(
                self._get_components(clearing_model),
                supported_types,
                forbidden_types,
            ),
            short_term=get_supported_components(
                self._get_components(short_term_model),
                supported_types,
                forbidden_types,
            ),
            medium_term=get_supported_components(
                self._get_components(medium_term_model),
                supported_types,
                forbidden_types,
            ),
            long_term=get_supported_components(
                self._get_components(long_term_model),
                supported_types,
                forbidden_types,
            ),
        )
        self.send_debug_event(f"NodeFlowGraphs init time: {round(time() - t, 2)} seconds")

        t = time()
        graph_infos = GraphInfos(
            clearing={k: ComponentInfo() for k in graphs.clearing},
            short_term={k: ComponentInfo() for k in graphs.short_term},
            medium_term={k: ComponentInfo() for k in graphs.medium_term},
            long_term={k: ComponentInfo() for k in graphs.long_term},
        )
        self.send_debug_event(f"GraphInfos init time: {round(time() - t, 2)} seconds")

        # we check that that aggregated models don't have different storages
        t = time()
        aggregator.assert_equal_storages(
            graphs.short_term,
            graphs.medium_term,
            graphs.long_term,
        )
        self.send_debug_event(f"assert_equal_storages time: {round(time() - t, 2)} seconds")

        t = time()
        constructor = CacheDB if config.is_cache_db() else ModelDB
        db = constructor(
            domain_models.clearing,
            domain_models.short_term,
            domain_models.medium_term,
            domain_models.long_term,
        )
        self.send_debug_event(f"DB init time: {round(time() - t, 2)} seconds")

        t = time()
        self.fill_graph_infos(graph_infos, graphs, names, aggregator, config, db)
        self.send_debug_event(f"fill_graph_infos time: {round(time() - t, 2)} seconds")

        # Finally, we set the member data
        self.folder: Path = folder
        self.config: JulESConfig = config
        self.names: JulESNames = names
        self.domain_models: DomainModels = domain_models
        self.graphs: NodeFlowGraphs = graphs
        self.graph_infos: GraphInfos = graph_infos
        # NB! will be freed after self.configure()
        # so we don't hold up memory during run
        self.db: QueryDB = db

    def build(self) -> None:
        """Build input files for JulES."""
        handler = self.create_build_handler()
        handler.build()

    def configure(self) -> None:
        """Build configuration file for JulES."""
        handler = self.create_config_handler()
        handler.configure()

        self.db = None
        gc.collect()

    def run(self) -> None:
        """Run Julia-JulES."""
        handler = self.create_run_handler()
        handler.run()

    def set_results(self) -> None:
        """Set results from Julia-JulES run into domain models."""
        handler = self.create_results_handler()
        handler.set_results()

    def create_build_handler(self) -> BuildHandler:
        """Create specialized BuildHandler for the chosen simulation mode."""
        if self.config.is_simulation_mode_serial():
            handler_constructor = SerialBuildHandler
        else:
            raise NotImplementedError

        return handler_constructor(
            folder=self.folder,
            config=self.config,
            names=self.names,
            domain_models=self.domain_models,
            graphs=self.graphs,
            graph_infos=self.graph_infos,
            db=self.db,
        )

    def create_config_handler(self) -> ConfigHandler:
        """Create specialized ConfigHandler for the chosen simulation mode."""
        if self.config.is_simulation_mode_serial():
            handler_constructor = SerialConfigHandler
        else:
            raise NotImplementedError

        return handler_constructor(
            folder=self.folder,
            config=self.config,
            names=self.names,
            graph_infos=self.graph_infos,
        )

    def create_run_handler(self) -> SerialRunHandler:
        """Create specialized RunHandler for the chosen simulation mode."""
        dependencies = []

        tulipa_version = self.config.get_tulipa_version()
        if tulipa_version is not None:
            if Path.exists(Path(tulipa_version)):
                dependencies.append(tulipa_version)
            else:
                dependencies.append(("https://github.com/NVE/TuLiPa.git", tulipa_version))

        jules_version = self.config.get_jules_version()
        if jules_version is not None:
            if Path.exists(Path(jules_version)):
                dependencies.append(jules_version)
            else:
                dependencies.append(("https://github.com/NVE/JulES.git", jules_version))

        dependencies.extend(["YAML", "HDF5", "JSON", "PythonCall"])

        if self.config.is_simulation_mode_serial():
            handler_constructor = SerialRunHandler
        else:
            message = "JulES Parallel simulation mode is not yet supported."
            raise NotImplementedError(message)
        handler_constructor.ENV_NAME = self.names.JULIA_ENV_NAME
        return handler_constructor(folder=self.folder, config=self.config, names=self.names, dependencies=dependencies)

    def create_results_handler(self) -> SerialResultsHandler:
        """Create a SerialResultsHandler."""
        if self.config.is_simulation_mode_serial():
            handler_constructor = SerialResultsHandler
        else:
            message = "JulES Parallel simulation mode is not yet supported."
            raise NotImplementedError(message)
        return handler_constructor(
            folder=self.folder,
            config=self.config,
            names=self.names,
            graphs=self.graphs,
            graph_infos=self.graph_infos,
        )

    def _get_components(self, model: Model) -> dict[str, Component]:
        return {k: v for k, v in model.get_data().items() if isinstance(v, Component)}

    def fill_graph_infos(
        self,
        graph_infos: GraphInfos,
        graphs: NodeFlowGraphs,
        names: JulESNames,
        aggregator: JulESAggregator,
        config: JulESConfig,
        db: QueryDB,
    ) -> None:
        """Fill graph_info with derived info."""
        # Intent is to gather complex derivations in just one place

        # NB! Order of below method calls matter
        t = time()
        self.set_basic_node_flow_info(graph_infos.clearing, graphs.clearing, db, config)
        self.set_basic_node_flow_info(graph_infos.short_term, graphs.short_term, db, config)
        self.set_basic_node_flow_info(graph_infos.medium_term, graphs.medium_term, db, config)
        self.set_basic_node_flow_info(graph_infos.long_term, graphs.long_term, db, config)
        self.send_debug_event(f"set_basic_node_flow_info time: {round(time() - t, 2)} seconds")

        t = time()
        self.set_agg_storage_node_info(graph_infos.clearing, aggregator, graphs.clearing, graphs.short_term)
        self.send_debug_event(f"set_agg_storage_node_info time: {round(time() - t, 2)} seconds")

        t = time()  # opposite order to check if higher level is short term sss
        self.set_sss_info(graphs.short_term, graph_infos.short_term, graph_infos.medium_term, names)
        self.set_sss_info(graphs.clearing, graph_infos.clearing, graph_infos.short_term, names)
        self.send_debug_event(f"set_sss_info time: {round(time() - t, 2)} seconds")

        t = time()
        self.set_market_info(graph_infos.clearing, graphs.clearing, names)
        self.set_market_info(graph_infos.short_term, graphs.short_term, names)
        self.set_market_info(graph_infos.medium_term, graphs.medium_term, names)
        self.set_market_info(graph_infos.long_term, graphs.long_term, names)
        self.send_debug_event(f"set_market_info time: {round(time() - t, 2)} seconds")

        # self.set_agg_market_node_info(graph_infos.clearing, aggregator, graphs.clearing, graphs.short_term)

        t = time()
        self.set_jules_id_info(graph_infos.clearing, is_aggregated=False, names=names)
        self.set_jules_id_info(graph_infos.short_term, is_aggregated=True, names=names)
        self.send_debug_event(f"set_jules_id_info time: {round(time() - t, 2)} seconds")

        t = time()
        self.set_unit_info(graph_infos.clearing, graphs.clearing, config, names)
        self.set_unit_info(graph_infos.short_term, graphs.short_term, config, names)
        self.send_debug_event(f"set_unit_info time: {round(time() - t, 2)} seconds")

        t = time()
        self.set_sss_global_eneq_info(graph_infos.clearing, graphs.clearing, db, config)
        self.set_sss_global_eneq_info(graph_infos.short_term, graphs.short_term, db, config)
        self.send_debug_event(f"set_sss_global_eneq_info time: {round(time() - t, 2)} seconds")

        t = time()
        self.set_sss_initial_storage(graph_infos.clearing, graphs.clearing, db, config)
        self.set_agg_initial_storage(graph_infos.short_term, graph_infos.clearing)
        self.send_debug_event(f"set_sss_initial_storage time: {round(time() - t, 2)} seconds")

        t = time()
        # assert that graph_infos has expected content
        assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.clearing.values())
        assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.short_term.values())
        assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.medium_term.values())
        assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.long_term.values())

        self.send_debug_event(f"validation time: {round(time() - t, 2)} seconds")

    def set_basic_node_flow_info(
        self,
        out_graph_info: dict[str, ComponentInfo],
        graph: dict[str, Flow | Node],
        db: QueryDB,
        config: JulESConfig,
    ) -> None:
        """Info directly accessible from Node and Flow API.

        We also set domain_commodity for Flow as main_node.get_commodity().
        """
        for component_id, c in graph.items():
            info = out_graph_info[component_id]

            info.is_node = isinstance(c, Node)
            if info.is_node:
                info.main_node_id = component_id
                info.domain_commodity = c.get_commodity()

            info.is_flow = isinstance(c, Flow)
            if info.is_flow:
                info.main_node_id = c.get_main_node()
                info.domain_commodity = graph[info.main_node_id].get_commodity()
                info.num_arrows = len(c.get_arrows())

            info.is_storage_node = isinstance(c, Node) and c.get_storage() is not None
            if info.is_storage_node:
                info.is_short_term_storage = False  # TODO: Should be based on storage duration # noqa FIX002

            info.is_exogenous = c.is_exogenous()

    def set_agg_storage_node_info(
        self,
        out: dict[str, ComponentInfo],
        aggregator: JulESAggregator,
        detailed_graph: dict[str, Flow | Node],
        aggregated_graph: dict[str, Flow | Node],
    ) -> None:
        """Aggregate storages and update info.agg_storage_node_id."""
        agg_storage_node_ids = {
            n: c for n, c in aggregated_graph.items() if isinstance(c, Node) and c.get_storage() is not None
        }
        graph_map = aggregator.get_short_term_graph_map(detailed_graph, agg_storage_node_ids)
        for member_id, agg_node_ids in graph_map.items():
            assert len(agg_node_ids) > 0
            assert sum(int(n in agg_storage_node_ids) for n in agg_node_ids) == 1
            info = out[member_id]
            if info.is_storage_node:
                for agg_node_id in agg_node_ids:
                    if agg_node_id in agg_storage_node_ids:
                        info.agg_storage_node_id = agg_node_id
                        break

    def set_agg_market_node_info(
        self,
        out: dict[str, ComponentInfo],
        aggregator: JulESAggregator,
        detailed_graph: dict[str, Flow | Node],
        aggregated_graph: dict[str, Flow | Node],
    ) -> None:
        """Aggregate market nodes and update info.agg_market_node_id."""
        market_nodes = {n: c for n, c in aggregated_graph.items() if out[n].is_market_node}
        graph_map = aggregator.get_medium_term_graph_map(detailed_graph, market_nodes)
        for agg_market_node_id, member_node_ids in graph_map.items():
            agg_component = aggregated_graph[agg_market_node_id]
            if not isinstance(agg_component, Node):
                continue
            if agg_component.get_storage() is not None:
                continue
            for node_id in member_node_ids:
                info = out[node_id]
                if info.is_market_node:
                    info.agg_market_node_id = agg_market_node_id

    def set_sss_info(
        self,
        detailed_graph: dict[str, Flow | Node],
        detailed_graph_info: dict[str, ComponentInfo],
        agg_graph_info: dict[str, ComponentInfo] | None,
        names: JulESNames,
    ) -> None:
        """Storage SubSystem (sss) info."""
        include_boundaries = False  # so market nodes at the boundary is not incorrectly classified
        subsystems = get_one_commodity_storage_subsystems(detailed_graph, include_boundaries)

        for info in detailed_graph_info.values():
            info.has_storage_resolution = False

        for subsystem_id, (__, subsystem, boundary_domain_commodities) in subsystems.items():
            is_short_term = self._is_short_term_storage_subsystem(subsystem, detailed_graph_info, agg_graph_info)

            jules_commodity = names.SHORT_TERM_STORAGE if is_short_term else names.STORAGE_SYSTEM

            if len(boundary_domain_commodities) == 0:
                message = (
                    f"Warning! No boundary domain commodity found for storage subsystem {subsystem_id} "
                    f"with members {subsystem}.\n"
                )
                self.send_warning_event(message)
                for component_id in subsystem:
                    info = detailed_graph_info[component_id]
                    info.jules_commodity = jules_commodity
                continue
            assert len(boundary_domain_commodities) == 1
            market_commodity = next(iter(boundary_domain_commodities))

            for component_id in subsystem:
                info = detailed_graph_info[component_id]

                info.is_sss_member = True

                info.sss_id = subsystem_id
                info.sss_is_short_term = is_short_term
                info.sss_market_commodity = market_commodity
                info.sss_members = subsystem

                info.jules_commodity = jules_commodity

            # all nodes in subsystem get True below since include_boundaries = False
            # Flow get False if any arrow points to market commodity
            assert include_boundaries is False
            for component_id in subsystem:
                component = detailed_graph[component_id]
                info = detailed_graph_info[component_id]
                info.has_storage_resolution = bool(not info.sss_is_short_term)
                if isinstance(component, Flow):
                    for arrow in component.get_arrows():
                        node_id = arrow.get_node()
                        node_info = detailed_graph_info[node_id]
                        if node_info.jules_commodity == info.sss_market_commodity:
                            info.has_storage_resolution = False
                            break

    def _is_short_term_storage_subsystem(
        self,
        subsystem: set[str],
        detailed_graph_info: dict[str, ComponentInfo],
        agg_graph_info: dict[str, ComponentInfo] | None,
    ) -> bool:
        """Return True if all storage nodes in subsystem are short term."""
        for component_id in subsystem:
            info = detailed_graph_info[component_id]
            if info.is_storage_node and not info.is_short_term_storage:
                return False
        for component_id in subsystem:
            info = detailed_graph_info[component_id]
            if info.is_storage_node and info.agg_storage_node_id and agg_graph_info:
                agg_info = agg_graph_info[info.agg_storage_node_id]
                if not agg_info.is_short_term_storage:
                    message = (
                        f"Detailed storages in subsystem are short term but aggregated is not for {component_id}.",
                        " Set storage system to long term. This is a limitation in JulES.jl where detailed and",
                        " aggregated models must have the same storage system type to be able to interact.",
                    )
                    self.send_debug_event(message)
                    return False
        return True

    def set_market_info(
        self,
        out_graph_info: dict[str, ComponentInfo],
        graph: dict[str, Flow | Node],
        names: JulESNames,
    ) -> None:
        """Set is_market_node and if so, also set jules_commodity to market."""
        for component_id, info in out_graph_info.items():
            info.is_market_node = info.is_node and not info.is_storage_node and not info.is_sss_member

            if info.is_market_node:
                info.jules_commodity = names.MARKET

        for component_id, info in out_graph_info.items():
            info.is_market_flow = False
            info.is_market_flow_to_exogenous = False
            if info.is_flow:
                flow = graph[component_id]
                for arrow in flow.get_arrows():
                    node_info = out_graph_info[arrow.get_node()]
                    if node_info.is_market_node:
                        info.is_market_flow = True
                        if node_info.is_exogenous:
                            info.is_market_flow_to_exogenous = True
                        break

    def set_jules_id_info(
        self,
        out: dict[str, ComponentInfo],
        is_aggregated: bool,
        names: JulESNames,
    ) -> None:
        """Add jules ids in compliance with required format. Warning! Julia-JulES currently requires this format."""
        for node_id, info in out.items():
            info.jules_global_eneq_id = f"{names.GLOBALENEQ}_{node_id}"

            if info.is_storage_node:
                if is_aggregated:
                    info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}_hydro_reservoir"
                    info.jules_storage_id = f"Reservoir_{node_id}_hydro_reservoir"
                else:
                    info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}"
                    info.jules_storage_id = f"Reservoir_{node_id}"

            elif info.is_node:
                info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}"

    def set_unit_info(  # noqa: C901
        self,
        out: dict[str, ComponentInfo],
        graph: dict[str, Flow | Node],
        config: JulESConfig,
        names: JulESNames,
    ) -> None:
        """Calculate all types of target units.

        Need from config:
        - unit_money
        - unit_stock per commodity for each storage_node
        - unit_flow per commodity for each flow

        Will derive:
        - unit_price per commodity for each market_node
        - unit_cost for each flow
        - unit_coeffs for each flow
        - unit_eneq for each sss_member in each sss

        And also for each flow, we derive:
        - unit_param_type
        - unit_param_flow_unit
        - unit_param_flow_unit

        """
        unit_money = config.get_currency()

        node_info = {k: info for k, info in out.items() if info.is_node}
        flow_info = {k: info for k, info in out.items() if info.is_flow}
        market_node_info = {k: info for k, info in node_info.items() if info.is_market_node}
        storage_node_info = {k: info for k, info in node_info.items() if info.is_storage_node}
        sss_member_info = {k: info for k, info in node_info.items() if info.is_sss_member}

        for info in market_node_info.values():
            unit_stock = config.get_unit_stock(info.domain_commodity)
            info.unit_price = f"{unit_money}/{unit_stock}"

        for info in storage_node_info.values():
            unit_flow = config.get_unit_flow(out[info.main_node_id].domain_commodity)
            unit_stock = config.get_unit_stock(out[info.main_node_id].domain_commodity)
            info.unit_flow = unit_flow
            info.unit_stock = unit_stock

        for d in [flow_info, storage_node_info]:
            for info in d.values():
                unit_flow = config.get_unit_flow(out[info.main_node_id].domain_commodity)
                unit_stock = config.get_unit_stock(out[info.main_node_id].domain_commodity)
                info.unit_flow = unit_flow
                info.unit_stock = unit_stock
                info.unit_cost = f"{unit_money}/{unit_stock}"
                if is_convertable(info.unit_flow, "MW"):
                    info.unit_param_type = names.MWTOGWHPARAM
                    info.unit_param_unit_flow = "MW"
                    info.unit_param_unit_stock = "GWh"
                elif is_convertable(info.unit_flow, "m3/s"):
                    info.unit_param_type = names.M3STOMM3PARAM
                    info.unit_param_unit_flow = "m3/s"
                    info.unit_param_unit_stock = "Mm3"
                else:
                    message = f"Unsupported unit_flow: {info.unit_flow}"
                    raise ValueError(message)

                if info.is_market_flow:
                    seconds = config.get_time_resolution().get_clearing_market_minutes() * 60
                else:
                    seconds = config.get_time_resolution().get_clearing_storage_minutes() * 60
                info.unit_flow_result = f"{unit_stock}/({seconds} * s)"

        for flow_id, info in flow_info.items():
            flow: Flow = graph[flow_id]
            info.unit_coeffs = dict()
            for arrow in flow.get_arrows():
                from_node_id = arrow.get_node()
                unit_coeff = None
                if from_node_id != info.main_node_id:
                    from_node_unit = config.get_unit_stock(out[from_node_id].domain_commodity)
                    unit_coeff = None if from_node_unit == info.unit_stock else f"{from_node_unit}/{info.unit_stock}"
                info.unit_coeffs[from_node_id] = unit_coeff

        for info in sss_member_info.values():
            if info.sss_global_eneq_unit is not None:
                continue
            unit_market = config.get_unit_stock(info.sss_market_commodity)
            unit_stock = config.get_unit_stock(info.domain_commodity)
            unit_eneq = f"{unit_market}/{unit_stock}"
            for component_id in info.sss_members:
                member_info = out[component_id]
                member_info.sss_global_eneq_unit = unit_eneq

    def set_sss_global_eneq_info(
        self,
        out: dict[str, ComponentInfo],
        graph: dict[str, Flow | Node],
        db: QueryDB,
        config: JulESConfig,
    ) -> dict[str, float]:
        """Set global_energy_coefficient using metadata. Convert to usable unit."""
        for component_id, info in out.items():
            if not info.is_sss_member or not info.is_storage_node:
                continue

            if is_convertable(info.sss_global_eneq_unit, "1"):
                info.sss_global_eneq_value = 1.0
                continue

            data_dim: FixedFrequencyTimeIndex = config.get_data_period()

            start_year, num_years = config.get_weather_years()
            scen_dim = AverageYearRange(start_year, num_years)

            metakeys = graph[component_id].get_meta_keys()
            if "EnergyEqDownstream" in metakeys:
                metadata = graph[component_id].get_meta("EnergyEqDownstream")
            else:
                message = (
                    f"Missing metadata EnergyEqDownstream for {component_id}, only metadata keys {list(metakeys)}."
                )
                message = message + f" Object info: {info}"
                raise ValueError(message)
            expr = metadata.get_value()

            info.sss_global_eneq_value = get_level_value(
                expr=expr,
                unit=info.sss_global_eneq_unit,
                db=db,
                data_dim=data_dim,
                scen_dim=scen_dim,
                is_max=False,
            )

    def set_sss_initial_storage(
        self,
        out: dict[str, ComponentInfo],
        graph: dict[str, Flow | Node],
        db: QueryDB,
        config: JulESConfig,
    ) -> dict[str, float]:
        """Set sss_initial_storage. Convert to usable unit."""
        for node_id, info in out.items():
            if not info.is_storage_node:
                continue

            node: Node = graph[node_id]

            try:
                percentage = node.get_storage().get_initial_storage_percentage()
                assert 0 <= percentage <= 1
            except Exception:
                percentage = 0.6
                self.send_warning_event(
                    f"Missing initial storage for {node_id}. Using 60 % of capacity.",
                )

            info.sss_initial_storage = self._get_initial_storage_capacity(
                node_id,
                node,
                info,
                percentage,
                db,
                config,
            )

    def _get_initial_storage_capacity(
        self,
        node_id: str,
        node: Node,
        info: ComponentInfo,
        percentage: float,
        db: QueryDB,
        config: JulESConfig,
    ) -> float:
        data_dim: FixedFrequencyTimeIndex = config.get_data_period()

        if data_dim.get_num_periods() > 1:
            raise NotImplementedError

        start_year, num_years = config.get_weather_years()
        scen_dim = AverageYearRange(start_year, num_years)

        capacity: StockVolume = node.get_storage().get_capacity()
        data_value: float = capacity.get_data_value(
            db=db,
            level_period=data_dim,
            scenario_horizon=scen_dim,
            unit=info.unit_stock,
            is_max_level=True,
        )

        return data_value * percentage

    def set_agg_initial_storage(
        self,
        agg_graph_info: dict[str, ComponentInfo],
        det_graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Set global_eneq and initial_storage in aggregated graph_info from detailed graph_info."""
        for det_id, det in det_graph_info.items():
            if not det.is_storage_node:
                continue
            if det.agg_storage_node_id is None:
                continue

            agg = agg_graph_info[det.agg_storage_node_id]

            if agg.sss_initial_storage is None:
                agg.sss_initial_storage = 0.0

            add_value = det.sss_initial_storage
            if det.sss_global_eneq_value:
                add_value *= det.sss_global_eneq_value

            agg.sss_initial_storage += add_value

__init__(folder: Path, clearing_model: Model, config: JulESConfig) -> None

Hold all data and methods needed to solve JulES.

Source code in framjules/solve_handler/SolveHandler.py

def __init__(self, folder: Path, clearing_model: Model, config: JulESConfig) -> None:
    """Hold all data and methods needed to solve JulES."""
    names = JulESNames()

    t = time()
    short_aggregations = config.get_short_term_aggregations()  # list of Aggregators
    self.send_debug_event(f"get_short_term_aggregations time: {round(time() - t, 2)} seconds")
    mid_aggregations = []  # config.get_mid_term_aggregations()  # not supported yet
    long_aggregations = []  # config.get_long_term_aggregations()  # not supported yet

    t = time()
    aggregator = JulESAggregator(
        clearing=clearing_model,
        short=short_aggregations,
        medium=mid_aggregations,
        long=long_aggregations,
    )
    self.send_debug_event(f"JulESAggregator init time: {round(time() - t, 2)} seconds")

    t = time()
    short_term_model = aggregator.get_short_term_model()
    self.send_debug_event(f"get_short_term_model init time: {round(time() - t, 2)} seconds")
    t = time()
    medium_term_model = aggregator.get_medium_term_model()
    self.send_debug_event(f"get_medium_term_model init time: {round(time() - t, 2)} seconds")
    t = time()
    long_term_model = aggregator.get_long_term_model()
    self.send_debug_event(f"get_long_term_model init time: {round(time() - t, 2)} seconds")

    t = time()
    domain_models = DomainModels(
        clearing=clearing_model,
        short_term=short_term_model,
        medium_term=medium_term_model,
        long_term=long_term_model,
    )
    self.send_debug_event(f"DomainModels init time: {round(time() - t, 2)} seconds")

    t = time()
    supported_types = (Flow, Node)
    forbidden_types = tuple()
    graphs = NodeFlowGraphs(
        clearing=get_supported_components(
            self._get_components(clearing_model),
            supported_types,
            forbidden_types,
        ),
        short_term=get_supported_components(
            self._get_components(short_term_model),
            supported_types,
            forbidden_types,
        ),
        medium_term=get_supported_components(
            self._get_components(medium_term_model),
            supported_types,
            forbidden_types,
        ),
        long_term=get_supported_components(
            self._get_components(long_term_model),
            supported_types,
            forbidden_types,
        ),
    )
    self.send_debug_event(f"NodeFlowGraphs init time: {round(time() - t, 2)} seconds")

    t = time()
    graph_infos = GraphInfos(
        clearing={k: ComponentInfo() for k in graphs.clearing},
        short_term={k: ComponentInfo() for k in graphs.short_term},
        medium_term={k: ComponentInfo() for k in graphs.medium_term},
        long_term={k: ComponentInfo() for k in graphs.long_term},
    )
    self.send_debug_event(f"GraphInfos init time: {round(time() - t, 2)} seconds")

    # we check that that aggregated models don't have different storages
    t = time()
    aggregator.assert_equal_storages(
        graphs.short_term,
        graphs.medium_term,
        graphs.long_term,
    )
    self.send_debug_event(f"assert_equal_storages time: {round(time() - t, 2)} seconds")

    t = time()
    constructor = CacheDB if config.is_cache_db() else ModelDB
    db = constructor(
        domain_models.clearing,
        domain_models.short_term,
        domain_models.medium_term,
        domain_models.long_term,
    )
    self.send_debug_event(f"DB init time: {round(time() - t, 2)} seconds")

    t = time()
    self.fill_graph_infos(graph_infos, graphs, names, aggregator, config, db)
    self.send_debug_event(f"fill_graph_infos time: {round(time() - t, 2)} seconds")

    # Finally, we set the member data
    self.folder: Path = folder
    self.config: JulESConfig = config
    self.names: JulESNames = names
    self.domain_models: DomainModels = domain_models
    self.graphs: NodeFlowGraphs = graphs
    self.graph_infos: GraphInfos = graph_infos
    # NB! will be freed after self.configure()
    # so we don't hold up memory during run
    self.db: QueryDB = db

build() -> None

Build input files for JulES.

Source code in framjules/solve_handler/SolveHandler.py

def build(self) -> None:
    """Build input files for JulES."""
    handler = self.create_build_handler()
    handler.build()

configure() -> None

Build configuration file for JulES.

Source code in framjules/solve_handler/SolveHandler.py

def configure(self) -> None:
    """Build configuration file for JulES."""
    handler = self.create_config_handler()
    handler.configure()

    self.db = None
    gc.collect()

create_build_handler() -> BuildHandler

Create specialized BuildHandler for the chosen simulation mode.

Source code in framjules/solve_handler/SolveHandler.py

def create_build_handler(self) -> BuildHandler:
    """Create specialized BuildHandler for the chosen simulation mode."""
    if self.config.is_simulation_mode_serial():
        handler_constructor = SerialBuildHandler
    else:
        raise NotImplementedError

    return handler_constructor(
        folder=self.folder,
        config=self.config,
        names=self.names,
        domain_models=self.domain_models,
        graphs=self.graphs,
        graph_infos=self.graph_infos,
        db=self.db,
    )

create_config_handler() -> ConfigHandler

Create specialized ConfigHandler for the chosen simulation mode.

Source code in framjules/solve_handler/SolveHandler.py

def create_config_handler(self) -> ConfigHandler:
    """Create specialized ConfigHandler for the chosen simulation mode."""
    if self.config.is_simulation_mode_serial():
        handler_constructor = SerialConfigHandler
    else:
        raise NotImplementedError

    return handler_constructor(
        folder=self.folder,
        config=self.config,
        names=self.names,
        graph_infos=self.graph_infos,
    )

create_results_handler() -> SerialResultsHandler

Create a SerialResultsHandler.

Source code in framjules/solve_handler/SolveHandler.py

def create_results_handler(self) -> SerialResultsHandler:
    """Create a SerialResultsHandler."""
    if self.config.is_simulation_mode_serial():
        handler_constructor = SerialResultsHandler
    else:
        message = "JulES Parallel simulation mode is not yet supported."
        raise NotImplementedError(message)
    return handler_constructor(
        folder=self.folder,
        config=self.config,
        names=self.names,
        graphs=self.graphs,
        graph_infos=self.graph_infos,
    )

create_run_handler() -> SerialRunHandler

Create specialized RunHandler for the chosen simulation mode.

Source code in framjules/solve_handler/SolveHandler.py

def create_run_handler(self) -> SerialRunHandler:
    """Create specialized RunHandler for the chosen simulation mode."""
    dependencies = []

    tulipa_version = self.config.get_tulipa_version()
    if tulipa_version is not None:
        if Path.exists(Path(tulipa_version)):
            dependencies.append(tulipa_version)
        else:
            dependencies.append(("https://github.com/NVE/TuLiPa.git", tulipa_version))

    jules_version = self.config.get_jules_version()
    if jules_version is not None:
        if Path.exists(Path(jules_version)):
            dependencies.append(jules_version)
        else:
            dependencies.append(("https://github.com/NVE/JulES.git", jules_version))

    dependencies.extend(["YAML", "HDF5", "JSON", "PythonCall"])

    if self.config.is_simulation_mode_serial():
        handler_constructor = SerialRunHandler
    else:
        message = "JulES Parallel simulation mode is not yet supported."
        raise NotImplementedError(message)
    handler_constructor.ENV_NAME = self.names.JULIA_ENV_NAME
    return handler_constructor(folder=self.folder, config=self.config, names=self.names, dependencies=dependencies)

fill_graph_infos(graph_infos: GraphInfos, graphs: NodeFlowGraphs, names: JulESNames, aggregator: JulESAggregator, config: JulESConfig, db: QueryDB) -> None

Fill graph_info with derived info.

Source code in framjules/solve_handler/SolveHandler.py

def fill_graph_infos(
    self,
    graph_infos: GraphInfos,
    graphs: NodeFlowGraphs,
    names: JulESNames,
    aggregator: JulESAggregator,
    config: JulESConfig,
    db: QueryDB,
) -> None:
    """Fill graph_info with derived info."""
    # Intent is to gather complex derivations in just one place

    # NB! Order of below method calls matter
    t = time()
    self.set_basic_node_flow_info(graph_infos.clearing, graphs.clearing, db, config)
    self.set_basic_node_flow_info(graph_infos.short_term, graphs.short_term, db, config)
    self.set_basic_node_flow_info(graph_infos.medium_term, graphs.medium_term, db, config)
    self.set_basic_node_flow_info(graph_infos.long_term, graphs.long_term, db, config)
    self.send_debug_event(f"set_basic_node_flow_info time: {round(time() - t, 2)} seconds")

    t = time()
    self.set_agg_storage_node_info(graph_infos.clearing, aggregator, graphs.clearing, graphs.short_term)
    self.send_debug_event(f"set_agg_storage_node_info time: {round(time() - t, 2)} seconds")

    t = time()  # opposite order to check if higher level is short term sss
    self.set_sss_info(graphs.short_term, graph_infos.short_term, graph_infos.medium_term, names)
    self.set_sss_info(graphs.clearing, graph_infos.clearing, graph_infos.short_term, names)
    self.send_debug_event(f"set_sss_info time: {round(time() - t, 2)} seconds")

    t = time()
    self.set_market_info(graph_infos.clearing, graphs.clearing, names)
    self.set_market_info(graph_infos.short_term, graphs.short_term, names)
    self.set_market_info(graph_infos.medium_term, graphs.medium_term, names)
    self.set_market_info(graph_infos.long_term, graphs.long_term, names)
    self.send_debug_event(f"set_market_info time: {round(time() - t, 2)} seconds")

    # self.set_agg_market_node_info(graph_infos.clearing, aggregator, graphs.clearing, graphs.short_term)

    t = time()
    self.set_jules_id_info(graph_infos.clearing, is_aggregated=False, names=names)
    self.set_jules_id_info(graph_infos.short_term, is_aggregated=True, names=names)
    self.send_debug_event(f"set_jules_id_info time: {round(time() - t, 2)} seconds")

    t = time()
    self.set_unit_info(graph_infos.clearing, graphs.clearing, config, names)
    self.set_unit_info(graph_infos.short_term, graphs.short_term, config, names)
    self.send_debug_event(f"set_unit_info time: {round(time() - t, 2)} seconds")

    t = time()
    self.set_sss_global_eneq_info(graph_infos.clearing, graphs.clearing, db, config)
    self.set_sss_global_eneq_info(graph_infos.short_term, graphs.short_term, db, config)
    self.send_debug_event(f"set_sss_global_eneq_info time: {round(time() - t, 2)} seconds")

    t = time()
    self.set_sss_initial_storage(graph_infos.clearing, graphs.clearing, db, config)
    self.set_agg_initial_storage(graph_infos.short_term, graph_infos.clearing)
    self.send_debug_event(f"set_sss_initial_storage time: {round(time() - t, 2)} seconds")

    t = time()
    # assert that graph_infos has expected content
    assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.clearing.values())
    assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.short_term.values())
    assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.medium_term.values())
    assert all({True, False} == {x.is_node, x.is_flow} for x in graph_infos.long_term.values())

    self.send_debug_event(f"validation time: {round(time() - t, 2)} seconds")

run() -> None

Run Julia-JulES.

Source code in framjules/solve_handler/SolveHandler.py

def run(self) -> None:
    """Run Julia-JulES."""
    handler = self.create_run_handler()
    handler.run()

set_agg_initial_storage(agg_graph_info: dict[str, ComponentInfo], det_graph_info: dict[str, ComponentInfo]) -> None

Set global_eneq and initial_storage in aggregated graph_info from detailed graph_info.

Source code in framjules/solve_handler/SolveHandler.py

def set_agg_initial_storage(
    self,
    agg_graph_info: dict[str, ComponentInfo],
    det_graph_info: dict[str, ComponentInfo],
) -> None:
    """Set global_eneq and initial_storage in aggregated graph_info from detailed graph_info."""
    for det_id, det in det_graph_info.items():
        if not det.is_storage_node:
            continue
        if det.agg_storage_node_id is None:
            continue

        agg = agg_graph_info[det.agg_storage_node_id]

        if agg.sss_initial_storage is None:
            agg.sss_initial_storage = 0.0

        add_value = det.sss_initial_storage
        if det.sss_global_eneq_value:
            add_value *= det.sss_global_eneq_value

        agg.sss_initial_storage += add_value

set_agg_market_node_info(out: dict[str, ComponentInfo], aggregator: JulESAggregator, detailed_graph: dict[str, Flow | Node], aggregated_graph: dict[str, Flow | Node]) -> None

Aggregate market nodes and update info.agg_market_node_id.

Source code in framjules/solve_handler/SolveHandler.py

def set_agg_market_node_info(
    self,
    out: dict[str, ComponentInfo],
    aggregator: JulESAggregator,
    detailed_graph: dict[str, Flow | Node],
    aggregated_graph: dict[str, Flow | Node],
) -> None:
    """Aggregate market nodes and update info.agg_market_node_id."""
    market_nodes = {n: c for n, c in aggregated_graph.items() if out[n].is_market_node}
    graph_map = aggregator.get_medium_term_graph_map(detailed_graph, market_nodes)
    for agg_market_node_id, member_node_ids in graph_map.items():
        agg_component = aggregated_graph[agg_market_node_id]
        if not isinstance(agg_component, Node):
            continue
        if agg_component.get_storage() is not None:
            continue
        for node_id in member_node_ids:
            info = out[node_id]
            if info.is_market_node:
                info.agg_market_node_id = agg_market_node_id

set_agg_storage_node_info(out: dict[str, ComponentInfo], aggregator: JulESAggregator, detailed_graph: dict[str, Flow | Node], aggregated_graph: dict[str, Flow | Node]) -> None

Aggregate storages and update info.agg_storage_node_id.

Source code in framjules/solve_handler/SolveHandler.py

def set_agg_storage_node_info(
    self,
    out: dict[str, ComponentInfo],
    aggregator: JulESAggregator,
    detailed_graph: dict[str, Flow | Node],
    aggregated_graph: dict[str, Flow | Node],
) -> None:
    """Aggregate storages and update info.agg_storage_node_id."""
    agg_storage_node_ids = {
        n: c for n, c in aggregated_graph.items() if isinstance(c, Node) and c.get_storage() is not None
    }
    graph_map = aggregator.get_short_term_graph_map(detailed_graph, agg_storage_node_ids)
    for member_id, agg_node_ids in graph_map.items():
        assert len(agg_node_ids) > 0
        assert sum(int(n in agg_storage_node_ids) for n in agg_node_ids) == 1
        info = out[member_id]
        if info.is_storage_node:
            for agg_node_id in agg_node_ids:
                if agg_node_id in agg_storage_node_ids:
                    info.agg_storage_node_id = agg_node_id
                    break

set_basic_node_flow_info(out_graph_info: dict[str, ComponentInfo], graph: dict[str, Flow | Node], db: QueryDB, config: JulESConfig) -> None

Info directly accessible from Node and Flow API.

We also set domain_commodity for Flow as main_node.get_commodity().

Source code in framjules/solve_handler/SolveHandler.py

def set_basic_node_flow_info(
    self,
    out_graph_info: dict[str, ComponentInfo],
    graph: dict[str, Flow | Node],
    db: QueryDB,
    config: JulESConfig,
) -> None:
    """Info directly accessible from Node and Flow API.

    We also set domain_commodity for Flow as main_node.get_commodity().
    """
    for component_id, c in graph.items():
        info = out_graph_info[component_id]

        info.is_node = isinstance(c, Node)
        if info.is_node:
            info.main_node_id = component_id
            info.domain_commodity = c.get_commodity()

        info.is_flow = isinstance(c, Flow)
        if info.is_flow:
            info.main_node_id = c.get_main_node()
            info.domain_commodity = graph[info.main_node_id].get_commodity()
            info.num_arrows = len(c.get_arrows())

        info.is_storage_node = isinstance(c, Node) and c.get_storage() is not None
        if info.is_storage_node:
            info.is_short_term_storage = False  # TODO: Should be based on storage duration # noqa FIX002

        info.is_exogenous = c.is_exogenous()

set_jules_id_info(out: dict[str, ComponentInfo], is_aggregated: bool, names: JulESNames) -> None

Add jules ids in compliance with required format. Warning! Julia-JulES currently requires this format.

Source code in framjules/solve_handler/SolveHandler.py

def set_jules_id_info(
    self,
    out: dict[str, ComponentInfo],
    is_aggregated: bool,
    names: JulESNames,
) -> None:
    """Add jules ids in compliance with required format. Warning! Julia-JulES currently requires this format."""
    for node_id, info in out.items():
        info.jules_global_eneq_id = f"{names.GLOBALENEQ}_{node_id}"

        if info.is_storage_node:
            if is_aggregated:
                info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}_hydro_reservoir"
                info.jules_storage_id = f"Reservoir_{node_id}_hydro_reservoir"
            else:
                info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}"
                info.jules_storage_id = f"Reservoir_{node_id}"

        elif info.is_node:
            info.jules_balance_id = f"{info.jules_commodity}Balance_{node_id}"

set_market_info(out_graph_info: dict[str, ComponentInfo], graph: dict[str, Flow | Node], names: JulESNames) -> None

Set is_market_node and if so, also set jules_commodity to market.

Source code in framjules/solve_handler/SolveHandler.py

def set_market_info(
    self,
    out_graph_info: dict[str, ComponentInfo],
    graph: dict[str, Flow | Node],
    names: JulESNames,
) -> None:
    """Set is_market_node and if so, also set jules_commodity to market."""
    for component_id, info in out_graph_info.items():
        info.is_market_node = info.is_node and not info.is_storage_node and not info.is_sss_member

        if info.is_market_node:
            info.jules_commodity = names.MARKET

    for component_id, info in out_graph_info.items():
        info.is_market_flow = False
        info.is_market_flow_to_exogenous = False
        if info.is_flow:
            flow = graph[component_id]
            for arrow in flow.get_arrows():
                node_info = out_graph_info[arrow.get_node()]
                if node_info.is_market_node:
                    info.is_market_flow = True
                    if node_info.is_exogenous:
                        info.is_market_flow_to_exogenous = True
                    break

set_results() -> None

Set results from Julia-JulES run into domain models.

Source code in framjules/solve_handler/SolveHandler.py

def set_results(self) -> None:
    """Set results from Julia-JulES run into domain models."""
    handler = self.create_results_handler()
    handler.set_results()

set_sss_global_eneq_info(out: dict[str, ComponentInfo], graph: dict[str, Flow | Node], db: QueryDB, config: JulESConfig) -> dict[str, float]

Set global_energy_coefficient using metadata. Convert to usable unit.

Source code in framjules/solve_handler/SolveHandler.py

def set_sss_global_eneq_info(
    self,
    out: dict[str, ComponentInfo],
    graph: dict[str, Flow | Node],
    db: QueryDB,
    config: JulESConfig,
) -> dict[str, float]:
    """Set global_energy_coefficient using metadata. Convert to usable unit."""
    for component_id, info in out.items():
        if not info.is_sss_member or not info.is_storage_node:
            continue

        if is_convertable(info.sss_global_eneq_unit, "1"):
            info.sss_global_eneq_value = 1.0
            continue

        data_dim: FixedFrequencyTimeIndex = config.get_data_period()

        start_year, num_years = config.get_weather_years()
        scen_dim = AverageYearRange(start_year, num_years)

        metakeys = graph[component_id].get_meta_keys()
        if "EnergyEqDownstream" in metakeys:
            metadata = graph[component_id].get_meta("EnergyEqDownstream")
        else:
            message = (
                f"Missing metadata EnergyEqDownstream for {component_id}, only metadata keys {list(metakeys)}."
            )
            message = message + f" Object info: {info}"
            raise ValueError(message)
        expr = metadata.get_value()

        info.sss_global_eneq_value = get_level_value(
            expr=expr,
            unit=info.sss_global_eneq_unit,
            db=db,
            data_dim=data_dim,
            scen_dim=scen_dim,
            is_max=False,
        )

set_sss_info(detailed_graph: dict[str, Flow | Node], detailed_graph_info: dict[str, ComponentInfo], agg_graph_info: dict[str, ComponentInfo] | None, names: JulESNames) -> None

Storage SubSystem (sss) info.

Source code in framjules/solve_handler/SolveHandler.py

def set_sss_info(
    self,
    detailed_graph: dict[str, Flow | Node],
    detailed_graph_info: dict[str, ComponentInfo],
    agg_graph_info: dict[str, ComponentInfo] | None,
    names: JulESNames,
) -> None:
    """Storage SubSystem (sss) info."""
    include_boundaries = False  # so market nodes at the boundary is not incorrectly classified
    subsystems = get_one_commodity_storage_subsystems(detailed_graph, include_boundaries)

    for info in detailed_graph_info.values():
        info.has_storage_resolution = False

    for subsystem_id, (__, subsystem, boundary_domain_commodities) in subsystems.items():
        is_short_term = self._is_short_term_storage_subsystem(subsystem, detailed_graph_info, agg_graph_info)

        jules_commodity = names.SHORT_TERM_STORAGE if is_short_term else names.STORAGE_SYSTEM

        if len(boundary_domain_commodities) == 0:
            message = (
                f"Warning! No boundary domain commodity found for storage subsystem {subsystem_id} "
                f"with members {subsystem}.\n"
            )
            self.send_warning_event(message)
            for component_id in subsystem:
                info = detailed_graph_info[component_id]
                info.jules_commodity = jules_commodity
            continue
        assert len(boundary_domain_commodities) == 1
        market_commodity = next(iter(boundary_domain_commodities))

        for component_id in subsystem:
            info = detailed_graph_info[component_id]

            info.is_sss_member = True

            info.sss_id = subsystem_id
            info.sss_is_short_term = is_short_term
            info.sss_market_commodity = market_commodity
            info.sss_members = subsystem

            info.jules_commodity = jules_commodity

        # all nodes in subsystem get True below since include_boundaries = False
        # Flow get False if any arrow points to market commodity
        assert include_boundaries is False
        for component_id in subsystem:
            component = detailed_graph[component_id]
            info = detailed_graph_info[component_id]
            info.has_storage_resolution = bool(not info.sss_is_short_term)
            if isinstance(component, Flow):
                for arrow in component.get_arrows():
                    node_id = arrow.get_node()
                    node_info = detailed_graph_info[node_id]
                    if node_info.jules_commodity == info.sss_market_commodity:
                        info.has_storage_resolution = False
                        break

set_sss_initial_storage(out: dict[str, ComponentInfo], graph: dict[str, Flow | Node], db: QueryDB, config: JulESConfig) -> dict[str, float]

Set sss_initial_storage. Convert to usable unit.

Source code in framjules/solve_handler/SolveHandler.py

def set_sss_initial_storage(
    self,
    out: dict[str, ComponentInfo],
    graph: dict[str, Flow | Node],
    db: QueryDB,
    config: JulESConfig,
) -> dict[str, float]:
    """Set sss_initial_storage. Convert to usable unit."""
    for node_id, info in out.items():
        if not info.is_storage_node:
            continue

        node: Node = graph[node_id]

        try:
            percentage = node.get_storage().get_initial_storage_percentage()
            assert 0 <= percentage <= 1
        except Exception:
            percentage = 0.6
            self.send_warning_event(
                f"Missing initial storage for {node_id}. Using 60 % of capacity.",
            )

        info.sss_initial_storage = self._get_initial_storage_capacity(
            node_id,
            node,
            info,
            percentage,
            db,
            config,
        )

set_unit_info(out: dict[str, ComponentInfo], graph: dict[str, Flow | Node], config: JulESConfig, names: JulESNames) -> None

Calculate all types of target units.

Need from config: - unit_money - unit_stock per commodity for each storage_node - unit_flow per commodity for each flow

Will derive: - unit_price per commodity for each market_node - unit_cost for each flow - unit_coeffs for each flow - unit_eneq for each sss_member in each sss

And also for each flow, we derive: - unit_param_type - unit_param_flow_unit - unit_param_flow_unit

Source code in framjules/solve_handler/SolveHandler.py

def set_unit_info(  # noqa: C901
    self,
    out: dict[str, ComponentInfo],
    graph: dict[str, Flow | Node],
    config: JulESConfig,
    names: JulESNames,
) -> None:
    """Calculate all types of target units.

    Need from config:
    - unit_money
    - unit_stock per commodity for each storage_node
    - unit_flow per commodity for each flow

    Will derive:
    - unit_price per commodity for each market_node
    - unit_cost for each flow
    - unit_coeffs for each flow
    - unit_eneq for each sss_member in each sss

    And also for each flow, we derive:
    - unit_param_type
    - unit_param_flow_unit
    - unit_param_flow_unit

    """
    unit_money = config.get_currency()

    node_info = {k: info for k, info in out.items() if info.is_node}
    flow_info = {k: info for k, info in out.items() if info.is_flow}
    market_node_info = {k: info for k, info in node_info.items() if info.is_market_node}
    storage_node_info = {k: info for k, info in node_info.items() if info.is_storage_node}
    sss_member_info = {k: info for k, info in node_info.items() if info.is_sss_member}

    for info in market_node_info.values():
        unit_stock = config.get_unit_stock(info.domain_commodity)
        info.unit_price = f"{unit_money}/{unit_stock}"

    for info in storage_node_info.values():
        unit_flow = config.get_unit_flow(out[info.main_node_id].domain_commodity)
        unit_stock = config.get_unit_stock(out[info.main_node_id].domain_commodity)
        info.unit_flow = unit_flow
        info.unit_stock = unit_stock

    for d in [flow_info, storage_node_info]:
        for info in d.values():
            unit_flow = config.get_unit_flow(out[info.main_node_id].domain_commodity)
            unit_stock = config.get_unit_stock(out[info.main_node_id].domain_commodity)
            info.unit_flow = unit_flow
            info.unit_stock = unit_stock
            info.unit_cost = f"{unit_money}/{unit_stock}"
            if is_convertable(info.unit_flow, "MW"):
                info.unit_param_type = names.MWTOGWHPARAM
                info.unit_param_unit_flow = "MW"
                info.unit_param_unit_stock = "GWh"
            elif is_convertable(info.unit_flow, "m3/s"):
                info.unit_param_type = names.M3STOMM3PARAM
                info.unit_param_unit_flow = "m3/s"
                info.unit_param_unit_stock = "Mm3"
            else:
                message = f"Unsupported unit_flow: {info.unit_flow}"
                raise ValueError(message)

            if info.is_market_flow:
                seconds = config.get_time_resolution().get_clearing_market_minutes() * 60
            else:
                seconds = config.get_time_resolution().get_clearing_storage_minutes() * 60
            info.unit_flow_result = f"{unit_stock}/({seconds} * s)"

    for flow_id, info in flow_info.items():
        flow: Flow = graph[flow_id]
        info.unit_coeffs = dict()
        for arrow in flow.get_arrows():
            from_node_id = arrow.get_node()
            unit_coeff = None
            if from_node_id != info.main_node_id:
                from_node_unit = config.get_unit_stock(out[from_node_id].domain_commodity)
                unit_coeff = None if from_node_unit == info.unit_stock else f"{from_node_unit}/{info.unit_stock}"
            info.unit_coeffs[from_node_id] = unit_coeff

    for info in sss_member_info.values():
        if info.sss_global_eneq_unit is not None:
            continue
        unit_market = config.get_unit_stock(info.sss_market_commodity)
        unit_stock = config.get_unit_stock(info.domain_commodity)
        unit_eneq = f"{unit_market}/{unit_stock}"
        for component_id in info.sss_members:
            member_info = out[component_id]
            member_info.sss_global_eneq_unit = unit_eneq

build_handler

BuildHandler

BuildHandler

Bases: Base, ABC

Responsible for implementing shared functionality in build method.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

class BuildHandler(Base, ABC):
    """Responsible for implementing shared functionality in build method."""

    def __init__(
        self,
        folder: Path,
        config: JulESConfig,
        names: JulESNames,
        domain_models: DomainModels,
        graphs: NodeFlowGraphs,
        graph_infos: GraphInfos,
        db: QueryDB,
    ) -> None:
        """Initialize handler.

        Use inputs passed down from SolveHandler

        and create extra fields only relevant for
        the build phase.
        """
        self.folder = folder
        self.config = config
        self.names = names
        self.domain_models = domain_models
        self.graphs = graphs
        self.graph_infos = graph_infos
        self.db = db

        self.append = DataElementAppender(names)

        self.errors: set[str] = set()
        self.timevectors: dict[FixedFrequencyTimeIndex, dict[str, NDArray]] = defaultdict(dict)

    def build(self) -> None:
        """Build input and configuration files for JulES."""
        t = time()
        self.build_data_elements(self.names.CLEARING, self.graphs.clearing, self.graph_infos.clearing)
        self.send_debug_event(f"build_data_elements clearing time: {round(time() - t, 2)} seconds")

        t = time()
        self.build_data_elements(self.names.AGGREGATED, self.graphs.short_term, self.graph_infos.short_term)
        self.send_debug_event(f"build_data_elements aggregated time: {round(time() - t, 2)} seconds")

        # Intent is to build data elements for aggregated models. Pending changes in Julia-JulES
        # self.build_data_elements(self.names.SHORT_TERM, self.graphs.short_term, self.graph_infos.short_term)
        # self.build_data_elements(self.names.MEDIUM_TERM, self.graphs.medium_term, self.graph_infos.medium_term)
        # self.build_data_elements(self.names.LONG_TERM, self.graphs.long_term, self.graph_infos.long_term)

        t = time()
        self.build_time_vectors()
        self.send_debug_event(f"build_time_vectors time: {round(time() - t, 2)} seconds")

        t = time()
        self.build_storage_mapping(self.graph_infos.clearing)
        self.send_debug_event(f"build_storage_mapping time: {round(time() - t, 2)} seconds")

        t = time()
        self.build_start_storage(self.names.FILENAME_START_STORAGES_CLEARING, self.graph_infos.clearing)
        self.send_debug_event(f"build_start_storage clearing time: {round(time() - t, 2)} seconds")

        t = time()
        self.build_start_storage(self.names.FILENAME_START_STORAGES_AGGREGATED, self.graph_infos.short_term)
        self.send_debug_event(f"build_start_storage aggregated time: {round(time() - t, 2)} seconds")

    def build_start_storage(
        self,
        filename: str,
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Write start storag json file to folder."""
        data = {
            info.jules_storage_id: info.sss_initial_storage for k, info in graph_info.items() if info.is_storage_node
        }
        self.write_json_file(data, filename)

    def build_data_elements(
        self,
        model_id: str,
        graph: dict[str, Flow | Node],
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Write json file with data elements for a graph belonging to a given model_id."""
        self.fill_data_elements(model_id, graph, graph_info)
        self.stop_if_errors()
        filename = f"{self.names.ROOT_FILENAME_DATAELEMENTS}_{model_id}.json"
        self.write_json_file(self.append.data_elements, filename)

    def stop_if_errors(self) -> None:
        """Throw RunTimeError if any errors."""
        if self.errors:
            error_string = "\n".join(self.errors)
            message = f"Errors found:\n{error_string}"
            raise RuntimeError(message)

    def build_time_vectors(self) -> None:
        """Write json file with time vector data elements and csv file for each unique time index."""
        self.append.data_elements = []
        for i, timeindex in enumerate(self.timevectors, start=1):
            time_index_id = f"timeindex_{i}"

            milliseconds: float = timeindex.get_period_duration().total_seconds() * 1000.0
            time_delta_id = f"{time_index_id}_timedelta"
            self.append.ms_time_delta(time_delta_id, milliseconds)

            self.append.range_time_index(
                time_index_id,
                timeindex.get_start_time(),
                timeindex.get_num_periods(),
                time_delta_id,
            )

            table_id = f"{time_index_id}_table"
            path_table, column_names = self.write_table(time_index_id, timeindex)
            self.append.base_table(table_id, path_table, column_names)

            is_one_year = timeindex.is_one_year()

            for column_name in column_names:
                time_vector_id = column_name
                time_values_id = f"{time_vector_id}_values"

                self.append.column_time_values(time_values_id, table_id, column_name)

                if is_one_year:
                    self.append.one_year_time_vector(time_vector_id, time_index_id, time_values_id)
                else:
                    self.append.rotating_time_vector(time_vector_id, time_index_id, time_values_id)

        self.write_json_file(self.append.data_elements, self.names.FILENAME_DATAELEMENTS_TIMEVECTORS)

    def get_time_index_id(self, timeindex: FixedFrequencyTimeIndex) -> str:
        """Return id that works in file name."""
        type_name = type(timeindex).__name__
        num_periods = timeindex.get_num_periods()
        resolution = int(timeindex.get_period_duration().total_seconds() * 1000.0)
        is_52 = timeindex.is_52_week_years()
        extr_first = timeindex.extrapolate_first_point()
        extr_last = timeindex.extrapolate_last_point()
        return f"timeindex_{type_name}_periods_{num_periods}_ms_{resolution}_{is_52}_{extr_first}_{extr_last}"

    def write_table(
        self,
        time_index_id: str,
        timeindex: FixedFrequencyTimeIndex,
    ) -> tuple[Path, list[str]]:
        """Write all vectors corresponding to time_index_id to a csv file."""
        vectors = self.timevectors[timeindex]
        column_names = list(vectors.keys())
        matrix = np.column_stack([vectors[c] for c in column_names])
        matrix = np.round(matrix, decimals=6)  # quick fix for negative values in time vectors
        # if negative values give warning
        if np.any(matrix < 0):
            self.errors.add(f"Negative values found in time vector for {time_index_id}. This might cause issues.")
        filename = f"timevector_{time_index_id}.csv"
        path = self.folder / filename
        np.savetxt(path, matrix, delimiter=",")
        return path, column_names

    def build_storage_mapping(
        self,
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Write the mapping of storages from Clearing to Aggregated Model to json."""
        data = {
            k: info.agg_storage_node_id
            for k, info in graph_info.items()
            if info.is_storage_node and info.agg_storage_node_id
        }
        self.write_json_file(data, self.names.FILENAME_STORAGE_MAPPING)

    def fill_data_elements(
        self,
        model_id: str,
        graph: dict[str, Flow | Node],
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Reset and fill self.append.data_elements with data element json data."""
        self.append.data_elements = []  # Important to reset the list

        nodes: dict[str, Node] = {k: v for k, v in graph.items() if isinstance(v, Node)}
        flows: dict[str, Flow] = {k: v for k, v in graph.items() if isinstance(v, Flow)}

        exogenous_nodes = {k: v for k, v in nodes.items() if v.is_exogenous()}
        exogenous_flows = {k: v for k, v in flows.items() if v.is_exogenous()}
        endogenous_nodes = {k: v for k, v in nodes.items() if not v.is_exogenous()}
        endogenous_flows = {k: v for k, v in flows.items() if not v.is_exogenous()}

        t = time()
        self.add_exogenous_nodes(exogenous_nodes, graph_info)
        self.send_debug_event(f"add_exogenous_nodes time: {round(time() - t, 2)} seconds")
        t = time()
        self.add_exogenous_flows(exogenous_flows, graph_info)
        self.send_debug_event(f"add_exogenous_flows time: {round(time() - t, 2)} seconds")
        t = time()
        self.add_endogenous_nodes(endogenous_nodes, graph_info, model_id)
        self.send_debug_event(f"add_endogenous_nodes time: {round(time() - t, 2)} seconds")
        t = time()
        self.add_endogenous_flows(endogenous_flows, graph_info)
        self.send_debug_event(f"add_endogenous_flows time: {round(time() - t, 2)} seconds")

        t = time()
        self.add_dummy_exogenous_balance()
        self.send_debug_event(f"add_dummy_exogenous_balance time: {round(time() - t, 2)} seconds")

    def add_dummy_exogenous_balance(self) -> None:
        """Add a dummy exogenous Node for JulES."""
        balance_id = "PowerBalance_DummyNode"
        profile_id = f"{balance_id}_Profile"
        # Find longest name.
        longest_name = ""
        for name in self.graphs.clearing:
            if len(name) > len(longest_name):
                longest_name = name
        if len(longest_name) >= len(balance_id):
            unique_postfix = "_" + "x" * (len(longest_name) - len(balance_id))  # Fill to guarantee uniqueness.
            balance_id += unique_postfix
            profile_id += unique_postfix

        # Set balance and profile ids in names so other parts of system have access to them.
        self.names.dummy_exogenous_balance_name = balance_id
        self.names.dummy_exogenous_profile_id = profile_id

        price_param_id = f"{balance_id}_price_param"
        self.append.exogenous_balance(balance_id, self.names.MARKET, price_param_id)
        self.append.mean_series_param(price_param_id, 1.0, profile_id)

        # Set index and vector so they are added to the dataset in build_time_vectors.

        first_scenario_year, num_scenario_years = self.config.get_weather_years()

        period_duration = timedelta(minutes=self.config.get_time_resolution().get_clearing_market_minutes())

        dummy_timeindex = ProfileTimeIndex(
            start_year=first_scenario_year,
            num_years=num_scenario_years,
            period_duration=self._get_closest_valid_profile_duration(period_duration),
            is_52_week_years=True,
        )

        if dummy_timeindex not in self.timevectors:
            default_vector = np.arange(0, dummy_timeindex.get_num_periods(), 1, dtype=np.float64)
            np.divide(default_vector, default_vector.max(), out=default_vector)

            self.timevectors[dummy_timeindex] = {profile_id: default_vector}

    def add_exogenous_nodes(
        self,
        exogenous_nodes: dict[str, Node],
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Append exogenous balance related data elements for exogenous node."""
        for node_id, node in exogenous_nodes.items():
            info = graph_info[node_id]

            balance_id = info.jules_balance_id

            price_param_id = f"{balance_id}_price_param"

            self.append.exogenous_balance(balance_id, info.jules_commodity, price_param_id)

            price: Price = node.get_price()

            if not price.has_level():
                message = f"Node {node_id} is exogenous but has not price."
                raise RuntimeError(message)

            if price.has_profile():
                units = get_units_from_expr(self.db, price.get_profile())
                if units:
                    message = f"Node {node_id} has exogenous price profile with units {units}."
                    raise RuntimeError(message)

            level = self.get_price_level(price_param_id, price, info)
            profile = self.get_price_profile(price_param_id, price, info)

            self.append.mean_series_param(price_param_id, level, profile)

    def add_endogenous_nodes(
        self,
        endogenous_nodes: dict[str, Node],
        graph_info: dict[str, ComponentInfo],
        model_id: str,
    ) -> None:
        """Append endogenous balance related data elements for endogenous node."""
        for node_id, node in endogenous_nodes.items():
            info = graph_info[node_id]

            self.append.endogenous_balance(info.jules_balance_id, info.jules_commodity)

            storage = node.get_storage()
            if storage is not None:
                self.add_storage(model_id, storage, info)

    def add_exogenous_flows(
        self,
        exogenous_flows: dict[str, Flow],
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Append data elements related to an exogenous flow."""
        for flow_id, flow in exogenous_flows.items():
            for arrow in flow.get_arrows():
                node_info = graph_info[arrow.get_node()]
                if not node_info.is_exogenous:
                    self.add_rhs_term(flow_id, flow, arrow, node_info, graph_info[flow_id])

    def add_endogenous_flows(
        self,
        endogenous_flows: dict[str, Flow],
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Append data elements for endogenous flows and related attributes."""
        for flow_id, flow in endogenous_flows.items():
            self.append.base_flow(flow_id)

            flow_info = graph_info[flow_id]

            self.add_flow_lower_bound(flow_id, flow, flow_info)
            self.add_flow_upper_bound(flow_id, flow, flow_info)
            self.add_flow_arrows(flow_id, flow, graph_info)
            self.add_flow_costs(flow_id, flow, flow_info)

    def add_storage(
        self,
        model_id: str,
        storage: Storage,
        info: ComponentInfo,
    ) -> None:
        """Append data elements related to a storage."""
        storage_id = info.jules_storage_id
        balance_id = info.jules_balance_id

        self.append.base_storage(storage_id, balance_id)

        capacity = storage.get_capacity()

        self.add_positive_capacity(storage_id, info, capacity, f"{storage_id}_upper_bound", False)

        self.append.lower_zero_capacity(f"{storage_id}_lower_bound", info.is_flow, storage_id)

        if model_id == self.names.CLEARING:
            if info.sss_global_eneq_value is not None:
                self.append.global_eneq(info.jules_global_eneq_id, info.jules_balance_id, info.sss_global_eneq_value)

            if info.sss_is_short_term and info.is_short_term_storage:
                self.append.storage_hint(
                    storage_id,
                    round(info.sss_storage_duration.total_seconds() * 1000),
                )

    def add_rhs_term(
        self,
        flow_id: str,
        flow: Flow,
        arrow: Arrow,
        node_info: ComponentInfo,
        flow_info: ComponentInfo,
    ) -> None:
        """Append data elements related to rhs term."""
        node_id = arrow.get_node()

        rhs_term_id = f"exogenous_flow_{flow_id}_{node_id}"
        unit_param_id = f"{rhs_term_id}_unit_param"
        series_param_id = f"{rhs_term_id}_series_param"
        balance_id = node_info.jules_balance_id

        self.append.base_rhs_term(rhs_term_id, balance_id, arrow.is_ingoing(), unit_param_id)

        level = self.get_rhs_term_level(rhs_term_id, flow, arrow, flow_info)

        profile = self.get_rhs_term_profile(rhs_term_id, flow, arrow, flow_info)

        # unit is actually flipped from main to target node
        # using conversion factor inside get_rhs_term_level
        # so the jules unit param might say GWh but the values
        # have been converted so that the result will become e.g. Mm3

        self.append.unit_param(unit_param_id, series_param_id, flow_info)
        self.append.mean_series_param(series_param_id, level, profile)

    def add_flow_lower_bound(
        self,
        flow_id: str,
        flow: Flow,
        flow_info: ComponentInfo,
    ) -> None:
        """Append lower bound related data elements for a flow."""
        capacity = flow.get_min_capacity()
        bound_id = f"{flow_id}_lower_bound"
        profile = None if capacity is None else capacity.get_profile()
        if profile is None:
            self.append.lower_zero_capacity(bound_id, flow_info.is_flow, flow_or_storage_id=flow_id)
            return
        self.add_positive_capacity(flow_id, flow_info, capacity, bound_id, is_lower_bound=True)

    def add_flow_upper_bound(
        self,
        flow_id: str,
        flow: Flow,
        flow_info: ComponentInfo,
    ) -> None:
        """Append upper bound related data elements for a flow."""
        capacity = flow.get_max_capacity()
        if capacity is None:
            return
        bound_id = f"{flow_id}_upper_bound"
        self.add_positive_capacity(flow_id, flow_info, capacity, bound_id, is_lower_bound=False)

    def add_flow_arrows(
        self,
        flow_id: str,
        flow: Flow,
        graph_info: dict[str, ComponentInfo],
    ) -> None:
        """Append arrow related data elements for each arrow in flow."""
        flow_info = graph_info[flow_id]
        for arrow in flow.get_arrows():
            assert arrow.has_profile() is False, "Currently not supported, will be implemented later"

            arrow_id = f"{flow_id}_arrow_{arrow.get_node()}->{flow_info.main_node_id}"

            level = self.get_coefficient_level(arrow_id, arrow, flow_info)

            balance_id = graph_info[arrow.get_node()].jules_balance_id

            self.append.base_arrow(arrow_id, flow_id, balance_id, arrow.is_ingoing(), level)

    def add_flow_costs(
        self,
        flow_id: str,
        flow: Flow,
        flow_info: ComponentInfo,
    ) -> None:
        """Append cost data element for each cost in flow."""
        cost_terms = flow.get_cost_terms()
        for cost_term_id, cost_term in cost_terms.items():
            level = self.get_cost_term_level(cost_term_id, cost_term, flow_info)

            has_profile = cost_term.get_profile() is None

            profile = self.get_cost_term_profile(cost_term_id, cost_term, flow_info) if has_profile else 1.0

            extended_cost_term_id = f"{flow_id}_{cost_term_id}"
            param_id = f"{extended_cost_term_id}_param"
            self.append.cost_term(extended_cost_term_id, flow_id, flow_info.is_flow, cost_term.is_cost(), param_id)

            self.append.mean_series_param(param_id, level, profile)

    def add_positive_capacity(
        self,
        flow_or_storage_id: str,
        info: ComponentInfo,
        capacity: FlowVolume | StockVolume,
        bound_id: str,
        is_lower_bound: bool,
    ) -> None:
        """Append data elements related to positive capacity."""
        series_param_id = f"{bound_id}_series_param"

        if isinstance(capacity, FlowVolume):
            unit_param_id = f"{bound_id}_unit_param"
            self.append.unit_param(unit_param_id, series_param_id, info)
            self.append.positive_capacity(bound_id, info.is_flow, flow_or_storage_id, is_lower_bound, unit_param_id)
        else:
            assert isinstance(capacity, StockVolume)
            self.append.positive_capacity(bound_id, info.is_flow, flow_or_storage_id, is_lower_bound, series_param_id)

        level = self.get_capacity_level(series_param_id, capacity, info)
        profile = self.get_capacity_profile(series_param_id, capacity, info)

        self.append.mean_series_param(series_param_id, level, profile)

    def write_json_file(self, data: object, filename: str) -> None:
        """Write data to json."""
        with Path.open(self.folder / filename, "w") as f:
            json.dump(data, f, indent=self.names.JSON_INDENT)

    def _get_closest_valid_profile_duration(self, period_duration: timedelta) -> timedelta:
        input_seconds = period_duration.total_seconds()
        data = [h * 3600 for h in [168, 84, 56, 42, 28, 24, 21, 14, 12, 8, 7, 6, 4, 3, 2, 1]]
        for profile_seconds in data:
            if profile_seconds <= input_seconds:
                break
        return timedelta(seconds=profile_seconds)

    # Must be implemented for each simulation mode

    @abstractmethod
    def get_price_level(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
        """Query price level."""
        pass

    @abstractmethod
    def get_price_profile(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
        """Query price profile."""
        pass

    @abstractmethod
    def get_capacity_level(self, root_id: str, capacity: FlowVolume | StockVolume, info: ComponentInfo) -> str | float:
        """Query capacity level."""
        pass

    @abstractmethod
    def get_capacity_profile(
        self,
        root_id: str,
        capacity: FlowVolume | StockVolume,
        info: ComponentInfo,
    ) -> str | float:
        """Query capacity profile."""
        pass

    @abstractmethod
    def get_coefficient_level(self, root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float:
        """Query arrow coefficient level."""
        pass

    @abstractmethod
    def get_cost_term_level(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
        """Query cost term level."""
        pass

    @abstractmethod
    def get_cost_term_profile(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
        """Query cost term profile."""
        pass

    @abstractmethod
    def get_rhs_term_level(
        self,
        rhs_term_id: str,
        flow: Flow,
        arrow: Arrow,
        flow_info: ComponentInfo,
    ) -> str | float:
        """Query rhs term level."""
        pass

    @abstractmethod
    def get_rhs_term_profile(
        self,
        rhs_term_id: str,
        flow: Flow,
        arrow: Arrow,
        flow_info: ComponentInfo,
    ) -> str | float:
        """Query rhs term profile."""
        pass

__init__(folder: Path, config: JulESConfig, names: JulESNames, domain_models: DomainModels, graphs: NodeFlowGraphs, graph_infos: GraphInfos, db: QueryDB) -> None

Initialize handler.

Use inputs passed down from SolveHandler

and create extra fields only relevant for the build phase.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def __init__(
    self,
    folder: Path,
    config: JulESConfig,
    names: JulESNames,
    domain_models: DomainModels,
    graphs: NodeFlowGraphs,
    graph_infos: GraphInfos,
    db: QueryDB,
) -> None:
    """Initialize handler.

    Use inputs passed down from SolveHandler

    and create extra fields only relevant for
    the build phase.
    """
    self.folder = folder
    self.config = config
    self.names = names
    self.domain_models = domain_models
    self.graphs = graphs
    self.graph_infos = graph_infos
    self.db = db

    self.append = DataElementAppender(names)

    self.errors: set[str] = set()
    self.timevectors: dict[FixedFrequencyTimeIndex, dict[str, NDArray]] = defaultdict(dict)

add_dummy_exogenous_balance() -> None

Add a dummy exogenous Node for JulES.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_dummy_exogenous_balance(self) -> None:
    """Add a dummy exogenous Node for JulES."""
    balance_id = "PowerBalance_DummyNode"
    profile_id = f"{balance_id}_Profile"
    # Find longest name.
    longest_name = ""
    for name in self.graphs.clearing:
        if len(name) > len(longest_name):
            longest_name = name
    if len(longest_name) >= len(balance_id):
        unique_postfix = "_" + "x" * (len(longest_name) - len(balance_id))  # Fill to guarantee uniqueness.
        balance_id += unique_postfix
        profile_id += unique_postfix

    # Set balance and profile ids in names so other parts of system have access to them.
    self.names.dummy_exogenous_balance_name = balance_id
    self.names.dummy_exogenous_profile_id = profile_id

    price_param_id = f"{balance_id}_price_param"
    self.append.exogenous_balance(balance_id, self.names.MARKET, price_param_id)
    self.append.mean_series_param(price_param_id, 1.0, profile_id)

    # Set index and vector so they are added to the dataset in build_time_vectors.

    first_scenario_year, num_scenario_years = self.config.get_weather_years()

    period_duration = timedelta(minutes=self.config.get_time_resolution().get_clearing_market_minutes())

    dummy_timeindex = ProfileTimeIndex(
        start_year=first_scenario_year,
        num_years=num_scenario_years,
        period_duration=self._get_closest_valid_profile_duration(period_duration),
        is_52_week_years=True,
    )

    if dummy_timeindex not in self.timevectors:
        default_vector = np.arange(0, dummy_timeindex.get_num_periods(), 1, dtype=np.float64)
        np.divide(default_vector, default_vector.max(), out=default_vector)

        self.timevectors[dummy_timeindex] = {profile_id: default_vector}

add_endogenous_flows(endogenous_flows: dict[str, Flow], graph_info: dict[str, ComponentInfo]) -> None

Append data elements for endogenous flows and related attributes.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_endogenous_flows(
    self,
    endogenous_flows: dict[str, Flow],
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Append data elements for endogenous flows and related attributes."""
    for flow_id, flow in endogenous_flows.items():
        self.append.base_flow(flow_id)

        flow_info = graph_info[flow_id]

        self.add_flow_lower_bound(flow_id, flow, flow_info)
        self.add_flow_upper_bound(flow_id, flow, flow_info)
        self.add_flow_arrows(flow_id, flow, graph_info)
        self.add_flow_costs(flow_id, flow, flow_info)

add_endogenous_nodes(endogenous_nodes: dict[str, Node], graph_info: dict[str, ComponentInfo], model_id: str) -> None

Append endogenous balance related data elements for endogenous node.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_endogenous_nodes(
    self,
    endogenous_nodes: dict[str, Node],
    graph_info: dict[str, ComponentInfo],
    model_id: str,
) -> None:
    """Append endogenous balance related data elements for endogenous node."""
    for node_id, node in endogenous_nodes.items():
        info = graph_info[node_id]

        self.append.endogenous_balance(info.jules_balance_id, info.jules_commodity)

        storage = node.get_storage()
        if storage is not None:
            self.add_storage(model_id, storage, info)

add_exogenous_flows(exogenous_flows: dict[str, Flow], graph_info: dict[str, ComponentInfo]) -> None

Append data elements related to an exogenous flow.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_exogenous_flows(
    self,
    exogenous_flows: dict[str, Flow],
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Append data elements related to an exogenous flow."""
    for flow_id, flow in exogenous_flows.items():
        for arrow in flow.get_arrows():
            node_info = graph_info[arrow.get_node()]
            if not node_info.is_exogenous:
                self.add_rhs_term(flow_id, flow, arrow, node_info, graph_info[flow_id])

add_exogenous_nodes(exogenous_nodes: dict[str, Node], graph_info: dict[str, ComponentInfo]) -> None

Append exogenous balance related data elements for exogenous node.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_exogenous_nodes(
    self,
    exogenous_nodes: dict[str, Node],
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Append exogenous balance related data elements for exogenous node."""
    for node_id, node in exogenous_nodes.items():
        info = graph_info[node_id]

        balance_id = info.jules_balance_id

        price_param_id = f"{balance_id}_price_param"

        self.append.exogenous_balance(balance_id, info.jules_commodity, price_param_id)

        price: Price = node.get_price()

        if not price.has_level():
            message = f"Node {node_id} is exogenous but has not price."
            raise RuntimeError(message)

        if price.has_profile():
            units = get_units_from_expr(self.db, price.get_profile())
            if units:
                message = f"Node {node_id} has exogenous price profile with units {units}."
                raise RuntimeError(message)

        level = self.get_price_level(price_param_id, price, info)
        profile = self.get_price_profile(price_param_id, price, info)

        self.append.mean_series_param(price_param_id, level, profile)

add_flow_arrows(flow_id: str, flow: Flow, graph_info: dict[str, ComponentInfo]) -> None

Append arrow related data elements for each arrow in flow.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_flow_arrows(
    self,
    flow_id: str,
    flow: Flow,
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Append arrow related data elements for each arrow in flow."""
    flow_info = graph_info[flow_id]
    for arrow in flow.get_arrows():
        assert arrow.has_profile() is False, "Currently not supported, will be implemented later"

        arrow_id = f"{flow_id}_arrow_{arrow.get_node()}->{flow_info.main_node_id}"

        level = self.get_coefficient_level(arrow_id, arrow, flow_info)

        balance_id = graph_info[arrow.get_node()].jules_balance_id

        self.append.base_arrow(arrow_id, flow_id, balance_id, arrow.is_ingoing(), level)

add_flow_costs(flow_id: str, flow: Flow, flow_info: ComponentInfo) -> None

Append cost data element for each cost in flow.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_flow_costs(
    self,
    flow_id: str,
    flow: Flow,
    flow_info: ComponentInfo,
) -> None:
    """Append cost data element for each cost in flow."""
    cost_terms = flow.get_cost_terms()
    for cost_term_id, cost_term in cost_terms.items():
        level = self.get_cost_term_level(cost_term_id, cost_term, flow_info)

        has_profile = cost_term.get_profile() is None

        profile = self.get_cost_term_profile(cost_term_id, cost_term, flow_info) if has_profile else 1.0

        extended_cost_term_id = f"{flow_id}_{cost_term_id}"
        param_id = f"{extended_cost_term_id}_param"
        self.append.cost_term(extended_cost_term_id, flow_id, flow_info.is_flow, cost_term.is_cost(), param_id)

        self.append.mean_series_param(param_id, level, profile)

add_flow_lower_bound(flow_id: str, flow: Flow, flow_info: ComponentInfo) -> None

Append lower bound related data elements for a flow.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_flow_lower_bound(
    self,
    flow_id: str,
    flow: Flow,
    flow_info: ComponentInfo,
) -> None:
    """Append lower bound related data elements for a flow."""
    capacity = flow.get_min_capacity()
    bound_id = f"{flow_id}_lower_bound"
    profile = None if capacity is None else capacity.get_profile()
    if profile is None:
        self.append.lower_zero_capacity(bound_id, flow_info.is_flow, flow_or_storage_id=flow_id)
        return
    self.add_positive_capacity(flow_id, flow_info, capacity, bound_id, is_lower_bound=True)

add_flow_upper_bound(flow_id: str, flow: Flow, flow_info: ComponentInfo) -> None

Append upper bound related data elements for a flow.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_flow_upper_bound(
    self,
    flow_id: str,
    flow: Flow,
    flow_info: ComponentInfo,
) -> None:
    """Append upper bound related data elements for a flow."""
    capacity = flow.get_max_capacity()
    if capacity is None:
        return
    bound_id = f"{flow_id}_upper_bound"
    self.add_positive_capacity(flow_id, flow_info, capacity, bound_id, is_lower_bound=False)

add_positive_capacity(flow_or_storage_id: str, info: ComponentInfo, capacity: FlowVolume | StockVolume, bound_id: str, is_lower_bound: bool) -> None

Append data elements related to positive capacity.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_positive_capacity(
    self,
    flow_or_storage_id: str,
    info: ComponentInfo,
    capacity: FlowVolume | StockVolume,
    bound_id: str,
    is_lower_bound: bool,
) -> None:
    """Append data elements related to positive capacity."""
    series_param_id = f"{bound_id}_series_param"

    if isinstance(capacity, FlowVolume):
        unit_param_id = f"{bound_id}_unit_param"
        self.append.unit_param(unit_param_id, series_param_id, info)
        self.append.positive_capacity(bound_id, info.is_flow, flow_or_storage_id, is_lower_bound, unit_param_id)
    else:
        assert isinstance(capacity, StockVolume)
        self.append.positive_capacity(bound_id, info.is_flow, flow_or_storage_id, is_lower_bound, series_param_id)

    level = self.get_capacity_level(series_param_id, capacity, info)
    profile = self.get_capacity_profile(series_param_id, capacity, info)

    self.append.mean_series_param(series_param_id, level, profile)

add_rhs_term(flow_id: str, flow: Flow, arrow: Arrow, node_info: ComponentInfo, flow_info: ComponentInfo) -> None

Append data elements related to rhs term.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_rhs_term(
    self,
    flow_id: str,
    flow: Flow,
    arrow: Arrow,
    node_info: ComponentInfo,
    flow_info: ComponentInfo,
) -> None:
    """Append data elements related to rhs term."""
    node_id = arrow.get_node()

    rhs_term_id = f"exogenous_flow_{flow_id}_{node_id}"
    unit_param_id = f"{rhs_term_id}_unit_param"
    series_param_id = f"{rhs_term_id}_series_param"
    balance_id = node_info.jules_balance_id

    self.append.base_rhs_term(rhs_term_id, balance_id, arrow.is_ingoing(), unit_param_id)

    level = self.get_rhs_term_level(rhs_term_id, flow, arrow, flow_info)

    profile = self.get_rhs_term_profile(rhs_term_id, flow, arrow, flow_info)

    # unit is actually flipped from main to target node
    # using conversion factor inside get_rhs_term_level
    # so the jules unit param might say GWh but the values
    # have been converted so that the result will become e.g. Mm3

    self.append.unit_param(unit_param_id, series_param_id, flow_info)
    self.append.mean_series_param(series_param_id, level, profile)

add_storage(model_id: str, storage: Storage, info: ComponentInfo) -> None

Append data elements related to a storage.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def add_storage(
    self,
    model_id: str,
    storage: Storage,
    info: ComponentInfo,
) -> None:
    """Append data elements related to a storage."""
    storage_id = info.jules_storage_id
    balance_id = info.jules_balance_id

    self.append.base_storage(storage_id, balance_id)

    capacity = storage.get_capacity()

    self.add_positive_capacity(storage_id, info, capacity, f"{storage_id}_upper_bound", False)

    self.append.lower_zero_capacity(f"{storage_id}_lower_bound", info.is_flow, storage_id)

    if model_id == self.names.CLEARING:
        if info.sss_global_eneq_value is not None:
            self.append.global_eneq(info.jules_global_eneq_id, info.jules_balance_id, info.sss_global_eneq_value)

        if info.sss_is_short_term and info.is_short_term_storage:
            self.append.storage_hint(
                storage_id,
                round(info.sss_storage_duration.total_seconds() * 1000),
            )

build() -> None

Build input and configuration files for JulES.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def build(self) -> None:
    """Build input and configuration files for JulES."""
    t = time()
    self.build_data_elements(self.names.CLEARING, self.graphs.clearing, self.graph_infos.clearing)
    self.send_debug_event(f"build_data_elements clearing time: {round(time() - t, 2)} seconds")

    t = time()
    self.build_data_elements(self.names.AGGREGATED, self.graphs.short_term, self.graph_infos.short_term)
    self.send_debug_event(f"build_data_elements aggregated time: {round(time() - t, 2)} seconds")

    # Intent is to build data elements for aggregated models. Pending changes in Julia-JulES
    # self.build_data_elements(self.names.SHORT_TERM, self.graphs.short_term, self.graph_infos.short_term)
    # self.build_data_elements(self.names.MEDIUM_TERM, self.graphs.medium_term, self.graph_infos.medium_term)
    # self.build_data_elements(self.names.LONG_TERM, self.graphs.long_term, self.graph_infos.long_term)

    t = time()
    self.build_time_vectors()
    self.send_debug_event(f"build_time_vectors time: {round(time() - t, 2)} seconds")

    t = time()
    self.build_storage_mapping(self.graph_infos.clearing)
    self.send_debug_event(f"build_storage_mapping time: {round(time() - t, 2)} seconds")

    t = time()
    self.build_start_storage(self.names.FILENAME_START_STORAGES_CLEARING, self.graph_infos.clearing)
    self.send_debug_event(f"build_start_storage clearing time: {round(time() - t, 2)} seconds")

    t = time()
    self.build_start_storage(self.names.FILENAME_START_STORAGES_AGGREGATED, self.graph_infos.short_term)
    self.send_debug_event(f"build_start_storage aggregated time: {round(time() - t, 2)} seconds")

build_data_elements(model_id: str, graph: dict[str, Flow | Node], graph_info: dict[str, ComponentInfo]) -> None

Write json file with data elements for a graph belonging to a given model_id.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def build_data_elements(
    self,
    model_id: str,
    graph: dict[str, Flow | Node],
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Write json file with data elements for a graph belonging to a given model_id."""
    self.fill_data_elements(model_id, graph, graph_info)
    self.stop_if_errors()
    filename = f"{self.names.ROOT_FILENAME_DATAELEMENTS}_{model_id}.json"
    self.write_json_file(self.append.data_elements, filename)

build_start_storage(filename: str, graph_info: dict[str, ComponentInfo]) -> None

Write start storag json file to folder.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def build_start_storage(
    self,
    filename: str,
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Write start storag json file to folder."""
    data = {
        info.jules_storage_id: info.sss_initial_storage for k, info in graph_info.items() if info.is_storage_node
    }
    self.write_json_file(data, filename)

build_storage_mapping(graph_info: dict[str, ComponentInfo]) -> None

Write the mapping of storages from Clearing to Aggregated Model to json.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def build_storage_mapping(
    self,
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Write the mapping of storages from Clearing to Aggregated Model to json."""
    data = {
        k: info.agg_storage_node_id
        for k, info in graph_info.items()
        if info.is_storage_node and info.agg_storage_node_id
    }
    self.write_json_file(data, self.names.FILENAME_STORAGE_MAPPING)

build_time_vectors() -> None

Write json file with time vector data elements and csv file for each unique time index.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def build_time_vectors(self) -> None:
    """Write json file with time vector data elements and csv file for each unique time index."""
    self.append.data_elements = []
    for i, timeindex in enumerate(self.timevectors, start=1):
        time_index_id = f"timeindex_{i}"

        milliseconds: float = timeindex.get_period_duration().total_seconds() * 1000.0
        time_delta_id = f"{time_index_id}_timedelta"
        self.append.ms_time_delta(time_delta_id, milliseconds)

        self.append.range_time_index(
            time_index_id,
            timeindex.get_start_time(),
            timeindex.get_num_periods(),
            time_delta_id,
        )

        table_id = f"{time_index_id}_table"
        path_table, column_names = self.write_table(time_index_id, timeindex)
        self.append.base_table(table_id, path_table, column_names)

        is_one_year = timeindex.is_one_year()

        for column_name in column_names:
            time_vector_id = column_name
            time_values_id = f"{time_vector_id}_values"

            self.append.column_time_values(time_values_id, table_id, column_name)

            if is_one_year:
                self.append.one_year_time_vector(time_vector_id, time_index_id, time_values_id)
            else:
                self.append.rotating_time_vector(time_vector_id, time_index_id, time_values_id)

    self.write_json_file(self.append.data_elements, self.names.FILENAME_DATAELEMENTS_TIMEVECTORS)

fill_data_elements(model_id: str, graph: dict[str, Flow | Node], graph_info: dict[str, ComponentInfo]) -> None

Reset and fill self.append.data_elements with data element json data.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def fill_data_elements(
    self,
    model_id: str,
    graph: dict[str, Flow | Node],
    graph_info: dict[str, ComponentInfo],
) -> None:
    """Reset and fill self.append.data_elements with data element json data."""
    self.append.data_elements = []  # Important to reset the list

    nodes: dict[str, Node] = {k: v for k, v in graph.items() if isinstance(v, Node)}
    flows: dict[str, Flow] = {k: v for k, v in graph.items() if isinstance(v, Flow)}

    exogenous_nodes = {k: v for k, v in nodes.items() if v.is_exogenous()}
    exogenous_flows = {k: v for k, v in flows.items() if v.is_exogenous()}
    endogenous_nodes = {k: v for k, v in nodes.items() if not v.is_exogenous()}
    endogenous_flows = {k: v for k, v in flows.items() if not v.is_exogenous()}

    t = time()
    self.add_exogenous_nodes(exogenous_nodes, graph_info)
    self.send_debug_event(f"add_exogenous_nodes time: {round(time() - t, 2)} seconds")
    t = time()
    self.add_exogenous_flows(exogenous_flows, graph_info)
    self.send_debug_event(f"add_exogenous_flows time: {round(time() - t, 2)} seconds")
    t = time()
    self.add_endogenous_nodes(endogenous_nodes, graph_info, model_id)
    self.send_debug_event(f"add_endogenous_nodes time: {round(time() - t, 2)} seconds")
    t = time()
    self.add_endogenous_flows(endogenous_flows, graph_info)
    self.send_debug_event(f"add_endogenous_flows time: {round(time() - t, 2)} seconds")

    t = time()
    self.add_dummy_exogenous_balance()
    self.send_debug_event(f"add_dummy_exogenous_balance time: {round(time() - t, 2)} seconds")

get_capacity_level(root_id: str, capacity: FlowVolume | StockVolume, info: ComponentInfo) -> str | float abstractmethod

Query capacity level.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_capacity_level(self, root_id: str, capacity: FlowVolume | StockVolume, info: ComponentInfo) -> str | float:
    """Query capacity level."""
    pass

get_capacity_profile(root_id: str, capacity: FlowVolume | StockVolume, info: ComponentInfo) -> str | float abstractmethod

Query capacity profile.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_capacity_profile(
    self,
    root_id: str,
    capacity: FlowVolume | StockVolume,
    info: ComponentInfo,
) -> str | float:
    """Query capacity profile."""
    pass

get_coefficient_level(root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float abstractmethod

Query arrow coefficient level.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_coefficient_level(self, root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float:
    """Query arrow coefficient level."""
    pass

get_cost_term_level(root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float abstractmethod

Query cost term level.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_cost_term_level(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
    """Query cost term level."""
    pass

get_cost_term_profile(root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float abstractmethod

Query cost term profile.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_cost_term_profile(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
    """Query cost term profile."""
    pass

get_price_level(root_id: str, price: Price, info: ComponentInfo) -> str | float abstractmethod

Query price level.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_price_level(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
    """Query price level."""
    pass

get_price_profile(root_id: str, price: Price, info: ComponentInfo) -> str | float abstractmethod

Query price profile.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_price_profile(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
    """Query price profile."""
    pass

get_rhs_term_level(rhs_term_id: str, flow: Flow, arrow: Arrow, flow_info: ComponentInfo) -> str | float abstractmethod

Query rhs term level.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_rhs_term_level(
    self,
    rhs_term_id: str,
    flow: Flow,
    arrow: Arrow,
    flow_info: ComponentInfo,
) -> str | float:
    """Query rhs term level."""
    pass

get_rhs_term_profile(rhs_term_id: str, flow: Flow, arrow: Arrow, flow_info: ComponentInfo) -> str | float abstractmethod

Query rhs term profile.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

@abstractmethod
def get_rhs_term_profile(
    self,
    rhs_term_id: str,
    flow: Flow,
    arrow: Arrow,
    flow_info: ComponentInfo,
) -> str | float:
    """Query rhs term profile."""
    pass

get_time_index_id(timeindex: FixedFrequencyTimeIndex) -> str

Return id that works in file name.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def get_time_index_id(self, timeindex: FixedFrequencyTimeIndex) -> str:
    """Return id that works in file name."""
    type_name = type(timeindex).__name__
    num_periods = timeindex.get_num_periods()
    resolution = int(timeindex.get_period_duration().total_seconds() * 1000.0)
    is_52 = timeindex.is_52_week_years()
    extr_first = timeindex.extrapolate_first_point()
    extr_last = timeindex.extrapolate_last_point()
    return f"timeindex_{type_name}_periods_{num_periods}_ms_{resolution}_{is_52}_{extr_first}_{extr_last}"

stop_if_errors() -> None

Throw RunTimeError if any errors.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def stop_if_errors(self) -> None:
    """Throw RunTimeError if any errors."""
    if self.errors:
        error_string = "\n".join(self.errors)
        message = f"Errors found:\n{error_string}"
        raise RuntimeError(message)

write_json_file(data: object, filename: str) -> None

Write data to json.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def write_json_file(self, data: object, filename: str) -> None:
    """Write data to json."""
    with Path.open(self.folder / filename, "w") as f:
        json.dump(data, f, indent=self.names.JSON_INDENT)

write_table(time_index_id: str, timeindex: FixedFrequencyTimeIndex) -> tuple[Path, list[str]]

Write all vectors corresponding to time_index_id to a csv file.

Source code in framjules/solve_handler/build_handler/BuildHandler.py

def write_table(
    self,
    time_index_id: str,
    timeindex: FixedFrequencyTimeIndex,
) -> tuple[Path, list[str]]:
    """Write all vectors corresponding to time_index_id to a csv file."""
    vectors = self.timevectors[timeindex]
    column_names = list(vectors.keys())
    matrix = np.column_stack([vectors[c] for c in column_names])
    matrix = np.round(matrix, decimals=6)  # quick fix for negative values in time vectors
    # if negative values give warning
    if np.any(matrix < 0):
        self.errors.add(f"Negative values found in time vector for {time_index_id}. This might cause issues.")
    filename = f"timevector_{time_index_id}.csv"
    path = self.folder / filename
    np.savetxt(path, matrix, delimiter=",")
    return path, column_names

DataElementAppender

Functionality to reate data element to JulES.

DataElementAppender

Used to generate list of data elements for JulES.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

class DataElementAppender:
    """Used to generate list of data elements for JulES."""

    def __init__(self, names: JulESNames) -> None:
        """Initialize new DataElementAppender."""
        self.data_elements: list = []
        self.names: JulESNames = names

    def exogenous_balance(
        self,
        balance_id: str,
        commodiy: str,
        price_param_id: str | float,
    ) -> None:
        """Append exogenous balance data element."""
        self.data_elements.append(
            [
                self.names.BALANCE,
                self.names.EXOGENBALANCE,
                balance_id,
                [self.names.COMMODITY, commodiy],
                [self.names.PRICE, price_param_id],
            ],
        )

    def endogenous_balance(
        self,
        balance_id: str,
        commodiy: str,
    ) -> None:
        """Append endogenous balance data element."""
        self.data_elements.append(
            [
                self.names.BALANCE,
                self.names.BASEBALANCE,
                balance_id,
                [self.names.COMMODITY, commodiy],
            ],
        )

    def mean_series_param(
        self,
        param_id: str,
        level: str | float,
        profile: str | float,
    ) -> None:
        """Append mean series param data element."""
        self.data_elements.append(
            [
                self.names.PARAM,
                self.names.MEANSERIESPARAM,
                param_id,
                [self.names.LEVEL, level],
                [self.names.PROFILE, profile],
            ],
        )

    def base_flow(self, flow_id: str) -> None:
        """Append base flow data element."""
        self.data_elements.append([self.names.FLOW, self.names.BASEFLOW, flow_id])

    def lower_zero_capacity(
        self,
        lower_bound_id: str,
        is_flow: bool,
        flow_or_storage_id: str,
    ) -> None:
        """Append zero lower capacity data element."""
        self.data_elements.append(
            [
                self.names.CAPACITY,
                self.names.LOWERZEROCAPACITY,
                lower_bound_id,
                [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
                [self.names.WHICHINSTANCE, flow_or_storage_id],
            ],
        )

    def positive_capacity(
        self,
        bound_id: str,
        is_flow: bool,
        flow_or_storage_id: str,
        is_lower_bound: bool,
        param_id: str,
    ) -> None:
        """Append positive capacity data element."""
        self.data_elements.append(
            [
                self.names.CAPACITY,
                self.names.POSITIVECAPACITY,
                bound_id,
                [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
                [self.names.WHICHINSTANCE, flow_or_storage_id],
                [self.names.BOUNDKEY, self.names.BOUNDLOWER if is_lower_bound else self.names.BOUNDUPPER],
                [self.names.PARAM, param_id],
            ],
        )

    def unit_param(
        self,
        unit_param_id: str,
        series_param_id: str,
        info: ComponentInfo,
    ) -> None:
        """Append unit param data element."""
        self.data_elements.append(
            [
                self.names.PARAM,
                info.unit_param_type,
                unit_param_id,
                [self.names.PARAM, series_param_id],
            ],
        )

    def base_arrow(
        self,
        arrow_id: str,
        flow_id: str,
        balance_id: str,
        is_ingoing: bool,
        conversion: str | float,
    ) -> None:
        """Append base arrow data element."""
        self.data_elements.append(
            [
                self.names.ARROW,
                self.names.BASEARROW,
                arrow_id,
                [self.names.FLOW, flow_id],
                [self.names.BALANCE, balance_id],
                [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_ingoing else self.names.DIRECTIONOUT],
                [self.names.CONVERSION, conversion],
            ],
        )

    def cost_term(
        self,
        cost_term_id: str,
        flow_or_storage_id: str,
        is_flow: str,
        is_cost: bool,
        cost: str | float,
    ) -> None:
        """Append cost term data element."""
        self.data_elements.append(
            [
                self.names.COST,
                self.names.COSTTERM,
                cost_term_id,
                [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_cost else self.names.DIRECTIONOUT],
                [self.names.PARAM, cost],
                [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
                [self.names.WHICHINSTANCE, flow_or_storage_id],
            ],
        )

    def base_rhs_term(
        self,
        rhs_term_id: str,
        balance_id: str,
        is_ingoing: bool,
        unit_param_id: str,
    ) -> None:
        """Append base rhs term data element."""
        # TODO: Add residualhint # noqa FIX002
        self.data_elements.append(
            [
                self.names.RHSTERM,
                self.names.BASERHSTERM,
                rhs_term_id,
                [self.names.BALANCE, balance_id],
                [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_ingoing else self.names.DIRECTIONOUT],
                [self.names.PARAM, unit_param_id],
            ],
        )

    def base_storage(
        self,
        storage_id: str,
        balance_id: str,
    ) -> None:
        """Append base storage data element."""
        self.data_elements.append(
            [
                self.names.STORAGE,
                self.names.BASESTORAGE,
                storage_id,
                [self.names.BALANCE, balance_id],
            ],
        )

    def global_eneq(
        self,
        global_eneq_id: str,
        balance_id: str,
        value: float,
    ) -> None:
        """Append global energy equivalent data element."""
        self.data_elements.append(
            [
                self.names.METADATA,
                self.names.GLOBALENEQ,
                global_eneq_id,
                [self.names.BALANCE, balance_id],
                [self.names.VALUE, value],
            ],
        )

    def storage_hint(self, storage_id: str, period: int) -> None:
        """Append storage hint data element to indicate storage duration in milliseconds."""
        storage_hint_id = f"StorageHint_{storage_id}"
        self.data_elements.append(
            [
                self.names.METADATA,
                self.names.STORAGEHINT,
                storage_hint_id,
                [self.names.STORAGE, storage_id],
                [self.names.PERIOD, period],
            ],
        )

    def ms_time_delta(
        self,
        time_delta_id: str,
        milliseconds: float,
    ) -> None:
        """Append ms time delta data element."""
        self.data_elements.append(
            [
                self.names.TIMEDELTA,
                self.names.MSTIMEDELTA,
                time_delta_id,
                [self.names.PERIOD, milliseconds],
            ],
        )

    def range_time_index(
        self,
        time_index_id: str,
        start_time: datetime,
        num_steps: int,
        time_delta_id: str,
    ) -> None:
        """Append range time index data element."""
        self.data_elements.append(
            [
                self.names.TIMEINDEX,
                self.names.RANGETIMEINDEX,
                time_index_id,
                [self.names.START, start_time.strftime(r"%Y-%m-%d %H:%M:%S")],
                [self.names.STEPS, num_steps],
                [self.names.DELTA, time_delta_id],
            ],
        )

    def base_table(
        self,
        table_id: str,
        path_table: str | Path,
        column_names: list[str],
    ) -> None:
        """Append base table data element."""
        self.data_elements.append(
            [
                self.names.TABLE,
                self.names.BASETABLE,
                table_id,
                [self.names.MATRIX, str(path_table)],
                [self.names.NAMES, column_names],
            ],
        )

    def column_time_values(
        self,
        time_values_id: str,
        table_id: str,
        column_id: str,
    ) -> None:
        """Append column time values data element."""
        self.data_elements.append(
            [
                self.names.TIMEVALUES,
                self.names.COLUMNTIMEVALUES,
                time_values_id,
                [self.names.TABLE, table_id],
                [self.names.NAME, column_id],
            ],
        )

    def rotating_time_vector(
        self,
        time_vector_id: str,
        time_index_id: str,
        time_values_id: str,
    ) -> None:
        """Append rotating time vector data element."""
        self.data_elements.append(
            [
                self.names.TIMEVECTOR,
                self.names.ROTATINGTIMEVECTOR,
                time_vector_id,
                [self.names.TIMEINDEX, time_index_id],
                [self.names.TIMEVALUES, time_values_id],
            ],
        )

    def one_year_time_vector(
        self,
        time_vector_id: str,
        time_index_id: str,
        time_values_id: str,
    ) -> None:
        """Append one year time vector data element."""
        self.data_elements.append(
            [
                self.names.TIMEVECTOR,
                self.names.ONEYEARTIMEVECTOR,
                time_vector_id,
                [self.names.TIMEINDEX, time_index_id],
                [self.names.TIMEVALUES, time_values_id],
            ],
        )

__init__(names: JulESNames) -> None

Initialize new DataElementAppender.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def __init__(self, names: JulESNames) -> None:
    """Initialize new DataElementAppender."""
    self.data_elements: list = []
    self.names: JulESNames = names

base_arrow(arrow_id: str, flow_id: str, balance_id: str, is_ingoing: bool, conversion: str | float) -> None

Append base arrow data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def base_arrow(
    self,
    arrow_id: str,
    flow_id: str,
    balance_id: str,
    is_ingoing: bool,
    conversion: str | float,
) -> None:
    """Append base arrow data element."""
    self.data_elements.append(
        [
            self.names.ARROW,
            self.names.BASEARROW,
            arrow_id,
            [self.names.FLOW, flow_id],
            [self.names.BALANCE, balance_id],
            [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_ingoing else self.names.DIRECTIONOUT],
            [self.names.CONVERSION, conversion],
        ],
    )

base_flow(flow_id: str) -> None

Append base flow data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def base_flow(self, flow_id: str) -> None:
    """Append base flow data element."""
    self.data_elements.append([self.names.FLOW, self.names.BASEFLOW, flow_id])

base_rhs_term(rhs_term_id: str, balance_id: str, is_ingoing: bool, unit_param_id: str) -> None

Append base rhs term data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def base_rhs_term(
    self,
    rhs_term_id: str,
    balance_id: str,
    is_ingoing: bool,
    unit_param_id: str,
) -> None:
    """Append base rhs term data element."""
    # TODO: Add residualhint # noqa FIX002
    self.data_elements.append(
        [
            self.names.RHSTERM,
            self.names.BASERHSTERM,
            rhs_term_id,
            [self.names.BALANCE, balance_id],
            [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_ingoing else self.names.DIRECTIONOUT],
            [self.names.PARAM, unit_param_id],
        ],
    )

base_storage(storage_id: str, balance_id: str) -> None

Append base storage data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def base_storage(
    self,
    storage_id: str,
    balance_id: str,
) -> None:
    """Append base storage data element."""
    self.data_elements.append(
        [
            self.names.STORAGE,
            self.names.BASESTORAGE,
            storage_id,
            [self.names.BALANCE, balance_id],
        ],
    )

base_table(table_id: str, path_table: str | Path, column_names: list[str]) -> None

Append base table data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def base_table(
    self,
    table_id: str,
    path_table: str | Path,
    column_names: list[str],
) -> None:
    """Append base table data element."""
    self.data_elements.append(
        [
            self.names.TABLE,
            self.names.BASETABLE,
            table_id,
            [self.names.MATRIX, str(path_table)],
            [self.names.NAMES, column_names],
        ],
    )

column_time_values(time_values_id: str, table_id: str, column_id: str) -> None

Append column time values data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def column_time_values(
    self,
    time_values_id: str,
    table_id: str,
    column_id: str,
) -> None:
    """Append column time values data element."""
    self.data_elements.append(
        [
            self.names.TIMEVALUES,
            self.names.COLUMNTIMEVALUES,
            time_values_id,
            [self.names.TABLE, table_id],
            [self.names.NAME, column_id],
        ],
    )

cost_term(cost_term_id: str, flow_or_storage_id: str, is_flow: str, is_cost: bool, cost: str | float) -> None

Append cost term data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def cost_term(
    self,
    cost_term_id: str,
    flow_or_storage_id: str,
    is_flow: str,
    is_cost: bool,
    cost: str | float,
) -> None:
    """Append cost term data element."""
    self.data_elements.append(
        [
            self.names.COST,
            self.names.COSTTERM,
            cost_term_id,
            [self.names.DIRECTIONKEY, self.names.DIRECTIONIN if is_cost else self.names.DIRECTIONOUT],
            [self.names.PARAM, cost],
            [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
            [self.names.WHICHINSTANCE, flow_or_storage_id],
        ],
    )

endogenous_balance(balance_id: str, commodiy: str) -> None

Append endogenous balance data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def endogenous_balance(
    self,
    balance_id: str,
    commodiy: str,
) -> None:
    """Append endogenous balance data element."""
    self.data_elements.append(
        [
            self.names.BALANCE,
            self.names.BASEBALANCE,
            balance_id,
            [self.names.COMMODITY, commodiy],
        ],
    )

exogenous_balance(balance_id: str, commodiy: str, price_param_id: str | float) -> None

Append exogenous balance data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def exogenous_balance(
    self,
    balance_id: str,
    commodiy: str,
    price_param_id: str | float,
) -> None:
    """Append exogenous balance data element."""
    self.data_elements.append(
        [
            self.names.BALANCE,
            self.names.EXOGENBALANCE,
            balance_id,
            [self.names.COMMODITY, commodiy],
            [self.names.PRICE, price_param_id],
        ],
    )

global_eneq(global_eneq_id: str, balance_id: str, value: float) -> None

Append global energy equivalent data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def global_eneq(
    self,
    global_eneq_id: str,
    balance_id: str,
    value: float,
) -> None:
    """Append global energy equivalent data element."""
    self.data_elements.append(
        [
            self.names.METADATA,
            self.names.GLOBALENEQ,
            global_eneq_id,
            [self.names.BALANCE, balance_id],
            [self.names.VALUE, value],
        ],
    )

lower_zero_capacity(lower_bound_id: str, is_flow: bool, flow_or_storage_id: str) -> None

Append zero lower capacity data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def lower_zero_capacity(
    self,
    lower_bound_id: str,
    is_flow: bool,
    flow_or_storage_id: str,
) -> None:
    """Append zero lower capacity data element."""
    self.data_elements.append(
        [
            self.names.CAPACITY,
            self.names.LOWERZEROCAPACITY,
            lower_bound_id,
            [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
            [self.names.WHICHINSTANCE, flow_or_storage_id],
        ],
    )

mean_series_param(param_id: str, level: str | float, profile: str | float) -> None

Append mean series param data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def mean_series_param(
    self,
    param_id: str,
    level: str | float,
    profile: str | float,
) -> None:
    """Append mean series param data element."""
    self.data_elements.append(
        [
            self.names.PARAM,
            self.names.MEANSERIESPARAM,
            param_id,
            [self.names.LEVEL, level],
            [self.names.PROFILE, profile],
        ],
    )

ms_time_delta(time_delta_id: str, milliseconds: float) -> None

Append ms time delta data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def ms_time_delta(
    self,
    time_delta_id: str,
    milliseconds: float,
) -> None:
    """Append ms time delta data element."""
    self.data_elements.append(
        [
            self.names.TIMEDELTA,
            self.names.MSTIMEDELTA,
            time_delta_id,
            [self.names.PERIOD, milliseconds],
        ],
    )

one_year_time_vector(time_vector_id: str, time_index_id: str, time_values_id: str) -> None

Append one year time vector data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def one_year_time_vector(
    self,
    time_vector_id: str,
    time_index_id: str,
    time_values_id: str,
) -> None:
    """Append one year time vector data element."""
    self.data_elements.append(
        [
            self.names.TIMEVECTOR,
            self.names.ONEYEARTIMEVECTOR,
            time_vector_id,
            [self.names.TIMEINDEX, time_index_id],
            [self.names.TIMEVALUES, time_values_id],
        ],
    )

positive_capacity(bound_id: str, is_flow: bool, flow_or_storage_id: str, is_lower_bound: bool, param_id: str) -> None

Append positive capacity data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def positive_capacity(
    self,
    bound_id: str,
    is_flow: bool,
    flow_or_storage_id: str,
    is_lower_bound: bool,
    param_id: str,
) -> None:
    """Append positive capacity data element."""
    self.data_elements.append(
        [
            self.names.CAPACITY,
            self.names.POSITIVECAPACITY,
            bound_id,
            [self.names.WHICHCONCEPT, self.names.FLOW if is_flow else self.names.STORAGE],
            [self.names.WHICHINSTANCE, flow_or_storage_id],
            [self.names.BOUNDKEY, self.names.BOUNDLOWER if is_lower_bound else self.names.BOUNDUPPER],
            [self.names.PARAM, param_id],
        ],
    )

range_time_index(time_index_id: str, start_time: datetime, num_steps: int, time_delta_id: str) -> None

Append range time index data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def range_time_index(
    self,
    time_index_id: str,
    start_time: datetime,
    num_steps: int,
    time_delta_id: str,
) -> None:
    """Append range time index data element."""
    self.data_elements.append(
        [
            self.names.TIMEINDEX,
            self.names.RANGETIMEINDEX,
            time_index_id,
            [self.names.START, start_time.strftime(r"%Y-%m-%d %H:%M:%S")],
            [self.names.STEPS, num_steps],
            [self.names.DELTA, time_delta_id],
        ],
    )

rotating_time_vector(time_vector_id: str, time_index_id: str, time_values_id: str) -> None

Append rotating time vector data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def rotating_time_vector(
    self,
    time_vector_id: str,
    time_index_id: str,
    time_values_id: str,
) -> None:
    """Append rotating time vector data element."""
    self.data_elements.append(
        [
            self.names.TIMEVECTOR,
            self.names.ROTATINGTIMEVECTOR,
            time_vector_id,
            [self.names.TIMEINDEX, time_index_id],
            [self.names.TIMEVALUES, time_values_id],
        ],
    )

storage_hint(storage_id: str, period: int) -> None

Append storage hint data element to indicate storage duration in milliseconds.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def storage_hint(self, storage_id: str, period: int) -> None:
    """Append storage hint data element to indicate storage duration in milliseconds."""
    storage_hint_id = f"StorageHint_{storage_id}"
    self.data_elements.append(
        [
            self.names.METADATA,
            self.names.STORAGEHINT,
            storage_hint_id,
            [self.names.STORAGE, storage_id],
            [self.names.PERIOD, period],
        ],
    )

unit_param(unit_param_id: str, series_param_id: str, info: ComponentInfo) -> None

Append unit param data element.

Source code in framjules/solve_handler/build_handler/DataElementAppender.py

def unit_param(
    self,
    unit_param_id: str,
    series_param_id: str,
    info: ComponentInfo,
) -> None:
    """Append unit param data element."""
    self.data_elements.append(
        [
            self.names.PARAM,
            info.unit_param_type,
            unit_param_id,
            [self.names.PARAM, series_param_id],
        ],
    )

SerialBuildHandler

SerialBuildHandler

Bases: BuildHandler

Specialized methods for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

class SerialBuildHandler(BuildHandler):
    """Specialized methods for serial simulation."""

    _PROFILE_VALUE_WHEN_ZERO = 1.0

    def __init__(
        self,
        folder: Path,
        config: JulESConfig,
        names: JulESNames,
        domain_models: DomainModels,
        graphs: NodeFlowGraphs,
        graph_infos: GraphInfos,
        db: QueryDB,
    ) -> None:
        """See BuildHandler."""
        super().__init__(folder, config, names, domain_models, graphs, graph_infos, db)

        # some validations spesific to serial simulation
        self._data_period = self.config.get_data_period()
        self._check_type(self._data_period, SinglePeriodTimeIndex)

        start_year, num_years = self.config.get_weather_years()
        self._avg_year_range = AverageYearRange(start_year=start_year, num_years=num_years)
        if self._avg_year_range.is_52_week_years():
            message = "Expected AverageYearRange to have is_52_week_years() == False."
            raise ValueError(message)

        self._is_float32 = self.config.is_float32()

    def get_attribute_level(
        self,
        root_id: str,
        attribute: LevelProfile,
        target_unit: str | None,
    ) -> float:
        """Get level value."""
        is_max_level = False
        value = attribute.get_data_value(
            db=self.db,
            unit=target_unit,
            level_period=self._data_period,
            scenario_horizon=self._avg_year_range,
            is_max_level=is_max_level,
        )
        if value < 0:
            self.send_warning_event(f"Attribute {root_id} of type {type(attribute).__name__} returned {value}")
        return value

    def get_attribute_profile(
        self,
        root_id: str,
        attribute: LevelProfile | Arrow,
        default: float,
        unit: str | None,
        info: ComponentInfo,
    ) -> str | float:
        """Add profile vector to timevectors and return profile_id."""
        if not attribute.has_profile():
            return default

        profile_id = f"{root_id}_profile"

        timeindex = self._find_profile_timeindex(attribute, info.has_storage_resolution)

        vector: NDArray = attribute.get_scenario_vector(
            db=self.db,
            level_period=self._data_period,
            scenario_horizon=timeindex,
            is_float32=self._is_float32,
            unit=unit,
        )
        vector = np.round(vector, decimals=6)  # avoid very small numerical noise, e.g negative prices from jules run

        if np.isnan(vector).any():
            message = (
                f"Profile {profile_id} in time index {timeindex} contains NaN values. "
                "This may indicate a problem with the data or the configuration."
            )
            raise ValueError(message)

        if (vector < 0).any():
            message = (
                f"Profile {profile_id} in time index {timeindex} has negative values. "
                "This may indicate a problem with the data or the configuration."
            )
            raise ValueError(message)

        denominator = vector.mean()

        if denominator == 0:
            vector.fill(self._PROFILE_VALUE_WHEN_ZERO)
        else:
            np.multiply(vector, 1 / denominator, out=vector)

        self.timevectors[timeindex][profile_id] = vector

        return profile_id

    def _find_profile_timeindex(
        self,
        attribute: LevelProfile | Arrow,
        has_storage_resolution: bool,
    ) -> ProfileTimeIndex:
        is_52_week_years = False

        start_year, num_years = self.config.get_weather_years()

        if has_storage_resolution:
            period_duration = timedelta(minutes=self.config.get_time_resolution().get_clearing_storage_minutes())
        else:
            period_duration = timedelta(minutes=self.config.get_time_resolution().get_clearing_market_minutes())

        fallback = ProfileTimeIndex(
            start_year,
            num_years,
            period_duration=self._get_closest_valid_profile_duration(period_duration),
            is_52_week_years=is_52_week_years,
        )

        ix_set: set[FixedFrequencyTimeIndex] = attribute.get_profile_timeindex_set(db=self.db)
        if not all(isinstance(ix, FixedFrequencyTimeIndex) for ix in ix_set):
            return fallback

        is_one_year = all(ix.is_one_year() for ix in ix_set)

        candidate = min((ix for ix in ix_set), key=lambda ix: ix.get_period_duration().total_seconds())

        s_fallback = fallback.get_period_duration().total_seconds()
        s_candidate = candidate.get_period_duration().total_seconds()
        best_duration = (candidate if s_fallback < s_candidate else fallback).get_period_duration()

        if is_one_year:
            return ProfileTimeIndex(
                start_year=candidate.get_start_time().isocalendar().year,
                num_years=1,
                period_duration=self._get_closest_valid_profile_duration(best_duration),
                is_52_week_years=is_52_week_years,
            )
        return ProfileTimeIndex(
            start_year=start_year,
            num_years=num_years,
            period_duration=self._get_closest_valid_profile_duration(best_duration),
            is_52_week_years=is_52_week_years,
        )

    def get_price_level(self, root_id: str, price: Price, info: ComponentInfo) -> float:
        """get_price_level for serial simulation."""
        return self.get_attribute_level(root_id, price, info.unit_price)

    def get_cost_term_level(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> float:
        """get_cost_term_level for serial simulation."""
        return self.get_attribute_level(root_id, cost_term, info.unit_cost)

    def get_capacity_level(self, root_id: str, capacity: StockVolume | FlowVolume, info: ComponentInfo) -> str | float:
        """get_capacity_level for serial simulation. Handles stock or flow based on info."""
        return self.get_attribute_level(root_id, capacity, info.unit_flow if info.is_flow else info.unit_stock)

    def get_coefficient_level(self, root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float:
        """get_coefficient_level for serial simulation."""
        return self.get_attribute_level(root_id, arrow, info.unit_coeffs[arrow.get_node()])

    def get_price_profile(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
        """get_price_profile for serial simulation."""
        return self.get_attribute_profile(root_id, price, 1.0, info.unit_price, info)

    def get_cost_term_profile(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
        """get_cost_term_profile for serial simulation."""
        return self.get_attribute_profile(root_id, cost_term, 1.0, info.unit_cost, info)

    def get_capacity_profile(
        self,
        root_id: str,
        capacity: FlowVolume | StockVolume,
        info: ComponentInfo,
    ) -> str | float:
        """get_capacity_profile for serial simulation."""
        unit = info.unit_flow if info.is_flow else info.unit_stock
        return self.get_attribute_profile(root_id, capacity, 1.0, unit, info)

    def get_rhs_term_level(
        self,
        rhs_term_id: str,
        flow: Flow,
        arrow: Arrow,
        flow_info: ComponentInfo,
    ) -> float:
        """Convert volume (main node) to target node volume.

        This may scale the volume using arrow coefficient,
        e.g. due to transportation loss.

        This may also change unit, if target node belongs to
        different commodity than main node, such as for hydropower.
        """
        coeff_value = self.get_coefficient_level(rhs_term_id, arrow, flow_info)

        volume = flow.get_volume()
        if volume.has_level():
            volume_value = self.get_attribute_level(
                root_id=rhs_term_id,
                attribute=volume,
                target_unit=flow_info.unit_flow,
            )
        else:
            max_cap = flow.get_max_capacity()
            min_cap = flow.get_min_capacity()
            if max_cap is not None and max_cap == min_cap:
                volume_value = self.get_attribute_level(
                    root_id=rhs_term_id,
                    attribute=max_cap,
                    target_unit=flow_info.unit_flow,
                )
            else:
                message = f"{rhs_term_id} is not exogenous"
                raise ValueError(message)

        if coeff_value == 0:
            message = (
                f"Got zero coeff_value for {rhs_term_id}.\n"
                f"volume_value = {volume_value}\n"
                f"coeff_value = {coeff_value}\n"
                f"flow.get_main_node() = {flow.get_main_node()}\n"
                f"arrow.get_node() = {arrow.get_node()}\n"
            )
            raise RuntimeError(message)

        return volume_value / coeff_value  # convert from main_unit to target_unit

    def _rank_profile_timeindex(self, ix: ProfileTimeIndex) -> tuple[bool, float]:
        return ix.is_one_year(), ix.get_period_duration().total_seconds()

    def _select_profile_timeindex(self, *candidates: ProfileTimeIndex) -> ProfileTimeIndex:
        """Select the one with not-is_one_year (if any) and finest period duration."""
        return min(candidates, key=self._rank_profile_timeindex)

    def get_rhs_term_profile(
        self,
        rhs_term_id: str,
        flow: Flow,
        arrow: Arrow,
        flow_info: ComponentInfo,
    ) -> str | float:
        """Create profile (possibly) representing volume_profile * coefficient_profile."""
        volume = flow.get_volume()

        if not volume.has_profile():
            volume = flow.get_max_capacity()

        not_volume_profile = volume is None or not volume.has_profile()
        not_arrow_profile = not arrow.has_profile()
        has_volume_profile = not not_volume_profile
        has_arrow_profile = not not_arrow_profile

        if not_volume_profile and not_arrow_profile:
            return 1.0

        if has_volume_profile and not_arrow_profile:
            return self.get_attribute_profile(
                rhs_term_id,
                attribute=volume,
                default=1.0,
                unit=flow_info.unit_flow,
                info=flow_info,
            )

        if not_volume_profile and has_arrow_profile:
            unit = flow_info.unit_coeffs[arrow.get_node()]
            return self.get_attribute_profile(
                rhs_term_id,
                attribute=arrow,
                default=1.0,
                unit=unit,
                info=flow_info,
            )

        # Here we get both profiles (volume and coefficient) and muliply them
        # together and store the resulting product profile in self.timevectors

        profile_id = f"{rhs_term_id}_profile"

        timeindex = self._select_profile_timeindex(
            self._find_profile_timeindex(volume, flow_info.has_storage_resolution),
            self._find_profile_timeindex(arrow, flow_info.has_storage_resolution),
        )

        x: NDArray = volume.get_scenario_vector(
            db=self.db,
            level_period=self._data_period,
            scenario_horizon=timeindex,
            is_float32=self._is_float32,
            unit=flow_info.unit_flow,
        )

        y: NDArray = arrow.get_scenario_vector(
            db=self.db,
            level_period=self._data_period,
            scenario_horizon=timeindex,
            is_float32=self._is_float32,
            unit=flow_info.unit_coeffs[arrow.get_node()],
        )

        x_mean = x.mean()
        y_mean = y.mean()
        if x_mean == 0 or y_mean == 0:
            x.fill(self._PROFILE_VALUE_WHEN_ZERO)
            return x

        np.multiply(x, 1.0 / x_mean, out=x)
        np.multiply(y, 1.0 / y_mean, out=y)

        np.multiply(x, y, out=x)

        self.timevectors[timeindex][profile_id] = x

        return profile_id

__init__(folder: Path, config: JulESConfig, names: JulESNames, domain_models: DomainModels, graphs: NodeFlowGraphs, graph_infos: GraphInfos, db: QueryDB) -> None

See BuildHandler.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def __init__(
    self,
    folder: Path,
    config: JulESConfig,
    names: JulESNames,
    domain_models: DomainModels,
    graphs: NodeFlowGraphs,
    graph_infos: GraphInfos,
    db: QueryDB,
) -> None:
    """See BuildHandler."""
    super().__init__(folder, config, names, domain_models, graphs, graph_infos, db)

    # some validations spesific to serial simulation
    self._data_period = self.config.get_data_period()
    self._check_type(self._data_period, SinglePeriodTimeIndex)

    start_year, num_years = self.config.get_weather_years()
    self._avg_year_range = AverageYearRange(start_year=start_year, num_years=num_years)
    if self._avg_year_range.is_52_week_years():
        message = "Expected AverageYearRange to have is_52_week_years() == False."
        raise ValueError(message)

    self._is_float32 = self.config.is_float32()

get_attribute_level(root_id: str, attribute: LevelProfile, target_unit: str | None) -> float

Get level value.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_attribute_level(
    self,
    root_id: str,
    attribute: LevelProfile,
    target_unit: str | None,
) -> float:
    """Get level value."""
    is_max_level = False
    value = attribute.get_data_value(
        db=self.db,
        unit=target_unit,
        level_period=self._data_period,
        scenario_horizon=self._avg_year_range,
        is_max_level=is_max_level,
    )
    if value < 0:
        self.send_warning_event(f"Attribute {root_id} of type {type(attribute).__name__} returned {value}")
    return value

get_attribute_profile(root_id: str, attribute: LevelProfile | Arrow, default: float, unit: str | None, info: ComponentInfo) -> str | float

Add profile vector to timevectors and return profile_id.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_attribute_profile(
    self,
    root_id: str,
    attribute: LevelProfile | Arrow,
    default: float,
    unit: str | None,
    info: ComponentInfo,
) -> str | float:
    """Add profile vector to timevectors and return profile_id."""
    if not attribute.has_profile():
        return default

    profile_id = f"{root_id}_profile"

    timeindex = self._find_profile_timeindex(attribute, info.has_storage_resolution)

    vector: NDArray = attribute.get_scenario_vector(
        db=self.db,
        level_period=self._data_period,
        scenario_horizon=timeindex,
        is_float32=self._is_float32,
        unit=unit,
    )
    vector = np.round(vector, decimals=6)  # avoid very small numerical noise, e.g negative prices from jules run

    if np.isnan(vector).any():
        message = (
            f"Profile {profile_id} in time index {timeindex} contains NaN values. "
            "This may indicate a problem with the data or the configuration."
        )
        raise ValueError(message)

    if (vector < 0).any():
        message = (
            f"Profile {profile_id} in time index {timeindex} has negative values. "
            "This may indicate a problem with the data or the configuration."
        )
        raise ValueError(message)

    denominator = vector.mean()

    if denominator == 0:
        vector.fill(self._PROFILE_VALUE_WHEN_ZERO)
    else:
        np.multiply(vector, 1 / denominator, out=vector)

    self.timevectors[timeindex][profile_id] = vector

    return profile_id

get_capacity_level(root_id: str, capacity: StockVolume | FlowVolume, info: ComponentInfo) -> str | float

get_capacity_level for serial simulation. Handles stock or flow based on info.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_capacity_level(self, root_id: str, capacity: StockVolume | FlowVolume, info: ComponentInfo) -> str | float:
    """get_capacity_level for serial simulation. Handles stock or flow based on info."""
    return self.get_attribute_level(root_id, capacity, info.unit_flow if info.is_flow else info.unit_stock)

get_capacity_profile(root_id: str, capacity: FlowVolume | StockVolume, info: ComponentInfo) -> str | float

get_capacity_profile for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_capacity_profile(
    self,
    root_id: str,
    capacity: FlowVolume | StockVolume,
    info: ComponentInfo,
) -> str | float:
    """get_capacity_profile for serial simulation."""
    unit = info.unit_flow if info.is_flow else info.unit_stock
    return self.get_attribute_profile(root_id, capacity, 1.0, unit, info)

get_coefficient_level(root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float

get_coefficient_level for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_coefficient_level(self, root_id: str, arrow: Arrow, info: ComponentInfo) -> str | float:
    """get_coefficient_level for serial simulation."""
    return self.get_attribute_level(root_id, arrow, info.unit_coeffs[arrow.get_node()])

get_cost_term_level(root_id: str, cost_term: Cost, info: ComponentInfo) -> float

get_cost_term_level for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_cost_term_level(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> float:
    """get_cost_term_level for serial simulation."""
    return self.get_attribute_level(root_id, cost_term, info.unit_cost)

get_cost_term_profile(root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float

get_cost_term_profile for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_cost_term_profile(self, root_id: str, cost_term: Cost, info: ComponentInfo) -> str | float:
    """get_cost_term_profile for serial simulation."""
    return self.get_attribute_profile(root_id, cost_term, 1.0, info.unit_cost, info)

get_price_level(root_id: str, price: Price, info: ComponentInfo) -> float

get_price_level for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_price_level(self, root_id: str, price: Price, info: ComponentInfo) -> float:
    """get_price_level for serial simulation."""
    return self.get_attribute_level(root_id, price, info.unit_price)

get_price_profile(root_id: str, price: Price, info: ComponentInfo) -> str | float

get_price_profile for serial simulation.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_price_profile(self, root_id: str, price: Price, info: ComponentInfo) -> str | float:
    """get_price_profile for serial simulation."""
    return self.get_attribute_profile(root_id, price, 1.0, info.unit_price, info)

get_rhs_term_level(rhs_term_id: str, flow: Flow, arrow: Arrow, flow_info: ComponentInfo) -> float

Convert volume (main node) to target node volume.

This may scale the volume using arrow coefficient, e.g. due to transportation loss.

This may also change unit, if target node belongs to different commodity than main node, such as for hydropower.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_rhs_term_level(
    self,
    rhs_term_id: str,
    flow: Flow,
    arrow: Arrow,
    flow_info: ComponentInfo,
) -> float:
    """Convert volume (main node) to target node volume.

    This may scale the volume using arrow coefficient,
    e.g. due to transportation loss.

    This may also change unit, if target node belongs to
    different commodity than main node, such as for hydropower.
    """
    coeff_value = self.get_coefficient_level(rhs_term_id, arrow, flow_info)

    volume = flow.get_volume()
    if volume.has_level():
        volume_value = self.get_attribute_level(
            root_id=rhs_term_id,
            attribute=volume,
            target_unit=flow_info.unit_flow,
        )
    else:
        max_cap = flow.get_max_capacity()
        min_cap = flow.get_min_capacity()
        if max_cap is not None and max_cap == min_cap:
            volume_value = self.get_attribute_level(
                root_id=rhs_term_id,
                attribute=max_cap,
                target_unit=flow_info.unit_flow,
            )
        else:
            message = f"{rhs_term_id} is not exogenous"
            raise ValueError(message)

    if coeff_value == 0:
        message = (
            f"Got zero coeff_value for {rhs_term_id}.\n"
            f"volume_value = {volume_value}\n"
            f"coeff_value = {coeff_value}\n"
            f"flow.get_main_node() = {flow.get_main_node()}\n"
            f"arrow.get_node() = {arrow.get_node()}\n"
        )
        raise RuntimeError(message)

    return volume_value / coeff_value  # convert from main_unit to target_unit

get_rhs_term_profile(rhs_term_id: str, flow: Flow, arrow: Arrow, flow_info: ComponentInfo) -> str | float

Create profile (possibly) representing volume_profile * coefficient_profile.

Source code in framjules/solve_handler/build_handler/SerialBuildHandler.py

def get_rhs_term_profile(
    self,
    rhs_term_id: str,
    flow: Flow,
    arrow: Arrow,
    flow_info: ComponentInfo,
) -> str | float:
    """Create profile (possibly) representing volume_profile * coefficient_profile."""
    volume = flow.get_volume()

    if not volume.has_profile():
        volume = flow.get_max_capacity()

    not_volume_profile = volume is None or not volume.has_profile()
    not_arrow_profile = not arrow.has_profile()
    has_volume_profile = not not_volume_profile
    has_arrow_profile = not not_arrow_profile

    if not_volume_profile and not_arrow_profile:
        return 1.0

    if has_volume_profile and not_arrow_profile:
        return self.get_attribute_profile(
            rhs_term_id,
            attribute=volume,
            default=1.0,
            unit=flow_info.unit_flow,
            info=flow_info,
        )

    if not_volume_profile and has_arrow_profile:
        unit = flow_info.unit_coeffs[arrow.get_node()]
        return self.get_attribute_profile(
            rhs_term_id,
            attribute=arrow,
            default=1.0,
            unit=unit,
            info=flow_info,
        )

    # Here we get both profiles (volume and coefficient) and muliply them
    # together and store the resulting product profile in self.timevectors

    profile_id = f"{rhs_term_id}_profile"

    timeindex = self._select_profile_timeindex(
        self._find_profile_timeindex(volume, flow_info.has_storage_resolution),
        self._find_profile_timeindex(arrow, flow_info.has_storage_resolution),
    )

    x: NDArray = volume.get_scenario_vector(
        db=self.db,
        level_period=self._data_period,
        scenario_horizon=timeindex,
        is_float32=self._is_float32,
        unit=flow_info.unit_flow,
    )

    y: NDArray = arrow.get_scenario_vector(
        db=self.db,
        level_period=self._data_period,
        scenario_horizon=timeindex,
        is_float32=self._is_float32,
        unit=flow_info.unit_coeffs[arrow.get_node()],
    )

    x_mean = x.mean()
    y_mean = y.mean()
    if x_mean == 0 or y_mean == 0:
        x.fill(self._PROFILE_VALUE_WHEN_ZERO)
        return x

    np.multiply(x, 1.0 / x_mean, out=x)
    np.multiply(y, 1.0 / y_mean, out=y)

    np.multiply(x, y, out=x)

    self.timevectors[timeindex][profile_id] = x

    return profile_id

dataclasses

ComponentInfo dataclass

All derived info we need during solve.

Source code in framjules/solve_handler/dataclasses.py

@dataclass
class ComponentInfo:
    """All derived info we need during solve."""

    is_flow: bool | None = None
    is_node: bool | None = None
    is_storage_node: bool | None = None
    is_short_term_storage: bool | None = None  # used to find if the whole subsystem should be short term
    is_market_node: bool | None = None
    is_market_flow: bool | None = None
    is_market_flow_to_exogenous: bool | None = None
    is_sss_member: bool | None = None
    is_exogenous: bool | None = None

    has_storage_resolution: bool | None = None

    jules_commodity: str | None = None
    domain_commodity: str | None = None

    # sss = storage subsystem
    sss_id: str | None = None
    sss_market_commodity: str | None = None
    sss_members: set[str] | None = None
    sss_is_short_term: int | None = None
    sss_storage_duration: timedelta | None = None
    sss_global_eneq_value: float | None = None
    sss_global_eneq_unit: str | None = None
    sss_initial_storage: float | None = None

    jules_balance_id: str | None = None
    jules_storage_id: str | None = None
    jules_global_eneq_id: str | None = None

    main_node_id: str | None = None
    num_arrows: int | None = None

    unit_price: str | None = None
    unit_stock: str | None = None
    unit_flow: str | None = None
    unit_flow_result: str | None = None
    unit_cost: str | None = None
    unit_coeffs: dict[str : str | None] | None = None
    unit_param_type: str | None = None
    unit_param_unit_flow: str | None = None
    unit_param_unit_stock: str | None = None

    agg_storage_node_id: str | None = None
    agg_market_node_id: str | None = None

DomainModels dataclass

Model instance for each term.

Source code in framjules/solve_handler/dataclasses.py

@dataclass
class DomainModels:
    """Model instance for each term."""

    clearing: Model
    short_term: Model
    medium_term: Model
    long_term: Model

GraphInfos dataclass

Hold all component info for all graphs.

Source code in framjules/solve_handler/dataclasses.py

@dataclass
class GraphInfos:
    """Hold all component info for all graphs."""

    clearing: dict[str, ComponentInfo]
    short_term: dict[str, ComponentInfo]
    medium_term: dict[str, ComponentInfo]
    long_term: dict[str, ComponentInfo]

NodeFlowGraphs dataclass

Node-Flow representation of domain model components via get_supported_components.

Source code in framjules/solve_handler/dataclasses.py

@dataclass
class NodeFlowGraphs:
    """Node-Flow representation of domain model components via get_supported_components."""

    clearing: dict[str, Node | Flow]
    short_term: dict[str, Node | Flow]
    medium_term: dict[str, Node | Flow]
    long_term: dict[str, Node | Flow]

results_handler

SerialResultsHandler

Handling of results produced by running JulES in Serial mode.

SerialResultsHandler

Set serial simulation results.

Source code in framjules/solve_handler/results_handler/SerialResultsHandler.py

class SerialResultsHandler:
    """Set serial simulation results."""

    def __init__(
        self,
        folder: Path | str,
        config: JulESConfig,
        names: JulESNames,
        graphs: NodeFlowGraphs,
        graph_infos: GraphInfos,
    ) -> None:
        """Handle retrieval of results from a JulES Serial simulation.

        Args:
            folder (Path | str): Path to folder of the JulES simulation.
            config (JulESConfig): Simulation config.
            names (JulESNames): JulES namespace.
            graphs (NodeFlowGraphs): Graphs used in the simulation.
            graph_infos (GraphInfos): JulES specific info of each Component in the graphs.

        """
        self._folder = Path(folder)
        self._config = config
        self._names = names

        self.graphs = graphs
        self.graph_infos = graph_infos

        self.units = self._set_units()

    def set_results(self) -> None:
        """Set JulES results of all Components in the clearing graph."""
        is_whole_years = self._config.is_simulation_mode_serial()

        loader = JulESH5TimeVectorLoader(
            source=self._folder,
            units=self.units,
            relative_loc=self._names.FILENAME_H5_OUTPUT,
            is_whole_years=is_whole_years,
        )

        supply_loader = SupplyJulESH5TimeVectorLoader(
            source=self._folder,
            units=self.units,
            relative_loc=self._names.FILENAME_H5_OUTPUT,
            is_whole_years=is_whole_years,
        )

        demand_loader = DemandJulESH5TimeVectorLoader(
            source=self._folder,
            units=self.units,
            relative_loc=self._names.FILENAME_H5_OUTPUT,
            is_whole_years=is_whole_years,
        )

        power_nodes = [
            name
            for name, c in self.graphs.clearing.items()
            if isinstance(c, Node) and c.get_commodity() == JulESNames.POWER
        ]
        for name, c in self.graphs.clearing.items():
            info: ComponentInfo = self.graph_infos.clearing[name]

            if isinstance(c, Node):
                if c.is_exogenous():
                    continue
                self._set_node_results(c, name, loader)
            if isinstance(c, Flow):
                self._set_flow_results(
                    c,
                    info,
                    name,
                    loader,
                    power_nodes,
                    supply_loader=supply_loader,
                    demand_loader=demand_loader,
                )

    def _set_node_results(self, node: Node, name: str, loader: JulESH5TimeVectorLoader) -> None:
        info = self.graph_infos.clearing[name]

        if info.is_market_node:
            level, profile = self._get_decomposed_level_profile(name, loader, self.units[name])
            price = node.get_price()
            price.clear()
            price.set_level(level)
            price.set_profile(profile)

        if info.is_storage_node:
            level, profile = self._get_decomposed_level_profile(
                info.jules_storage_id,
                loader,
                self.units[info.jules_storage_id],
                is_stock=True,
            )
            storage = node.get_storage()
            volume = storage.get_volume()
            volume.clear()
            volume.set_level(level)
            volume.set_profile(profile)

            level, profile = self._get_decomposed_level_profile(
                info.jules_storage_id + "_sv",
                loader,
                self.units[info.jules_storage_id + "_sv"],
            )
            price = node.get_price()
            price.clear()
            price.set_level(level)
            price.set_profile(profile)

    def _set_flow_results(
        self,
        flow: Flow,
        info: ComponentInfo,
        name: str,
        loader: JulESH5TimeVectorLoader,
        power_nodes: list[str],
        supply_loader: SupplyJulESH5TimeVectorLoader,
        demand_loader: DemandJulESH5TimeVectorLoader,
    ) -> None:
        if info.is_exogenous and flow.get_volume().get_level():
            return
        if info.is_exogenous:
            max_capacity = flow.get_max_capacity()
            level = max_capacity.get_level()
            profile = max_capacity.get_profile()
        else:
            level, profile = self._get_decomposed_level_profile(
                name,
                loader,
                self.units[name],
                is_flow=True,
            )
        volume = flow.get_volume()
        volume.clear()
        volume.set_level(level)
        volume.set_profile(profile)

        # temporary workaround, will be fixed in future versions
        n = len(name)
        for arrow, volume in flow.get_arrow_volumes().items():
            arrow_node = arrow.get_node()
            if arrow_node not in power_nodes:
                continue

            is_supply = arrow.is_ingoing()
            loader = supply_loader if is_supply else demand_loader

            unit_flow = self.units[name]
            unit_coeff = info.unit_coeffs[arrow.get_node()]
            unit = f"({unit_flow})*({unit_coeff})" if unit_coeff is not None else unit_flow

            for i in range(n):
                subname = name[: n - i]
                level, profile = self._get_decomposed_level_profile(
                    subname,
                    loader,
                    unit,
                    is_flow=True,
                )
                volume.clear()
                volume.set_level(level)
                volume.set_profile(profile)
                break

    def _get_decomposed_level_profile(
        self,
        name: str,
        loader: JulESH5TimeVectorLoader,
        unit: str | None = None,
        is_flow: bool = False,
        is_stock: bool = False,
    ) -> tuple[Expr, Expr]:
        """Decompose result vector into level and profile expressions.

        Note! Support for negative prices will come in the next version.
        """
        timevector = self._get_timevector(jules_id=name, loader=loader)

        mean_value = timevector.get_vector(self._config.is_float32()).mean()

        scale = float(1 / mean_value) if mean_value != 0 else 1.0
        mean_one_profile_timevector = LinearTransformTimeVector(
            timevector=timevector,
            scale=scale,
            shift=0.0,
            unit=None,
            is_zero_one_profile=False,
        )

        profile_expr = None
        reference_period = None
        if mean_value != 0:
            reference_period = self._get_reference_period()
            profile_expr = ensure_expr(mean_one_profile_timevector, is_profile=True)

            level_timevector = ConstantTimeVector(
                scalar=mean_value,
                unit=unit,
                is_max_level=False,
                reference_period=reference_period,
            )
        else:
            level_timevector = ConstantTimeVector(
                scalar=0.0,
                unit=unit,
                is_max_level=True,
            )
            profile_expr = ensure_expr(
                ConstantTimeVector(
                    scalar=1.0,
                    is_zero_one_profile=True,
                ),
                is_profile=True,
            )

        level_expr = ensure_expr(
            level_timevector,
            is_level=True,
            is_flow=is_flow,
            is_stock=is_stock,
            profile=profile_expr,
        )

        return level_expr, profile_expr

    def _set_units(self) -> dict[str : str | None]:
        units = {name: info.unit_price for name, info in self.graph_infos.clearing.items() if info.is_market_node}
        units.update(
            {
                info.jules_storage_id: info.unit_stock
                for info in self.graph_infos.clearing.values()
                if info.is_storage_node
            },
        )
        units.update(
            {
                info.jules_storage_id + "_sv": info.unit_cost
                for info in self.graph_infos.clearing.values()
                if info.is_storage_node and info.unit_cost is not None
            },
        )
        units.update(
            {name: info.unit_flow_result for name, info in self.graph_infos.clearing.items() if info.is_flow},
        )
        return units

    def _get_reference_period(self) -> ReferencePeriod:
        first_year, num_years = self._config.get_simulation_years()
        return ReferencePeriod(first_year, num_years)

    def _get_timevector(self, jules_id: str, loader: JulESH5TimeVectorLoader) -> LoadedTimeVector:
        try:
            return LoadedTimeVector(vector_id=jules_id, loader=loader)
        except Exception:
            raise AssertionError from None

__init__(folder: Path | str, config: JulESConfig, names: JulESNames, graphs: NodeFlowGraphs, graph_infos: GraphInfos) -> None

Handle retrieval of results from a JulES Serial simulation.

Parameters:

Name	Type	Description	Default
folder	Path | str	Path to folder of the JulES simulation.	required
config	JulESConfig	Simulation config.	required
names	JulESNames	JulES namespace.	required
graphs	NodeFlowGraphs	Graphs used in the simulation.	required
graph_infos	GraphInfos	JulES specific info of each Component in the graphs.	required

Source code in framjules/solve_handler/results_handler/SerialResultsHandler.py

def __init__(
    self,
    folder: Path | str,
    config: JulESConfig,
    names: JulESNames,
    graphs: NodeFlowGraphs,
    graph_infos: GraphInfos,
) -> None:
    """Handle retrieval of results from a JulES Serial simulation.

    Args:
        folder (Path | str): Path to folder of the JulES simulation.
        config (JulESConfig): Simulation config.
        names (JulESNames): JulES namespace.
        graphs (NodeFlowGraphs): Graphs used in the simulation.
        graph_infos (GraphInfos): JulES specific info of each Component in the graphs.

    """
    self._folder = Path(folder)
    self._config = config
    self._names = names

    self.graphs = graphs
    self.graph_infos = graph_infos

    self.units = self._set_units()

set_results() -> None

Set JulES results of all Components in the clearing graph.

Source code in framjules/solve_handler/results_handler/SerialResultsHandler.py

def set_results(self) -> None:
    """Set JulES results of all Components in the clearing graph."""
    is_whole_years = self._config.is_simulation_mode_serial()

    loader = JulESH5TimeVectorLoader(
        source=self._folder,
        units=self.units,
        relative_loc=self._names.FILENAME_H5_OUTPUT,
        is_whole_years=is_whole_years,
    )

    supply_loader = SupplyJulESH5TimeVectorLoader(
        source=self._folder,
        units=self.units,
        relative_loc=self._names.FILENAME_H5_OUTPUT,
        is_whole_years=is_whole_years,
    )

    demand_loader = DemandJulESH5TimeVectorLoader(
        source=self._folder,
        units=self.units,
        relative_loc=self._names.FILENAME_H5_OUTPUT,
        is_whole_years=is_whole_years,
    )

    power_nodes = [
        name
        for name, c in self.graphs.clearing.items()
        if isinstance(c, Node) and c.get_commodity() == JulESNames.POWER
    ]
    for name, c in self.graphs.clearing.items():
        info: ComponentInfo = self.graph_infos.clearing[name]

        if isinstance(c, Node):
            if c.is_exogenous():
                continue
            self._set_node_results(c, name, loader)
        if isinstance(c, Flow):
            self._set_flow_results(
                c,
                info,
                name,
                loader,
                power_nodes,
                supply_loader=supply_loader,
                demand_loader=demand_loader,
            )

run_handler

SerialRunHandler

SerialRunHandler

Bases: JuliaModel

Handle running the JulES solver in serial simulation mode.

Source code in framjules/solve_handler/run_handler/SerialRunHandler.py

class SerialRunHandler(JuliaModel):
    """Handle running the JulES solver in serial simulation mode."""

    # ENV_NAME = "JulES_julia_env"

    def __init__(
        self,
        folder: Path,
        config: JulESConfig,
        names: JulESNames,
        dependencies: list[str | tuple[str, str | None]] | None = None,
    ) -> None:
        """Initialize JulES serial folder.

        The three parameters env_path, depot_path and julia_path sets environment variables for locations of your Julia
        environment, packages and language.
            - If user has not specified locations, the default is to use the current python/conda environment.
            - If a system installation of Python is used, the default is set to the current user location.

        Args:
            folder (Path): Location of JulES model dataset.
            config (JulESConfig): Simulaiton config.
            names (JulESNames): JulES namespace object.
            dependencies (list[str]): Julia packages dependencies. List of str, either package names or urls.

        """
        self._folder = folder
        self._config = config
        self._names = names
        self.ENV_NAME = self._names.JULIA_ENV_NAME
        super().__init__(
            julia_path=self._config.get_julia_exe_path(),
            env_path=self._config.get_julia_env_path(),
            depot_path=self._config.get_julia_depot_path(),
            dependencies=dependencies,
            skip_install_dependencies=config.is_skip_install_dependencies(),
            force_julia_install=config.get_force_julia_install(),
        )

    def run(self) -> None:
        """Run JulES in Series mode."""
        data_year = self._config.get_data_period().get_start_time().isocalendar().year
        weather_year = self._config.get_weather_years()[0]

        config_path = self._folder / self._names.JULES_CONFIG
        output_path = self._folder / self._names.FILENAME_H5_OUTPUT

        def get_all_attrs(obj) -> dict:  # noqa: ANN001
            result = {}
            result.update(
                {k: v for k, v in obj.__class__.__dict__.items() if not k.startswith("__") and not callable(v)},
            )
            result.update(obj.__dict__)
            return result

        names_dict = get_all_attrs(self._names)
        filename_clearing = f"{self._names.ROOT_FILENAME_DATAELEMENTS}_{self._names.CLEARING}.json"
        filename_aggregated = f"{self._names.ROOT_FILENAME_DATAELEMENTS}_{self._names.AGGREGATED}.json"

        self._jl.seval(f"""
        using Distributed, YAML, HDF5
        config = YAML.load_file(\"{config_path.as_posix()}\")
        println("Add cores")
        const numcores = config["main"]["numcores"]
        if nprocs() < numcores
            addprocs(numcores - nprocs())
        end
        @show nprocs()
        println("Load JulES")
        @time @everywhere using JulES
        using Pkg; Pkg.status()
        """)

        self._jl.JulES.run_jules(
            config_path.as_posix(),
            data_year,
            weather_year,
            output_path.as_posix(),
            names_dict,
            filename_clearing,
            filename_aggregated,
        )

__init__(folder: Path, config: JulESConfig, names: JulESNames, dependencies: list[str | tuple[str, str | None]] | None = None) -> None

Initialize JulES serial folder.

The three parameters env_path, depot_path and julia_path sets environment variables for locations of your Julia environment, packages and language. - If user has not specified locations, the default is to use the current python/conda environment. - If a system installation of Python is used, the default is set to the current user location.

Parameters:

Name	Type	Description	Default
folder	Path	Location of JulES model dataset.	required
config	JulESConfig	Simulaiton config.	required
names	JulESNames	JulES namespace object.	required
dependencies	list[str]	Julia packages dependencies. List of str, either package names or urls.	None

Source code in framjules/solve_handler/run_handler/SerialRunHandler.py

def __init__(
    self,
    folder: Path,
    config: JulESConfig,
    names: JulESNames,
    dependencies: list[str | tuple[str, str | None]] | None = None,
) -> None:
    """Initialize JulES serial folder.

    The three parameters env_path, depot_path and julia_path sets environment variables for locations of your Julia
    environment, packages and language.
        - If user has not specified locations, the default is to use the current python/conda environment.
        - If a system installation of Python is used, the default is set to the current user location.

    Args:
        folder (Path): Location of JulES model dataset.
        config (JulESConfig): Simulaiton config.
        names (JulESNames): JulES namespace object.
        dependencies (list[str]): Julia packages dependencies. List of str, either package names or urls.

    """
    self._folder = folder
    self._config = config
    self._names = names
    self.ENV_NAME = self._names.JULIA_ENV_NAME
    super().__init__(
        julia_path=self._config.get_julia_exe_path(),
        env_path=self._config.get_julia_env_path(),
        depot_path=self._config.get_julia_depot_path(),
        dependencies=dependencies,
        skip_install_dependencies=config.is_skip_install_dependencies(),
        force_julia_install=config.get_force_julia_install(),
    )

run() -> None

Run JulES in Series mode.

Source code in framjules/solve_handler/run_handler/SerialRunHandler.py

def run(self) -> None:
    """Run JulES in Series mode."""
    data_year = self._config.get_data_period().get_start_time().isocalendar().year
    weather_year = self._config.get_weather_years()[0]

    config_path = self._folder / self._names.JULES_CONFIG
    output_path = self._folder / self._names.FILENAME_H5_OUTPUT

    def get_all_attrs(obj) -> dict:  # noqa: ANN001
        result = {}
        result.update(
            {k: v for k, v in obj.__class__.__dict__.items() if not k.startswith("__") and not callable(v)},
        )
        result.update(obj.__dict__)
        return result

    names_dict = get_all_attrs(self._names)
    filename_clearing = f"{self._names.ROOT_FILENAME_DATAELEMENTS}_{self._names.CLEARING}.json"
    filename_aggregated = f"{self._names.ROOT_FILENAME_DATAELEMENTS}_{self._names.AGGREGATED}.json"

    self._jl.seval(f"""
    using Distributed, YAML, HDF5
    config = YAML.load_file(\"{config_path.as_posix()}\")
    println("Add cores")
    const numcores = config["main"]["numcores"]
    if nprocs() < numcores
        addprocs(numcores - nprocs())
    end
    @show nprocs()
    println("Load JulES")
    @time @everywhere using JulES
    using Pkg; Pkg.status()
    """)

    self._jl.JulES.run_jules(
        config_path.as_posix(),
        data_year,
        weather_year,
        output_path.as_posix(),
        names_dict,
        filename_clearing,
        filename_aggregated,
    )