import logging from logging import Logger LOG_FORMATTER = logging.Formatter( fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt="%Y-%m-%d %H:%M:%S" ) def setup_logger( name: str, log_file: str = None, ) -> Logger: """ general logger configurator :param name: logger name, generally the script or class name :param log_file: logger file name. If None, no file is written :return: the logger, with debug level in development (change into info in production) """ logger = logging.getLogger(name) # noinspection SpellCheckingInspection logging.basicConfig(format="%(asctime)s - %(name)-7s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S") logger.setLevel(logging.DEBUG) logger.handlers = [] logger.addHandler(get_console_handler()) if log_file: logger.addHandler(get_file_handler(log_file)) logger.propagate = False return logger def get_console_handler() -> logging.StreamHandler: console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(LOG_FORMATTER) return console_handler def get_file_handler(log_file: str) -> logging.FileHandler: file_handler = logging.FileHandler(log_file) file_handler.setFormatter(LOG_FORMATTER) return file_handler import pandas as pd import numpy as np import random import string import os import sys logger = setup_logger('TEST') TRN_PATH = '' TRN_ASSETS_PATH = os.path.join(TRN_PATH, 'assets') TRN_QUERIES_PATH = os.path.join(TRN_PATH, 'queries') DEMO_TYPE = 'POC_ETL' epsilon = 10 NUM_ITEMS = 50 NUM_MAX_CAPACITY = 1000 # Max capacity of each container NUM_SAFETY = int(NUM_MAX_CAPACITY * .2) # Safety margin for each container NUM_CONTAINERS_IN_UNIT = int(NUM_ITEMS * .1) # Number of containers in each unit WEIGHT_TOTAL_COST = 1 # Weight for the total cost in the objective function WEIGHT_USED_CONTAINERS = 1 # Weight for the number of used containers WEIGHT_ABS_SAFETY_DISTANCE = 1 # Weight for the absolute safety distance # This is a simple example of a bin-packing problem. The problem is to pack a set of items of different sizes into a set of bins of equal capacity. The objective is to minimize the number of bins used. The problem is formulated as a mixed-integer programming problem. def genera_stringa_alfanumerica(): caratteri = string.ascii_letters + string.digits # Lettere maiuscole, minuscole e numeri return ''.join(random.choices(caratteri, k=8)) def populate_trn(demo_type, id_run, num_items=10, logger=None): """ Crea un dataset di esempio con articoli, dimensioni e costi. :param demo_type: Tipo di demo da eseguire ('POC_ETL', 'DB_ETL') :param id_run: ID della run :param num_items: Numero di items da generare :param logger: Logger per registrare le informazioni """ logger.info(f"Populating transformation data for demo type: {demo_type}, id_run: {id_run}, num_items: {num_items}") list_items = [genera_stringa_alfanumerica() for _ in range(num_items)] list_sizes = [random.randint(20, 70) for _ in range(num_items)] np.random.seed(42) matrix_costs = np.random.randint(1, 11, size=(num_items, num_items)) df_costs_data = pd.DataFrame(matrix_costs, index=range(num_items), columns=list_items) df_costs_data['id_container'] = df_costs_data.index df_costs_data = pd.melt( df_costs_data, id_vars=['id_container'], var_name='cod_item', value_name='val_cost' ).copy() df_data = pd.DataFrame({'cod_item': list_items, 'num_size': list_sizes}) df_constraints = pd.DataFrame({'num_max_capacity': [NUM_MAX_CAPACITY], 'num_safety': [NUM_SAFETY], 'num_containers_in_unit': [NUM_CONTAINERS_IN_UNIT]}) df_fobj_weights = pd.DataFrame({'total_cost': [WEIGHT_TOTAL_COST], 'num_used_containers': [WEIGHT_USED_CONTAINERS], 'abs_safety_distance': [WEIGHT_ABS_SAFETY_DISTANCE]}) if demo_type == 'POC_ETL': os.makedirs(os.path.join(TRN_ASSETS_PATH, 'demo_data'), exist_ok=True) df_data.to_csv(os.path.join(TRN_ASSETS_PATH, 'demo_data', f'items_{id_run}.csv'), index=False) df_costs_data.to_csv(os.path.join(TRN_ASSETS_PATH, 'demo_data', f'items_costs_{id_run}.csv'), index=False) df_constraints.to_csv(os.path.join(TRN_ASSETS_PATH, 'demo_data', f'constraints_{id_run}.csv'), index=False) df_fobj_weights.to_csv(os.path.join(TRN_ASSETS_PATH, 'demo_data', f'fobj_weights_{id_run}.csv'), index=False) elif demo_type == 'DB_ETL': df_data['id_run'] = id_run df_costs_data['id_run'] = id_run df_constraints['id_run'] = id_run df_fobj_weights['id_run'] = id_run # TODO: Implement database saving logic here else: raise NotImplementedError( f"Demo type {demo_type} not implemented. Supported types: 'POC_ETL', 'DB_ETL'" ) populate_trn( demo_type=DEMO_TYPE, id_run=1, num_items=NUM_ITEMS, logger=logger, ) from typing import ( List, Literal, ) from typing import ( Optional, ) # import external libraries from dataclasses import ( dataclass, field, ) from jsonargparse import ActionConfigFile stepsEnum = Literal['stg', 'dtn', 'trn', 'mdl', 'prs'] stepsType = List[stepsEnum] @dataclass # It's a container for all the parameters that are needed to configure a solver class SolverConfig: num_search_workers: int = 16 minutes_to_solve: int = 10 min_perc_objective_delta: float = 1 num_max_solutions_no_significant_change: int = 10 @dataclass class CfgParams: id_run: Optional[int] steps: stepsType solver: SolverConfig cfg: ActionConfigFile # Import external libraries import os from logging import Logger import pandas as pd TRN_DATA_FOLDER = 'assets/demo_data' class InputManager: def __init__( self, cfg: CfgParams, logger: Logger, id_run: int = 0, ): self.cfg = cfg self.logger = logger self.id_run = id_run self.logger.info("Start initializing Input Manager") self.get_data() self.create_utils() self.logger.info("Done initialization Input Manager") def get_data(self): self.logger.info("Getting data") items_path = os.path.join(TRN_DATA_FOLDER, f'items_{self.id_run}.csv') costs_path = os.path.join(TRN_DATA_FOLDER, f'items_costs_{self.id_run}.csv') constraints_path = os.path.join(TRN_DATA_FOLDER, f'constraints_{self.id_run}.csv') fobj_weights_path = os.path.join(TRN_DATA_FOLDER, f'fobj_weights_{self.id_run}.csv') df_items = pd.read_csv(items_path) df_costs = pd.read_csv(costs_path) df_constraints = pd.read_csv(constraints_path) df_fobj_weights = pd.read_csv(fobj_weights_path) setattr(self, 'df_items', df_items) setattr(self, 'df_costs', df_costs) setattr(self, 'df_constraints', df_constraints) setattr(self, 'df_fobj_weights', df_fobj_weights) def create_utils(self): self.logger.info("Create utils") self.create_scalars() self.create_lists() self.create_dictionaries() def create_scalars(self): self.logger.info("\tCreate scalars") num_items = len(self.df_items) num_containers = len(self.df_costs['id_container'].unique()) num_max_capacity = self.df_constraints['num_max_capacity'].iloc[0] num_safety = self.df_constraints['num_safety'].iloc[0] num_containers_in_unit = self.df_constraints['num_containers_in_unit'].iloc[0] setattr(self, 'num_items', num_items) setattr(self, 'num_containers', num_containers) setattr(self, 'num_max_capacity', num_max_capacity) setattr(self, 'num_safety', num_safety) setattr(self, 'num_containers_in_unit', num_containers_in_unit) def create_lists(self): self.logger.info("\tCreate lists") list_items = self.df_items['cod_item'].tolist() list_containers = self.df_costs['id_container'].unique().tolist() setattr(self, 'list_items', list_items) setattr(self, 'list_containers', list_containers) def create_dictionaries(self): self.logger.info("\tCreate dictionaries") dict_item_to_size = self.df_items.set_index('cod_item')['num_size'].to_dict() dict_item_container_to_cost = self.df_costs.set_index(['cod_item', 'id_container'])['val_cost'].to_dict() dict_fobj_terms_to_weights = self.df_fobj_weights.iloc[0].to_dict() setattr(self, 'dict_item_to_size', dict_item_to_size) setattr(self, 'dict_item_container_to_cost', dict_item_container_to_cost) setattr(self, 'dict_fobj_terms_to_weights', dict_fobj_terms_to_weights) # Import typing libraries from logging import Logger from typing import ( Dict, Optional, ) # Import external libraries import numpy as np from ortools.sat.python import cp_model import time from threading import Timer class EarlyStoppingSolutionPrinter(cp_model.CpSolverSolutionCallback): def __init__( self, logger: Logger, dict_fobj_terms: Dict, timer_limit: Optional[int], max_time: Optional[int] = None, max_solutions: Optional[int] = None, min_perc_objective_delta: Optional[float] = None, num_max_solutions_no_significant_change: Optional[int] = None, ): """ This function initializes the class `CpSolverSolutionCallback` and sets the attributes `logger`, `dict_fobj_terms`, `timer_limit`, `max_time`, `timer`, `end_time`, and `solution_count`. :param logger: A logger object to log messages to :type logger: Logger :param dict_fobj_terms: This is a dictionary of the terms in the objective function :type dict_fobj_terms: Dict :param timer_limit: The time limit in seconds :type timer_limit: int :param max_time: the maximum time in seconds to run the solver :type max_time: int :param max_solutions: The maximum number of solutions to find, defaults to None :type max_solutions: Optional[int] :param min_perc_objective_delta: The minimum percentage change in the objective function value to consider as significant, defaults to None :type min_perc_objective_delta: Optional[float] :param num_max_solutions_no_significant_change: The maximum number of solutions that can be found without a significant change in the objective function value, defaults to None :type num_max_solutions_no_significant_change: Optional[int] """ cp_model.CpSolverSolutionCallback.__init__(self) self.logger = logger self.dict_fobj_terms = dict_fobj_terms self.timer_limit = timer_limit if timer_limit else max_time self.max_time = max_time self.max_solutions = max_solutions if max_solutions else float('inf') self.timer = None self.end_time = time.time() + max_time if max_time else None self.solution_count = 0 self.val_previous_objective_function = None self.min_perc_objective_delta = min_perc_objective_delta if min_perc_objective_delta else 0.0 self.num_max_solutions_no_significant_change = num_max_solutions_no_significant_change if num_max_solutions_no_significant_change else 10000 self.counter_no_significant_change = 0 def on_solution_callback(self): """ The function creates a dictionary of the characters in the solution, and then logs the solution to the console """ self.create_dict_callback() if self.solution_count == 0: self.create_dict_callback_characters() setattr(self, 'val_first_objective_function', self.ObjectiveValue()) self.log_string_callback() self.cancel() if self.end_time and time.time() >= self.end_time or self.solution_count >= self.max_solutions: self.logger.debug("Time limit reached - stop search") self.StopSearch() else: self.timer = Timer(self.timer_limit, self.stop) self.timer.start() self.solution_count += 1 # if the objective function does not change significantly in some solutions, stop the search if self.val_previous_objective_function: if abs(self.perc_variation) < self.min_perc_objective_delta: self.counter_no_significant_change += 1 if self.counter_no_significant_change >= self.num_max_solutions_no_significant_change: self.logger.debug( f"Objective function not changed more than {self.min_perc_objective_delta}% in {self.num_max_solutions_no_significant_change} " f"solutions - stop search" ) self.StopSearch() else: self.counter_no_significant_change = 0 self.val_previous_objective_function = self.ObjectiveValue() def get_string_to_print(self): """ It takes the dictionary of objective function terms, and for each term, it gets the name, the variable and the weight, then it creates a string that looks like this: `name: weight * value` """ list_to_print = [ f'{name}: {weight:.0f} * {self.Value(var):.0f}' for name, [var, weight] in self.dict_fobj_terms.items() ] string_to_print = ', '.join(list_to_print) setattr(self, 'string_to_print', string_to_print) def create_dict_callback(self): """ It creates a dictionary of the objective function terms and their values """ dict_callback = { "num_solution": f"{self.solution_count:.0f}", "val_wall_time[s]": f"{self.WallTime():.2f}", "val_objective": f"{self.ObjectiveValue():.0f}", } if self.solution_count == 0: dict_callback["perc_variation"] = "" dict_callback["perc_variation_first_sol"] = "" perc_variation = 0 else: perc_variation = ( (self.ObjectiveValue() - self.val_previous_objective_function) / self.val_previous_objective_function * 100 ) perc_variation_first = ( (self.ObjectiveValue() - self.val_first_objective_function) / self.val_first_objective_function * 100 ) dict_callback["perc_variation"] = f"{perc_variation:.2f} %" dict_callback["perc_variation_first_sol"] = f"{perc_variation_first:.2f} %" for name, [var, weight] in self.dict_fobj_terms.items(): dict_callback[name] = f"{weight:.0f} * {self.Value(var):.0f}" setattr(self, 'dict_callback', dict_callback) setattr(self, 'perc_variation', perc_variation) def create_dict_callback_characters( self, num_log_characters: int = 4, ): """ It creates a dictionary of the number of characters needed to print the callback name and value :param num_log_characters: The number of characters to use for each line of the log, defaults to 4 :type num_log_characters: int (optional) """ dict_callback_characters = { name: int( np.ceil( max( [ len(name), len(value), ] ) / num_log_characters ) ) * num_log_characters + 4 for name, value in self.dict_callback.items() } setattr(self, 'dict_callback_characters', dict_callback_characters) def log_string_callback(self): """ It takes the dictionary of callback values and prints them out in a nice table format """ if self.solution_count == 0: list_str_callback_names = [ f"{name:-^{self.dict_callback_characters[name]}}" for name in self.dict_callback.keys() ] str_callback_names = '|'.join(list_str_callback_names) self.logger.info(f"|{str_callback_names}|") list_str_callback_values = [ f"{value: ^{self.dict_callback_characters[name]}}" for name, value in self.dict_callback.items() ] str_callback_values = '|'.join(list_str_callback_values) self.logger.info(f"|{str_callback_values}|") def cancel(self): """ The cancel function cancels the timer """ if self.timer: self.timer.cancel() def stop(self): """ If the objective function has not changed in the last `timer_limit` seconds, stop the search """ self.logger.debug(f"Objective function not changed in {self.timer_limit} sec - stop search") self.StopSearch() def reset_end_time(self): """ If the end_time attribute is not None, then set the end_time attribute to the current time plus the max_time attribute """ if self.end_time: self.end_time = time.time() + self.max_time from typing import List import itertools def binary_tree_sum(list_variables: List): """ It takes a list of variables and sums them up in a binary tree fashion :param list_variables: a list of numbers :type list_variables: List :return: The sum of the list of variables. """ if not isinstance(list_variables, list): list_variables = list(list_variables) if len(list_variables) == 0: return 0 elif len(list_variables) == 1: return list_variables[0] else: list_tuples = list( itertools.zip_longest( list_variables[:len(list_variables) // 2], list_variables[(len(list_variables) // 2):], ) ) while len(list_variables) > 1: list_variables = [ sum(t) if t[0] is not None else t[1] for t in list_tuples ] list_tuples = list( itertools.zip_longest( list_variables[:len(list_variables) // 2], list_variables[(len(list_variables) // 2):], ) ) return list_variables[0] class Config(): def __init__(self): pass class solver(): pass cfg = Config() cfg.id_run = 1 cfg.solver.num_max_solutions_no_significant_change = 10 cfg.solver.min_perc_objective_delta = 0.1 cfg.solver.num_search_workers = 16 cfg.solver.minutes_to_solve = 10 import rmm pool = rmm.mr.PoolMemoryResource( rmm.mr.ManagedMemoryResource(), initial_pool_size="1GiB", maximum_pool_size="20GiB" ) rmm.mr.set_current_device_resource(pool) logger.info("\n\n\n\nCUOPT EXAMPLE") from cuopt.linear_programming.problem import Problem, INTEGER, MAXIMIZE from cuopt.linear_programming.solver_settings import SolverSettings # Create a new MIP problem problem = Problem("Simple MIP") # Add integer variables with bounds x = problem.addVariable(vtype=INTEGER, name="V_x") y = problem.addVariable(lb=10, ub=50, vtype=INTEGER, name="V_y") # Add constraints problem.addConstraint(2 * x + 4 * y >= 1, name="C1") problem.addConstraint(3 * x + 2 * y <= 200, name="C2") # Set objective function problem.setObjective(5 * x + 3 * y, sense=MAXIMIZE) # Configure solver settings settings = SolverSettings() settings.set_parameter("time_limit", 60) # Solve the problem problem.solve(settings) # Check solution status and results if problem.Status.name == "Optimal": print(f"Optimal solution found in {problem.SolveTime:.2f} seconds") print(f"x = {x.getValue()}") print(f"y = {y.getValue()}") print(f"Objective value = {problem.ObjValue}") else: print(f"Problem status: {problem.Status.name}") logger.info("\n\n\n\nCUOPT") # Import typing libraries from logging import Logger import pandas as pd # Import external libraries import numpy as np # Import internal libraries from cuopt.linear_programming.problem import Problem, INTEGER, MINIMIZE from cuopt.linear_programming.solver_settings import SolverSettings class Optimizer: def __init__( self, im: InputManager, cfg: CfgParams, logger: Logger, id_run: int = 0, ): self.im = im self.cfg = cfg self.logger = logger self.id_run = id_run self.logger.info("Start initializing Optimizer") self.create_model() def create_model(self): self.logger.info("Creating model") setattr(self, 'model', Problem("Knapsack")) self.create_variables() self.set_constraints() def create_variables(self): self.logger.info("Creating variables") self.create_bool_used_container_variables() self.create_bool_item_in_container_variables() self.create_container_abs_safety_distance_variables() def create_bool_used_container_variables(self): self.logger.info("\tCreating bool used containers variables") dict_containers_to_bool_used_variables = { container: self.model.addVariable(vtype=INTEGER, lb=0, ub=1, name=f'bool_container_{container}_is_used') for container in self.im.list_containers } setattr(self, 'dict_containers_to_bool_used_variables', dict_containers_to_bool_used_variables) def create_bool_item_in_container_variables(self): self.logger.info("\tCreating bool item in container variables") dict_item_container_to_bool_in_variables = { (item, container): self.model.addVariable(vtype=INTEGER, lb=0, ub=1, name=f'bool_item_{item}_in_container_{container}') for item in self.im.list_items for container in self.im.list_containers } setattr(self, 'dict_item_container_to_bool_in_variables', dict_item_container_to_bool_in_variables) def create_container_abs_safety_distance_variables(self): self.logger.info("\tCreating container abs safety distance variables") dict_container_to_abs_safety_distance_variables = { container: self.model.addVariable(vtype=INTEGER, lb=0, ub=self.im.num_max_capacity * 2, name=f'abs_safety_distance_container_{container}') for container in self.im.list_containers } dict_container_to_bool_distance_is_positive_variables = { container: self.model.addVariable(vtype=INTEGER, lb=0, ub=1, name=f'bool_distance_is_positive_container_{container}') for container in self.im.list_containers } dict_container_to_positive_safety_distance_variables = { container: self.model.addVariable(vtype=INTEGER, lb=0, ub=self.im.num_max_capacity, name=f'positive_distance_container_{container}') for container in self.im.list_containers } dict_container_to_negative_safety_distance_variables = { container: self.model.addVariable(vtype=INTEGER, lb=0, ub=self.im.num_max_capacity, name=f'negative_distance_container_{container}') for container in self.im.list_containers } setattr(self, 'dict_container_to_abs_safety_distance_variables', dict_container_to_abs_safety_distance_variables) setattr(self, 'dict_container_to_bool_distance_is_positive_variables', dict_container_to_bool_distance_is_positive_variables) setattr(self, 'dict_container_to_positive_safety_distance_variables', dict_container_to_positive_safety_distance_variables) setattr(self, 'dict_container_to_negative_safety_distance_variables', dict_container_to_negative_safety_distance_variables) def set_constraints(self): self.logger.info("Setting constraints") self.set_one_container_per_item_constraint() self.set_container_capacity_constraint() self.set_used_containers_constraint() # self.set_units_constraint() self.set_safety_distance_constraints() # self.set_infeasible_constraint() # uncomment to test how aopt library handles an infeasible model def set_one_container_per_item_constraint(self): self.logger.info("\tSetting one container per item constraint") for item in self.im.list_items: self.model.addConstraint( binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] for container in self.im.list_containers ] ) == 1, ) def set_container_capacity_constraint(self): self.logger.info("\tSetting container capacity constraint") for container in self.im.list_containers: self.model.addConstraint( binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] * self.im.dict_item_to_size[item] for item in self.im.list_items ] ) <= int(self.im.num_max_capacity), ) def set_used_containers_constraint(self): self.logger.info("\tSetting used containers constraint") for container in self.im.list_containers: self.model.addConstraint( self.dict_containers_to_bool_used_variables[container] * self.im.num_items >= binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] for item in self.im.list_items ] ), ) self.model.addConstraint( binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] for item in self.im.list_items ] ) >= self.dict_containers_to_bool_used_variables[container], ) def set_units_constraint(self): self.logger.info("\tSetting units constraint") # This is an example of how to use AddLinearExpressionInDomain, I want the number of used containers to be a multiple of the number of containers in a unit num_used_containers = binary_tree_sum( list(self.dict_containers_to_bool_used_variables.values()) ) list_values = [i * self.im.num_containers_in_unit for i in range(0, int(np.ceil(self.im.num_containers / self.im.num_containers_in_unit)) + 1)] domain = cp_model.Domain.FromValues(values=list_values) self.model.AddLinearExpressionInDomain( num_used_containers, domain, ) def set_safety_distance_constraints(self): self.logger.info("\tSetting safety distance constraints") for container in self.im.list_containers: sum_sizes_in_container = binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] * self.im.dict_item_to_size[item] for item in self.im.list_items ] ) val_safety_distance = self.im.num_safety * self.dict_containers_to_bool_used_variables[container] - sum_sizes_in_container # There are two ways to write the following constraints, one using the OnlyEnforceIf method and the other without it. # The first one leads to half the constraints, but they are more complex. # It's debatable which one is better, but I will use the second one to show how to linearize the constraints. self.model.addConstraint( self.dict_container_to_positive_safety_distance_variables[container] <= val_safety_distance + (self.im.num_max_capacity * (1 - self.dict_container_to_bool_distance_is_positive_variables[container])), ) self.model.addConstraint( self.dict_container_to_positive_safety_distance_variables[container] >= val_safety_distance - (self.im.num_max_capacity * (1 - self.dict_container_to_bool_distance_is_positive_variables[container])), ) self.model.addConstraint( self.dict_container_to_positive_safety_distance_variables[container] <= self.im.num_max_capacity * self.dict_container_to_bool_distance_is_positive_variables[container], ) self.model.addConstraint( self.dict_container_to_positive_safety_distance_variables[container] >= - self.im.num_max_capacity * self.dict_container_to_bool_distance_is_positive_variables[container], ) self.model.addConstraint( self.dict_container_to_negative_safety_distance_variables[container] <= (self.im.num_max_capacity * self.dict_container_to_bool_distance_is_positive_variables[container]) - val_safety_distance, ) self.model.addConstraint( self.dict_container_to_negative_safety_distance_variables[container] + self.im.num_max_capacity * self.dict_container_to_bool_distance_is_positive_variables[container] + val_safety_distance >= 0, ) self.model.addConstraint( self.dict_container_to_negative_safety_distance_variables[container] <= self.im.num_max_capacity * (1 - self.dict_container_to_bool_distance_is_positive_variables[container]), ) self.model.addConstraint( self.dict_container_to_negative_safety_distance_variables[container] + self.im.num_max_capacity * (1 - self.dict_container_to_bool_distance_is_positive_variables[container]) >= 0, ) self.model.addConstraint( self.dict_container_to_abs_safety_distance_variables[container] == self.dict_container_to_positive_safety_distance_variables[container] + self.dict_container_to_negative_safety_distance_variables[container], ) def set_infeasible_constraint(self): self.model.addConstraint( binary_tree_sum(self.dict_containers_to_bool_used_variables.values()) <= -1, ) def optimize(self): self.logger.info("Optimize model") self.define_objective_function() # self.add_hints() des_status = self.solve_model() return des_status def define_objective_function(self): self.logger.info("\tDefining objective function") dict_fobj_terms_to_weights = self.im.dict_fobj_terms_to_weights sum_costs = binary_tree_sum( [ self.dict_item_container_to_bool_in_variables[(item, container)] * self.im.dict_item_container_to_cost[(item, container)] for item in self.im.list_items for container in self.im.list_containers ] ) sum_abs_values = binary_tree_sum( [ self.dict_container_to_abs_safety_distance_variables[container] for container in self.im.list_containers ] ) num_used_containers = binary_tree_sum( list(self.dict_containers_to_bool_used_variables.values()) ) # aaa dict_fobj_terms_to_value_weight = { # 'total_cost': [sum_costs, dict_fobj_terms_to_weights['total_cost']], # 'num_used_containers': [num_used_containers, dict_fobj_terms_to_weights['num_used_containers']], 'abs_safety_distance': [sum_abs_values, dict_fobj_terms_to_weights['abs_safety_distance']], } dict_fobj_terms_to_weighted_value = { term_name: term_variable * term_weight for term_name, [term_variable, term_weight] in dict_fobj_terms_to_value_weight.items() } self.model.setObjective( binary_tree_sum( list(dict_fobj_terms_to_weighted_value.values()) ), sense=MINIMIZE, ) setattr(self, 'dict_fobj_terms_to_value_weight', dict_fobj_terms_to_value_weight) def add_hints(self): self.logger.info("\tAdding hints") # Hint of trivial solution: first item in first container, second item in second container, etc. for i, item in enumerate(self.im.list_items): if i < len(self.im.list_containers): container = self.im.list_containers[i] self.model.AddHint( self.dict_item_container_to_bool_in_variables[(item, container)], 1, ) else: # If there are more items than containers, we cannot add a hint for this item self.logger.warning(f'No hint for item {item} as there are not enough containers') def solve_model(self): self.logger.info("\tSolving model") # Configure solver settings settings = SolverSettings() settings.set_parameter("time_limit", self.cfg.solver.minutes_to_solve * 60 + 1) # Solve the problem self.logger.info( f'Optimization with {self.cfg.solver.minutes_to_solve * 60} seconds to solve' ) self.model.solve(settings) def extract_solutions(self): self.logger.info("Extracting solutions") # Extract items in containers dict_items_in_container = { (item, container): self.solver.Value(self.dict_item_container_to_bool_in_variables[(item, container)]) for item in self.im.list_items for container in self.im.list_containers } df_items_in_container = pd.DataFrame.from_dict(dict_items_in_container, orient='index', columns=['flg_is_in_container']) df_items_in_container = df_items_in_container[df_items_in_container['flg_is_in_container'] == 1].reset_index().copy() df_items_in_container['cod_item'] = df_items_in_container['index'].apply(lambda x: x[0]) df_items_in_container['id_container'] = df_items_in_container['index'].apply(lambda x: x[1]) df_items_in_container = df_items_in_container.drop(columns=['index', 'flg_is_in_container']) return df_items_in_container def extract_mus_solutions(self): self.logger.info("Extracting MUS solutions") mus = AOPTMus( self.model, self.logger, ) dict_mus = mus.compute_mus_fast() df_mus = pd.DataFrame.from_dict( dict_mus, orient='index', columns=['mus'], ) df_mus['id_mus'] = df_mus.index df_mus['family'] = df_mus['mus'].apply(lambda x: x.family) df_mus['constraint'] = df_mus['mus'].apply(lambda x: x.name) df_mus = df_mus.drop(columns=['mus']).copy() return df_mus # Import typing libraries from logging import Logger # Import external libraries import os # Import internal libraries RESULTS_FOLDER = 'assets/' class MdlExecute: def __init__( self, cfg: CfgParams, logger: Logger, ) -> None: self.cfg = cfg self.logger = logger self.id_run = cfg.id_run def main(self): # initialize input manager im = InputManager( cfg=self.cfg, logger=self.logger, id_run=self.id_run, ) setattr(self, 'im', im) # run optimizer self.run_optimizer() def run_optimizer(self): optimizer = Optimizer( im=self.im, cfg=self.cfg, logger=self.logger, id_run=self.id_run, ) # solve model optimizer.optimize() execute = MdlExecute( cfg, logger, ) execute.main()