''' @ Date: 2022-06-20 15:03:50 @ Author: Qing Shuai @ Mail: s_q@zju.edu.cn @ LastEditors: Qing Shuai @ LastEditTime: 2022-08-16 20:24:07 @ FilePath: /EasyMocapPublic/easymocap/mytools/colmap_wrapper.py ''' import shutil import sys import os import sqlite3 import numpy as np from os.path import join import cv2 from .debug_utils import mkdir, run_cmd, log, mywarn from .colmap_structure import Camera, Image, CAMERA_MODEL_NAMES from .colmap_structure import rotmat2qvec from .colmap_structure import read_points3d_binary IS_PYTHON3 = sys.version_info[0] >= 3 MAX_IMAGE_ID = 2**31 - 1 CREATE_CAMERAS_TABLE = """CREATE TABLE IF NOT EXISTS cameras ( camera_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, model INTEGER NOT NULL, width INTEGER NOT NULL, height INTEGER NOT NULL, params BLOB, prior_focal_length INTEGER NOT NULL)""" CREATE_DESCRIPTORS_TABLE = """CREATE TABLE IF NOT EXISTS descriptors ( image_id INTEGER PRIMARY KEY NOT NULL, rows INTEGER NOT NULL, cols INTEGER NOT NULL, data BLOB, FOREIGN KEY(image_id) REFERENCES images(image_id) ON DELETE CASCADE)""" CREATE_IMAGES_TABLE = """CREATE TABLE IF NOT EXISTS images ( image_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, name TEXT NOT NULL UNIQUE, camera_id INTEGER NOT NULL, prior_qw REAL, prior_qx REAL, prior_qy REAL, prior_qz REAL, prior_tx REAL, prior_ty REAL, prior_tz REAL, CONSTRAINT image_id_check CHECK(image_id >= 0 and image_id < {}), FOREIGN KEY(camera_id) REFERENCES cameras(camera_id)) """.format(MAX_IMAGE_ID) CREATE_TWO_VIEW_GEOMETRIES_TABLE = """ CREATE TABLE IF NOT EXISTS two_view_geometries ( pair_id INTEGER PRIMARY KEY NOT NULL, rows INTEGER NOT NULL, cols INTEGER NOT NULL, data BLOB, config INTEGER NOT NULL, F BLOB, E BLOB, H BLOB) """ CREATE_KEYPOINTS_TABLE = """CREATE TABLE IF NOT EXISTS keypoints ( image_id INTEGER PRIMARY KEY NOT NULL, rows INTEGER NOT NULL, cols INTEGER NOT NULL, data BLOB, FOREIGN KEY(image_id) REFERENCES images(image_id) ON DELETE CASCADE) """ CREATE_MATCHES_TABLE = """CREATE TABLE IF NOT EXISTS matches ( pair_id INTEGER PRIMARY KEY NOT NULL, rows INTEGER NOT NULL, cols INTEGER NOT NULL, data BLOB)""" CREATE_NAME_INDEX = \ "CREATE UNIQUE INDEX IF NOT EXISTS index_name ON images(name)" CREATE_ALL = "; ".join([ CREATE_CAMERAS_TABLE, CREATE_IMAGES_TABLE, CREATE_KEYPOINTS_TABLE, CREATE_DESCRIPTORS_TABLE, CREATE_MATCHES_TABLE, CREATE_TWO_VIEW_GEOMETRIES_TABLE, CREATE_NAME_INDEX ]) def image_ids_to_pair_id(image_id1, image_id2): if image_id1 > image_id2: image_id1, image_id2 = image_id2, image_id1 return image_id1 * MAX_IMAGE_ID + image_id2 def pair_id_to_image_ids(pair_id): image_id2 = pair_id % MAX_IMAGE_ID image_id1 = (pair_id - image_id2) // MAX_IMAGE_ID return image_id1, image_id2 def array_to_blob(array): if IS_PYTHON3: return array.tobytes() else: return np.getbuffer(array) def blob_to_array(blob, dtype, shape=(-1,)): if blob is None: return np.empty((0, 2), dtype=dtype) if IS_PYTHON3: return np.frombuffer(blob, dtype=dtype).reshape(*shape) else: return np.frombuffer(blob, dtype=dtype).reshape(*shape) class COLMAPDatabase(sqlite3.Connection): @staticmethod def connect(database_path): return sqlite3.connect(database_path, factory=COLMAPDatabase) def __init__(self, *args, **kwargs): super(COLMAPDatabase, self).__init__(*args, **kwargs) self.create_tables = lambda: self.executescript(CREATE_ALL) self.create_cameras_table = \ lambda: self.executescript(CREATE_CAMERAS_TABLE) self.create_descriptors_table = \ lambda: self.executescript(CREATE_DESCRIPTORS_TABLE) self.create_images_table = \ lambda: self.executescript(CREATE_IMAGES_TABLE) self.create_two_view_geometries_table = \ lambda: self.executescript(CREATE_TWO_VIEW_GEOMETRIES_TABLE) self.create_keypoints_table = \ lambda: self.executescript(CREATE_KEYPOINTS_TABLE) self.create_matches_table = \ lambda: self.executescript(CREATE_MATCHES_TABLE) self.create_name_index = lambda: self.executescript(CREATE_NAME_INDEX) def add_camera(self, model, width, height, params, prior_focal_length=False, camera_id=None): params = np.asarray(params, np.float64) cursor = self.execute( "INSERT INTO cameras VALUES (?, ?, ?, ?, ?, ?)", (camera_id, model, width, height, array_to_blob(params), prior_focal_length)) return cursor.lastrowid def add_image(self, name, camera_id, prior_q=np.full(4, np.NaN), prior_t=np.full(3, np.NaN), image_id=None): cursor = self.execute( "INSERT INTO images VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (image_id, name, camera_id, prior_q[0], prior_q[1], prior_q[2], prior_q[3], prior_t[0], prior_t[1], prior_t[2])) return cursor.lastrowid def add_keypoints(self, image_id, keypoints): assert(len(keypoints.shape) == 2) assert(keypoints.shape[1] in [2, 4, 6]) keypoints = np.asarray(keypoints, np.float32) self.execute( "INSERT INTO keypoints VALUES (?, ?, ?, ?)", (image_id,) + keypoints.shape + (array_to_blob(keypoints),)) def add_descriptors(self, image_id, descriptors): descriptors = np.ascontiguousarray(descriptors, np.uint8) self.execute( "INSERT INTO descriptors VALUES (?, ?, ?, ?)", (image_id,) + descriptors.shape + (array_to_blob(descriptors),)) def add_matches(self, image_id1, image_id2, matches): assert(len(matches.shape) == 2) assert(matches.shape[1] == 2) if image_id1 > image_id2: matches = matches[:,::-1] pair_id = image_ids_to_pair_id(image_id1, image_id2) matches = np.asarray(matches, np.uint32) self.execute( "INSERT INTO matches VALUES (?, ?, ?, ?)", (pair_id,) + matches.shape + (array_to_blob(matches),)) def add_two_view_geometry(self, image_id1, image_id2, matches, F=np.eye(3), E=np.eye(3), H=np.eye(3), config=2): assert(len(matches.shape) == 2) assert(matches.shape[1] == 2) if image_id1 > image_id2: matches = matches[:,::-1] pair_id = image_ids_to_pair_id(image_id1, image_id2) matches = np.asarray(matches, np.uint32) F = np.asarray(F, dtype=np.float64) E = np.asarray(E, dtype=np.float64) H = np.asarray(H, dtype=np.float64) self.execute( "INSERT INTO two_view_geometries VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (pair_id,) + matches.shape + (array_to_blob(matches), config, array_to_blob(F), array_to_blob(E), array_to_blob(H))) def read_images(self): image_id_to_name, name_to_image_id = {}, {} image_results = self.execute("SELECT * FROM images") for result in image_results: image_id, name, camera_id, q0, q1, q2, q3, t0, t1, t2 = result image_id_to_name[image_id] = name name_to_image_id[name] = image_id return image_id_to_name, name_to_image_id def read_keypoints(self, mapping=None): image_id_to_keypoints = {} keypoints_results = self.execute("SELECT * FROM keypoints") for keypoints_result in keypoints_results: image_id, rows, cols, keypoints = keypoints_result keypoints = blob_to_array(keypoints, np.float32, (rows, cols)) if mapping is None: image_id_to_keypoints[image_id] = keypoints else: image_id_to_keypoints[mapping[image_id]] = keypoints return image_id_to_keypoints def read_matches(self, mapping=None): matches_results = self.execute("SELECT * FROM matches") matches = {} for matches_result in matches_results: pair_id, rows, cols, match = matches_result image_id0, image_id1 = pair_id_to_image_ids(pair_id) if rows == 0: continue match = blob_to_array(match, dtype=np.uint32, shape=(rows, cols)) if mapping is not None: image_id0 = mapping[image_id0] image_id1 = mapping[image_id1] matches[(image_id0, image_id1)] = match return matches def read_two_view_geometry(self, mapping=None): geometry = self.execute("SELECT * FROM two_view_geometries") geometries = {} for pair_id, rows, cols, data, config, F, E, H in geometry: F = blob_to_array(F, dtype=np.float64) E = blob_to_array(E, dtype=np.float64) H = blob_to_array(H, dtype=np.float64) image_id0, image_id1 = pair_id_to_image_ids(pair_id) match = blob_to_array(data, dtype=np.uint32, shape=(rows, cols)) if rows == 0:continue if mapping is not None: image_id0 = mapping[image_id0] image_id1 = mapping[image_id1] geometries[(image_id0, image_id1)] = {'matches': match, 'F':F, 'E':E, 'H':H, 'config': config} return geometries def create_empty_db(database_path): if os.path.exists(database_path): mywarn('Removing old database: {}'.format(database_path)) os.remove(database_path) print('Creating an empty database...') db = COLMAPDatabase.connect(database_path) db.create_tables() db.commit() db.close() def create_cameras(db, cameras, subs, width, height, share_intri=True): model = 'OPENCV' if share_intri: cam_id = 1 K = cameras[subs[0]]['K'] D = cameras[subs[0]]['dist'].reshape(1, 5) fx, fy, cx, cy, k1, k2, p1, p2, k3, k4, k5, k6 = K[0, 0], K[1, 1], K[0, 2], K[1, 2], D[0, 0], D[0, 1], D[0, 2], D[0, 3], D[0, 4], 0, 0, 0 params = [fx, fy, cx, cy, k1, k2, p1, p2] # params = [fx, fy, cx, cy, 0, 0, 0, 0] camera = Camera( id=cam_id, model=model, width=width, height=height, params=params ) cameras_colmap = {cam_id: camera} cameras_map = {sub:cam_id for sub in subs} # db.add_camera(CAMERA_MODEL_NAMES[model].model_id, width, height, params, prior_focal_length=False, camera_id=cam_id) else: raise NotImplementedError return cameras_colmap, cameras_map def create_images(db, cameras, cameras_map, image_names): subs = sorted(list(image_names.keys())) images = {} for sub, image_name in image_names.items(): img_id = subs.index(sub) + 1 R = cameras[sub]['R'] T = cameras[sub]['T'] qvec = rotmat2qvec(R) tvec = T.T[0] image = Image( id=img_id, qvec=qvec, tvec=tvec, camera_id=cameras_map[sub], name=os.path.basename(image_name), xys=[], point3D_ids=[] ) images[img_id] = image db.add_image(image.name, camera_id=image.camera_id, prior_q=image.qvec, prior_t=image.tvec, image_id=img_id) return images def copy_images(data, out, nf=0, copy_func=shutil.copyfile, mask='mask', add_mask=True): subs = sorted(os.listdir(join(data, 'images'))) image_names = {} for sub in subs: srcname = join(data, 'images', sub, '{:06d}.jpg'.format(nf)) if not os.path.exists(srcname): mywarn('{} not exists, skip'.format(srcname)) return False dstname = join(out, 'images', '{}.jpg'.format(sub)) image_names[sub] = dstname if os.path.exists(dstname): continue os.makedirs(os.path.dirname(dstname), exist_ok=True) copy_func(srcname, dstname) mskname = join(data, mask, sub, '{:06d}.png'.format(nf)) dstname = join(out, 'mask', '{}.jpg.png'.format(sub)) if os.path.exists(mskname) and add_mask: os.makedirs(os.path.dirname(dstname), exist_ok=True) copy_func(mskname, dstname) return True, image_names def colmap_feature_extract(colmap, path, share_camera, add_mask): ''' struct SiftMatchingOptions { // Number of threads for feature matching and geometric verification. int num_threads = -1; // Whether to use the GPU for feature matching. bool use_gpu = true; // Index of the GPU used for feature matching. For multi-GPU matching, // you should separate multiple GPU indices by comma, e.g., "0,1,2,3". std::string gpu_index = "-1"; // Maximum distance ratio between first and second best match. double max_ratio = 0.8; // Maximum distance to best match. double max_distance = 0.7; // Whether to enable cross checking in matching. bool cross_check = true; // Maximum number of matches. int max_num_matches = 32768; // Maximum epipolar error in pixels for geometric verification. double max_error = 4.0; // Confidence threshold for geometric verification. double confidence = 0.999; // Minimum/maximum number of RANSAC iterations. Note that this option // overrules the min_inlier_ratio option. int min_num_trials = 100; int max_num_trials = 10000; // A priori assumed minimum inlier ratio, which determines the maximum // number of iterations. double min_inlier_ratio = 0.25; // Minimum number of inliers for an image pair to be considered as // geometrically verified. int min_num_inliers = 15; // Whether to attempt to estimate multiple geometric models per image pair. bool multiple_models = false; // Whether to perform guided matching, if geometric verification succeeds. bool guided_matching = false; bool Check() const; }; ''' cmd = f'{colmap} feature_extractor --database_path {path}/database.db \ --image_path {path}/images \ --SiftExtraction.peak_threshold 0.006 \ --ImageReader.camera_model OPENCV \ ' if share_camera: cmd += ' --ImageReader.single_camera 1' if add_mask: cmd += f' --ImageReader.mask_path {path}/mask' cmd += f' >> {path}/log.txt' run_cmd(cmd) def colmap_feature_match(colmap, path): cmd = f'{colmap} exhaustive_matcher --database_path {path}/database.db \ --SiftMatching.guided_matching 0 \ --SiftMatching.max_ratio 0.8 \ --SiftMatching.max_distance 0.5 \ --SiftMatching.cross_check 1 \ --SiftMatching.max_error 4 \ --SiftMatching.max_num_matches 32768 \ --SiftMatching.confidence 0.9999 \ --SiftMatching.max_num_trials 10000 \ --SiftMatching.min_inlier_ratio 0.25 \ --SiftMatching.min_num_inliers 30 \ >> {path}/log.txt' run_cmd(cmd) def colmap_ba(colmap, path, with_init=False): if with_init: cmd = f'{colmap} point_triangulator --database_path {path}/database.db \ --image_path {path}/images \ --input_path {path}/sparse/0 \ --output_path {path}/sparse/0 \ --Mapper.tri_merge_max_reproj_error 3 \ --Mapper.ignore_watermarks 1 \ --Mapper.filter_max_reproj_error 2 \ >> {path}/log.txt' run_cmd(cmd) cmd = f'{colmap} bundle_adjuster \ --input_path {path}/sparse/0 \ --output_path {path}/sparse/0 \ >> {path}/log.txt' run_cmd(cmd) points3d = read_points3d_binary(join(path, 'sparse', '0', 'points3D.bin')) pids = list(points3d.keys()) mean_error = np.mean([points3d[p].error for p in pids]) log('Triangulate {} points, mean error: {:.2f} pixel'.format(len(pids), mean_error)) else: mkdir(join(path, 'sparse')) cmd = f'{colmap} mapper --database_path {path}/database.db --image_path {path}/images --output_path {path}/sparse \ --Mapper.ba_refine_principal_point 1 \ --Mapper.ba_global_max_num_iterations 1000 \ >> {path}/log.txt' run_cmd(cmd) def colmap_dense(colmap, path): mkdir(join(path, 'dense')) cmd = f'{colmap} image_undistorter --image_path {path}/images --input_path {path}/sparse/0 --output_path {path}/dense --output_type COLMAP --max_image_size 2000' run_cmd(cmd) cmd = f'{colmap} patch_match_stereo \ --workspace_path {path}/dense \ --workspace_format COLMAP \ --PatchMatchStereo.geom_consistency true \ >> {path}/log.txt' run_cmd(cmd) cmd = f'{colmap} stereo_fusion \ --workspace_path {path}/dense \ --workspace_format COLMAP \ --input_type geometric \ --output_path {path}/dense/fused.ply \ >> {path}/log.txt' run_cmd(cmd)