BehaviorPredictionUsingLTL/sc_separations.py at master · Sanaarasteh/BehaviorPredictionUsingLTL · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
import random

from typing import List
from typing import Tuple

from sc_utils import GameLoader
from sc_utils import PlayerTracker
from sc_utils import AllTracker
from sc_utils import PassAnalyzer


def team_possession_separation(traces: List, game: GameLoader, player_tracker: PlayerTracker) -> Tuple[List, List]:
    """
    This function receives the traces, the game information, and the target player information.
    The function splits the traces into positive and negative traces where positive traces are
    the instances in which the player's team possesses the ball and the negative traces are the
    instances where the opponent team possesses the ball.

    :param traces (List): a list of traces made by the Pipeline.
    :param game (GameLoader): an instance of GameLoader which contains the information about the
                              the game including the game tracking information
    :param player_tracker (PlayerTracker): an instance of PlayerTracker which contains the target
                                           player information including their position and whether
                                           they belong to the home team or away team
    """
    # crating empty lists for positive and negative traces
    positive_traces = []
    negative_traces = []

    # detemining the player team
    player_team = 'home team' if player_tracker.home else 'away team'

    # creating an empty list to only store the tracks in which the target player exists
    player_possession_instances = []
    # creating a list to store the instance numbers at which the ball possession changes from one team to another
    changing_points = []

    # fill the player_possession_instances list
    for index in range(len(game.game_tracks)):
        if index in player_tracker.present_instances:
            player_possession_instances.append(game.game_tracks[index])

    # finding the instances at which the ball possession changes
    changing_points.append(0)
    possessions = []
    for index in range(len(player_possession_instances) - 1):
        current_track = player_possession_instances[index]
        next_track = player_possession_instances[index + 1]

        if next_track['possession']['group'] != current_track['possession']['group']:
            if current_track['possession']['group'] != 'out':
                changing_points.append(index)
                possessions.append(current_track['possession']['group'])
            if next_track['possession']['group'] != 'out':
                changing_points.append(index + 1)

        if index == len(player_possession_instances) - 2:
            possessions.append(current_track['possession']['group'])
            changing_points.append(len(player_possession_instances) - 1)

    # separting the instances into positive and negative sequences
    # if at the beginning the ball is possessed by the player's team
    # it is considered as a positive sequence and at each changing point
    # the possession alternates
    # If at the beginning the ball is possessed by the opponent team
    # it is considered as a negative sequence and at each changing point
    # the possession alternates
    print('[*] Separating the traces based on which team possesses the ball...')
    for i in range(len(possessions)):
        if possessions[i] == player_team:
            positive_traces.extend(traces[changing_points[2 * i]: changing_points[2 * i + 1] + 1])
        else:
            negative_traces.extend(traces[changing_points[2 * i]: changing_points[2 * i + 1] + 1])

    return positive_traces, negative_traces


def attack_defense_separation(traces: List, game: GameLoader, all_tracker: AllTracker, player_tracker: PlayerTracker,
                              attack_mode: str='penetration', defense_mode: str='retreat') -> Tuple[List, List]:
    """
    This function receives the traces, the game information, and the target player information.
    The function splits the traces into positive and negative traces where positive traces are
    the instances in which the player's team possesses the ball and the negative traces are the
    instances where the opponent team possesses the ball.

    :param traces (List): a list of traces made by the Pipeline.
    :param game (GameLoader): an instance of GameLoader which contains the information about the
                              the game including the game tracking information
    :param player_tracker (PlayerTracker): an instance of PlayerTracker which contains the target
                                           player information including their position and whether
                                           they belong to the home team or away team
    """
    # crating empty lists for positive and negative traces
    positive_traces = []
    negative_traces = []

    # getting the attacking indices of the player's team
    attacking_indices = all_tracker.attack_detection(player_tracker.home, mode=attack_mode)

    # getting the defense indices of the player's team
    defense_indices = all_tracker.defense_detection(player_tracker.home, mode=defense_mode)

    # pruning the indices so that we can just have the indices in which the player exists
    positive_indices = []

    for index_pair in attacking_indices:
        for i in range(index_pair[0], index_pair[1] + 1):
            if game.player_exists(i, player_tracker.track_id)[0]:
                positive_indices.append(player_tracker.present_instances.index(i))

    negative_indices = []
    for index_pair in defense_indices:
        for i in range(index_pair[0], index_pair[1] + 1):
            if game.player_exists(i, player_tracker.track_id)[0]:
                negative_indices.append(player_tracker.present_instances.index(i))

    for index in positive_indices:
        try:
            positive_traces.append(traces[index])
        except:
            continue

    for index in negative_indices:
        try:
            negative_traces.append(traces[index])
        except:
            continue


    return positive_traces, negative_traces


def passing_style_separation(traces: List, game: GameLoader, player_tracker: PlayerTracker, all_tracker: AllTracker) -> Tuple[List, List]:
    # crating empty lists for positive and negative traces
    positive_traces = []
    negative_traces = []


    pass_analyzer = PassAnalyzer('pass_instances.csv', game, all_tracker, n_cluster=2)

    cluster_instances = pass_analyzer.instance_labels

    # extract the instances of positive and negative traces
    possible_positive_indices = cluster_instances[0]
    possible_negative_indices = cluster_instances[1]


    # pruning the indices so that we can just have the indices in which the player exists
    positive_indices = []

    for index in possible_positive_indices:
        if game.player_exists(index, player_tracker.track_id)[0]:
            positive_indices.append(player_tracker.present_instances.index(index))

    negative_indices = []
    for index in possible_negative_indices:
        if game.player_exists(index, player_tracker.track_id)[0]:
            negative_indices.append(player_tracker.present_instances.index(index))

    for index in positive_indices:
        try:
            positive_traces.append(traces[index])
        except:
            continue

    for index in negative_indices:
        try:
            negative_traces.append(traces[index])
        except:
            continue


    if len(positive_traces) / len(negative_traces) > 1.5:
        positive_traces = random.sample(positive_traces, len(negative_traces))

    if len(negative_traces) / len(positive_indices) > 1.5:
        negative_traces = random.sample(negative_traces, len(positive_traces))


    return positive_traces, negative_traces