diff --git a/src/snc/agents/activity_rate_to_mpc_actions/fox_mpc.py b/src/snc/agents/activity_rate_to_mpc_actions/fox_mpc.py new file mode 100644 index 0000000..f82cd75 --- /dev/null +++ b/src/snc/agents/activity_rate_to_mpc_actions/fox_mpc.py @@ -0,0 +1,179 @@ +import numpy as np +import random +from typing import List, Optional, Tuple + +from snc.agents.activity_rate_to_mpc_actions.action_mpc_policy import ActionMPCPolicy +import snc.agents.agents_utils as agents_utils +import snc.agents.hedgehog.policies.policy_utils as policy_utils +import snc.utils.snc_types as types + + +class FoxMpcPolicy(ActionMPCPolicy): + def __init__(self, + physical_constituency_matrix: types.ConstituencyMatrix, + buffer_processing_matrix: types.BufferMatrix, + mpc_seed: Optional[int] = None) -> None: + """ + Obtain feasible binary actions from activity rates with feedback on how many actions have + been performed so far for a given horizon. The actions are drawn from a probability + distribution that aims to match the activity rates after some horizon. The feedback allows + to adjust the distribution so that underperformed activities are emphasised. Feasible + actions refer to those that drain nonempty buffers. In the case that some action is + infeasible, then the corresponding resource performs other activity. + + :param physical_constituency_matrix: Constituency matrix from environment. We assume it has + orthogonal rows. + :param buffer_processing_matrix: Buffer processing matrix from environment. + :return: None. + """ + # Ensure that constituency_matrix has a single one per column. + assert agents_utils.has_orthogonal_rows(physical_constituency_matrix), \ + "Physical constituency matrix must have orthogonal rows." + super().__init__(physical_constituency_matrix, mpc_seed) + + self.buffer_processing_matrix = buffer_processing_matrix + self.n_activities = buffer_processing_matrix.shape[1] + self.activities_per_resource, self.num_activities_per_resource \ + = self.get_ind_activities_per_resource(physical_constituency_matrix) + self.exit_activities = self.get_exit_activities(buffer_processing_matrix) + self.activities_to_target_buffers, self.activities_to_source_buffers \ + = self.get_target_and_source_buffers(buffer_processing_matrix,self.exit_activities) + + @staticmethod + def get_exit_activities(buffer_processing_matrix): + exit_activities = set() + for a in range(buffer_processing_matrix.shape[1]): + if np.all(buffer_processing_matrix[:,a] <= 0): + exit_activities.add(a) + return exit_activities + + @staticmethod + def get_target_and_source_buffers(buffer_processing_matrix,exit_activities): + n_activities = buffer_processing_matrix.shape[1] + activities_to_target_buffers = {} + activities_to_source_buffers = {} + for a in range(n_activities): + act_vector = buffer_processing_matrix[:,a] + activities_to_source_buffers[a] = int(np.where(act_vector < 0)[0][0]) + if a in exit_activities: + continue + activities_to_target_buffers[a] = int(np.where(act_vector > 0)[0][0]) + return activities_to_target_buffers, activities_to_source_buffers + + @staticmethod + def get_ind_activities_per_resource(physical_constituency_matrix: types.ConstituencyMatrix) -> \ + Tuple[List[List[int]], List[int]]: + """ + Return the index of activities per resource and the total number of activities per resource. + + :param physical_constituency_matrix: Constituency matrix from environment. We assume it has + orthogonal rows. + :return: (activities_per_resource, num_activities_per_resource): + - activities_per_resource: List of lists of activities per resource. + - num_activities_per_resource: List of number of activities per resource. + """ + assert agents_utils.has_orthogonal_rows(physical_constituency_matrix), \ + "Physical constituency matrix must have orthogonal rows." + + activities_per_resource = [] # type: List[List[int]] + num_activities_per_resource = [] # type: List[int] + + for c in physical_constituency_matrix: + activities_c = np.nonzero(c)[0] + assert activities_c.size > 0 + + activities_per_resource += [activities_c.tolist()] + num_activities_per_resource += [activities_c.size] + + return activities_per_resource, num_activities_per_resource + + def obtain_actions(self, **kwargs) -> types.ActionProcess: + """ + This method implements the abstract method from super class `ActionMPCPolicy`. + It first gathers the feedback information namely the number of times each activity has to + be performed (i.e. 'sum_actions') and the current state ('state'). Then, it calls its own + method to return a single action vector. + + :return: actions: binary indicator matrix with number of rows equal to the number of + non-idling activities (i.e. num of columns of the constituency matrix), and number of + columns equal to number of time steps to perform MPC. + """ + assert 'state' in kwargs, "Current state ('state') is required to be passed as parameter." + + state = kwargs["state"] + x_eff = kwargs["x_eff"] + x_star = kwargs["x_star"] + print(state.ravel().astype(int),x_eff.ravel().astype(int),x_star.ravel().astype(int)) + idling_set = kwargs["r_idling_set"] + draining_resources = kwargs["draining_resources"] + print(idling_set) + print(self.activities_per_resource) + x_target = x_star if len(idling_set) > 0 else x_eff + buffer_weights = list(np.maximum(x_target - state, 0).astype(int).ravel()) + print(buffer_weights) + actions = np.zeros((self.n_activities,1)) + + for r in draining_resources: + actions_list = self._get_actions_list(r,state,buffer_weights,False) + #random.shuffle(actions_list) + decided_action,_ = actions_list[0] + actions[decided_action,0] = 1 + + + for r in range(len(self.activities_per_resource)): + if r in draining_resources: + continue + actions_list = self._get_actions_list(r,state,buffer_weights, r in idling_set) + if not actions_list: + continue + random.shuffle(actions_list) + print(r,actions_list) + decided_action,_ = actions_list[0] + + + actions[decided_action,0] = 1 + print() + return actions + + def _get_actions_list(self,r,state,buffer_weights,r_in_idling_set): + + actions_list = [] + max_weight = 0 + + has_starved_activities = False + for a in self.activities_per_resource[r]: + if a in self.exit_activities: + continue + source_buffer = self.activities_to_source_buffers[a] + target_buffer = self.activities_to_target_buffers[a] + buffer_weight = buffer_weights[target_buffer] + if buffer_weight > 0 and state[source_buffer,0] == 0: + has_starved_activities = True + + for a in self.activities_per_resource[r]: + if state[self.activities_to_source_buffers[a]] == 0: + continue + if a in self.exit_activities: + source_buffer = self.activities_to_source_buffers[a] + buffer_weight = buffer_weights[source_buffer] + #weight = max(1,buffer_weight) + weight = max(1,int(state[source_buffer,0])) + if weight == max_weight: + actions_list.append((a,weight)) + elif weight > max_weight: + actions_list = [(a,weight)] + max_weight = weight + continue + + source_buffer = self.activities_to_target_buffers[a] + target_buffer = self.activities_to_target_buffers[a] + buffer_weight = buffer_weights[target_buffer] + if r_in_idling_set and buffer_weight == 0:# and not has_starved_activities: + continue + if buffer_weight == max_weight: + actions_list.append((a,buffer_weight)) + elif buffer_weight > max_weight: + actions_list = [(a,buffer_weight)] + max_weight = buffer_weight + print(has_starved_activities) + return actions_list diff --git a/src/snc/agents/hedgehog/hh_agents/fox_agent.py b/src/snc/agents/hedgehog/hh_agents/fox_agent.py new file mode 100644 index 0000000..0c31224 --- /dev/null +++ b/src/snc/agents/hedgehog/hh_agents/fox_agent.py @@ -0,0 +1,309 @@ +import numpy as np +from typing import Optional, Type, Union, Dict, Tuple, Any + +from snc.agents.activity_rate_to_mpc_actions.feedback_mip_feasible_mpc_policy import \ + FeedbackMipFeasibleMpcPolicy +from snc.agents.hedgehog import safety_stocks +from snc.agents.hedgehog.strategic_idling.strategic_idling_utils import get_dynamic_bottlenecks +from snc.agents.activity_rate_to_mpc_actions.fox_mpc import FoxMpcPolicy +from snc.agents.activity_rate_to_mpc_actions.feedback_stationary_feasible_mpc_policy import \ + FeedbackStationaryFeasibleMpcPolicy +from snc.agents.hedgehog.class_loader import get_class_from_name +from snc.agents.hedgehog.hh_agents.hedgehog_agent_interface import AsymptoticCovarianceParams, \ + HedgehogAgentInterface, HedgehogHyperParams, WorkloadRelaxationParams +import snc.agents.hedgehog.minimal_draining_time as mdt +from snc.agents.hedgehog.params import \ + BigStepLayeredPolicyParams, \ + BigStepPenaltyPolicyParams, \ + DemandPlanningParams, \ + StrategicIdlingParams +from snc.agents.hedgehog.strategic_idling.strategic_idling import StrategicIdlingCore +from snc.agents.hedgehog.strategic_idling.strategic_idling_fox import StrategicIdlingFox +from snc.environments.controlled_random_walk import ControlledRandomWalk +import snc.utils.snc_types as types +from snc.environments import controlled_random_walk as crw +import snc.simulation.store_data.reporter as rep + + +class FoxAgent(HedgehogAgentInterface): + def __init__(self, + env: ControlledRandomWalk, + discount_factor: float, + workload_relaxation_params: WorkloadRelaxationParams, + hedgehog_hyperparams: HedgehogHyperParams, + asymptotic_covariance_params: AsymptoticCovarianceParams, + strategic_idling_params: StrategicIdlingParams = StrategicIdlingParams(), + policy_params: Optional[ + Union[BigStepLayeredPolicyParams, BigStepPenaltyPolicyParams]] = None, + strategic_idling_class: Type[StrategicIdlingCore] = StrategicIdlingFox, + demand_planning_params: DemandPlanningParams = DemandPlanningParams(), + name: str = "FoxAgent", + debug_info: bool = False, + agent_seed: Optional[int] = None, + mpc_seed: Optional[int] = None) -> None: + """ + :param env: Environment to stepped through. + :param discount_factor: Discount factor for the cost per time step. + :param workload_relaxation_params: Tuple of parameters defining the first workload + relaxation. + :param hedgehog_hyperparams: Named tuple including the following parameters: + - activity_rates_policy_class_name: + - mpc_policy_class_name: + - theta_0: Tuning parameter to compute the safety stock threshold. + - horizon_drain_time_ratio: Ratio num steps of the planning horizon (a.k.a. size of the + Big step) over the minimal draining time. If horizon_drain_time_ratio == 0, then + the horizon equals minimum_horizon (described below). + - horizon_mpc_ratio: Ratio num steps to follow the activity rates over horizon. + - minimum_horizon: Minimum horizon length i.e. minimum step size of the big step policy. + :param asymptotic_covariance_params: Named tuple of parameters used to generate an estimate + of the asymptotic covariance matrix. It includes + - num_batch: Number of batches to group data samples for covariance estimation. + - num_presimulation_steps: Total number of simulated steps for covariance estimation. + :param strategic_idling_params: Named tuple with the following parameters: + - convex_solver: Solver to be used at the different steps when finding potential idling + directions. + - epsilon: Nonnegative float that gives the angle used to create the artificial cone + around the monotone region when it is a ray (i.e. it has empty interior). + - shift_eps: Nonnegative float used as distance that we go before and beyond w_star in + order to obtain the level sets. It has to be large enough to avoid the lack of + precision of the convex optimisation solver, and small enough to remain in the + closest linear part of the piece-wise linear cost. + - hedging_scaling_factor: Nonnegative float used as scaling factor that multiplies + the hedging threshold given by the diffusion heuristic. + - penalty_coeff_w_star: Nonnegative float used as penalty coefficient to encourage the + solution w_star to be close to the lower boundary of the solution set (useful when + this is not a singleton). + :param policy_params: Named tuple with the following parameters. It is specific for each + policy class. + - convex_solver: String with the solver to be used by the policy. + - boolean_action_flag: Indicates if the optimisation variable of the policy synthesis + problem should be a real or a binary vector. + - nonidling_penalty_coeff: Coefficient to scale the nonidling penalty. + - safety_penalty_coeff: Coefficient to scale the safety stock penalty. + :param strategic_idling_class: Class to be used to make strategic idling decisions (e.g., + 'StrategicIdlingCore', 'StrategicIdlingHedging', 'StrategicIdlingForesight', etc.) + :param demand_planning_params: Tuple of parameters that specify demand_planning_class and + its parameters (if any). + :param name: Agent identifier used when storing the results. + :param debug_info: Flat to print instantaneous calculations useful for debugging. + :param agent_seed: Random seed for agent's random number generator. + :param mpc_seed: Random seed for MPC policy's random number generator. + """ + mpc_policy_class = self.get_mpc_policy_class(hedgehog_hyperparams.mpc_policy_class_name) + mpc_policy_object = mpc_policy_class( + env.physical_constituency_matrix, + env.job_generator.buffer_processing_matrix, + mpc_seed + ) + super().__init__( + env, + mpc_policy_object, + discount_factor, + workload_relaxation_params, + hedgehog_hyperparams, + asymptotic_covariance_params, + strategic_idling_params, + policy_params, + strategic_idling_class, # TODO: Get from class name in hyperparams. + demand_planning_params, + name, + debug_info, + agent_seed + ) + self.strategic_idling_class = StrategicIdlingFox + + @staticmethod + def get_horizon(**kwargs) -> int: + """ + Returns the size of the big step, i.e. the horizon, for which the big step policy will + compute a schedule. + + :return: num_time_steps: Horizon as a number of time steps. + """ + pass + + def serialise_init_policy_kwargs(self): + """ + Serialise the parameters for the initialisation of the activity rates class. Depending on + the activity rates class, it passes different parameters. + It allows: BigStepLayeredPolicy, BigStepPolicy and BigStepSurplusLayeredPolicy classes. + + :return: Dictionary of parameters. + """ + pass + + @staticmethod + def get_mpc_policy_class(class_name: str) -> Type: + """ + Return MPC policy class from class name. + + :param class_name: String with MPC policy class name. + :return: MPC policy class. + """ + return FoxMpcPolicy + + def _fill_init_big_step_policy_kwargs(self): + """ + Build dictionary of parameters to be passed when constructing a BigStepLayeredPolicy object. + + :return: Dictionary of parameters. + """ + kwargs_init = { + 'cost_per_buffer': self.env.cost_per_buffer, + 'constituency_matrix': self.env.constituency_matrix, + 'demand_rate': self.env.job_generator.demand_rate, + 'buffer_processing_matrix': self.env.job_generator.buffer_processing_matrix, + 'workload_mat': self.workload_tuple.workload_mat, + 'nu': self.workload_tuple.nu, + 'list_boundary_constraint_matrices': self.list_boundary_constraint_matrices, + 'policy_params': self.policy_params, + 'debug_info': self.debug_info + } + return kwargs_init + + def reset_mpc_variables(self, **kwargs) -> None: + self.mpc_variables["sum_actions"] = np.round(max(1, self.num_steps_to_recompute_policy) + * self.current_policy) + if self.debug_info: + print("Current policy: ", self.current_policy.ravel()) + print("Current sum of actions: ", self.mpc_variables["sum_actions"].ravel()) + if "total_fluid_sum_actions" in self.mpc_variables: + print("Total sum of actions: ", self.mpc_variables["total_sum_actions"].ravel()) + print("Total sum of fluid actions", + self.mpc_variables["total_fluid_sum_actions"].ravel().astype(int)) + print("Actual to fluid ratio: ", + (self.mpc_variables["total_sum_actions"] / + self.mpc_variables["total_fluid_sum_actions"]).flatten(), "\n") + + def update_mpc_variables(self, **kwargs) -> None: + assert "actions" in kwargs + actions = kwargs["actions"] + self.mpc_variables["sum_actions"] -= actions + + # The section below collects diagnostic information and is not used by the policy + if "total_sum_actions" in self.mpc_variables: + self.mpc_variables["total_sum_actions"] += actions + self.mpc_variables["total_timesteps"] += 1 + else: + self.mpc_variables["total_sum_actions"] = actions.copy() + self.mpc_variables["total_timesteps"] = 1 + # The section above collects diagnostic information and is not used by the policy + print(self.mpc_variables["total_sum_actions"].ravel()) + print(np.sqrt(self.mpc_variables["total_sum_actions"][4]/self.mpc_variables["total_sum_actions"][3])) + + def query_hedgehog_policy(self, + state: types.StateSpace, + env: crw.ControlledRandomWalk, + safety_stocks_vec: types.ResourceSpace, + draining_time_solver: str, + reporter: Optional[rep.Reporter]) \ + -> Tuple[types.ActionProcess, int]: + """ + Return activity rates for the current state and their horizon. + + :param state: current state. + :param env: the environment specifying the topology and constraints. + :param safety_stocks_vec: Safety stocks vector. + :param draining_time_solver: Convex solver for computing the minimal draining time. + :param reporter: reporter to store all data. + :return: (z_star, horizon) + - z_star: Activity rates. + - horizon: Horizon for which the activity rates have been computed. + """ + strategic_idling_tuple = self.strategic_idling_object.get_allowed_idling_directions(state, + safety_stocks_vec) + + draining_bottlenecks = get_dynamic_bottlenecks( + strategic_idling_tuple.w, self.workload_tuple.workload_mat, self.workload_tuple.load) + + horizon = 1 + + # Find activity rates for some horizon given nonidling and safety stock penalties. + kwargs = { + 'state': state, + 'safety_stocks_vec': safety_stocks_vec, + 'x_eff': strategic_idling_tuple.x_eff, + 'x_star': strategic_idling_tuple.x_star, + 'w_star': strategic_idling_tuple.w_star-strategic_idling_tuple.w, + 'k_idling_set': strategic_idling_tuple.k_idling_set, + 'draining_bottlenecks': draining_bottlenecks, + 'horizon': horizon, + 'demand_plan': self.get_demand_plan() + } + + if self.debug_info: + print(f"horizon: {horizon}") + print(f"z_star: {np.squeeze(z_star)}") + + if reporter is not None: + stored_vars = {'strategic_idling_tuple': strategic_idling_tuple, 'horizon': horizon} + reporter.store(**stored_vars) + + return None, None, kwargs + + + def map_state_to_actions(self, state: types.StateSpace, **override_args: Any) \ + -> types.ActionProcess: + """ + Returns actions (possibly many) given current state. Can take a kwarg dictionary + of overriding arguments that may be policy specific. + + :param state: Current state of the system. + :return: Schedule of actions. + """ + # Compute safety stock target. + safety_stocks_vec = safety_stocks.obtain_safety_stock_vector( + self.theta, self.load_ph, self.sigma_2_ph, state, self.debug_info) + if self.env.model_type == "pull": + safety_stocks_vec += self.obtain_safety_stock_for_surplus_buffers() + + # Recompute activity rates + args = { + "state": state, + "env": self.env, + "safety_stocks_vec": safety_stocks_vec, + "draining_time_solver": self.policy_params.convex_solver, + "reporter": None + } + args.update(override_args) + _, _, kwargs = self.query_hedgehog_policy(**args) + self.reset_mpc_variables() + + # Store retrospectively the actual number of MPC steps performed in the previous + # iteration before recomputing the activity rates. We skip zero since it corresponds + # with the first iteration, before actually having performed any action yet. + if self.actual_num_mpc_steps > 0 and args['reporter'] is not None: + stored_vars = {'num_mpc_steps': self.actual_num_mpc_steps} + args['reporter'].store(**stored_vars) + self.actual_num_mpc_steps = 0 + + r_idling_set = self._get_resource_idling_set(kwargs['k_idling_set'], kwargs['draining_bottlenecks']) + draining_resources = set() + for w_dir in kwargs['draining_bottlenecks']: + draining_resources = draining_resources.union(self.w_dirs_to_resources[w_dir]) + + # Obtain physically feasible actions from MPC policy. + actions = self.mpc_policy.obtain_actions( + state=state, + x_star = kwargs['x_star'], + x_eff = kwargs['x_eff'], + w_star = kwargs['w_star'], + r_idling_set = r_idling_set, + draining_resources = draining_resources) + actions.setflags(write=False) + + # Update remaining number of actions to be performed, countdown before recomputing activity + # rates, and actual number of steps following the current activity rates. + self.update_mpc_variables(actions=actions) + self.actual_num_mpc_steps += 1 + + return actions + + def _get_resource_idling_set(self,k_idling_set, draining_bottlenecks): + r_idling_set = set() + for w_dir in k_idling_set: + if w_dir in draining_bottlenecks: + continue + r_idling_set = r_idling_set.union(self.w_dirs_to_resources[w_dir]) + + return r_idling_set diff --git a/src/snc/agents/hedgehog/hh_agents/hedgehog_agent_interface.py b/src/snc/agents/hedgehog/hh_agents/hedgehog_agent_interface.py index ba6bbf1..de31860 100644 --- a/src/snc/agents/hedgehog/hh_agents/hedgehog_agent_interface.py +++ b/src/snc/agents/hedgehog/hh_agents/hedgehog_agent_interface.py @@ -20,6 +20,7 @@ from snc.agents.hedgehog.policies.big_step_policy import BigStepPolicy from snc.agents.hedgehog.strategic_idling.strategic_idling import StrategicIdlingCore from snc.agents.hedgehog.strategic_idling.strategic_idling_foresight import StrategicIdlingForesight +from snc.agents.hedgehog.strategic_idling.strategic_idling_fox import StrategicIdlingFox from snc.agents.hedgehog.strategic_idling.strategic_idling_hedgehog_gto import \ StrategicIdlingHedgehogGTO, \ StrategicIdlingHedgehogGTO2, \ @@ -97,6 +98,7 @@ def __init__(self, assert strategic_idling_class in [StrategicIdlingCore, StrategicIdlingForesight, + StrategicIdlingFox, StrategicIdlingGTO, StrategicIdlingHedgehogGTO, StrategicIdlingHedgehogGTO2, @@ -250,6 +252,9 @@ def _initialize_strategic_idling_object(self) -> None: if self.strategic_idling_class == StrategicIdlingForesight: init_vars.update({'policy_object': self.policy_obj}) + if self.strategic_idling_class == StrategicIdlingFox: + init_vars.update({'list_boundary_constraint_matrices': self.env.list_boundary_constraint_matrices}) + self.strategic_idling_object = self.strategic_idling_class(**init_vars) @staticmethod @@ -284,8 +289,8 @@ def map_workload_to_physical_resources(workload_tuple: WorkloadTuple, sigma_2 = workload_cov.diagonal() # Variance of the workload process load_sig = safety_stocks.map_workload_to_physical_resources_with_conservative_max_heuristic( workload_tuple.nu, workload_tuple.load, sigma_2) - load_ph, sigma_2_ph = load_sig - return load_ph, sigma_2_ph + load_ph, sigma_2_ph, w_dirs_to_resources = load_sig + return load_ph, sigma_2_ph, w_dirs_to_resources def perform_offline_calculations(self) -> None: """ @@ -299,7 +304,8 @@ def perform_offline_calculations(self) -> None: ) # Initialise policy, as implemented by children classes. - self.policy_obj = self.activity_rates_policy_class(**self.serialise_init_policy_kwargs()) + if not self.strategic_idling_class == StrategicIdlingFox: + self.policy_obj = self.activity_rates_policy_class(**self.serialise_init_policy_kwargs()) self.workload_cov = self.asymptotic_workload_cov_estimator.estimate_asymptotic_workload_cov( self.env.job_generator.buffer_processing_matrix, @@ -314,8 +320,8 @@ def perform_offline_calculations(self) -> None: self._initialize_strategic_idling_object() - self.load_ph, self.sigma_2_ph = self.map_workload_to_physical_resources(self.workload_tuple, - self.workload_cov) + self.load_ph, self.sigma_2_ph, self.w_dirs_to_resources = self.map_workload_to_physical_resources(self.workload_tuple, + self.workload_cov) # Reset trigger to recompute big step policy LP and remaining set of actions (they might've # been set if the estimation of the asymptotic covariance was done with this self agent). @@ -396,7 +402,7 @@ def query_hedgehog_policy(self, stored_vars = {'strategic_idling_tuple': strategic_idling_tuple, 'horizon': horizon} reporter.store(**stored_vars) - return z_star, horizon + return z_star, horizon, kwargs @staticmethod def get_num_steps_to_recompute_policy(current_horizon: float, @@ -448,7 +454,7 @@ def map_state_to_actions(self, state: types.StateSpace, **override_args: Any) \ "reporter": None } args.update(override_args) - self.current_policy, current_horizon = self.query_hedgehog_policy(**args) + self.current_policy, current_horizon, kwargs = self.query_hedgehog_policy(**args) # Reset countdown timer to recomputing the activity rates. self.num_steps_to_recompute_policy = self.get_num_steps_to_recompute_policy( current_horizon, diff --git a/src/snc/agents/hedgehog/safety_stocks.py b/src/snc/agents/hedgehog/safety_stocks.py index cbb8f70..a1110f7 100644 --- a/src/snc/agents/hedgehog/safety_stocks.py +++ b/src/snc/agents/hedgehog/safety_stocks.py @@ -2,6 +2,7 @@ from typing import List, Tuple import snc.utils.snc_types as types import math +from collections import defaultdict def map_workload_to_physical_resources_with_conservative_max_heuristic( @@ -30,15 +31,18 @@ def map_workload_to_physical_resources_with_conservative_max_heuristic( # Initialise output load_ph = np.zeros((num_resources, 1)) sigma_2_ph = np.zeros((num_resources, 1)) + w_dirs_to_resources = defaultdict(set) for s in range(num_resources): - for load_wl_i, sigma_2_wl_i, nu_i in zip(load_wl, sigma_2_wl, nu): + for i, (load_wl_i, sigma_2_wl_i, nu_i) in enumerate(zip(load_wl, sigma_2_wl, nu)): if nu_i[s] > 0: load_ph[s] = max(load_ph[s], load_wl_i) sigma_2_ph[s] = max(sigma_2_ph[s], sigma_2_wl_i) + w_dirs_to_resources[i].add(s) + assert np.all(load_ph >= 0) assert np.all(sigma_2_ph >= 0) - return load_ph, sigma_2_ph + return load_ph, sigma_2_ph, w_dirs_to_resources def obtain_safety_stock_per_resource(theta: float, load_ph_s: float, sigma_2_ph_s: float, diff --git a/src/snc/agents/hedgehog/strategic_idling/compute_primal_effective_cost.py b/src/snc/agents/hedgehog/strategic_idling/compute_primal_effective_cost.py new file mode 100644 index 0000000..63be39d --- /dev/null +++ b/src/snc/agents/hedgehog/strategic_idling/compute_primal_effective_cost.py @@ -0,0 +1,77 @@ +from copy import deepcopy +from typing import Tuple, Dict +import cvxpy as cvx +import numpy as np +import snc.utils.snc_types as types +from snc.agents.solver_names import SolverNames +from snc.simulation.store_data.numpy_encoder import clean_to_serializable + + +class ComputePrimalEffectiveCost: + + def __init__(self, + workload_mat: types.WorkloadMatrix, + cost_per_buffer: types.StateSpace, + list_boundary_constraint_matrices, + convex_solver: str): + + assert convex_solver in SolverNames.CVX + self.convex_solver = convex_solver + + self.workload_mat = workload_mat + self.cost_per_buffer = cost_per_buffer + self.list_boundary_constraint_matrices = list_boundary_constraint_matrices + + self._lp_problem, self._x_eff_var, self._w_param, self._safety_stocks_vec = \ + self._create_compute_primal_effective_cost_lp_program() + + def _create_compute_primal_effective_cost_lp_program(self) \ + -> Tuple[cvx.Problem, cvx.Variable, cvx.Parameter]: + """ + Creates the linear program that will defines the dual of the effective cost problem. + + :return: (lp_problem, c_bar_var, w_param, constraints): + - lp_problem: Linear program structure. + - c_bar_var: c_bar vector variable normal to the the hyperplane that defines the + effective cost for the current workload. + - w_param: Current workload parameter. + """ + num_workload_dirs,num_buffers = self.workload_mat.shape + x_eff_var = cvx.Variable((num_buffers, 1),nonneg=True) + w_param = cvx.Parameter((num_workload_dirs, 1)) + safety_stocks_vec = cvx.Parameter((num_workload_dirs, 1)) + + objective = cvx.Minimize(self.cost_per_buffer.T @ x_eff_var) + constraints = [self.workload_mat @ x_eff_var == w_param] + + a_mat = np.vstack(self.list_boundary_constraint_matrices) + + constraints.append(a_mat @ x_eff_var >= safety_stocks_vec) + constraints.append(x_eff_var >= 1) + + lp_problem = cvx.Problem(objective, constraints) + return lp_problem, x_eff_var, w_param, safety_stocks_vec + + def solve(self, w: types.WorkloadSpace, safety_stocks_vec): + """ + Solves the linear program with the current workload using warm start. + + :param w: Current workload. + :return: (c_bar, x_star, eff_cost) + - c_bar: vector defining level set of the effective cost at w. None is returned if the + optimisation is unsuccessful. + - x_star = effective state, solution to primal program, + - eff_cost = actual value of the effective cost. + """ + self._w_param.value = w # Update parameter value with actual current workload. + self._safety_stocks_vec.value = safety_stocks_vec + eff_cost = self._lp_problem.solve(solver=eval(self.convex_solver), warm_start=True) + c_bar = self._lp_problem.constraints[0].dual_value + x_eff = self._x_eff_var.value + return c_bar, x_eff, eff_cost + + def to_serializable(self) -> Dict: + """ + Return a serializable object, that can be used by a JSON encoder. + """ + return clean_to_serializable(self) diff --git a/src/snc/agents/hedgehog/strategic_idling/strategic_idling.py b/src/snc/agents/hedgehog/strategic_idling/strategic_idling.py index 9c8084a..2183e5c 100644 --- a/src/snc/agents/hedgehog/strategic_idling/strategic_idling.py +++ b/src/snc/agents/hedgehog/strategic_idling/strategic_idling.py @@ -12,6 +12,8 @@ StrategicIdlingOutput = NamedTuple('StrategicIdlingOutput', [('w', WorkloadSpace), + ('x_eff', StateSpace), + ('x_star', StateSpace), ('beta_star', float), ('k_idling_set', Array1D), ('sigma_2_h', float), @@ -106,6 +108,8 @@ def _get_null_strategic_idling_output(self, **overrides) -> StrategicIdlingOutpu """ assert 'w' in overrides, "Current workload variable is not being returned" w = overrides['w'] + x_eff = overrides.get('x_eff', np.array([])) + x_star = overrides.get('x_star', np.array([])) beta_star = overrides.get('beta_star', 0) k_idling_set = overrides.get('k_idling_set', np.array([])) sigma_2_h = overrides.get('sigma_2_h', 0) @@ -121,7 +125,7 @@ def _get_null_strategic_idling_output(self, **overrides) -> StrategicIdlingOutpu delta_h = overrides.get('delta_h', 0) lambda_star = overrides.get('lambda_star', 0) theta_roots = overrides.get('theta_roots', None) - return StrategicIdlingOutput(w, beta_star, k_idling_set, sigma_2_h, psi_plus, + return StrategicIdlingOutput(w, x_eff, x_star, beta_star, k_idling_set, sigma_2_h, psi_plus, height_process, w_star, c_plus, c_bar, psi_plus_cone_list, beta_star_cone_list, delta_h, lambda_star, theta_roots) @@ -217,8 +221,8 @@ def _get_level_set_for_current_workload(self, w: WorkloadSpace) -> Optional[Work :return: c_bar: vector defining level set of the effective cost at current w. None is returned if the optimisation is unsuccessful. """ - c_bar, _, _ = self.c_bar_solver.solve(w) - return c_bar + c_bar, x_eff, _ = self.c_bar_solver.solve(w) + return c_bar, x_eff @staticmethod def _get_vector_defining_possible_idling_direction(w_star: WorkloadSpace, @@ -253,7 +257,7 @@ def _find_workload_with_min_eff_cost_by_idling(self, w: WorkloadSpace) -> Worklo w_star = self._workload_mat @ x_star # Workload in the boundary of the monotone region. tol = 1e-6 assert np.all(w_star >= w - tol) - return w_star + return w_star, x_star def _non_negative_workloads(self, w: WorkloadSpace, eps: float = 1e-6) -> Dict[str, Any]: """ @@ -269,28 +273,29 @@ def _non_negative_workloads(self, w: WorkloadSpace, eps: float = 1e-6) -> Dict[s if not np.any(w > eps): return {'w': w, 'w_star': w, 'k_idling_set': np.array([])} - c_bar = self._get_level_set_for_current_workload(w) + c_bar, x_eff = self._get_level_set_for_current_workload(w) if self._is_infeasible(c_bar): - return {'w': w, 'w_star': w, 'k_idling_set': np.array([])} + return {'w': w, 'w_star': w, 'x_eff': x_eff, 'k_idling_set': np.array([])} elif self._is_defining_a_monotone_region(c_bar): current_workload_vars = {'w': w, 'w_star': w, 'c_bar': c_bar, + 'x_eff': x_eff, 'k_idling_set': np.array([])} return current_workload_vars - w_star = self._find_workload_with_min_eff_cost_by_idling(w) + w_star, x_star = self._find_workload_with_min_eff_cost_by_idling(w) if self._is_w_inside_monotone_region(w, w_star, c_bar): # Since c_bar doesn't define a monotone region, w is already at the boundary. - current_workload_vars = {'w': w, 'w_star': w_star, 'c_bar': c_bar, - 'k_idling_set': np.array([])} + current_workload_vars = {'w': w, 'w_star': w_star, 'c_bar': c_bar, 'x_eff': x_eff, + 'x_star': x_star, 'k_idling_set': np.array([])} return current_workload_vars v_star = self._get_vector_defining_possible_idling_direction(w_star, w) k_idling_set = np.where(v_star > eps)[0] - current_workload_vars = {'w': w, 'w_star': w_star, 'c_bar': c_bar, 'v_star': v_star, - 'k_idling_set': k_idling_set} + current_workload_vars = {'w': w, 'w_star': w_star, 'c_bar': c_bar, 'v_star': v_star, 'x_star': x_star, + 'x_eff': x_eff, 'k_idling_set': k_idling_set} return current_workload_vars diff --git a/src/snc/agents/hedgehog/strategic_idling/strategic_idling_fox.py b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_fox.py new file mode 100644 index 0000000..fc0fe3e --- /dev/null +++ b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_fox.py @@ -0,0 +1,167 @@ +import cvxpy as cvx +import numpy as np +from typing import Optional, Set, Dict, Any + +from snc.agents.hedgehog.minimal_draining_time import compute_minimal_draining_time_from_workload \ + as compute_min_drain_time +from snc.agents.hedgehog.strategic_idling.strategic_idling import StrategicIdlingOutput, \ + StrategicIdlingCore +from snc.agents.hedgehog.params import StrategicIdlingParams +from snc.agents.hedgehog.strategic_idling.compute_dual_effective_cost \ + import ComputeDualEffectiveCost +from snc.agents.hedgehog.strategic_idling.compute_primal_effective_cost \ + import ComputePrimalEffectiveCost +from snc.agents.hedgehog.strategic_idling.strategic_idling_hedging import StrategicIdlingHedging +from snc.agents.hedgehog.strategic_idling.strategic_idling_utils import get_dynamic_bottlenecks, \ + is_pull_model +import snc.utils.snc_types as types +from snc.utils.snc_types import WorkloadSpace, StateSpace + + +class StrategicIdlingFox(StrategicIdlingHedging): + + def __init__(self, + workload_mat: types.WorkloadMatrix, + neg_log_discount_factor: float, + load: WorkloadSpace, + cost_per_buffer: types.StateSpace, + model_type: str, + list_boundary_constraint_matrices, + strategic_idling_params: Optional[StrategicIdlingParams] = None, + workload_cov: Optional[types.WorkloadCov] = None, + debug_info: bool = False) -> None: + """ + StrategicIdling class is responsible for online identification of idling directions for + bottlenecks to reduce effective running cost in the network. + + :param workload_mat: workload matrix, with rows being workload vectors. + :param neg_log_discount_factor: negative log of the discount factor given by environment. + :param load: vector with loads for every workload vector. + :param cost_per_buffer: cost per unit of inventory per buffer. + :param model_type: String indicating if this is a `'pull'` or `'push'` model. + :param strategic_idling_params: tolerance levels, convex solver and other params + for navigating effective cost space. + :param workload_cov: asymptotic covariance of the workload process. + :param debug_info: Boolean flag that indicates whether printing useful debug info. + """ + self._workload_mat = workload_mat + self._load = load + self._cost_per_buffer = cost_per_buffer + + assert model_type in ['push', 'pull'] + self.model_type = model_type + + self.check_strategic_idling_parameters(strategic_idling_params) + self.strategic_idling_params = strategic_idling_params + + self.debug_info = debug_info + + self._num_bottlenecks, self._num_buffers = workload_mat.shape + + convex_solver = strategic_idling_params.convex_solver + self.c_bar_solver = ComputeDualEffectiveCost(workload_mat, cost_per_buffer, convex_solver) + + self._workload_cov = workload_cov + self._neg_log_discount_factor = neg_log_discount_factor + + self.check_strategic_idling_parameters(strategic_idling_params) + self.strategic_idling_params = strategic_idling_params + + self._psi_plus_cone_list: Optional[List[WorkloadSpace]] = None + self._beta_star_cone_list: Optional[List[float]] = None + + # Create linear programs that will be used at each iteration. + convex_solver = strategic_idling_params.convex_solver + self.c_minus_solver = ComputeDualEffectiveCost(workload_mat, cost_per_buffer, convex_solver) + self.c_plus_solver = ComputeDualEffectiveCost(workload_mat, cost_per_buffer, convex_solver) + + if workload_cov is not None: + self.update_workload_cov(workload_cov) + + + self.list_boundary_constraint_matrices = list_boundary_constraint_matrices + self.c_bar_solver = ComputePrimalEffectiveCost(workload_mat, cost_per_buffer, list_boundary_constraint_matrices, convex_solver) + self._w_star_lp_problem, self._x_star, self._w_param, self._safety_stocks_param,= \ + self._create_find_workload_with_min_eff_cost_by_idling_lp_program() + + def _find_workload_with_min_eff_cost_by_idling(self, w: WorkloadSpace) -> WorkloadSpace: + self._w_param.value = w + self._safety_stocks_param.value = np.zeros_like(self._safety_stocks_vec) + self._w_star_lp_problem.solve(solver=eval(self.strategic_idling_params.convex_solver), + warm_start=True) # Solve LP. + x_star = self._x_star.value + w_star = self._workload_mat @ x_star # Workload in the boundary of the monotone region. + self._safety_stocks_param.value = self._safety_stocks_vec + self._w_star_lp_problem.solve(solver=eval(self.strategic_idling_params.convex_solver), + warm_start=True) # Solve LP. + x_star = self._x_star.value + tol = 1e-6 + assert np.all(w_star >= w - tol) + return w_star, x_star + + def _get_level_set_for_current_workload(self, w: WorkloadSpace) -> Optional[WorkloadSpace]: + """ + The effective cost can be represented as a piecewise linear function, + with coefficients given by the vertexes of the feasible set of the dual + program of the LP that computes the effective cost. Indeed, the solution to such + dual program for a given w, gives the linear coefficient at w. + + :param w: current state in workload space, i.e. w = Xi x. + :return: c_bar: vector defining level set of the effective cost at current w. None is + returned if the optimisation is unsuccessful. + """ + c_bar, x_eff, _ = self.c_bar_solver.solve(w, self._safety_stocks_vec) + return c_bar, x_eff + + def _create_find_workload_with_min_eff_cost_by_idling_lp_program(self): + x_var = cvx.Variable((self._num_buffers, 1), nonneg=True) # Variable + w_par = cvx.Parameter((self._num_bottlenecks, 1)) # Parameter + safety_stocks_vec = cvx.Parameter((self._num_bottlenecks, 1)) + penalty_coeff_w_star = self.strategic_idling_params.penalty_coeff_w_star + objective = cvx.Minimize( + self._cost_per_buffer.T @ x_var + + penalty_coeff_w_star * cvx.sum(self._workload_mat @ x_var - w_par)) + constraints = [self._workload_mat @ x_var >= w_par] + constraints.append(self._workload_mat[0,:] @ x_var == w_par[0]) + + a_mat = np.vstack(self.list_boundary_constraint_matrices) + constraints.append(a_mat @ x_var >= safety_stocks_vec) + + constraints.append(x_var >= 1) + constraints.append(x_var[1:4] >= 20) + constraints.append(x_var[3] >= 25) + #constraints.append(x_var[2:4] >= 50) + + + lp_problem = cvx.Problem(objective, constraints) + return lp_problem, x_var, w_par, safety_stocks_vec + + def get_allowed_idling_directions(self, x: StateSpace, safety_stocks_vec) -> StrategicIdlingOutput: + """ + Method projects current worload onto the full monotone effective cost cone, or + projects onto the precomputed envelope of the monotone effective cost cone. + + :param x: current buffer state of the network. + :return: set of allowed idling resources with auxiliary variables + """ + w = self._workload_mat @ x + self._safety_stocks_vec = safety_stocks_vec + self._verify_offline_preliminaries() + if self._is_negative_orthant(w): + idling_decision_dict = self._negative_workloads(w) + else: + current_workload_variables = self._non_negative_workloads(w) + + if self._is_decision_not_to_idle(current_workload_variables['k_idling_set']): + idling_decision_dict = current_workload_variables + else: + idling_decision_dict = self._add_standard_hedging(w, current_workload_variables) + + idling_decision = self._get_null_strategic_idling_output(**idling_decision_dict) + + if self.debug_info: + print(f"beta_star: {idling_decision.beta_star}, " + f"k_iling_set: {idling_decision.k_idling_set}, " + f"sigma_2_h: {idling_decision.sigma_2_h}, " + f"delta_h: {idling_decision.delta_h}") + return idling_decision diff --git a/src/snc/agents/hedgehog/strategic_idling/strategic_idling_hedging.py b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_hedging.py index e597f3e..5c029da 100644 --- a/src/snc/agents/hedgehog/strategic_idling/strategic_idling_hedging.py +++ b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_hedging.py @@ -481,15 +481,12 @@ def _add_standard_hedging(self, w: WorkloadSpace, hedging_case = 'empty_interior' height_process = self._compute_height_process(w, psi_plus) + current_workload_variables['psi_plus'] = psi_plus + current_workload_variables['c_plus'] = c_plus + current_workload_variables['hedging_case'] = hedging_case + current_workload_variables['height_process'] = height_process if self._is_w_inside_artificial_monotone_region(w, psi_plus): - current_workload_variables = {'w': w, - 'w_star': w_star, - 'c_plus': c_plus, - 'c_bar': c_bar, - 'psi_plus': psi_plus, - 'height_process': height_process, - 'hedging_case': hedging_case} return current_workload_variables beta_star, sigma_2_h, delta_h, lambda_star, theta_roots \ @@ -500,21 +497,13 @@ def _add_standard_hedging(self, w: WorkloadSpace, # Update cone envelope with the current closest face if needed. self._add_face_to_cone_envelope(psi_plus, beta_star) - current_workload_variables = {'w': w, - 'beta_star': beta_star, - 'k_idling_set': k_idling_set, - 'sigma_2_h': sigma_2_h, - 'psi_plus': psi_plus, - 'height_process': height_process, - 'w_star':w_star, - 'c_plus': c_plus, - 'c_bar': c_bar, - 'psi_plus_cone_list': self.psi_plus_cone_list, - 'beta_star_cone_list':self.beta_star_cone_list, - 'delta_h':delta_h, - 'lambda_star': lambda_star, - 'theta_roots': theta_roots, - 'hedging_case': hedging_case} + current_workload_variables['beta_star'] = beta_star + current_workload_variables['k_idling_set'] = k_idling_set + current_workload_variables['sigma_2_h'] = sigma_2_h + current_workload_variables['delta_h'] = delta_h + current_workload_variables['lambda_star'] = lambda_star + current_workload_variables['theta_roots'] = theta_roots + return current_workload_variables diff --git a/src/snc/agents/hedgehog/strategic_idling/strategic_idling_horizon.py b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_horizon.py index 974a8c1..644cd57 100644 --- a/src/snc/agents/hedgehog/strategic_idling/strategic_idling_horizon.py +++ b/src/snc/agents/hedgehog/strategic_idling/strategic_idling_horizon.py @@ -47,13 +47,13 @@ def _non_negative_workloads(self, w: WorkloadSpace, eps: float = 1e-6) -> Dict[s """ w_drift = w - self._horizon * self.drift - c_bar = self._get_level_set_for_current_workload(w_drift) + c_bar, x_eff = self._get_level_set_for_current_workload(w_drift) if not self._is_infeasible(c_bar) and self._is_defining_a_monotone_region(c_bar): current_workload_vars = {'w': w, 'w_star': w_drift, 'c_bar': c_bar, 'k_idling_set': np.array([])} return current_workload_vars - w_star = self._find_workload_with_min_eff_cost_by_idling(w_drift) + w_star, x_star = self._find_workload_with_min_eff_cost_by_idling(w_drift) v_star = self._get_vector_defining_possible_idling_direction(w_star, w_drift) k_idling_set = np.where(v_star > eps)[0] diff --git a/src/snc/agents/steady_state_agents/steady_state_policy_agent.py b/src/snc/agents/steady_state_agents/steady_state_policy_agent.py index 7a8ba52..e800fb1 100644 --- a/src/snc/agents/steady_state_agents/steady_state_policy_agent.py +++ b/src/snc/agents/steady_state_agents/steady_state_policy_agent.py @@ -81,7 +81,12 @@ def map_state_to_actions(self, state: types.StateSpace, **override_args: Dict) \ # Obtain physically feasible actions from MPC policy. actions = self.mpc_policy.obtain_actions( - state=state, mpc_variables=self.mpc_variables, + state=state, + x_star = state, + x_eff = state, + r_idling_set = np.array([]), + draining_resources = set(), + mpc_variables=self.mpc_variables, num_steps_to_recompute_policy=self.num_steps_to_recompute_policy, z_star=self.policy, demand_rate=self.env.job_generator.demand_rate) diff --git a/src/snc/simulation/utils/load_agents.py b/src/snc/simulation/utils/load_agents.py index 6fbebaf..4e04d21 100644 --- a/src/snc/simulation/utils/load_agents.py +++ b/src/snc/simulation/utils/load_agents.py @@ -5,6 +5,7 @@ from snc.agents.general_heuristics.random_nonidling_agent import RandomNonIdlingAgent import snc.agents.hedgehog.hh_agents.hedgehog_agent_interface as hh_int from snc.agents.hedgehog.hh_agents.big_step_hedgehog_agent import BigStepHedgehogAgent +from snc.agents.hedgehog.hh_agents.fox_agent import FoxAgent from snc.agents.hedgehog.hh_agents.pure_feedback_mip_hedgehog_agent \ import PureFeedbackMIPHedgehogAgent from snc.agents.hedgehog.hh_agents.pure_feedback_stationary_hedgehog_agent \ @@ -18,6 +19,7 @@ WorkloadRelaxationParams from snc.agents.hedgehog.strategic_idling.strategic_idling import StrategicIdlingCore from snc.agents.hedgehog.strategic_idling.strategic_idling_foresight import StrategicIdlingForesight +from snc.agents.hedgehog.strategic_idling.strategic_idling_fox import StrategicIdlingFox from snc.agents.hedgehog.strategic_idling.strategic_idling_hedgehog_gto import \ StrategicIdlingHedgehogGTO, StrategicIdlingHedgehogGTO2, StrategicIdlingHedgehogNaiveGTO from snc.agents.hedgehog.strategic_idling.strategic_idling_hedging import StrategicIdlingHedging @@ -43,6 +45,7 @@ def get_strategic_idling_class(si_class_name: str) -> Type[StrategicIdlingCore]: classes = [ StrategicIdlingCore, StrategicIdlingForesight, + StrategicIdlingFox, StrategicIdlingHedgehogGTO, StrategicIdlingHedgehogGTO2, StrategicIdlingHedgehogNaiveGTO, @@ -206,6 +209,41 @@ def build_pf_stationary_hedgehog( mpc_seed ) +def build_fox_agent( + env: ControlledRandomWalk, + discount_factor: float, + hh_overrides: Dict[str, Any], + debug_info: bool = False, agent_seed: Optional[int] = None, + mpc_seed: Optional[int] = None) -> hh_int.HedgehogAgentInterface: + """ + Sets up an instantiation of a Pure Feedback with Stationary MPC Hedgehog agent. + + :param env: Environment the hedgehog agent will run in. + :param discount_factor: Discount factor to future rewards/costs. + :param hh_overrides: Dictionary of hedgehog parameter overrides. + :param debug_info: Boolean flag that indicates whether printing useful debug info. + :param agent_seed: Agent random seed. + :param mpc_seed: MPC random seed. + :return: A PureFeedbackStationaryHedgehogAgent agent. + """ + ac_params, wk_params, si_params, po_params, hh_params, si_class, dp_params, name \ + = get_hedgehog_hyperparams(**hh_overrides) + return FoxAgent( + env, + discount_factor, + wk_params, + hh_params, + ac_params, + si_params, + po_params, + si_class, + dp_params, + name, + debug_info, + agent_seed, + mpc_seed + ) + def build_pf_mip_hedgehog(env: ControlledRandomWalk, discount_factor: float, @@ -358,7 +396,7 @@ def get_agent(agent_name: str, env: ControlledRandomWalk, **kwargs: Any) \ """ if agent_name not in AGENT_CONSTRUCTORS: raise NotImplementedError(f'Requested agent "{agent_name}" not implemented.') - if agent_name in ['bs_hedgehog', 'pf_stationary_hedgehog', 'pf_mip_hedgehog']: + if agent_name in ['bs_hedgehog', 'pf_stationary_hedgehog', 'pf_mip_hedgehog', 'fox']: return AGENT_CONSTRUCTORS[agent_name](env, discount_factor=kwargs['discount_factor'], hh_overrides=kwargs['hh_overrides'], @@ -400,10 +438,11 @@ def get_all_agent_names(env_name: str, with_rl_agent: bool = False) -> List[str] return agent_list -HEDGEHOG_AGENTS = ['bs_hedgehog', 'pf_stationary_hedgehog', 'pf_mip_hedgehog'] +HEDGEHOG_AGENTS = ['bs_hedgehog', 'pf_stationary_hedgehog', 'pf_mip_hedgehog', 'fox'] AGENT_CONSTRUCTORS: Dict[str, Callable] = { 'bs_hedgehog': build_bs_hedgehog_agent, + 'fox': build_fox_agent, 'pf_mip_hedgehog': build_pf_mip_hedgehog, 'pf_stationary_hedgehog': build_pf_stationary_hedgehog, 'distribution_with_rebalancing_heuristic': DistributionWithRebalancingLocalPriorityAgent,