import logging
import shutil
import subprocess
from importlib.resources import files
from pathlib import Path
import duckdb
import numpy as np
import pandas as pd
from .build_model import build_model_with_bids, postprocess_tulipa_model
from .config import TulipaConfigError, _TulipaConfig
from .utils import convert_data_for_tulipa, read_outputs_from_db, write_inputs_to_db
[docs]
logger = logging.getLogger(__name__)
[docs]
def run_julia_subprocess(tulipa_config: _TulipaConfig) -> None:
"""Run Julia subprocess with Tulipa configuration.
Args:
tulipa_config: :class:`_TulipaConfig` instance with paths and settings.
Raises:
RuntimeError: If Julia subprocess fails with non-zero exit code.
"""
tulipa_script = files("annular.market.tulipa").joinpath("tulipa.jl")
logger.info("Running Julia via subprocess")
proc = subprocess.run(
[
"julia",
f"--project={tulipa_config.julia_environment}",
tulipa_script,
str(tulipa_config.config_file),
],
capture_output=True,
text=True,
)
if proc.stdout != "":
logger.info(f"Julia output log:\n{proc.stdout}\n-------------------")
if proc.returncode != 0:
logger.info(f"Julia error log:\n{proc.stderr}")
raise RuntimeError(f"Julia subprocess failed with exit code {proc.returncode}")
[docs]
def run_market_model(
bids: pd.DataFrame,
timeseries_data: pd.DataFrame,
snapshots: pd.DatetimeIndex,
bidding_window: pd.DatetimeIndex,
output_path: Path,
iteration_id: int,
config: dict,
) -> tuple[np.ndarray, np.ndarray]:
"""Create the central market clearing model.
Args:
bids: MultiIndex DataFrame of bids for all satellites.
timeseries_data: DataFrame with timeseries_data for entire simulation time.
snapshots: Optimization window of the market model.
bidding_window: Bidding window that matters for bids.
output_path: Path to store the database file with the solved `TulipaEnergyModel`.
The file is named `tulipa_iteration-ID.duckdb` where `ID` is
the iteration ID (starting at 0).
iteration_id: Identifier of the iteration number. Used internally to
manage folder names.
config: config dict
Returns:
tuple: market clearing price and quantities allocated by the market.
Raises:
ValueError: If Tulipa configuration is incomplete.
RuntimeError:
- if Tulipa cofiguration caused other problems
- if the optimization problem is infeasible or unbounded
"""
optimization_window = snapshots # better naming
try:
config["iteration_id"] = iteration_id
tulipa_config = _TulipaConfig.from_config(**config)
except TypeError:
# This is supposed to capture missing positional arguments in config
msg = f"Could not initialize _TulipaConfig with provided configuration: {config}"
raise ValueError(msg)
except TulipaConfigError as exc:
msg = f"Error initializing tulipa_config: {exc.message}"
raise RuntimeError(msg) from exc
except Exception as exc:
msg = f"An unknown error occurred when initializing _TulipaConfig with configuration: {config}"
raise RuntimeError(msg) from exc
tulipa_config.write_to_yaml()
logger.info("tulipa_config: %s", tulipa_config)
bid_idx_cols = list(bids.index.names)
bids, timeseries_data, bidding_window, optimization_window_df = convert_data_for_tulipa(
bids, timeseries_data, optimization_window, bidding_window
)
with duckdb.connect(tulipa_config.db_path) as con:
write_inputs_to_db(con, bids, timeseries_data, optimization_window_df)
logger.info("Building TulipaEnergyModel tables.")
bid_manager, timestep_start = build_model_with_bids(tulipa_config)
run_julia_subprocess(tulipa_config)
logger.info("Postprocessing TulipaEnergyModel")
postprocess_tulipa_model(tulipa_config.db_path, bid_manager, timestep_start)
bid_idx_cols = [
col for col in bid_idx_cols if col != "sense"
] # `sense` is already summed over in the function below
with duckdb.connect(tulipa_config.db_path) as con:
market_price, bids = read_outputs_from_db(con, bidding_window, bid_idx_cols, bids)
logger.debug("market_price: \n%s", market_price)
logger.debug("scheduled_bids: \n%s", bids)
copy_path = output_path / f"tulipa_iteration-{iteration_id}.duckdb"
shutil.copy(tulipa_config.db_path, copy_path)
tulipa_config.cleanup()
return market_price.to_numpy().flatten(), bids.to_numpy().flatten()