Moving pad locations to field constants. Updated stolen boosts logic. (…

…#212)

* Moving pad locations to field constants. Updated stolen boosts logic.

Boost steals are now better calculated using the pad label rather than player location heuristic.

* Added comments to stolen boost logic.

* added a steal test

* adding more steals tests

carball/analysis/constants/field_constants.py

@@ -9,6 +9,7 @@
 class FieldType(Enum):
     STANDARD = 1
 
+
 STANDARD_FIELD_LENGTH_HALF = 5120
 STANDARD_FIELD_WIDTH_HALF = 4096
 STANDARD_GOAL_WIDTH_HALF = 893
@@ -24,7 +25,6 @@ class FieldType(Enum):
 
 
 class FieldConstants:
-
     field_type = FieldType.STANDARD
 
     corner = np.array([STANDARD_FIELD_WIDTH_HALF - STANDARD_GOAL_WIDTH_HALF,
@@ -43,6 +43,59 @@ def __init__(self, field_type=FieldType.STANDARD):
     return: Boolean series that can be used to index the original data_frame to sum deltas with.
     """
 
+    # The first two values are X,Y. The last value is (RLBot_label +1) and multiplied by 10 if its a bigboost.
+    # Bigboosts are *10 so that all big boost values are larger than all small boost values, for easy querying.
+    # If nobody needs these to be RLBot labels, this can be simplified.
+    def get_big_pads(self, field_type=None):
+        if field_type is None:
+            field_type = self.field_type
+        if field_type == FieldType.STANDARD:
+            return np.array([
+                (3072, -4096, 50),
+                (-3072, -4096, 40),
+                (3584, 0, 190),
+                (-3584, 0, 160),
+                (3072, 4096, 300),
+                (-3072, 4096, 310)])
+        else:
+            raise NotImplementedError
+
+    def get_small_pads(self, field_type=None):
+        if field_type is None:
+            field_type = self.field_type
+        if field_type == FieldType.STANDARD:
+            return np.array([
+                (0.0, -4240.0, 1),
+                (-1792.0, -4184.0, 2),
+                (1792.0, -4184.0, 3),
+                (- 940.0, -3308.0, 6),
+                (940.0, -3308.0, 7),
+                (0.0, -2816.0, 8),
+                (-3584.0, -2484.0, 9),
+                (3584.0, -2484.0, 10),
+                (-1788.0, -2300.0, 11),
+                (1788.0, -2300.0, 12),
+                (-2048.0, -1036.0, 13),
+                (0.0, -1024.0, 14),
+                (2048.0, -1036.0, 15),
+                (-1024.0, 0.0, 17),
+                (1024.0, 0.0, 18),
+                (-2048.0, 1036.0, 20),
+                (0.0, 1024.0, 21),
+                (2048.0, 1036.0, 22),
+                (-1788.0, 2300.0, 23),
+                (1788.0, 2300.0, 24),
+                (-3584.0, 2484.0, 25),
+                (3584.0, 2484.0, 26),
+                (0.0, 2816.0, 27),
+                (- 940.0, 3310.0, 28),
+                (940.0, 3308.0, 29),
+                (-1792.0, 4184.0, 32),
+                (1792.0, 4184.0, 33),
+                (0.0, 4240.0, 34)])
+        else:
+            raise NotImplementedError
+
     def get_neutral_zone(self, player_data_frame, **kwargs):
         return self.abs(player_data_frame.pos_y) < NEUTRAL_ZONE
 
@@ -68,7 +121,7 @@ def get_height_0_ball(self, player_data_frame, **kwargs):
         return player_data_frame.pos_z < HEIGHT_0_BALL_LIM
 
     def get_height_1(self, player_data_frame, **kwargs):
-        return (HEIGHT_0_LIM < player_data_frame.pos_z) & (player_data_frame.pos_z < HEIGHT_1_LIM)\
+        return (HEIGHT_0_LIM < player_data_frame.pos_z) & (player_data_frame.pos_z < HEIGHT_1_LIM) \
                & (player_data_frame.pos_x.abs() < self.on_wall[0]) & (player_data_frame.pos_y.abs() < self.on_wall[1])
 
     def get_height_2(self, player_data_frame, **kwargs):
@@ -94,9 +147,9 @@ def get_on_wall(self, player_data_frame, **kwargs):
 
     def get_corner_time(self, player_data_frame, **kwargs):
         return (((player_data_frame.pos_x >= self.corner[0]) |
-                (player_data_frame.pos_x <= -self.corner[0])) &
+                 (player_data_frame.pos_x <= -self.corner[0])) &
                 ((player_data_frame.pos_y >= self.corner[1]) |
-                (player_data_frame.pos_y <= - self.corner[1])))
+                 (player_data_frame.pos_y <= - self.corner[1])))
 
     def abs(self, value):
         if value is pd.DataFrame:

carball/analysis/events/boost_pad_detection/pickup_analysis.py

@@ -1,117 +1,72 @@
 import numpy as np
 import pandas as pd
 from carball.generated.api import game_pb2
-
-# The first two values are X,Y. The last value is (RLBot_label +1) and multiplied by 10 if its a bigboost.
-# Bigboosts are *10 so that all big boost values are larger than all small boost values, for easy querying.
-# If nobody needs these to be RLBot labels, this can be simplified.
-BIG_BOOST_POSITIONS = np.array([
-    (3072, -4096, 50),
-    (-3072, -4096, 40),
-    (3584, 0, 190),
-    (-3584, 0, 160),
-    (3072, 4096, 300),
-    (-3072, 4096, 310)])
-
-SMALL_BOOST_POSITIONS = np.array([
-    (0.0, -4240.0, 1),
-    (-1792.0, -4184.0, 2),
-    (1792.0, -4184.0, 3),
-    (- 940.0, -3308.0, 6),
-    (940.0, -3308.0, 7),
-    (0.0, -2816.0, 8),
-    (-3584.0, -2484.0, 9),
-    (3584.0, -2484.0, 10),
-    (-1788.0, -2300.0, 11),
-    (1788.0, -2300.0, 12),
-    (-2048.0, -1036.0, 13),
-    (0.0, -1024.0, 14),
-    (2048.0, -1036.0, 15),
-    (-1024.0, 0.0, 17),
-    (1024.0, 0.0, 18),
-    (-2048.0, 1036.0, 20),
-    (0.0, 1024.0, 21),
-    (2048.0, 1036.0, 22),
-    (-1788.0, 2300.0, 23),
-    (1788.0, 2300.0, 24),
-    (-3584.0, 2484.0, 25),
-    (3584.0, 2484.0, 26),
-    (0.0, 2816.0, 27),
-    (- 940.0, 3310.0, 28),
-    (940.0, 3308.0, 29),
-    (-1792.0, 4184.0, 32),
-    (1792.0, 4184.0, 33),
-    (0.0, 4240.0, 34)
-])
-
-BIG_BOOST_RADIUS = 208
-SMALL_BOOST_RADIUS = 149  # 144 doesn't work for some pickups that are very close to the edge.
-BIG_BOOST_HEIGHT = 168
-SMALL_BOOST_HEIGHT = 165
-# Choosing how many frames to be open to setting a pickup. Back is for when the player is ahead of the server (usually smaller)
-LAG_BACK = 6
-LAG_FORWARD = 14
-
-BOOST_POSITIONS = np.concatenate((BIG_BOOST_POSITIONS, SMALL_BOOST_POSITIONS))
+from carball.analysis.constants.field_constants import FieldConstants
 
 
 class PickupAnalysis:
+    field_constants = FieldConstants()
+    BIG_BOOST_POSITIONS = field_constants.get_big_pads()
+    SMALL_BOOST_POSITIONS = field_constants.get_small_pads()
+    BIG_BOOST_RADIUS = 208
+    SMALL_BOOST_RADIUS = 149  # 144 doesn't work for some pickups that are very close to the edge.
+    BIG_BOOST_HEIGHT = 168
+    SMALL_BOOST_HEIGHT = 165
+    # Choosing how many frames to be open to setting a pickup. Back is for when the player is ahead of the server (usually smaller)
+    LAG_BACK = 6
+    LAG_FORWARD = 14
 
-    @staticmethod
-    def add_pickups(proto_game: game_pb2.Game, data_frame: pd.DataFrame):
+    @classmethod
+    def add_pickups(cls, proto_game: game_pb2.Game, data_frame: pd.DataFrame):
 
         for player in proto_game.players:
             player_vals_df = data_frame[player.name][['pos_x', 'pos_y', 'pos_z', 'boost']].copy()
             player_vals_df['boost'] /= 2.55
             player_vals_df['boost'] = player_vals_df['boost'].round(5)
             player_vals_df = player_vals_df.dropna(axis=0, how='all')
             player_vals_df = player_vals_df.fillna(0)
-            player_vals_df['boost_collect'] = get_boost_collect(player_vals_df)
+            player_vals_df['boost_collect'] = cls.get_boost_collect(player_vals_df)
             data_frame[player.name, 'boost_collect'] = player_vals_df['boost_collect']
         return
 
-    @staticmethod
-    def something(proto_game: game_pb2.Game, data_frame: pd.DataFrame):
-        pass
-
-
-def get_boost_collect(player_vals_df):
-    # Get a series with indexes as a subset of the indexes of df, values being pad label picked up.
-    # Iterate through every pad, label each frame in the path with which boost pad it was in range of.
-    df = player_vals_df.copy()
-    path = df.drop(['pos_z', 'boost'], axis=1)
-    big_labels = np.zeros(len(path))
-    small_labels = np.zeros(len(path))
-    # Calculate the distances from each pad. Add label of the pad if distance <= radius
-    for pad in BIG_BOOST_POSITIONS:
-        distances = np.sqrt(np.square(path.values - pad[:2]).sum(axis=1, dtype=np.float32))
-        big_labels += (pad[2] * (distances <= BIG_BOOST_RADIUS))
+    @classmethod
+    def get_boost_collect(cls, player_vals_df):
+        # Get a series with indexes as a subset of the indexes of df, values being pad label picked up.
+        # Iterate through every pad, label each frame in the path with which boost pad it was in range of.
+        df = player_vals_df.copy()
+        path = df.drop(['pos_z', 'boost'], axis=1)
+        big_labels = np.zeros(len(path))
+        small_labels = np.zeros(len(path))
+        # Calculate the distances from each pad. Add label of the pad if distance <= radius
+        for pad in cls.BIG_BOOST_POSITIONS:
+            distances = np.sqrt(np.square(path.values - pad[:2]).sum(axis=1, dtype=np.float32))
+            big_labels += (pad[2] * (distances <= cls.BIG_BOOST_RADIUS))
 
-    for pad in SMALL_BOOST_POSITIONS:
-        distances = np.sqrt(np.square(path.values - pad[:2]).sum(axis=1, dtype=np.float32))
-        small_labels += (pad[2] * (distances <= SMALL_BOOST_RADIUS))
-    # Add labels and exclude labels with z too high. Didn't calculate this earlier because its a flat height)
-    df['pad_in_range'] = 0
-    df['pad_in_range'] += small_labels
-    df.loc[df['pos_z'] >= SMALL_BOOST_HEIGHT, 'pad_in_range'] = 0
-    df['pad_in_range'] += big_labels
-    df.loc[df['pos_z'] >= BIG_BOOST_HEIGHT, 'pad_in_range'] = 0
-    # Get the gains in boost per frame
-    df['gains'] = df['boost'].diff().clip(0)
-    # Get whether we entered or exited the range of a pad per frame
-    df['status_change'] = (df['pad_in_range'].diff(1))
-    df = df.fillna(0)
-    # Get the index of the frame we most recently entered a pad range, per frame.
-    df['recent_entry_index'] = df.index
-    df.loc[df['status_change'] <= 0, 'recent_entry_index'] = 0
-    df['recent_entry_index'] = df['recent_entry_index'].replace(0, np.nan).fillna(
-        method='bfill', limit=LAG_BACK).fillna(
-        method='ffill', limit=LAG_FORWARD)
-    gains_frames = df.loc[
-        ((df['gains'] > 5) & (df['boost'] != 33.33333)) | ((df['gains'] > 0) & (df['boost'] > 95.0))].copy()
+        for pad in cls.SMALL_BOOST_POSITIONS:
+            distances = np.sqrt(np.square(path.values - pad[:2]).sum(axis=1, dtype=np.float32))
+            small_labels += (pad[2] * (distances <= cls.SMALL_BOOST_RADIUS))
+        # Add labels and exclude labels with z too high. Didn't calculate this earlier because its a flat height)
+        df['pad_in_range'] = 0
+        df['pad_in_range'] += small_labels
+        df.loc[df['pos_z'] >= cls.SMALL_BOOST_HEIGHT, 'pad_in_range'] = 0
+        df['pad_in_range'] += big_labels
+        df.loc[df['pos_z'] >= cls.BIG_BOOST_HEIGHT, 'pad_in_range'] = 0
+        # Get the gains in boost per frame
+        df['gains'] = df['boost'].diff().clip(0)
+        # Get whether we entered or exited the range of a pad per frame
+        df['status_change'] = (df['pad_in_range'].diff(1))
+        df = df.fillna(0)
+        # Get the index of the frame we most recently entered a pad range, per frame.
+        df['recent_entry_index'] = df.index
+        df.loc[df['status_change'] <= 0, 'recent_entry_index'] = 0
+        df['recent_entry_index'] = df['recent_entry_index'].replace(0, np.nan).fillna(
+            method='bfill', limit=cls.LAG_BACK).fillna(
+            method='ffill', limit=cls.LAG_FORWARD)
+        gains_frames = df.loc[
+            ((df['gains'] > 5) & (df['boost'] != 33.33333)) | ((df['gains'] > 0) & (df['boost'] > 95.0))].copy()
 
-    gains_indexes = gains_frames['recent_entry_index'].dropna()
+        gains_indexes = gains_frames['recent_entry_index'].dropna()
 
-    pickups = df.loc[gains_indexes]['status_change'].copy()
-    pickups = pickups.loc[~pickups.index.duplicated(keep='first')]
-    return pickups
+        pickups = df.loc[gains_indexes]['status_change'].copy()
+        pickups = pickups.loc[~pickups.index.duplicated(keep='first')]
+        return pickups

carball/analysis/stats/boost/boost.py

@@ -85,13 +85,16 @@ def get_average_boost_level(player_dataframe: pd.DataFrame) -> np.float64:
 
     @classmethod
     def get_num_stolen_boosts(cls, player_dataframe: pd.DataFrame, is_orange):
-        big_pads_collected = player_dataframe[player_dataframe.boost_collect == True]
-        if is_orange == 1:
-            boost_collected_in_opposing_third = big_pads_collected[cls.field_constants.get_third_0(big_pads_collected)]
+        big = cls.field_constants.get_big_pads()
+        # Get big pads below or above 0 depending on team
+        # The index of y position is 1. The index of the label is 2.
+        if is_orange:
+            opponent_pad_labels = big[big[:, 1] < 0][:, 2] #big[where[y] is < 0][labels]
         else:
-            boost_collected_in_opposing_third = big_pads_collected[cls.field_constants.get_third_2(big_pads_collected)]
-
-        return len(boost_collected_in_opposing_third.index)
+            opponent_pad_labels = big[big[:, 1] > 0][:, 2] #big[where[y] is > 0][labels]
+        # Count all the places where isin = True by summing
+        stolen = player_dataframe.boost_collect.isin(opponent_pad_labels).sum()
+        return stolen
 
     @staticmethod
     def get_player_boost_usage_max_speed(player_dataframe: pd.DataFrame) -> np.float64:
@@ -135,11 +138,3 @@ def get_player_boost_collection(player_dataframe: pd.DataFrame) -> Dict[str, int
         except (AttributeError, KeyError):
             return {}
         return ret
-
-    @staticmethod
-    def get_player_boost_waste(usage: np.float64, collection: Dict[str, int]) -> float:
-        try:
-            total_collected = collection['big'] * 100 + collection['small'] * 12
-            return total_collected - usage
-        except KeyError:
-            return 0

carball/tests/stats/boost_test.py

@@ -56,6 +56,30 @@ def test(analysis: AnalysisManager):
 
         run_analysis_test_on_replay(test, get_raw_replays()["6_BIG_25_SMALL"], cache=replay_cache)
 
+    def test_boost_steals(self, replay_cache):
+        def test(analysis: AnalysisManager):
+            proto_game = analysis.get_protobuf_data()
+            player = proto_game.players[0]
+            boost = player.stats.boost
+            assert boost.num_stolen_boosts == 2
+
+        run_analysis_test_on_replay(test, get_raw_replays()["6_BIG_25_SMALL"], cache=replay_cache)
+
+    def test_boost_steals_post_goal(self, replay_cache):
+        def test(analysis: AnalysisManager):
+            proto_game = analysis.get_protobuf_data()
+            player = proto_game.players[0]
+            boost = player.stats.boost
+            assert [boost.num_small_boosts, boost.num_large_boosts,
+                    boost.num_stolen_boosts, boost.boost_usage] == [0, 0, 0, 0]
+
+            player = proto_game.players[1]
+            boost = player.stats.boost
+            assert [boost.num_large_boosts, boost.num_stolen_boosts] == [3, 3]
+            assert boost.boost_usage > 0
+
+        run_analysis_test_on_replay(test, get_raw_replays()["3_STEAL_ORANGE_0_STEAL_BLUE"], cache=replay_cache)
+
     def test_boost_used(self, replay_cache):
         case = unittest.TestCase('__init__')
 

carball/tests/utils.py

@@ -109,6 +109,7 @@ def get_raw_replays():
         "1_NORMAL_SAVE_FROM_SHOT_TOWARD_POST": ["1_NORMAL_SAVE.replay"],
 
         # Boost
+        "3_STEAL_ORANGE_0_STEAL_BLUE": ["3_STEALS.replay"],
         "12_BOOST_PAD_0_USED": ["12_BOOST_PAD_0_USED.replay"],
         "12_BOOST_PAD_45_USED": ["12_BOOST_PAD_45_USED.replay"],
         "100_BOOST_PAD_0_USED": ["100_BOOST_PAD_0_USED.replay"],