from __future__ import annotations
import logging
import os
from pathlib import Path
from typing import Literal
from typing import Optional
from typing import TYPE_CHECKING
from typing import Union
from gams import GamsOptions
from gams import GamsWorkspace
from gams import SymbolUpdateType
from pydantic import BaseModel
logger = logging.getLogger("Options")
logger.setLevel(logging.INFO)
formatter = logging.Formatter("[%(name)s - %(levelname)s] %(message)s")
handler = logging.StreamHandler()
handler.setFormatter(formatter)
handler.setLevel(logging.INFO)
logger.addHandler(handler)
if TYPE_CHECKING:
import io
multi_solve_map = {"replace": 0, "merge": 1, "clear": 2}
# GAMSPy to GAMS Control mapping
option_map = {
"cns": "cns",
"dnlp": "dnlp",
"emp": "emp",
"lp": "lp",
"mcp": "mcp",
"minlp": "minlp",
"mip": "mip",
"miqcp": "miqcp",
"mpec": "mpec",
"nlp": "nlp",
"qcp": "qcp",
"rminlp": "rminlp",
"rmip": "rmip",
"rmiqcp": "rmiqcp",
"rmpec": "rmpec",
"allow_suffix_in_equation": "suffixalgebravars",
"allow_suffix_in_limited_variables": "suffixdlvars",
"basis_detection_threshold": "bratio",
"compile_error_limit": "cerr",
"domain_violation_limit": "domlim",
"job_time_limit": "etlim",
"job_heap_limit": "heaplimit",
"hold_fixed_variables": "holdfixed",
"integer_variable_upper_bound": "intvarup",
"iteration_limit": "iterlim",
"keep_temporary_files": "keep",
"license": "license",
"listing_file": "output",
"log_file": "_logfile",
"variable_listing_limit": "limcol",
"equation_listing_limit": "limrow",
"node_limit": "nodlim",
"absolute_optimality_gap": "optca",
"relative_optimality_gap": "optcr",
"profile": "profile",
"profile_tolerance": "profiletol",
"time_limit": "reslim",
"savepoint": "savepoint",
"seed": "seed",
"report_solution": "solprint",
"show_os_memory": "showosmemory",
"solver_link_type": "solvelink",
"merge_strategy": "solveopt",
"step_summary": "stepsum",
"suppress_compiler_listing": "suppress",
"report_solver_status": "sysout",
"threads": "threads",
"trace_file": "trace",
"trace_file_format": "traceopt",
"write_listing_file": "_writeoutput",
"zero_rounding_threshold": "zerores",
"report_underflow": "zeroresrep",
}
[docs]
class Options(BaseModel):
cns: Optional[str] = None
dnlp: Optional[str] = None
emp: Optional[str] = None
lp: Optional[str] = None
mcp: Optional[str] = None
minlp: Optional[str] = None
mip: Optional[str] = None
miqcp: Optional[str] = None
mpec: Optional[str] = None
nlp: Optional[str] = None
qcp: Optional[str] = None
rminlp: Optional[str] = None
rmip: Optional[str] = None
rmiqcp: Optional[str] = None
rmpec: Optional[str] = None
allow_suffix_in_equation: Optional[bool] = None
allow_suffix_in_limited_variables: Optional[bool] = None
basis_detection_threshold: Optional[float] = None
compile_error_limit: int = 1
domain_violation_limit: Optional[int] = None
job_time_limit: Optional[float] = None
job_heap_limit: Optional[float] = None
hold_fixed_variables: Optional[bool] = None
integer_variable_upper_bound: Optional[int] = None
iteration_limit: Optional[int] = None
keep_temporary_files: bool = False
license: Optional[str] = None
listing_file: Optional[str] = None
log_file: Optional[str] = None
variable_listing_limit: Optional[int] = None
equation_listing_limit: Optional[int] = None
node_limit: Optional[int] = None
absolute_optimality_gap: Optional[float] = None
relative_optimality_gap: Optional[float] = None
profile: Optional[int] = None
profile_tolerance: Optional[float] = None
time_limit: Optional[float] = None
savepoint: Optional[Literal[0, 1, 2, 3, 4]] = None
seed: Optional[int] = None
report_solution: Literal[0, 1, 2] = 2
show_os_memory: Literal[0, 1, 2] = 0
solver_link_type: Optional[Literal[0, 1, 2, 3, 4, 5, 6, 7]] = None
merge_strategy: Optional[Literal["replace", "merge", "clear"]] = None
step_summary: Optional[bool] = None
suppress_compiler_listing: bool = False
report_solver_status: Optional[bool] = None
threads: Optional[int] = None
trace_file: Optional[str] = None
trace_file_format: Optional[Literal[0, 1, 2, 3, 4, 5]] = None
write_listing_file: bool = True
zero_rounding_threshold: Optional[float] = None
report_underflow: Optional[bool] = None
def _get_gams_compatible_options(self):
options_dict = self.model_dump()
if options_dict["allow_suffix_in_equation"] is not None:
allows_suffix = options_dict["allow_suffix_in_equation"]
options_dict["allow_suffix_in_equation"] = (
"on" if allows_suffix else "off"
)
if options_dict["allow_suffix_in_limited_variables"] is not None:
allows_suffix = options_dict["allow_suffix_in_limited_variables"]
options_dict["allow_suffix_in_limited_variables"] = (
"on" if allows_suffix else "off"
)
if options_dict["merge_strategy"] is not None:
strategy = options_dict["merge_strategy"]
options_dict["merge_strategy"] = multi_solve_map[strategy]
if options_dict["listing_file"] is not None:
os.makedirs(Path(options_dict["listing_file"]).parent.absolute(), exist_ok=True)
if not os.path.isabs(options_dict["listing_file"]):
options_dict["listing_file"] = os.path.abspath(options_dict["listing_file"])
if options_dict["log_file"] is not None:
os.makedirs(Path(options_dict["log_file"]).parent.absolute(), exist_ok=True)
if not os.path.isabs(options_dict["log_file"]):
options_dict["log_file"] = os.path.abspath(options_dict["log_file"])
options_dict = {
option_map[key]: value for key, value in options_dict.items() # type: ignore
}
return options_dict
def _fix_log_option(
output: Union[io.TextIOWrapper, None],
create_log_file: bool,
options: GamsOptions,
) -> GamsOptions:
if output is None:
options._logoption = 2 if create_log_file else 0
else:
options._logoption = 4 if create_log_file else 3
return options
def _set_options(
gams_options: GamsOptions,
options: Options,
is_seedable: bool = True,
):
options_dict = options._get_gams_compatible_options()
for option, value in options_dict.items():
if value is not None:
if option == "seed" and not is_seedable:
continue
setattr(gams_options, option.lower(), value)
return gams_options
def _set_trace_options(
gams_options: GamsOptions,
options: Optional[Options],
backend: str,
workspace: GamsWorkspace,
):
if options is not None:
if options.trace_file_format is not None:
if options.trace_file_format != 3:
logger.log(
logging.INFO,
"Trace file format is different than 3. GAMSPy will not"
" return any summary!",
)
trace_option = options.trace_file_format
else:
trace_option = 3
if options.trace_file is None:
trace_path = (
"trace.txt"
if backend in ["engine", "neos"]
else os.path.join(workspace.working_directory, "trace.txt")
)
else:
trace_path = (
options.trace_file
if backend in ["engine", "neos"]
else os.path.abspath(options.trace_file)
)
else:
trace_option = 3
trace_path = (
"trace.txt"
if backend in ["engine", "neos"]
else os.path.join(workspace.working_directory, "trace.txt")
)
gams_options.trace = trace_path
gams_options.traceopt = trace_option
return gams_options
def _map_options(
workspace: GamsWorkspace,
backend: str = "local",
options: Union[Options, None] = None,
global_options: Union[Options, None] = None,
is_seedable: bool = True,
output: Optional[io.TextIOWrapper] = None,
create_log_file: bool = False,
) -> GamsOptions:
"""
Maps given GAMSPy options to GamsOptions
Parameters
----------
options : Options | None
GAMSPy options
global_options : Options | None
Global options
is_seedable : bool, optional
only seedable at first run or in model.solve function, by default True
Returns
-------
GamsOptions
"""
gams_options = GamsOptions(workspace)
if global_options is not None:
gams_options = _set_options(
gams_options,
global_options,
is_seedable,
)
if options is not None:
gams_options = _set_options(gams_options, options, is_seedable)
gams_options = _set_trace_options(
gams_options, options, backend, workspace
)
gams_options = _fix_log_option(output, create_log_file, gams_options)
gams_options.previouswork = 1 # In case GAMS version differs on backend
return gams_options
update_type_map = {
"0": SymbolUpdateType.Zero,
"base_case": SymbolUpdateType.BaseCase,
"accumulate": SymbolUpdateType.Accumulate,
"inherit": SymbolUpdateType._Inherit,
}
[docs]
class ModelInstanceOptions(BaseModel):
solver: Optional[str] = None
opt_file: int = -1
no_match_limit: int = 0
debug: bool = False
update_type: Literal["0", "base_case", "accumulate", "inherit"] = (
"base_case"
)
def items(self):
dictionary = self.model_dump()
dictionary["update_type"] = update_type_map[dictionary["update_type"]]
return dictionary
