Enforce time end interval (#248)

ghiggi · web-flow · commit 0c14be962e31 · 2026-03-09T00:40:32.000+01:00
* Update resampler for time being measurement interval end

* Update time block creation for time being end of measurement interval
diff --git a/disdrodb/api/path.py b/disdrodb/api/path.py
@@ -705,6 +705,13 @@ def define_filename(
     """
     from disdrodb.api.checks import check_product, check_product_kwargs, check_temporal_resolution
 
+    # Retrieve temporal resolution
+    if product == "L0C":
+        temporal_resolution = product_kwargs.pop("temporal_resolution", "XXX")
+    else:
+        temporal_resolution = product_kwargs.get("temporal_resolution", "")
+
+    # Check product
     product = check_product(product)
     product_kwargs = check_product_kwargs(product, product_kwargs)
 
@@ -715,9 +722,8 @@ def define_filename(
     # Define product name
     product_name = f"{product}"
 
-    # L0C ... sample interval known only per-file
-    if product in ["L1", "L2E", "L2M"]:
-        temporal_resolution = product_kwargs.get("temporal_resolution")
+    # Temporal resolution required from L0C onwards
+    if product in ["L0C", "L1", "L2E", "L2M"]:
         check_temporal_resolution(temporal_resolution)
         product_name = f"{product}.{temporal_resolution}"
     if product in ["L2M"]:
@@ -810,10 +816,19 @@ def define_l0c_filename(ds, campaign_name: str, station_name: str, sample_interv
     """Define L0C file name."""
     temporal_resolution = define_temporal_resolution(sample_interval, rolling=False)
     starting_time, ending_time = get_file_start_end_time(ds)
-    starting_time = starting_time.strftime("%Y%m%d%H%M%S")
-    ending_time = ending_time.strftime("%Y%m%d%H%M%S")
-    version = ARCHIVE_VERSION
-    filename = f"L0C.{temporal_resolution}.{campaign_name}.{station_name}.s{starting_time}.e{ending_time}.{version}.nc"
+    filename = define_filename(
+        product="L0C",
+        campaign_name=campaign_name,
+        station_name=station_name,
+        # Filename options
+        start_time=starting_time,
+        end_time=ending_time,
+        add_version=True,
+        add_time_period=True,
+        add_extension=True,
+        # Product options
+        temporal_resolution=temporal_resolution,
+    )
     return filename
 
 
diff --git a/disdrodb/etc/products/L2E/LPM/global.yaml b/disdrodb/etc/products/L2E/LPM/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/LPM_V0/global.yaml b/disdrodb/etc/products/L2E/LPM_V0/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/ODM470/global.yaml b/disdrodb/etc/products/L2E/ODM470/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/PARSIVEL/global.yaml b/disdrodb/etc/products/L2E/PARSIVEL/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/PARSIVEL2/global.yaml b/disdrodb/etc/products/L2E/PARSIVEL2/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/PWS100/global.yaml b/disdrodb/etc/products/L2E/PWS100/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/etc/products/L2E/RD80/1MIN.yaml b/disdrodb/etc/products/L2E/RD80/1MIN.yaml
@@ -15,5 +15,5 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
diff --git a/disdrodb/etc/products/L2E/SWS250/1MIN.yaml b/disdrodb/etc/products/L2E/SWS250/1MIN.yaml
@@ -15,5 +15,5 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
diff --git a/disdrodb/etc/products/L2E/global.yaml b/disdrodb/etc/products/L2E/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/l0/l0c_processing.py b/disdrodb/l0/l0c_processing.py
@@ -26,6 +26,7 @@
 from disdrodb.l0.l0b_processing import set_l0b_encodings
 from disdrodb.l1.resampling import add_sample_interval
 from disdrodb.utils.attrs import set_disdrodb_attrs
+from disdrodb.utils.coords import add_time_bnds
 from disdrodb.utils.logger import log_info, log_warning
 from disdrodb.utils.time import ensure_sorted_by_time
 
@@ -34,7 +35,7 @@
 # L0C processing requires searching for data (per time blocks) into neighbouring files:
 # - to account for possible trailing seconds in previous/next files
 # - to get information if at the edges of the time blocks previous/next timesteps are available
-# - to shift the time to ensure reported L0C time is the start of the measurement interval
+# - to shift the time to ensure reported L0C time is the end of the measurement interval
 TOLERANCE_SECONDS = 60 * 3
 
 ####---------------------------------------------------------------------------------
@@ -65,7 +66,7 @@ def split_dataset_by_sampling_intervals(
     measurement_intervals,
     min_sample_interval=10,
     min_block_size=5,
-    time_is_end_interval=True,
+    time_is_interval_end=True,
 ):
     """
     Split a dataset into subsets where each subset has a consistent sampling interval.
@@ -83,7 +84,7 @@ def split_dataset_by_sampling_intervals(
         The minimum number of timesteps with a given sampling interval to be considered.
         Otherwise such portion of data is discarded !
         Defaults to 5 timesteps.
-    time_is_end_interval: bool
+    time_is_interval_end: bool
         Whether time refers to the end of the measurement interval.
         The default is True.
 
@@ -184,7 +185,7 @@ def split_dataset_by_sampling_intervals(
     change_points = np.where(mapped_intervals[:-1] != mapped_intervals[1:])[0] + 1
 
     # Split ds into segments according to change_points
-    offset = 1 if time_is_end_interval else 0
+    offset = 1 if time_is_interval_end else 0
     segments = np.split(np.arange(ds.sizes["time"]), change_points + offset)
 
     # Remove segments with less than min_block_size elements
@@ -623,13 +624,13 @@ def finalize_l0c_dataset(ds, sample_interval, sensor_name, verbose=True, logger=
     # - Just log warnings in the log file
     ds = check_timesteps_regularity(ds=ds, sample_interval=sample_interval, verbose=verbose, logger=logger)
 
-    # Shift timesteps to ensure time correspond to start of measurement interval
-    # TODO as function of sensor name
-
     # Set netCDF dimension order
     # --> Required for correct encoding !
     ds = ds.transpose("time", "diameter_bin_center", ...)
 
+    # Add time bounds
+    ds = add_time_bnds(ds)
+
     # Set encodings
     ds = set_l0b_encodings(ds=ds, sensor_name=sensor_name)
 
@@ -813,6 +814,7 @@ def generate_l0c_datasets(ds, measurement_intervals, ensure_variables_equality=T
         measurement_intervals=measurement_intervals,
         min_sample_interval=10,
         min_block_size=5,
+        time_is_interval_end=True,
     )
 
     # Log a warning if two sampling intervals are present within a given time block
diff --git a/disdrodb/l1/resampling.py b/disdrodb/l1/resampling.py
@@ -20,6 +20,7 @@
 import pandas as pd
 import xarray as xr
 
+from disdrodb.utils.coords import add_time_bnds
 from disdrodb.utils.time import (
     ensure_sample_interval_in_seconds,
     get_dataset_start_end_time,
@@ -138,7 +139,7 @@ def _rolling(ds, variables, window_size, op):
     return ds_subset
 
 
-def resample_dataset(ds, sample_interval, temporal_resolution):
+def resample_dataset(ds, sample_interval, temporal_resolution, time_is_interval_end=True):
     """
     Resample the dataset to a specified accumulation interval.
 
@@ -156,6 +157,9 @@ def resample_dataset(ds, sample_interval, temporal_resolution):
         It should be a string representing the accumulation interval,
         e.g., "5MIN" for 5 minutes, "1H" for 1 hour, "30S" for 30 seconds, etc.
         Prefixed with "ROLL" for rolling resampling, e.g., "ROLL5MIN".
+    time_is_interval_end: bool
+        Whether time correspond to the end of the measurement/accumulation interval/period.
+        The default is True.
 
     Returns
     -------
@@ -279,12 +283,19 @@ def resample_dataset(ds, sample_interval, temporal_resolution):
     var_to_min = [var for var in var_to_min if var in ds]
     var_to_max = [var for var in var_to_max if var in ds]
 
+    # -------------------------------------------------------------------------.
     # Resample the dataset
     # - Rolling currently does not allow direct rolling forward.
     # - We currently use center=False which means search for data backward (right-aligned) !
     # - We then drop the first 'window_size' NaN timesteps and we shift backward the timesteps.
     # - https://github.com/pydata/xarray/issues/9773
     # - https://github.com/pydata/xarray/issues/8958
+
+    if time_is_interval_end:
+        ds = ds.assign_coords(
+            {"time": ds["time"].data - pd.Timedelta(sample_interval, "s")},
+        )  # set temporary time as start of measurement/accumulation interval
+
     if not rolling:
         # Resample
         accumulation = pd.Timedelta(seconds=accumulation_interval)
@@ -304,6 +315,12 @@ def resample_dataset(ds, sample_interval, temporal_resolution):
             {"time": ds_resampled["time"].data[: -window_size + 1]},
         )
 
+    if time_is_interval_end:
+        ds_resampled = ds_resampled.assign_coords(
+            {"time": ds_resampled["time"].data + pd.Timedelta(accumulation_interval, "s")},
+        )  # set back time as end of accumulation interval
+
+    # -------------------------------------------------------------------------.
     # Finalize qc_resampling
     ds_resampled = _finalize_qc_resampling(
         ds_resampled,
@@ -326,4 +343,9 @@ def resample_dataset(ds, sample_interval, temporal_resolution):
 
     # Add accumulation_interval as new sample_interval coordinate
     ds_resampled = add_sample_interval(ds_resampled, sample_interval=accumulation_interval)
+
+    # Add time bounds
+    ds_resampled = add_time_bnds(ds_resampled)
+
+    # Return resampled dataset
     return ds_resampled
diff --git a/disdrodb/routines/l0.py b/disdrodb/routines/l0.py
@@ -939,6 +939,7 @@ def run_l0c_station(
         filepaths=filepaths,
         freq=product_frequency,
         tolerance_seconds=TOLERANCE_SECONDS,
+        time_is_interval_end=True,
     )
 
     # -----------------------------------------------------------------.
diff --git a/disdrodb/routines/options.py b/disdrodb/routines/options.py
@@ -173,7 +173,7 @@ def check_availability_radar_simulations(options):
     return options
 
 
-def _define_blocks_offsets(sample_interval, temporal_resolution):
+def _define_blocks_offsets(sample_interval, temporal_resolution, time_is_interval_end=True):
     """Define blocks offset for resampling logic."""
     # Retrieve accumulation_interval and rolling option
     accumulation_interval, rolling = get_sampling_information(temporal_resolution)
@@ -186,9 +186,16 @@ def _define_blocks_offsets(sample_interval, temporal_resolution):
     block_starts_offset = 0
     block_ends_offset = 0
 
-    # If rolling, need to search also in next time block ...
+    # If rolling, need to load extra data beyond the block boundary to fill the rolling window.
+    # - When time_is_interval_end=False: the rolling drops entries at the END, so extend block_ends forward.
+    # - When time_is_interval_end=True: the rolling drops entries at the START (after the internal
+    #   time shift), so extend block_starts backward.
     if rolling and sample_interval != accumulation_interval:
-        block_ends_offset = accumulation_interval - sample_interval
+        offset = accumulation_interval - sample_interval
+        if time_is_interval_end:
+            block_starts_offset = -offset
+        else:
+            block_ends_offset = offset
     return block_starts_offset, block_ends_offset
 
 
@@ -273,7 +280,12 @@ def __init__(self, filepaths, parallel, sensor_name, temporal_resolutions=None):
             key = json.dumps(archive_options, sort_keys=True)
             if key not in _cache_dict_temporal_partitions:
                 _cache_dict_temporal_partitions[key] = {
-                    sample_interval: define_temporal_partitions(filepaths, parallel=parallel, **archive_options)
+                    sample_interval: define_temporal_partitions(
+                        filepaths,
+                        parallel=parallel,
+                        time_is_interval_end=True,
+                        **archive_options,
+                    )
                     for sample_interval, filepaths in dict_filepaths.items()
                 }
             dict_temporal_partitions = _cache_dict_temporal_partitions[key].copy()  # To avoid in-place replacement
@@ -370,7 +382,12 @@ def __init__(self, product, filepaths, parallel, temporal_resolution, sensor_nam
         #   --> Does some data filtering on what to process !
         # - "strategy=time_block" does not require loading data into memory
         #   --> Does not do data filtering on what to process !
-        temporal_partitions = define_temporal_partitions(filepaths, parallel=parallel, **archive_options)
+        temporal_partitions = define_temporal_partitions(
+            filepaths,
+            parallel=parallel,
+            time_is_interval_end=True,
+            **archive_options,
+        )
 
         # ------------------------------------------------------------------.
         # Group filepaths by temporal partitions
diff --git a/disdrodb/summary/routines.py b/disdrodb/summary/routines.py
@@ -4545,6 +4545,10 @@ def prepare_summary_dataset(ds, velocity_method="theoretical_velocity", source="
         if dim in ds.dims and dim != "frequency":
             ds = ds.isel({dim: 0})
 
+    # Drop time bounds coordinate
+    if "time_bnds" in ds:
+        ds = ds.drop_vars("time_bnds")
+
     # Select only one elevation angle
     if "elevation_angle" in ds.dims:
         ds = ds.isel(elevation_angle=0)
diff --git a/disdrodb/tests/products/L2E/global.yaml b/disdrodb/tests/products/L2E/global.yaml
@@ -22,7 +22,7 @@ product_options:
   below_velocity_tolerance: 3
   maintain_drops_smaller_than: 1
   maintain_drops_slower_than: 2.5
-  maintain_smallest_drops: True
+  maintain_smallest_drops: False
   remove_splashing_drops: True
 radar_enabled: False
 radar_options:
diff --git a/disdrodb/tests/test_api/test_api_path.py b/disdrodb/tests/test_api/test_api_path.py
@@ -793,8 +793,9 @@ def test_l0c_with_sample_interval(self):
             station_name="STATION_NAME",
             start_time=start,
             end_time=end,
+            temporal_resolution="1MIN",
         )
-        assert fn == f"L0C.CAMPAIGN_NAME.STATION_NAME.s20220601000000.e20220601001000.{ARCHIVE_VERSION}.nc"
+        assert fn == f"L0C.1MIN.CAMPAIGN_NAME.STATION_NAME.s20220601000000.e20220601001000.{ARCHIVE_VERSION}.nc"
 
     def test_l2e_with_rolling(self):
         """Test L2E filename includes rolling temporal resolution."""
diff --git a/disdrodb/tests/test_l0/test_l0c_processing.py b/disdrodb/tests/test_l0/test_l0c_processing.py
diff --git a/disdrodb/tests/test_l1/test_l1_resampling.py b/disdrodb/tests/test_l1/test_l1_resampling.py
diff --git a/disdrodb/tests/test_routines/test_options.py b/disdrodb/tests/test_routines/test_options.py
diff --git a/disdrodb/tests/test_utils/test_utils_archiving.py b/disdrodb/tests/test_utils/test_utils_archiving.py
diff --git a/disdrodb/utils/archiving.py b/disdrodb/utils/archiving.py
diff --git a/disdrodb/utils/coords.py b/disdrodb/utils/coords.py
diff --git a/disdrodb/utils/encoding.py b/disdrodb/utils/encoding.py
