Timelapse EOWorkflow

eoworkflow_timelapse.py

# generic imports
import os
import imageio
import datetime
import numpy as np

# sentinelhub-py and eo-learn imports
from sentinelhub.geometry import BBox
from sentinelhub.constants import CRS, DataSource
from eolearn.core import EOTask, FeatureType, LinearWorkflow
from eolearn.io import SentinelHubInputTask
from eolearn.features import SimpleFilterTask
from eolearn.coregistration import ThunderRegistration


# Define predicate for filtering frames based on cloud coverage
class MaxCCPredicate:
    def __init__(self, maxcc):
        self.maxcc = maxcc

    def __call__(self, mask):
        height, width, depth = mask.shape
        cc = np.sum(mask) / (height * width)
        return cc <= self.maxcc


# Custom task to create a GIF from an EOPatch
class MakeGIFTask(EOTask):
    def __init__(self, feature, project_dir, frames_per_sec=5):
        self.feature = next(self._parse_features(feature)())
        self.project_dir = project_dir
        self.frames_per_sec = frames_per_sec

    def execute(self, eopatch, *, filename):
        """
        Generates a GIF animation from an EOPatch.
        """
        with imageio.get_writer(os.path.join(self.project_dir, filename), mode='I', fps=self.frames_per_sec) as writer:
            for image in eopatch[self.feature]:
                writer.append_data(np.array(image * 255, dtype=np.uint8))
        return eopatch


# Define BBox and time range
bbox = BBox(bbox=[-6.887, 31.075, -6.855, 31.047], crs=CRS.WGS84)
time_interval = ('2019-01-01', '2020-01-01')

# Initiliase EOTask objects defining the workflow
# 1. Add Sentinel-2 TRUE_COLOR images AND cloud masks
input_task = SentinelHubInputTask(
    bands_feature=(FeatureType.DATA, 'TRUE-COLOR-S2-L1C'),
    bands=['B04', 'B03', 'B02'],
    resolution=10,
    maxcc=1.0,
    time_difference=datetime.timedelta(hours=2),
    data_source=DataSource.SENTINEL2_L1C,
    max_threads=10,
    additional_data=[
        (FeatureType.MASK, 'dataMask'),
        (FeatureType.MASK, 'CLM')
    ])

# patch = input_task.execute(bbox=bbox, time_interval=time_interval)

# 2. Filter out from the series frames with cloud coverage greater than 1%
predicate = MaxCCPredicate(maxcc=0.01)
add_filter = SimpleFilterTask((FeatureType.MASK, 'CLM'), predicate)

# 3. Create GIFs
make_gif_timelapse = MakeGIFTask((FeatureType.DATA, 'TRUE-COLOR-S2-L1C'), './')
make_gif_coregistered = MakeGIFTask((FeatureType.DATA, 'TRUE-COLOR-S2-L1C'), './')

# 4. Coregister frames
coreg_task = ThunderRegistration((FeatureType.DATA, 'TRUE-COLOR-S2-L1C'), channel=0)

# Create workflow
workflow = LinearWorkflow(input_task,
                          add_filter,
                          make_gif_timelapse,
                          coreg_task,
                          make_gif_coregistered)

# Run workflow specifying input parameters at run time
result = workflow.execute({input_task: {'bbox': bbox, 'time_interval': time_interval},
                           make_gif_timelapse: {'filename': 'timelapse.gif'},
                           make_gif_coregistered: {'filename': 'coregistered.gif'}})

Code snippet update to support SH cloud masks and changes in sentinelhub-py.

To execute the code, the minimum eo-learn version is 0.7.4