Batch Processing Example

In this example, we use the micasense.imageset class to load a set of directories of images into a list of micasense.capture objects, and we iterate over that list, saving out each image as an aligned stack of images as separate bands in a single tiff file each. Part of this process (via imageutils.write_exif_to_stack) injects that the GPS, capture datetime, camera model, etc into the processed images, allowing us to stitch those images using commercial software such as Pix4DMapper or Agisoft Metashape.

Note: for this example to work, the images must have a valid RigRelatives tag. This requires RedEdge (3/M/MX) version of at least 3.4.0, or any version of RedEdge-P/Altum-PT/Altum/RedEdge-MX Dual. If your images don't meet that spec, you can also follow this support article to add the RigRelatives tag to your imagery: https://support.micasense.com/hc/en-us/articles/360006368574-Modifying-older-collections-for-Pix4Dfields-support

%load_ext autoreload
%autoreload 2

Load Images into ImageSet

from ipywidgets import FloatProgress, Layout
from IPython.display import display
import micasense.capture as capture
import os
from pathlib import Path

# set to True if you have an Altum-PT
# or RedEdge-P and wish to output pan-sharpened stacks
panSharpen = True

# If creating a lot of stacks, it is more efficient to save the metadata
# and then write all of the exif to the images after the stacks are created
write_exif_to_individual_stacks = False

panelNames = None
useDLS = True

# set your image path here. See more here: https://docs.python.org/3/library/pathlib.html
imagePath = Path("./data/REDEDGE-MX")

# these will return lists of image paths as strings. Comment out of you aren't using panels.
panelNames = list(imagePath.glob("IMG_0001_*.tif"))
panelNames = [x.as_posix() for x in panelNames]

if panelNames:
    panelCap = capture.Capture.from_filelist(panelNames)

# destinations on your computer to put the stacks
# and RGB thumbnails
outputPath = imagePath / ".." / "stacks"
thumbnailPath = outputPath / "thumbnails"

cam_model = panelCap.camera_model
cam_serial = panelCap.camera_serial

# determine if this sensor has a panchromatic band
if cam_model == "RedEdge-P" or cam_model == "Altum-PT":
    panchroCam = True
else:
    panchroCam = False
    panSharpen = False

# if this is a multicamera system like the RedEdge-MX Dual,
# we can combine the two serial numbers to help identify
# this camera system later.
if len(panelCap.camera_serials) > 1:
    cam_serial = "_".join(panelCap.camera_serials)
    print("Serial number:", cam_serial)
else:
    cam_serial = panelCap.camera_serial
    print("Serial number:", cam_serial)

overwrite = False  # can be set to set to False to continue interrupted processing
generateThumbnails = True

# Allow this code to align both radiance and reflectance images; but excluding
# a definition for panelNames above, radiance images will be used
# For panel images, efforts will be made to automatically extract the panel information
# but if the panel/firmware is before Altum 1.3.5, RedEdge 5.1.7 the panel reflectance
# will need to be set in the panel_reflectance_by_band variable.
# Note: radiance images will not be used to properly create NDVI/NDRE images below.
if panelNames is not None:
    panelCap = capture.Capture.from_filelist(panelNames)
else:
    panelCap = None

if panelCap is not None:
    if panelCap.panel_albedo() is not None and not any(
        v is None for v in panelCap.panel_albedo()
    ):
        panel_reflectance_by_band = panelCap.panel_albedo()
    else:
        panel_reflectance_by_band = [0.49] * len(
            panelCap.eo_band_names()
        )  # RedEdge band_index order

    panel_irradiance = panelCap.panel_irradiance(panel_reflectance_by_band)
    img_type = "reflectance"
else:
    if useDLS:
        img_type = "reflectance"
    else:
        img_type = "radiance"

Serial number: RX02-2023065-SC

import micasense.imageset as imageset

## This progress widget is used for display of the long-running process
f = FloatProgress(min=0, max=1, layout=Layout(width="100%"), description="Loading")
display(f)


def update_f(val):
    if (
        val - f.value
    ) > 0.005 or val == 1:  # reduces cpu usage from updating the progressbar by 10x
        f.value = val


# time is not recognized and would remove the imageset as unused
imgset = imageset.ImageSet.from_directory(imagePath, progress_callback=update_f)
%time imgset = imageset.ImageSet.from_directory(imagePath, progress_callback=update_f)
update_f(1.0)

CPU times: user 10.2 ms, sys: 8.64 ms, total: 18.8 ms
Wall time: 197 ms

Capture map

We can map out the capture GPS locations to ensure we are processing the right data. A GeoJSON of the captures will later be saved to the outputPath.

Define which warp method to use

For newer data sets with RigRelatives tags (images captured with RedEdge (3/M/MX) version 3.4.0 or greater with a valid calibration load, see https://support.micasense.com/hc/en-us/articles/360005428953-Updating-RedEdge-for-Pix4Dfields), we can use the RigRelatives for a simple alignment. To use this simple alignment, simply set warp_matrices=None

For sets without those tags, or sets that require a RigRelatives optimization, we can go through the Alignment.ipynb notebook and get a set of warp_matrices that we can use here to align.

from micasense.warp_io import load_warp_matrices

if panchroCam:
    warp_matrices_filename = cam_serial + "_warp_matrices_SIFT.npy"
else:
    warp_matrices_filename = cam_serial + "_warp_matrices_opencv.npy"

if Path("./" + warp_matrices_filename).is_file():
    print("Found existing warp matrices for camera", cam_serial)
    loaded = load_warp_matrices(warp_matrices_filename, as_projective=panchroCam)

    if panchroCam:
        warp_matrices_SIFT = loaded
    else:
        warp_matrices = loaded
    print("Loaded warp matrices from", Path("./" + warp_matrices_filename).resolve())
else:
    print("No warp matrices found at expected location:", warp_matrices_filename)
    warp_matrices = None
    warp_matrices_SIFT = None

No warp matrices found at expected location: RX02-2023065-SC_warp_matrices_opencv.npy

Align images and save each capture to a layered TIFF file

import datetime
import micasense.imageutils as imageutils

exif_list = []
## This progress widget is used for display of the long-running process
f2 = FloatProgress(min=0, max=1, layout=Layout(width="100%"), description="Saving")
display(f2)


def update_f2(val):
    f2.value = val


if not os.path.exists(outputPath):
    os.makedirs(outputPath)
if generateThumbnails and not os.path.exists(thumbnailPath):
    os.makedirs(thumbnailPath)


try:
    irradiance = panel_irradiance + [0]
except NameError:
    irradiance = None

start = datetime.datetime.now()
for i, capture in enumerate(imgset.captures):
    outputFilename = str(i).zfill(4) + "_" + capture.uuid + ".tif"
    thumbnailFilename = str(i).zfill(4) + "_" + capture.uuid + ".jpg"
    fullOutputPath = os.path.join(outputPath, outputFilename)
    fullThumbnailPath = os.path.join(thumbnailPath, thumbnailFilename)
    if (not os.path.exists(fullOutputPath)) or overwrite:
        if len(capture.images) == len(imgset.captures[0].images):
            if panchroCam:
                capture.radiometric_pan_sharpened_aligned_capture(
                    warp_matrices=warp_matrices_SIFT,
                    irradiance_list=capture.dls_irradiance(),
                    img_type=img_type,
                    write_exif=write_exif_to_individual_stacks,
                )
            else:
                capture.create_aligned_capture(
                    irradiance_list=irradiance, warp_matrices=warp_matrices
                )
            exif_list.append(
                imageutils.prepare_exif_for_stacks(capture, fullOutputPath)
            )
            capture.save_capture_as_stack(
                fullOutputPath,
                pansharpen=panSharpen,
                sort_by_wavelength=True,
                write_exif=write_exif_to_individual_stacks,
            )
            if generateThumbnails:
                capture.save_capture_as_rgb(fullThumbnailPath)
    capture.clear_image_data()
    update_f2(float(i) / float(len(imgset.captures)))
update_f2(1.0)
end = datetime.datetime.now()

print("Saving time: {}".format(end - start))
print(
    "Alignment+Saving rate: {:.2f} images per second".format(
        float(len(imgset.captures)) / float((end - start).total_seconds())
    )
)

Saving time: 0:00:01.444331
Alignment+Saving rate: 1.38 images per second

Write EXIF data to stacks

As mentioned above, it is more time intensive to write the exif data to each image as it is created. Here, we write the exif data after all of the TIFF files have been created. This should take a few seconds per stack.

if not write_exif_to_individual_stacks:
    start = datetime.datetime.now()
    for exif in exif_list:
        imageutils.write_exif_to_stack(existing_exif_list=exif)
    end = datetime.datetime.now()
    print("Saving time: {}".format(end - start))
    print(
        "Alignment+Saving rate: {:.2f} images per second".format(
            float(len(exif_list)) / float((end - start).total_seconds())
        )
    )

Saving time: 0:00:00.668209
Alignment+Saving rate: 2.99 images per second