Wednesday, October 21, 2026

 A simple orthomosaic program for nadir drone videos:

import cv2

import numpy as np

import os



def extract_frames(video_path, output_dir, seconds_interval=1):

    """

    Extract frames every N seconds.

    """

    if not os.path.exists(output_dir):

        os.makedirs(output_dir)


    cap = cv2.VideoCapture(video_path)


    if not cap.isOpened():

        print("Error: Could not open video.")

        return []


    fps = cap.get(cv2.CAP_PROP_FPS)


    if fps <= 0:

        fps = 30


    frame_interval = int(fps * seconds_interval)


    print(

        f"Using frame interval: "

        f"{frame_interval} frames ({seconds_interval} seconds)"

    )


    frame_paths = []


    frame_count = 0

    saved_count = 0


    while True:


        ret, frame = cap.read()


        if not ret:

            break


        if frame_count % frame_interval == 0:


            path = os.path.join(

                output_dir,

                f"frame_{saved_count:04d}.jpg"

            )


            cv2.imwrite(path, frame)


            frame_paths.append(path)


            saved_count += 1


        frame_count += 1


    cap.release()


    print(f"Extracted {saved_count} frames.")


    return frame_paths



def compute_affine_transforms(frame_paths):


    orb = cv2.ORB_create(nfeatures=5000)


    bf = cv2.BFMatcher(cv2.NORM_HAMMING)


    transforms = [np.eye(3, dtype=np.float64)]


    prev_img = cv2.imread(frame_paths[0])

    prev_gray = cv2.cvtColor(prev_img, cv2.COLOR_BGR2GRAY)


    cumulative_T = np.eye(3, dtype=np.float64)


    for i in range(1, len(frame_paths)):


        print(

            f"Computing affine transform "

            f"{i}/{len(frame_paths)-1}"

        )


        curr_img = cv2.imread(frame_paths[i])

        curr_gray = cv2.cvtColor(curr_img, cv2.COLOR_BGR2GRAY)


        kp1, des1 = orb.detectAndCompute(prev_gray, None)

        kp2, des2 = orb.detectAndCompute(curr_gray, None)


        if des1 is None or des2 is None:


            print("Skipping frame: descriptors missing")


            transforms.append(cumulative_T.copy())


            prev_gray = curr_gray

            continue


        matches = bf.knnMatch(des1, des2, k=2)


        good = []


        for m, n in matches:

            if m.distance < 0.75 * n.distance:

                good.append(m)


        print(f"Good matches: {len(good)}")


        if len(good) < 200:

            print(

                f"Rejecting frame "

                f"(only {len(good)} matches)"

            )


            transforms.append(cumulative_T.copy())


            prev_gray = curr_gray

            continue


        src_pts = np.float32(

            [kp1[m.queryIdx].pt for m in good]

        ).reshape(-1, 1, 2)


        dst_pts = np.float32(

            [kp2[m.trainIdx].pt for m in good]

        ).reshape(-1, 1, 2)


        A, inliers = cv2.estimateAffinePartial2D(

            dst_pts,

            src_pts,

            method=cv2.RANSAC,

            ransacReprojThreshold=3

        )


        if A is None:


            print("Skipping frame: affine failed")


            transforms.append(cumulative_T.copy())


            prev_gray = curr_gray

            continue


        T = np.eye(3, dtype=np.float64)


        T[:2, :] = A


        cumulative_T = cumulative_T @ T


        transforms.append(cumulative_T.copy())


        prev_gray = curr_gray


    return transforms



def compute_canvas_size(frame_paths, homographies):


    first = cv2.imread(frame_paths[0])


    h, w = first.shape[:2]


    all_corners = []


    corners = np.float32([

        [0, 0],

        [w, 0],

        [w, h],

        [0, h]

    ]).reshape(-1, 1, 2)


    for H in homographies:


        transformed = cv2.perspectiveTransform(

            corners,

            H

        )


        all_corners.append(transformed)


    all_corners = np.concatenate(all_corners, axis=0)


    min_x = np.min(all_corners[:, 0, 0])

    max_x = np.max(all_corners[:, 0, 0])


    min_y = np.min(all_corners[:, 0, 1])

    max_y = np.max(all_corners[:, 0, 1])


    width = int(np.ceil(max_x - min_x))

    height = int(np.ceil(max_y - min_y))


    print("\nMosaic bounds:")

    print(f"min_x={min_x:.0f}")

    print(f"max_x={max_x:.0f}")

    print(f"min_y={min_y:.0f}")

    print(f"max_y={max_y:.0f}")

    print(f"Canvas size = {width} x {height}")


    translation = np.array([

        [1, 0, -min_x],

        [0, 1, -min_y],

        [0, 0, 1]

    ])


    return width, height, translation



def build_mosaic(

    frame_paths,

    homographies,

    canvas_width,

    canvas_height,

    translation

):


    mosaic = np.zeros(

        (

            canvas_height,

            canvas_width,

            3

        ),

        dtype=np.uint8

    )


    for idx, frame_path in enumerate(frame_paths):


        print(

            f"Warping frame "

            f"{idx+1}/{len(frame_paths)}"

        )


        img = cv2.imread(frame_path)


        H_global = translation @ homographies[idx]


        warped = cv2.warpPerspective(

            img,

            H_global,

            (

                canvas_width,

                canvas_height

            )

        )


        mask = np.any(warped > 0, axis=2)


        mosaic[mask] = warped[mask]


    return mosaic



def stitch_frames(frame_paths):


    if len(frame_paths) == 0:

        return None


    print("\nComputing cumulative homographies...")

    transforms = compute_affine_transforms(frame_paths)


    print("\nComputing canvas size...")

    width, height, translation = compute_canvas_size(

        frame_paths,

        transforms

    )


    print("\nBuilding mosaic...")

    mosaic = build_mosaic(

        frame_paths,

        transforms,

        width,

        height,

        translation

    )


    return mosaic



def main():


    VIDEO_PATH = "mainindexedvideo.mp4"


    FRAMES_DIR = "extracted_frames"


    OUTPUT_MAP = "orthomosaic.jpg"


    frame_paths = extract_frames(

        VIDEO_PATH,

        FRAMES_DIR,

        seconds_interval=1

    )


    if not frame_paths:

        return


    mosaic = stitch_frames(frame_paths)


    if mosaic is not None:


        cv2.imwrite(

            OUTPUT_MAP,

            mosaic

        )


        print(

            f"\nOrthomosaic saved as "

            f"{OUTPUT_MAP}"

        )



if __name__ == "__main__":

    main()






No comments:

Post a Comment