ServerlessVideoAnalytics/run.py at main · STAR-Tsinghua/ServerlessVideoAnalytics · 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
191
192
193
194
195
196
197
198
199
200
201
202
import csv
import os
import json
import time

from main_pipeline import main_pipeline
from videos import get_dataset_class, get_seg_paths
import scheduler

scheduler_dict = {
    "baseline": scheduler.BaselineScheduler,
    "adaptive": scheduler.AdaptiveScheduler
}

DEBUG=False
def debug_print(*args):
    if DEBUG:
        print(*args)

def pipeline(dataset_class, seg_paths, runtime_cfg):
    pipeline_cfg = runtime_cfg["pipeline_cfg"]
    scheduler_class = scheduler_dict[pipeline_cfg["scheduler"]]()
    target_accuracy = pipeline_cfg["target_accuracy"]
    pipeline_mode = pipeline_cfg["pipeline_mode"]
    invoking_mode = pipeline_cfg["invoking_mode"]
    model_list = pipeline_cfg["model_list"]
    tracking_parallel = False
    if pipeline_mode == "tracking":
        tracking_parallel = True
    # load configs
    for seg_path in seg_paths:
        print(seg_path)
        seg_name = os.path.basename(seg_path)
        pipeline = main_pipeline(f"{seg_path}/result/profile.csv", scheduler_class, tracking_parallel, invoking_mode)
        f_out = open(f"{seg_path}/result/{pipeline_cfg['scheduler']}_result.csv", 'w', 1)
        writer = csv.writer(f_out)
        writer.writerow(["video_name", 'averaged_tracking_time', 'f1', 'precision', 'recall', 'target_accuracy', 'cost', 'triggered_times', 'pipeline_time'])
        print(seg_path)
        pipeline.scheduler.load_video_profile(seg_path)
        # loading videos
        video = dataset_class(seg_path, seg_name, model_list, runtime_cfg['ground_truth_model'], filter_flag=False)
        start_frame = 1
        end_frame = video.end_frame_index
        if DEBUG:
            start_frame = 1601
        print('Evaluate {} start={} end={}'.format(
            seg_name, start_frame, end_frame))
        start_t = time.perf_counter()
        frames_log, dt_glimpse, averaged_tracking_time, triggered, f1, precision, recall, total_cost, cost, interval_accuracy, interval_log, interval_frame_rate, model_list = pipeline.pipeline(seg_name, video, start_frame, end_frame, target_accuracy)
        end_t = time.perf_counter()
        interval_f1 = [item['f1'] for item in interval_accuracy]
        interval_recall = [item['recall'] for item in interval_accuracy]
        pipelien_time = end_t - start_t

        # start_t = time.perf_counter()
        # f1, precision, recall = pipeline.profile(video, start_frame, end_frame, "yolov5x", 1, 100)
        # end_t = time.perf_counter()
        # print(f"profile uses {end_t - start_t}, {f1}, {precision}, {recall}")

        writer.writerow([seg_name, averaged_tracking_time, f1, precision, recall, target_accuracy, total_cost, len(triggered), pipelien_time])
        with open(f"{seg_path}/result/{pipeline_cfg['scheduler']}_interval_information.csv", 'w') as f:
            interval_writer = csv.writer(f)
            interval_writer.writerow(['cost', *cost])
            interval_writer.writerow(['interval_f1', *interval_f1])
            interval_writer.writerow(['interval_recall', *interval_recall])
            interval_writer.writerow(['interval_frame_rate', *interval_frame_rate])
            interval_writer.writerow(['model', *model_list])
        print(cost)
        print(interval_f1)
        print(interval_recall)
        print(interval_frame_rate)
        print(f"pipeline uses {pipelien_time}, {f1}, {precision}, {recall}")
        if DEBUG:
            print("{} pipeline ended".format(seg_name))
            print("target accuracy is {}".format(target_accuracy))
            print("average tracking time is {}".format(averaged_tracking_time))
            print("average f1 score is {}".format(f1))
            print("average recall rate is {}".format(recall))
            print("average precision is {}".format(precision))
            print("overall_cost is {}".format(total_cost))
            print("triggered {}".format(len(triggered)))

def iteration(dataset_class, seg_paths, runtime_cfg):
    iteration_cfg = runtime_cfg["iteration_cfg"]
    scheduler_class = scheduler_dict[iteration_cfg["scheduler"]]()
    target_accuracy_list = iteration_cfg["target_accuracy_list"]
    pipeline_mode = iteration_cfg["pipeline_mode"]
    invoking_mode = iteration_cfg["invoking_mode"]
    model_list = iteration_cfg["model_list"]
    iteration_time = iteration_cfg["iteration_time"]
    tracking_parallel = False
    if pipeline_mode == "tracking":
        tracking_parallel = True
    for seg_path in seg_paths:
        print(seg_path)
        seg_name = os.path.basename(seg_path)
        pipeline = main_pipeline(f"{seg_path}/result/profile.csv", scheduler_class, tracking_parallel, invoking_mode)
        f_out = open(f"{seg_path}/result/{iteration_cfg['scheduler']}_iteration_result.csv", 'w', 1)
        writer = csv.writer(f_out)
        writer.writerow(["video_name", 'averaged_tracking_time', 'f1', 'precision', 'recall', 'target_accuracy', 'cost', 'triggered_times', 'pipeline_time'])
        print(seg_path)
        pipeline.scheduler.load_video_profile(seg_path)
        video = dataset_class(seg_path, seg_name, model_list, runtime_cfg['ground_truth_model'], filter_flag=False)
        for target_accuracy in target_accuracy_list:
            for _ in range(iteration_time):
                start_frame = 1
                end_frame = video.end_frame_index
                if DEBUG:
                    start_frame = 1601
                print('Evaluate {} start={} end={}'.format(
                    seg_name, start_frame, end_frame))
                start_t = time.perf_counter()
                frames_log, dt_glimpse, averaged_tracking_time, triggered, f1, precision, recall, total_cost, cost, interval_accuracy, interval_log, interval_frame_rate, model_list = pipeline.pipeline(seg_name, video, start_frame, end_frame, target_accuracy)
                interval_f1 = [item['f1'] for item in interval_accuracy]
                interval_recall = [item['recall'] for item in interval_accuracy]
                end_t = time.perf_counter()
                pipelien_time = end_t - start_t
                writer.writerow([seg_name, averaged_tracking_time, f1, precision, recall, target_accuracy, total_cost, len(triggered), pipelien_time])
                print(cost)
                print(interval_f1)
                print(interval_recall)
                print(interval_frame_rate)
                print(f"pipeline uses {end_t - start_t}")
                if DEBUG:
                    print("{} pipeline ended".format(seg_name))
                    print("target accuracy is {}".format(target_accuracy))
                    print("average tracking time is {}".format(averaged_tracking_time))
                    print("average f1 score is {}".format(f1))
                    print("average recall rate is {}".format(recall))
                    print("average precision is {}".format(precision))
                    print("overall_cost is {}".format(total_cost))
                    print("triggered {}".format(len(triggered)))

def profile(dataset_class, runtime_cfg):
    profile_cfg = runtime_cfg["profile_cfg"]
    pipeline_mode = profile_cfg["pipeline_mode"]
    invoking_mode = profile_cfg["invoking_mode"]
    model_list = profile_cfg["model_list"]
    tracking_parallel = False
    if pipeline_mode == "tracking":
        tracking_parallel = True
    profile_cfg = runtime_cfg["profile_cfg"]
    profile_paths = profile_cfg["profile_paths"]
    frame_slot = profile_cfg["frame_slot"]
    profile_model_frame_interval = profile_cfg["profile_model_frame_interval"]
    for path in profile_paths:
        print(path)
        seg_name = os.path.basename(path)
        pipeline = main_pipeline(f"{path}/result/profile.csv", scheduler.BaselineScheduler(), tracking_parallel, invoking_mode)
        f = open(f"{path}/profile/profile.csv", 'w')
        writer = csv.writer(f)
        writer.writerow(["profile_slot", "model", "accuracy", "recall", "frame_interval"])
        video = dataset_class(path, seg_name, model_list, runtime_cfg['ground_truth_model'], filter_flag=False)
        result = {}
        result['frame_slot'] = frame_slot
        profile_slot_count = video.frame_count // frame_slot + 1
        result["model_accuracy_to_frame_rate"] = {}
        for i in range(profile_slot_count):
            print(f"{i} of {profile_slot_count}")
            result["model_accuracy_to_frame_rate"][i] = {}
            for model in profile_model_frame_interval:
                result["model_accuracy_to_frame_rate"][i][model] = {"accuracy": [], "frame_rate": [], "recall": []}
                for frame_interval in profile_model_frame_interval[model]:
                    pipeline.scheduler.change_internal_state(model, frame_interval)
                    clip = seg_name + '_' + str(i)
                    start_frame = i * frame_slot + video.start_frame_index
                    end_frame = min((i + 1) * frame_slot, video.end_frame_index)
                    debug_print('{} start={} end={}'.format(clip, start_frame, end_frame))
                    debug_print(f"iteration: {frame_interval} {model}")
                    debug_print(f"scheduler: {pipeline.scheduler.get_next_frame_interval()} {pipeline.scheduler.get_next_frame_interval_model_name()}")
                    f1, precision, recall = pipeline.profile(video, start_frame, end_frame, model, frame_interval, frame_slot)
                    debug_print(f"{f1} {frame_interval} {recall}")
                    result["model_accuracy_to_frame_rate"][i][model]["accuracy"].append(f1)
                    result["model_accuracy_to_frame_rate"][i][model]["frame_rate"].append(frame_interval)
                    result["model_accuracy_to_frame_rate"][i][model]["recall"].append(recall)
                    writer.writerow([i, model, f1, recall, frame_interval])
                    if (f1 < 0.7):
                        break
        with open(f"{path}/profile/profile.json", 'w') as f:
            json.dump(result, f)

def run(args):
    dataset_class = get_dataset_class(args.dataset)
    seg_paths = get_seg_paths(args.data_root, args.dataset, args.video)
    # video is under data_root, if video is not None, data_root will join video
    config = args.config
    if config is None:
        config = './configs/runtime_cfg.json'
    runtime_cfg = None
    with open(config) as cfg:
        runtime_cfg = json.load(cfg)
    assert runtime_cfg != None

    mode = runtime_cfg["mode"]

    if mode == "pipeline":
        pipeline(dataset_class, seg_paths, runtime_cfg)
    if mode == "iteration":
        iteration(dataset_class, seg_paths, runtime_cfg)
    if mode == "profile":
        profile(dataset_class, runtime_cfg)