import logging
import shutil
from pathlib import Path
import yaml
[docs]
logger = logging.getLogger(__name__)
[docs]
class TulipaConfigError(Exception):
def __init__(self, msg):
"""Exception raised when the TulipaConfig configuration has issues.
Args:
msg: Explanation of the error.
"""
super().__init__(self.message)
[docs]
class _TulipaConfig:
def __init__(
self,
input_data_folder: Path | str,
julia_environment: Path | str,
communication_folder: Path | str,
db_path: Path | str,
debug_input_folder: Path | str | None = None,
):
"""Container for Tulipa configuration.
The container is used inside Annular to pass data from the simulation
configuration to Tulipa. It contains configuration from the parameters
defined via the `tulipa_config_file` key in the muscle3
configuration file, and additional data for internal logic.
Arguments:
input_data_folder: Path to data defining the Tulipa model.
julia_environment: Path to Julia project environment, passed
to Julia as the `--project` argument.
communication_folder: Folder that stores data that are sent between
annular and Tulipa. Needs to be a new folder, and it is deleted at
cleanup.
db_path: Path to the database file.
debug_input_folder: Path to folder to keep database tables as csv files
for debugging. Should exist inside communication_folder.
Details:
Data for communication (config file, input data, duckdb file) are
deleted at cleanup. However, if debug_input_folder is given,
the communication folder is persisted after cleanup.
"""
[docs]
self.julia_environment = Path(julia_environment)
[docs]
self.communication_folder = Path(communication_folder)
[docs]
self.db_path = Path(db_path)
@classmethod
[docs]
def from_config(
cls,
input_data_folder: Path | str,
self_base_folder: Path | str,
julia_environment: Path | str,
communication_folder: Path | str | None = None,
iteration_id: None | int = None,
db_file: str = "tulipa.duckdb",
):
"""Create a _TulipaConfig class from a configuration file.
Arguments:
input_data_folder: Path to data defining the Tulipa model, interpreted
relative to `self_base_folder`.
self_base_folder: The path to the `tulipa_config_file`. Used internally
to find the path to the `input_data_folder`.
julia_environment: Path to Julia project environment, passed
to Julia as the `--project` argument. Interpreted relative
to `self_base_folder`.
communication_folder: Folder that stores data that are sent between
Annular and Tulipa. Needs to be a new folder, and it is deleted at
cleanup. By default, a directory `./.tulipa-communication-0`
is used in the location of the main python process.
Otherwise, the path is interpreted relative to the location of
`self_base_folder`.
iteration_id: Identifier of the iteration. If given, used for creating
unique communication folders for each invocation of the market model
in a multi-period annular simulation.
db_file: Name of the database file inside `communication_folder`.
Raises:
TulipaConfigError: When (i) `input_data_folder` does not exist,
(ii) `communication_folder` already exists, or (iii) the provided
project environment for Julia cannot be found.
"""
self_base_folder = Path(self_base_folder)
input_data_folder = (self_base_folder / Path(input_data_folder)).resolve()
if not input_data_folder.exists():
raise TulipaConfigError("cannot find input_data_folder.")
julia_environment = cls._validate_julia_env(Path(julia_environment), self_base_folder)
use_communication_folder_relative_to_base = True
if not communication_folder:
use_communication_folder_relative_to_base = False
communication_folder = ".tulipa-communication"
if iteration_id is not None:
communication_folder = f"{communication_folder}-{iteration_id}"
communication_folder = Path(communication_folder)
if use_communication_folder_relative_to_base:
communication_folder = (self_base_folder / communication_folder).resolve()
try:
communication_folder.mkdir(parents=True)
except FileExistsError as exc:
raise TulipaConfigError("communication folder already exists") from exc
db_path = communication_folder / db_file
debug_folder = None
if logger.isEnabledFor(logging.DEBUG):
debug_folder = communication_folder / "debug/"
debug_folder.mkdir(exist_ok=True)
return cls(input_data_folder, julia_environment, communication_folder, db_path, debug_folder)
@property
[docs]
def config_file(self) -> Path:
"""Yaml file for communicating settings from python to Julia."""
return self.communication_folder / "config.yaml"
@staticmethod
[docs]
def _validate_julia_env(environment: Path, base_folder: Path) -> Path:
if str(environment).startswith("@"):
return environment
environment = (base_folder / environment).resolve()
if not environment.exists():
msg = f"Project {environment} should be a global environment `@SOMETHING` or a path"
raise TulipaConfigError(msg)
if not Path(environment / "Manifest.toml").is_file():
msg = f"'Manifest.toml' not found in '{environment}', make sure to instantiate the environment"
raise TulipaConfigError(msg)
return environment
[docs]
def write_to_yaml(self):
"""Write config to yaml, converting Path objects to strings."""
tulipa_config_dict = {}
for config_key, config_value in dict(vars(self)).items():
if isinstance(config_value, (Path, str)):
tulipa_config_dict[config_key] = str(config_value)
with self.config_file.open("w") as file:
yaml.dump(tulipa_config_dict, file)
[docs]
def cleanup(self):
"""Remove database and temp dir."""
logger.debug("Deleting %s", self.db_path)
self.db_path.unlink(missing_ok=True)
if self.debug_input_folder is not None:
logger.debug("Keeping data in %s", self.communication_folder)
return
shutil.rmtree(self.communication_folder)