def check_good_flow_pairs()

in flow.py [0:0]

• 33 lines of code
• 11 McCabe index (conditional complexity)

    def check_good_flow_pairs(self, frame_pairs, overlap_ratio):
        flow_list_path = pjoin(self.out_path, "flow_list_%.2f.json" % overlap_ratio)
        if os.path.isfile(flow_list_path):
            return flow_list_path

        def ratio(mask):
            return np.sum(mask > 0) / np.prod(mask.shape[:2])

        mask_fmt = pjoin(self.path, "mask", "mask_{:06d}_{:06d}.png")
        result_pairs = []
        checked_pairs = set()
        for pair in frame_pairs:
            if pair in checked_pairs:
                continue

            cur_pairs = [pair, pair[::-1]]
            checked_pairs.update(cur_pairs)

            mask_fns = [mask_fmt.format(*ids) for ids in cur_pairs]
            masks = [cv2.imread(fn, 0) for fn in mask_fns]
            mask_ratios = [ratio(m) for m in masks]
            if all(r >= overlap_ratio for r in mask_ratios):
                result_pairs.extend(cur_pairs)
            else:
                print("Bad frame pair(%d, %d). Overlap_ratio=" % (pair[0], pair[1]),
                    mask_ratios)

        print(f"Filtered {len(result_pairs)} / {len(frame_pairs)} good frame pairs")

        if len(result_pairs) == 0:
            raise Exception("No good frame pairs are found.")

        frame_dists = np.array([np.abs(i - j) for (i, j) in result_pairs])
        print(
            "Frame distance statistics: max = %d, mean = %d, median = %d" %
            (np.amax(frame_dists), np.mean(frame_dists), np.median(frame_dists))
        )

        with open(flow_list_path, "w") as f:
            json.dump(list(result_pairs), f)
        return flow_list_path