battlefield.py (10260B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\behaviors\actions\battlefield.py 2 from ballpark.entities.awareness.fleetmatching.player_fleet_report import convert_report_to_dict 3 from ballpark.fleets.fleet_types import FLEET_TYPE_SUPPORT, SUPPORTED_FLEET_TYPES, FLEET_TYPES_TO_COUNTER 4 from behaviors.tasks import Task 5 from behaviors.utility.context import get_extra_context 6 from behaviors.utility.owner import get_owner_id 7 from behaviors.utility.spawning import get_current_spawn_points, add_spawn_points 8 from behaviors.utility.threat import evaluate_target 9 from behaviors.utility.battlefield import generate_player_fleet_report, get_total_fleet_points, get_match_result 10 from ccpProfile import TimedFunction 11 from evemath import weighted_choice 12 from inventorycommon.const import categoryStructure 13 import logging 14 import mathext 15 import uthread2 16 THREAT_WINDOW_IN_SECONDS = 10 17 logger = logging.getLogger(__name__) 18 FULL_HEALTH = 1.0 19 20 class CalculateStructureSpawnPointMultiplier(Task): 21 22 @TimedFunction('behaviors::actions::battlefield::CalculateStructureSpawnPointMultiplier::OnEnter') 23 def OnEnter(self): 24 if self.is_structure(): 25 multiplier = self._get_spawn_point_multiplier() 26 self.SendBlackboardValue(self.attributes.spawnPointMultiplierAddress, multiplier) 27 self.SetStatusToSuccess() 28 else: 29 self.SetStatusToFailed() 30 31 def is_structure(self): 32 return self.context.mySlimItem.categoryID == categoryStructure 33 34 def _get_structure_health(self): 35 structure = self.context.myBall 36 if not structure.IsDamaged(): 37 return FULL_HEALTH 38 elif structure.IsStateShieldVulnerable(): 39 return structure.GetShield() 40 elif structure.IsStateArmorVulnerable(): 41 return structure.GetArmor() 42 elif structure.IsStateHullVulnerable(): 43 return structure.GetHull() 44 else: 45 return FULL_HEALTH 46 47 def _get_spawn_point_multiplier(self): 48 health = self._get_structure_health() 49 return self._get_multiplier_at(health) 50 51 def _get_multiplier_at(self, health): 52 min_val = self.attributes.minSpawnPointMultiplier 53 max_val = self.attributes.maxSpawnPointMultiplier 54 peak_health_fraction = self.attributes.peakHealthFraction 55 multiplier = min_val + (max_val - min_val) * (FULL_HEALTH - health) ** 2 / (FULL_HEALTH - peak_health_fraction) ** 2 56 return max(0.0, min(multiplier, max_val)) 57 58 59 class SelectFleetToCounter(Task): 60 61 @TimedFunction('behaviors::actions::battlefield::SelectFleetToCounter::OnEnter') 62 def OnEnter(self): 63 match_results = self.GetLastBlackboardValue(self.attributes.fleetMatchResultsAddress) 64 if match_results: 65 if len(match_results) > 1: 66 weighted_results = self._get_weighted_results(match_results) 67 match_result = weighted_choice(weighted_results) 68 logger.debug('chose fleet_type %s from the weighted options %s', match_result['fleet_type'], weighted_results) 69 else: 70 match_result = match_results[0] 71 logger.debug('chose fleet_type %s as it was the only option available', match_result['fleet_type']) 72 self.SendBlackboardValue(self.attributes.fleetTypeToCounterAddress, match_result['fleet_type']) 73 self.SetStatusToSuccess() 74 else: 75 logger.debug('No valid fleet type available to select from') 76 self.SendBlackboardValue(self.attributes.fleetTypeToCounterAddress, None) 77 self.SetStatusToSuccess() 78 79 def _get_weighted_results(self, match_results): 80 return [ (result, self._get_match_result_weight(result)) for result in match_results ] 81 82 def _get_match_result_weight(self, result): 83 threat = sum((self._get_threat(item_id) for item_id in result['item_ids'])) 84 return max(1, threat) 85 86 def _get_threat(self, item_id): 87 return evaluate_target(self.context.damageTracker, self.context.dogmaLM, item_id, THREAT_WINDOW_IN_SECONDS) 88 89 90 class AnalyzeBattlefieldFleetMatching(Task): 91 92 def __init__(self, attributes = None): 93 super(AnalyzeBattlefieldFleetMatching, self).__init__(attributes=attributes) 94 self.is_analysing = False 95 96 @TimedFunction('behaviors::actions::battlefield::AnalyzeBattlefieldFleetMatching::OnEnter') 97 def OnEnter(self): 98 self.SetStatusToFailed() 99 entity_group_id_by_fleet_type = self.get_entity_group_id_by_fleet_type_dict() 100 spawn_pool_id = self.GetLastBlackboardValue(self.attributes.spawnPoolIdAddress) 101 spawn_table_id = self.GetLastBlackboardValue(self.attributes.spawnTableIdAddress) 102 min_spawn_points = self.attributes.minimumSpawnPoints 103 max_spawn_points = self.attributes.maximumSpawnPoints 104 spawn_point_multiplier = self.GetLastBlackboardValue(self.attributes.spawnPointMultiplierAddress) 105 if spawn_pool_id and spawn_pool_id and spawn_point_multiplier: 106 self.start_analysis(entity_group_id_by_fleet_type, spawn_point_multiplier, spawn_pool_id, spawn_table_id, min_spawn_points, max_spawn_points) 107 self.SetStatusToSuccess() 108 109 def start_analysis(self, entity_group_id_by_fleet_type, spawn_point_multiplier, spawn_pool_id, spawn_table_id, min_spawn_points, max_spawn_points): 110 uthread2.start_tasklet(self.analyze_battlefield_fleet_matching, spawn_pool_id, spawn_table_id, spawn_point_multiplier, entity_group_id_by_fleet_type, min_spawn_points, max_spawn_points) 111 112 def get_entity_group_id_by_fleet_type_dict(self): 113 entity_group_id_by_fleet_type = {} 114 for fleet_type, entity_group_id_address in self.attributes.entityGroupIdAddressByFleetType.iteritems(): 115 if entity_group_id_address is None: 116 continue 117 entity_group_id = self.GetLastBlackboardValue(entity_group_id_address) 118 if entity_group_id is None: 119 continue 120 entity_group_id_by_fleet_type[fleet_type] = entity_group_id 121 122 return entity_group_id_by_fleet_type 123 124 @TimedFunction('behaviors::actions::battlefield::AnalyzeBattlefieldFleetMatching::analyze_battlefield_fleet_matching') 125 def analyze_battlefield_fleet_matching(self, spawn_pool_id, spawn_table_id, spawn_point_multiplier, entity_group_id_by_fleet_type, min_spawn_points, max_spawn_points): 126 if self.is_analysing: 127 logger.warning('analysis already in progress pool=%s table=%s multiplier=%s groups_by_fleet_type=%s min_amount=%s max_amount=%s', spawn_pool_id, spawn_table_id, spawn_point_multiplier, entity_group_id_by_fleet_type, min_spawn_points, max_spawn_points) 128 return 129 self.is_analysing = True 130 try: 131 self._analyze_battlefield_fleet_matching(spawn_pool_id, spawn_table_id, spawn_point_multiplier, entity_group_id_by_fleet_type, min_spawn_points, max_spawn_points) 132 except: 133 logger.exception('error while analyzing battle field') 134 finally: 135 self.is_analysing = False 136 137 def _analyze_battlefield_fleet_matching(self, spawn_pool_id, spawn_table_id, spawn_point_multiplier, entity_group_id_by_fleet_type, min_spawn_points, max_spawn_points): 138 player_report = generate_player_fleet_report(self, spawn_pool_id) 139 spawn_point_difference = self.calculate_spawn_points_difference(player_report, spawn_point_multiplier, spawn_pool_id, spawn_table_id) 140 self.assign_spawn_points(spawn_pool_id, spawn_point_difference, min_spawn_points, max_spawn_points) 141 match_results = self.match_fleets(entity_group_id_by_fleet_type, player_report, spawn_point_multiplier, spawn_table_id) 142 self.post_results(player_report, match_results) 143 144 def post_results(self, player_report, match_results): 145 self.SendBlackboardValue(self.attributes.playerFleetReportAddress, convert_report_to_dict(player_report)) 146 self.SendBlackboardValue(self.attributes.fleetMatchResultsAddress, match_results) 147 148 @TimedFunction('behaviors::actions::battlefield::AnalyzeBattlefieldFleetMatching::match_fleets') 149 def match_fleets(self, entity_group_id_by_fleet_type, player_report, spawn_point_multiplier, spawn_table_id): 150 support_points = player_report.fleet_by_type[FLEET_TYPE_SUPPORT].fleet_cost 151 supported_fleet_points = float(sum((r.fleet_cost for fleet_type, r in player_report.fleet_by_type.iteritems() if fleet_type in SUPPORTED_FLEET_TYPES))) 152 match_results = [] 153 for fleet_type in FLEET_TYPES_TO_COUNTER: 154 match_result = get_match_result(self, spawn_table_id, fleet_type, player_report, spawn_point_multiplier, support_points, supported_fleet_points, entity_group_id_by_fleet_type) 155 if match_result['player_fleet_points'] > 0: 156 match_results.append(match_result) 157 158 return match_results 159 160 @TimedFunction('behaviors::actions::battlefield::AnalyzeBattlefieldFleetMatching::calculate_spawn_points_difference') 161 def calculate_spawn_points_difference(self, player_report, spawn_point_multiplier, spawn_pool_id, spawn_table_id): 162 total_npc_fleet_cost = get_total_fleet_points(self, spawn_pool_id, spawn_table_id) 163 fleet_cost_difference = player_report.total_fleet_cost * spawn_point_multiplier - total_npc_fleet_cost 164 current_spawn_points = get_current_spawn_points(self, spawn_pool_id) 165 spawn_point_difference = fleet_cost_difference - current_spawn_points 166 logger.debug('assign_spawn_pool_points: total_npc_points=%s total_player_points=%s current_pool=%s differences=%s multiplier=%s', total_npc_fleet_cost, player_report.total_fleet_cost, current_spawn_points, spawn_point_difference, spawn_point_multiplier) 167 return spawn_point_difference 168 169 def assign_spawn_points(self, spawn_pool_id, requested_spawn_points, min_spawn_points, max_spawn_points): 170 actual_spawn_points = mathext.clamp(requested_spawn_points, min_spawn_points, max_spawn_points) 171 add_spawn_points(self, spawn_pool_id, actual_spawn_points) 172 self.context.eventLogger.log_adding_spawn_points(get_owner_id(self), spawn_pool_id, requested_spawn_points, min_spawn_points, max_spawn_points, actual_spawn_points, get_extra_context(self))