add function to subset tc by years

Author: NicolasColombi

CHANGELOG.md

@@ -60,6 +60,7 @@ Removed:
 
 ### Added
 
+- `climada.hazard.tc_tracks.TCTracks.subset_years` function
 - `climada.hazard.tc_tracks.TCTracks.from_FAST` function, add Australia basin (AU) [#993](https://github.com/CLIMADA-project/climada_python/pull/993)
 - Add `osm-flex` package to CLIMADA core [#981](https://github.com/CLIMADA-project/climada_python/pull/981)
 - `doc.tutorial.climada_entity_Exposures_osm.ipynb` tutorial explaining how to use `osm-flex`with CLIMADA

climada/hazard/tc_tracks.py

@@ -28,6 +28,7 @@
 import re
 import shutil
 import warnings
+from operator import itemgetter
 from pathlib import Path
 from typing import List, Optional
 
@@ -321,6 +322,74 @@ def subset(self, filterdict):
 
         return out
 
+    def subset_year(self, start_year: int = None, end_year: int = None):
+        """Subset TCTracks between start and end years, both included.
+
+        Parameters:
+        ----------
+        start_year: int
+            First year to include in the selection
+        end_year: int
+            Last year to include in the selection
+
+        Returns:
+        --------
+        subset: TCTracks
+            TCTtracks object containing the subset of tracks
+        Raises:
+        -------
+        TypeError
+            - If either `start_year` or `end_year` is not an integer.
+            - If `self` is not a `TCTracks` object.
+            - If `self.data` is empty (i.e., no tracks are available).
+        ValueError
+            - If `start_year` is greater than `end_year`.
+            - If the date format in a track is invalid and the year cannot be extracted.
+            - If no tracks are found within the specified year range.
+
+        """
+
+        subset = self.__class__()
+
+        if not isinstance(start_year, int) or not isinstance(end_year, int):
+            raise TypeError("Both start_year and end_year must be integers.")
+
+        if start_year > end_year:
+            raise ValueError(
+                f"start_year ({start_year}) cannot be greater than end_year ({end_year})."
+            )
+
+        if not isinstance(self, TCTracks):
+            raise TypeError(
+                f"self should be a TCTtracks object and not {self.__class__()}."
+            )
+
+        if len(self.data) == 0:
+            raise TypeError("self.data should be a non-empty list of tracks.")
+
+        # Find indices corresponding to the years
+        index: list = []
+        for i, track in enumerate(self.data):
+            try:
+                date_array = track.time[0].to_numpy()
+                year = date_array.astype("datetime64[Y]").item().year
+            except AttributeError:
+                raise ValueError(
+                    f"Invalid date format in track {i}, could not extract year."
+                )
+
+            if start_year <= year <= end_year:
+                index.append(i)
+
+        if not index:
+            raise ValueError(
+                f"No tracks found for the years between {start_year} and {end_year}."
+            )
+
+        subset.data = itemgetter(*index)(self.data)
+
+        return subset
+
     def tracks_in_exp(self, exposure, buffer=1.0):
         """Select only the tracks that are in the vicinity (buffer) of an exposure.
 

climada/hazard/test/test_tc_tracks.py

@@ -22,6 +22,7 @@
 import unittest
 from datetime import datetime as dt
 
+import cftime
 import geopandas as gpd
 import numpy as np
 import pandas as pd
@@ -763,6 +764,54 @@ def test_subset(self):
         tc_track = tc.TCTracks.from_ibtracs_netcdf(storm_id=storms)
         self.assertEqual(tc_track.subset({"basin": "SP"}).size, 2)
 
+    def test_subset_years(self):
+        """Test that subset_years correctly select tracks between year min and year max."""
+
+        tc_test = tc.TCTracks.from_simulations_emanuel(TEST_TRACK_EMANUEL)
+        for i in range(5):
+            date = cftime.DatetimeProlepticGregorian(
+                2000 + i, 2, 20, 0, 0, 0, 0, has_year_zero=True
+            )
+            tc_test.data[i]["time"] = np.full(tc_test.data[i].time.shape[0], date)
+
+        tc_subset = tc_test.subset_year(start_year=2001, end_year=2003)
+
+        self.assertEqual(len(tc_subset.data), 3)
+        self.assertEqual(tc_subset.data[0].time[0].item().year, 2001)
+        self.assertEqual(tc_subset.data[1].time[0].item().year, 2002)
+        self.assertEqual(tc_subset.data[2].time[0].item().year, 2003)
+
+        # Invalid input: non-integer start_year
+        with self.assertRaisesRegex(
+            TypeError, "Both start_year and end_year must be integers."
+        ):
+            tc_test.subset_year(start_year="2000", end_year=2003)
+
+        # Invalid input: non-integer end_year
+        with self.assertRaisesRegex(
+            TypeError, "Both start_year and end_year must be integers."
+        ):
+            tc_test.subset_year(start_year=2000, end_year=None)
+
+        # Invalid range: start_year greater than end_year
+        with self.assertRaisesRegex(
+            ValueError, r"start_year \(2005\) cannot be greater than end_year \(2000\)."
+        ):
+            tc_test.subset_year(start_year=2005, end_year=2000)
+
+        # No tracks match the year range
+        with self.assertRaisesRegex(
+            ValueError, "No tracks found for the years between 2050 and 2060."
+        ):
+            tc_test.subset_year(start_year=2050, end_year=2060)
+
+        # Empty data case
+        empty_tc = tc.TCTracks()
+        with self.assertRaisesRegex(
+            TypeError, "self.data should be a non-empty list of tracks."
+        ):
+            empty_tc.subset_year(start_year=2000, end_year=2010)
+
     def test_get_extent(self):
         """Test extent/bounds attributes."""
         storms = ["1988169N14259", "2002073S16161", "2002143S07157"]