tracking.py

import numpy as np
from abc import ABC, abstractmethod
import xarray as xr
# from enstools.feature.identification._proto_gen import identification_pb2


class TrackingTechnique(ABC):
    """
    Base abstract class for feature tracking algorithms.
    Implementations need to override the abstract track() method.
    """

    @abstractmethod
    def track(self, timesteps, subset: xr.Dataset):
        """
        Abstract tracking method. This gets called for each timesteps list of the feature descriptions.
        This timeline can be for example a reforecast or a member of an ensemble forecast, in which detected objects should
        be tracked over multiple timestamps. This method gets called in parallel for all timelines in the dataset.

        This method should compute links of corresponding objects of two consecutive timestamps. Each object in a
        timestep has its unique ID, and a computed tuple (id1, id2) remarks that object id1 from timestamp t is the
        same object as id2 from timestamp t+1. One tuple is an edge in a tracking graph.

        Parameters
        ----------
        timesteps : iterable of identification_pb2.Timestep
                The timeline to be tracked.
        subset : xarray.Dataset
                Subset to be tracked. Forwarded from identification

        Returns
        -------
        connections : list of tuples of int
                The connections forming the path of the objects in the timesteps.
        """
        connections = []
        return connections

    def execute(self, object_desc, dataset_ref: xr.Dataset):
        """
        Execute the tracking procedure.
        The description is split into the different timelines which can be executed in parallel.

        Parameters
        ----------
        object_desc : identification_pb2.DatasetDescription TODO?
                The description of the detected features from the identificaiton technique.
        dataset_ref : xarray.Dataset
                Reference to the dataset used in the pipeline.
        """
        tracking_sets = object_desc.sets
        from enstools.misc import get_ensemble_dim

        # TODO parallelize sets (intra-set is parallelized in comparer)
        for tracking_set in tracking_sets:

            # TODO split if init time exists?
            ensemble_dim = get_ensemble_dim(dataset_ref)
            if ensemble_dim is not None:
                dataset_sel = dataset_ref.sel({ensemble_dim: tracking_set.member})
            else:
                dataset_sel = dataset_ref

            connections = self.track(tracking_set.timesteps, dataset_sel)

            for connection in connections:
                co_ = tracking_set.connections.add()
                co_.id_1 = connection[0]
                co_.id_2 = connection[1]
        # TODO next link pairs together to "path"
        # track then has N-list of IDs

        # object_desc has now the links.
        # TODO more to do? is edge list of graph enough? maybe wrapper for graph
        # TODO what if tracks are the identification? e.g. AEW identification in Hovmoller
        pass