import agentpy as ap import math class FirmAgent(ap.Agent): def setup(self, code, name, type_region, revenue_log, a_lst_product): self.firm_network = self.model.firm_network self.product_network = self.model.product_network # self para self.code = code self.name = name self.type_region = type_region self.ori_size = revenue_log self.size = revenue_log self.a_lst_product = a_lst_product self.a_lst_product_removed = ap.AgentList(self.model, []) self.dct_prod_up_prod_stat = {} self.dct_prod_capacity = {} # para in trial self.dct_n_trial_up_prod_removed = {} self.dct_cand_alt_supply_up_prod_removed = {} self.dct_request_prod_from_firm = {} # external variable self.is_prf_size = self.model.is_prf_size self.is_prf_conn = bool(self.p.prf_conn) self.str_cap_limit_prob_type = str(self.p.cap_limit_prob_type) self.flt_cap_limit_level = float(self.p.cap_limit_level) self.flt_diff_new_conn = float(self.p.diff_new_conn) self.flt_crit_supplier = float(self.p.crit_supplier) # init dct_prod_up_prod_stat (self para) for prod in a_lst_product: self.dct_prod_up_prod_stat[prod] = { # Normal / Disrupted / Removed + time step 'status': [('N', 0)], # have or have no supply 'supply': dict.fromkeys(prod.a_predecessors(), True) } # init extra capacity (self para) for product in self.a_lst_product: # init extra capacity based on discrete uniform distribution assert self.str_cap_limit_prob_type in ['uniform', 'normal'], \ "cap_limit_prob_type other than uniform, normal" if self.str_cap_limit_prob_type == 'uniform': extra_cap_mean = \ self.size / self.flt_cap_limit_level extra_cap = self.model.nprandom.integers(extra_cap_mean-2, extra_cap_mean+2) extra_cap = 0 if round(extra_cap) < 0 else round(extra_cap) # print(firm_agent.name, extra_cap) self.dct_prod_capacity[product] = extra_cap elif self.str_cap_limit_prob_type == 'normal': extra_cap_mean = \ self.size / self.flt_cap_limit_level extra_cap = self.model.nprandom.normal(extra_cap_mean, 1) extra_cap = 0 if round(extra_cap) < 0 else round(extra_cap) # print(firm_agent.name, extra_cap) self.dct_prod_capacity[product] = extra_cap def remove_edge_to_cus_remove_cus_up_prod(self, remove_product): lst_out_edge = list( self.firm_network.graph.out_edges( self.firm_network.positions[self], keys=True, data='Product')) for n1, n2, key, product_code in lst_out_edge: if product_code == remove_product.code: # remove edge self.firm_network.graph.remove_edge(n1, n2, key) # remove customer up product conditionally customer = ap.AgentIter(self.model, n2).to_list()[0] lst_in_edge = list( self.firm_network.graph.in_edges(n2, keys=True, data='Product')) lst_select_in_edge = [ edge for edge in lst_in_edge if edge[-1] == remove_product.code ] prod_remove = math.exp(-1 * self.flt_crit_supplier * len(lst_select_in_edge)) if self.model.nprandom.choice([True, False], p=[prod_remove, 1 - prod_remove]): customer.dct_n_trial_up_prod_removed[remove_product] = 0 for prod in customer.dct_prod_up_prod_stat.keys(): if remove_product in \ customer.dct_prod_up_prod_stat[ prod]['supply'].keys(): customer.dct_prod_up_prod_stat[ prod]['supply'][remove_product] = False customer.dct_prod_up_prod_stat[ prod]['status'].append(('D', self.model.t)) print(self.name, remove_product.code, 'affect', customer.name, prod.code) def seek_alt_supply(self, product): # para product is the product that self is seeking # for product in self.a_lst_up_product_removed: print(f"{self.name} seek alt supply for {product.code}") if self.dct_n_trial_up_prod_removed[ product] <= self.model.int_n_max_trial: if self.dct_n_trial_up_prod_removed[product] == 0: # select a list of candidate firm that has the product self.dct_cand_alt_supply_up_prod_removed[product] = \ self.model.a_lst_total_firms.select([ product in firm.a_lst_product and product not in firm.a_lst_product_removed for firm in self.model.a_lst_total_firms ]) if self.dct_cand_alt_supply_up_prod_removed[product]: # select based on connection lst_firm_connect = [] if self.is_prf_conn: for firm in \ self.dct_cand_alt_supply_up_prod_removed[product]: out_edges = self.model.firm_network.graph.out_edges( self.model.firm_network.positions[firm], keys=True) in_edges = self.model.firm_network.graph.in_edges( self.model.firm_network.positions[firm], keys=True) lst_adj_firm = [] lst_adj_firm += \ [ap.AgentIter(self.model, edge[1]).to_list()[ 0].code for edge in out_edges] lst_adj_firm += \ [ap.AgentIter(self.model, edge[0]).to_list()[ 0].code for edge in in_edges] if self.code in lst_adj_firm: lst_firm_connect.append(firm) if len(lst_firm_connect) == 0: # select based on size or not if self.is_prf_size: lst_size = \ [size for size in self.dct_cand_alt_supply_up_prod_removed[ product].size] lst_prob = [size / sum(lst_size) for size in lst_size] select_alt_supply = self.model.nprandom.choice( self.dct_cand_alt_supply_up_prod_removed[product], p=lst_prob) else: select_alt_supply = self.model.nprandom.choice( self.dct_cand_alt_supply_up_prod_removed[product]) elif len(lst_firm_connect) > 0: # select based on size of connected firm or not if self.is_prf_size: lst_firm_size = \ [firm.size for firm in lst_firm_connect] lst_prob = \ [size / sum(lst_firm_size) for size in lst_firm_size] select_alt_supply = \ self.model.nprandom.choice(lst_firm_connect, p=lst_prob) else: select_alt_supply = \ self.model.nprandom.choice(lst_firm_connect) print( f"{self.name} selct alt supply for {product.code} " f"from {select_alt_supply.name}" ) assert product in select_alt_supply.a_lst_product, \ f"{select_alt_supply} \ does not produce requested product {product}" if product in select_alt_supply.dct_request_prod_from_firm.\ keys(): select_alt_supply.dct_request_prod_from_firm[ product].append(self) else: select_alt_supply.dct_request_prod_from_firm[product] = [ self ] print( select_alt_supply.name, 'dct_request_prod_from_firm', { key.code: [v.name for v in value] for key, value in select_alt_supply.dct_request_prod_from_firm.items() }) self.dct_n_trial_up_prod_removed[product] += 1 def handle_request(self): print(self.name, 'handle_request') for product, lst_firm in self.dct_request_prod_from_firm.items(): if self.dct_prod_capacity[product] > 0: if len(lst_firm) == 0: continue elif len(lst_firm) == 1: self.accept_request(lst_firm[0], product) elif len(lst_firm) > 1: # handling based on connection lst_firm_connect = [] if self.is_prf_conn: for firm in lst_firm: out_edges = \ self.model.firm_network.graph.out_edges( self.model.firm_network.positions[firm], keys=True) in_edges = \ self.model.firm_network.graph.in_edges( self.model.firm_network.positions[firm], keys=True) lst_adj_firm = [] lst_adj_firm += \ [ap.AgentIter(self.model, edge[1]).to_list()[ 0].code for edge in out_edges] lst_adj_firm += \ [ap.AgentIter(self.model, edge[0]).to_list()[ 0].code for edge in in_edges] if self.code in lst_adj_firm: lst_firm_connect.append(firm) if len(lst_firm_connect) == 0: # handling based on size or not if self.is_prf_size: lst_firm_size = \ [firm.size for firm in lst_firm] lst_prob = \ [size / sum(lst_firm_size) for size in lst_firm_size] select_customer = \ self.model.nprandom.choice(lst_firm, p=lst_prob) else: select_customer = \ self.model.nprandom.choice(lst_firm) self.accept_request(select_customer, product) elif len(lst_firm_connect) > 0: # handling based on size of connected firm or not if self.is_prf_size: lst_firm_size = \ [firm.size for firm in lst_firm_connect] lst_prob = \ [size / sum(lst_firm_size) for size in lst_firm_size] select_customer = \ self.model.nprandom.choice(lst_firm_connect, p=lst_prob) else: select_customer = \ self.model.nprandom.choice(lst_firm_connect) self.accept_request(select_customer, product) def accept_request(self, down_firm, product): # para product is the product that self is selling prod_accept = self.flt_diff_new_conn if self.model.nprandom.choice([True, False], p=[prod_accept, 1 - prod_accept]): self.firm_network.graph.add_edges_from([ (self.firm_network.positions[self], self.firm_network.positions[down_firm], { 'Product': product.code }) ]) self.dct_prod_capacity[product] -= 1 self.dct_request_prod_from_firm[product].remove(down_firm) for prod in down_firm.dct_prod_up_prod_stat.keys(): if product in down_firm.dct_prod_up_prod_stat[ prod]['supply'].keys(): down_firm.dct_prod_up_prod_stat[ prod]['supply'][product] = True down_firm.dct_prod_up_prod_stat[ prod]['status'].append(('N', self.model.t)) del down_firm.dct_n_trial_up_prod_removed[product] del down_firm.dct_cand_alt_supply_up_prod_removed[product] print( f"{self.name} accept {product.code} request " f"from {down_firm.name}" ) else: down_firm.dct_cand_alt_supply_up_prod_removed[product].remove(self) def clean_before_trial(self): self.dct_request_prod_from_firm = {} def clean_before_time_step(self): self.dct_n_trial_up_prod_removed = \ dict.fromkeys(self.dct_n_trial_up_prod_removed.keys(), 0) self.dct_cand_alt_supply_up_prod_removed = {} def get_firm_network_node(self): return self.firm_network.positions[self]