| import copy | |
| import os | |
| import cv2 | |
| import glob | |
| import logging | |
| import argparse | |
| import numpy as np | |
| from tqdm import tqdm | |
| from alike import ALike, configs | |
| class ImageLoader(object): | |
| def __init__(self, filepath: str): | |
| self.N = 3000 | |
| if filepath.startswith("camera"): | |
| camera = int(filepath[6:]) | |
| self.cap = cv2.VideoCapture(camera) | |
| if not self.cap.isOpened(): | |
| raise IOError(f"Can't open camera {camera}!") | |
| logging.info(f"Opened camera {camera}") | |
| self.mode = "camera" | |
| elif os.path.exists(filepath): | |
| if os.path.isfile(filepath): | |
| self.cap = cv2.VideoCapture(filepath) | |
| if not self.cap.isOpened(): | |
| raise IOError(f"Can't open video {filepath}!") | |
| rate = self.cap.get(cv2.CAP_PROP_FPS) | |
| self.N = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT)) - 1 | |
| duration = self.N / rate | |
| logging.info(f"Opened video {filepath}") | |
| logging.info(f"Frames: {self.N}, FPS: {rate}, Duration: {duration}s") | |
| self.mode = "video" | |
| else: | |
| self.images = ( | |
| glob.glob(os.path.join(filepath, "*.png")) | |
| + glob.glob(os.path.join(filepath, "*.jpg")) | |
| + glob.glob(os.path.join(filepath, "*.ppm")) | |
| ) | |
| self.images.sort() | |
| self.N = len(self.images) | |
| logging.info(f"Loading {self.N} images") | |
| self.mode = "images" | |
| else: | |
| raise IOError( | |
| "Error filepath (camerax/path of images/path of videos): ", filepath | |
| ) | |
| def __getitem__(self, item): | |
| if self.mode == "camera" or self.mode == "video": | |
| if item > self.N: | |
| return None | |
| ret, img = self.cap.read() | |
| if not ret: | |
| raise "Can't read image from camera" | |
| if self.mode == "video": | |
| self.cap.set(cv2.CAP_PROP_POS_FRAMES, item) | |
| elif self.mode == "images": | |
| filename = self.images[item] | |
| img = cv2.imread(filename) | |
| if img is None: | |
| raise Exception("Error reading image %s" % filename) | |
| return img | |
| def __len__(self): | |
| return self.N | |
| class SimpleTracker(object): | |
| def __init__(self): | |
| self.pts_prev = None | |
| self.desc_prev = None | |
| def update(self, img, pts, desc): | |
| N_matches = 0 | |
| if self.pts_prev is None: | |
| self.pts_prev = pts | |
| self.desc_prev = desc | |
| out = copy.deepcopy(img) | |
| for pt1 in pts: | |
| p1 = (int(round(pt1[0])), int(round(pt1[1]))) | |
| cv2.circle(out, p1, 1, (0, 0, 255), -1, lineType=16) | |
| else: | |
| matches = self.mnn_mather(self.desc_prev, desc) | |
| mpts1, mpts2 = self.pts_prev[matches[:, 0]], pts[matches[:, 1]] | |
| N_matches = len(matches) | |
| out = copy.deepcopy(img) | |
| for pt1, pt2 in zip(mpts1, mpts2): | |
| p1 = (int(round(pt1[0])), int(round(pt1[1]))) | |
| p2 = (int(round(pt2[0])), int(round(pt2[1]))) | |
| cv2.line(out, p1, p2, (0, 255, 0), lineType=16) | |
| cv2.circle(out, p2, 1, (0, 0, 255), -1, lineType=16) | |
| self.pts_prev = pts | |
| self.desc_prev = desc | |
| return out, N_matches | |
| def mnn_mather(self, desc1, desc2): | |
| sim = desc1 @ desc2.transpose() | |
| sim[sim < 0.9] = 0 | |
| nn12 = np.argmax(sim, axis=1) | |
| nn21 = np.argmax(sim, axis=0) | |
| ids1 = np.arange(0, sim.shape[0]) | |
| mask = ids1 == nn21[nn12] | |
| matches = np.stack([ids1[mask], nn12[mask]]) | |
| return matches.transpose() | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description="ALike Demo.") | |
| parser.add_argument( | |
| "input", | |
| type=str, | |
| default="", | |
| help='Image directory or movie file or "camera0" (for webcam0).', | |
| ) | |
| parser.add_argument( | |
| "--model", | |
| choices=["alike-t", "alike-s", "alike-n", "alike-l"], | |
| default="alike-t", | |
| help="The model configuration", | |
| ) | |
| parser.add_argument( | |
| "--device", type=str, default="cuda", help="Running device (default: cuda)." | |
| ) | |
| parser.add_argument( | |
| "--top_k", | |
| type=int, | |
| default=-1, | |
| help="Detect top K keypoints. -1 for threshold based mode, >0 for top K mode. (default: -1)", | |
| ) | |
| parser.add_argument( | |
| "--scores_th", | |
| type=float, | |
| default=0.2, | |
| help="Detector score threshold (default: 0.2).", | |
| ) | |
| parser.add_argument( | |
| "--n_limit", | |
| type=int, | |
| default=5000, | |
| help="Maximum number of keypoints to be detected (default: 5000).", | |
| ) | |
| parser.add_argument( | |
| "--no_display", | |
| action="store_true", | |
| help="Do not display images to screen. Useful if running remotely (default: False).", | |
| ) | |
| parser.add_argument( | |
| "--no_sub_pixel", | |
| action="store_true", | |
| help="Do not detect sub-pixel keypoints (default: False).", | |
| ) | |
| args = parser.parse_args() | |
| logging.basicConfig(level=logging.INFO) | |
| image_loader = ImageLoader(args.input) | |
| model = ALike( | |
| **configs[args.model], | |
| device=args.device, | |
| top_k=args.top_k, | |
| scores_th=args.scores_th, | |
| n_limit=args.n_limit, | |
| ) | |
| tracker = SimpleTracker() | |
| if not args.no_display: | |
| logging.info("Press 'q' to stop!") | |
| cv2.namedWindow(args.model) | |
| runtime = [] | |
| progress_bar = tqdm(image_loader) | |
| for img in progress_bar: | |
| if img is None: | |
| break | |
| img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) | |
| pred = model(img_rgb, sub_pixel=not args.no_sub_pixel) | |
| kpts = pred["keypoints"] | |
| desc = pred["descriptors"] | |
| runtime.append(pred["time"]) | |
| out, N_matches = tracker.update(img, kpts, desc) | |
| ave_fps = (1.0 / np.stack(runtime)).mean() | |
| status = f"Fps:{ave_fps:.1f}, Keypoints/Matches: {len(kpts)}/{N_matches}" | |
| progress_bar.set_description(status) | |
| if not args.no_display: | |
| cv2.setWindowTitle(args.model, args.model + ": " + status) | |
| cv2.imshow(args.model, out) | |
| if cv2.waitKey(1) == ord("q"): | |
| break | |
| logging.info("Finished!") | |
| if not args.no_display: | |
| logging.info("Press any key to exit!") | |
| cv2.waitKey() | |