# Copyright (c) 2020, salesforce.com, inc. # All rights reserved. # SPDX-License-Identifier: BSD-3-Clause # For full license text, see the LICENSE file in the repo root # or https://opensource.org/licenses/BSD-3-Clause import numpy as np from ai_economist.foundation.base.base_component import ( BaseComponent, component_registry, ) from ai_economist.foundation.entities.resources import Resource, resource_registry @component_registry.add class Craft(BaseComponent): """ Allows mobile agents to build house landmarks in the world using stone and wood, earning income. Can be configured to include heterogeneous building skill where agents earn different levels of income when building. Args: commodities (list(str)): list of commodities that can be crafted in the local world payment_max_skill_multiplier (int): Maximum skill multiplier that an agent can sample. Must be >= 1. Default is 1. skill_dist (str): Distribution type for sampling skills. Default ("none") gives all agents identical skill equal to a multiplier of 1. "pareto" and "lognormal" sample skills from the associated distributions. build_labor (float): Labor cost associated with building a house. Must be >= 0. Default is 10. """ name = "Craft" component_type = "Build" required_entities = ["Coin", "Labor"] agent_subclasses = ["BasicMobileAgent"] commodities=[] def __init__( self, *base_component_args, commodities=[], max_skill_amount_benefit=1, max_skill_labour_benefit=1, skill_dist="none", **base_component_kwargs ): assert len(commodities)>0 #setup commodities self.recip_map={} self.commodities=[] for v in commodities: res_class=resource_registry.get(v) res=res_class() if res.craft_recp!=None: # is craftable assert res.craft_recp!={} assert res.craft_labour_base >= 0 self.required_entities.append(v) self.recip_map[res.name]=res.craft_recp self.commodities.append(res) self.max_skill_amount_benefit=max_skill_amount_benefit self.max_skill_labour_benefit=max_skill_labour_benefit assert self.max_skill_amount_benefit >= 1 assert self.max_skill_labour_benefit <= 1 self.skill_dist = skill_dist.lower() assert self.skill_dist in ["none", "pareto"] self.sampled_skills = {} self.builds = [] super().__init__(*base_component_args, **base_component_kwargs) def agent_can_build(self, agent, res): """Return True if agent can actually build in its current location.""" # See if the agent has the resources necessary to complete the action if res in self.recip_map: recipe= self.recip_map[res] for resource, cost in recipe.items(): if agent.state["inventory"][resource] < cost: return False else: return True return False # Required methods for implementing components # -------------------------------------------- def get_n_actions(self, agent_cls_name): """ See base_component.py for detailed description. Add a single action (build) for mobile agents. """ # This component adds 1 action that mobile agents can take: build a house if agent_cls_name in self.agent_subclasses: return len(self.commodities) return None def get_additional_state_fields(self, agent_cls_name): """ See base_component.py for detailed description. For mobile agents, add state fields for building skill. """ if agent_cls_name not in self.agent_subclasses: return {} if agent_cls_name == "BasicMobileAgent": return {} raise NotImplementedError def component_step(self): """ See base_component.py for detailed description. Convert stone+wood to house+coin for agents that choose to build and can. """ world = self.world build = [] # Apply any building actions taken by the mobile agents for agent in world.get_random_order_agents(): action = agent.get_component_action(self.name) # This component doesn't apply to this agent! if action is None: continue # NO-OP! if action == 0: pass # Build! (If you can.) else: action-=1 comm=self.commodities[action] if self.agent_can_build(agent,comm.name): # Remove the resources for resource, cost in comm.craft_recp.items(): agent.state["inventory"][resource] -= cost # Receive crafted commodity agent.state["inventory"][comm.name] += agent.state["craft_amount"][comm.name] # Incur the labor cost for building agent.state["endogenous"]["Labor"] += agent.state["craft_labour"][comm.name] build.append( { "crafter": agent.idx, "craft_commodity": comm.name, "craft_skill": agent.state["craft_skill"][comm.name], "craft_amount": agent.state["craft_amount"][comm.name], "craft_labour": agent.state["craft_labour"][comm.name] } ) else: agent.bad_action=True self.builds.append(build) def generate_observations(self): """ See base_component.py for detailed description. Here, agents observe their build skill. The planner does not observe anything from this component. """ obs_dict = dict() for agent in self.world.agents: if agent.name in self.agent_subclasses: obs_dict[agent.idx]={} for k in self.commodities: obs_dict[agent.idx]["craft_skill_{}".format(k.name)] = agent.state["craft_skill"][k.name] return obs_dict def generate_masks(self, completions=0): """ See base_component.py for detailed description. Prevent building only if a landmark already occupies the agent's location. """ masks = {} # Mobile agents' build action is masked if they cannot build with their # current location and/or endowment for agent in self.world.agents: if agent.name in self.agent_subclasses: masks[agent.idx] = np.array([self.agent_can_build(agent,k.name) for k in self.commodities]) return masks # For non-required customization # ------------------------------ def get_metrics(self): """ Metrics that capture what happened through this component. Returns: metrics (dict): A dictionary of {"metric_name": metric_value}, where metric_value is a scalar. """ world = self.world """ build_stats = {a.idx: {"n_builds": 0} for a in world.agents} for builds in self.builds: for build in builds: idx = build["builder"] build_stats[idx]["n_builds"] += 1 out_dict = {} for a in world.agents: for k, v in build_stats[a.idx].items(): out_dict["{}/{}".format(a.idx, k)] = v num_houses = np.sum(world.maps.get("House") > 0) out_dict["total_builds"] = num_houses """ return {} def additional_reset_steps(self): """ See base_component.py for detailed description. Re-sample agents' building skills. """ world = self.world MSAB= self.max_skill_amount_benefit MSLB= self.max_skill_labour_benefit for agent in world.agents: if (agent.name not in self.agent_subclasses) | agent.is_setup: continue agent.state["craft_skill"]={} agent.state["craft_labour"]={} agent.state["craft_amount"]={} for comm in self.commodities: if self.skill_dist == "none": sampled_skill = 1 amount= 1 labour = 1 elif self.skill_dist == "pareto": labour = 1 sampled_skill = np.random.pareto(2) amount = 1+np.minimum(MSAB,(MSAB-1) * (sampled_skill) ) labour_modifier = 1 - np.minimum(1 - MSLB, (1 - MSLB) * sampled_skill) else: raise NotImplementedError agent.state["craft_skill"][comm.name]=sampled_skill agent.state["craft_labour"][comm.name]=comm.craft_labour_base*labour_modifier agent.state["craft_amount"][comm.name]=amount self.builds = [] def get_dense_log(self): """ Log builds. Returns: builds (list): A list of build events. Each entry corresponds to a single timestep and contains a description of any builds that occurred on that timestep. """ return self.builds