boggart/ClusteringPipelineEngine.py at master · Brave-Maker/boggart · 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
181
182
183
184
185
186
187
188
189
190
from operator import itemgetter

import numpy as np
import pandas as pd
from tqdm import tqdm

from configs import BackgroundConfig, TrajectoryConfig
from VideoData import VideoData
from ingest import IngestTimeProcessing
from utils import parallelize_update_dictionary
from QueryProcessor import QueryProcessor

class ClusteringPipelineEngine:
    def __init__(self, vid_label, bg_conf:BackgroundConfig=None, traj_conf:TrajectoryConfig=None, fps:int=30, query_conf=0.3, query_seg_size=1800, skip_no_traj=False):

        self.vid_label = vid_label

        self.bg_conf = BackgroundConfig(peak_thresh=0.1) if bg_conf is None else bg_conf
        self.traj_conf =  TrajectoryConfig(diff_thresh=16, chunk_size=1800, fps=fps) if traj_conf is None else traj_conf
        assert self.traj_conf.fps == fps

        self.fps = fps
        self.query_conf = query_conf
        self.query_seg_size = query_seg_size
        self.chunk_size = self.traj_conf.chunk_size

        self.skip_no_traj = skip_no_traj

        self.total_frames_per_hour = 60 * 60 * 30 # min/hour * sec/min * frames/sec

        self.mfs_sweep = [900, 450, 300, 200, 100, 50, 20, 10, 5, 2, 1, 0, -900, -300, -30, -10, -3, -2]

        self.all_vecs = None
        self._all_vecs = None
        self.hours = None # parameterize all vecs
        self._hours = None

        self.qp_sweep = None

    def _sweep_helper(self, sweep_idx):
        chunk_start, query_seg_start, qp = self.qp_sweep[sweep_idx]
        return qp.execute(chunk_start, query_seg_start, check_only=False, get_results_df=True)

    def _get_vec(self, query_seg_start):
        chunk_start = (query_seg_start//self.chunk_size) * self.chunk_size
        assert self.ip is not None
        return self.ip.get_feature_vector(chunk_start, query_seg_start, self.query_seg_size, self.skip_no_traj)

    def get_hour_vecs(self, hour, get_skips=False, get_df=False):
        vd = VideoData(self.vid_label, hour)
        self.ip = IngestTimeProcessing(vd, self.bg_conf, self.traj_conf)
        vecs = parallelize_update_dictionary(self._get_vec, range(0, self.total_frames_per_hour, self.query_seg_size), total_cpus=60, max_workers=30)
        self.ip = None

        skips = [k for k,v in vecs.items() if v is None]
        query_starts, vecs = zip(*[(k, v) for k, v in sorted(vecs.items()) if v is not None])

        vecs = np.array(vecs)

        assert not (get_skips and get_df)
        if get_skips:
            return vecs, skips

        if get_df:
            df = pd.DataFrame(query_starts, columns=["seg_start"])
            df["chunk_start"] = (df["seg_start"]//self.chunk_size) * self.chunk_size
            df["hour"] = hour
            return vecs, df

        return vecs

    def execute(self, hours, query_type, model, query_class, acc_target, percent_clusters, ioda, get_boggart_results=False):
        if type(hours) == int:
            hours = list(hours)

        if self._hours is None or self._hours != hours:
            self._hours = hours
            self._all_vecs, self._all_dfs = zip(*[self.get_hour_vecs(h, get_df=True) for h in hours])
            self._all_vecs = np.vstack(self._all_vecs)
            self._all_dfs = pd.concat(self._all_dfs).reset_index().drop(columns=["index"])

        self.hours = self._hours.copy()
        self.all_vecs = self._all_vecs.copy()
        self.all_dfs = self._all_dfs.copy()

        n_clusters = max(2, int(percent_clusters * len(self.all_vecs)))

        _, clusters, centroids, _, _ = IngestTimeProcessing.cluster_profile(self.all_vecs.copy(), n_clusters=n_clusters)

        if not np.array_equal(centroids[clusters][centroids], centroids):
            clusters[centroids] = list(range(n_clusters))

        self.all_dfs["cluster"] = clusters

        centroids_df = self.all_dfs.loc[centroids].copy()

        mfs_sweep_index = 0
        sweep_chunk_length = 8

        entire_df = None

        full_df = None

        while len(centroids_df) > 0:

            centroid_qps = []
            for (_, (query_seg_start, chunk_start, hour, cluster)) in centroids_df.iterrows():

                vd = VideoData(self.vid_label, hour)
                for mfs in self.mfs_sweep[mfs_sweep_index * sweep_chunk_length : (mfs_sweep_index + 1) * sweep_chunk_length]:
                    qp = QueryProcessor(query_type, vd, model, query_class, self.query_conf, mfs, self.bg_conf, self.traj_conf, ioda, self.query_seg_size)
                    centroid_qps.append([chunk_start, query_seg_start, qp])

            if len(centroid_qps) > 0:

                self.qp_sweep = centroid_qps
                c_sweep_res = parallelize_update_dictionary(self._sweep_helper, range(len(centroid_qps)), total_cpus=60, max_workers=min(60, len(centroid_qps)))
                self.qp_sweep = None
                _tempDF = list(map(itemgetter(1), sorted(c_sweep_res.items(), key=itemgetter(0))))
                c_sweep_res = pd.concat(_tempDF).reset_index().drop(columns="index")
                entire_df = pd.concat([entire_df, c_sweep_res.copy()]) if entire_df is not None else c_sweep_res
                c_sweep_res = c_sweep_res[c_sweep_res.score >= acc_target].sort_values(["hour", "seg_start"])
                c_sweep_res = c_sweep_res.loc[c_sweep_res.groupby(["hour", "seg_start"]).mfs_approach.idxmax()]

                centroids_df = centroids_df.merge(c_sweep_res, how='outer', indicator=True).loc[lambda x : x['_merge']=='left_only'][["seg_start", "chunk_start", "hour", "cluster"]]
                mfs_sweep_index += 1

            else:
                temp_df = centroids_df.merge(entire_df, on=["seg_start", "chunk_start", "hour"])
                c_sweep_res = temp_df.loc[temp_df.groupby(["seg_start", "chunk_start", "hour"]).score.idxmax()].drop(columns=["cluster"])
                centroids_df = []
            full_df = pd.concat([c_sweep_res, full_df]) if full_df is not None else c_sweep_res

        full_df = full_df.merge(self.all_dfs.loc[centroids], on=["hour", "chunk_start", "seg_start"]).sort_values("cluster").reset_index().drop(columns=["index"])

        assert len(full_df) == len(centroids)

        self.all_dfs["mfs_approach"] = full_df.mfs_approach.values[self.all_dfs["cluster"]]

        assert len(full_df.merge(self.all_dfs, how='outer', on=["chunk_start", "seg_start", "hour", "cluster"], indicator=True).loc[lambda x : x['_merge']=='both'].loc[lambda x : x['mfs_approach_x'] != x['mfs_approach_y']]) == 0

        remaining_segments_df = full_df.merge(self.all_dfs, how='outer', on=["chunk_start", "seg_start", "hour", "cluster", "mfs_approach"], indicator=True).loc[lambda x : x['_merge']=='right_only'][["hour", "chunk_start", "seg_start", "mfs_approach", "cluster"]]

        remaining_qps = []

        for (_, (hour, chunk_start, seg_start, mfs_approach, cluster)) in remaining_segments_df.iterrows():
            vd = VideoData(self.vid_label, hour)
            qp = QueryProcessor(query_type, vd, model, query_class, self.query_conf, mfs_approach, self.bg_conf, self.traj_conf, ioda, self.query_seg_size)
            remaining_qps.append([chunk_start, seg_start, qp])

        self.qp_sweep = remaining_qps
        remaining_query_results = parallelize_update_dictionary(self._sweep_helper, range(len(remaining_qps)), total_cpus=72, max_workers=36)
        self.qp_sweep = None

        _tempDF = list(map(itemgetter(1), sorted(remaining_query_results.items())))
        remaining_query_results = pd.concat(_tempDF).reset_index().drop(columns="index")

        mfs = remaining_query_results.min_frames.sum() + len(centroids) * self.query_seg_size * self.fps / 30
        full_results = pd.concat([full_df, remaining_query_results]).sort_values(["hour", "seg_start"]).reset_index().drop(columns="index")
        score = np.round(full_results.score.mean(), 4)

        assert len(full_results) == len(self.all_dfs)

        total_frames = len(full_results) * self.query_seg_size * self.fps / 30

        if not get_boggart_results:
            return score, mfs, total_frames, np.round(mfs/total_frames, 4), n_clusters

        assert self.query_seg_size == self.chunk_size

        all_rows = []
        for _, row in full_results.sort_values(["hour", "chunk_start"]).iterrows():
            vd = VideoData(row.vid, row.hour)
            qp = QueryProcessor(query_type, vd, model,query_class, self.query_conf, row.mfs_approach, self.bg_conf, self.traj_conf, ioda, self.query_seg_size)
            chunk_result = qp.load_boggart_results(row.chunk_start, row.seg_start)
            for frame_diff, frame_result in enumerate(chunk_result):
                if query_type == "bbox":
                    for det in frame_result:
                        all_rows.append([row.hour, row.chunk_start+frame_diff] + det)
                else:
                    all_rows.append([row.hour, row.chunk_start+frame_diff, frame_result])

        if query_type == "bbox":
            return pd.DataFrame(all_rows, columns=["hour", "frame_no", "x1", "y1", "x2", "y2"])
        elif query_type == "count":
            return pd.DataFrame(all_rows, columns=["hour", "frame_no", "count"])
        elif query_type == "binary":
            return pd.DataFrame(all_rows, columns=["hour", "frame_no", "found"])
        else:
            assert False