Automatically decrease the number of required past steps if data is missing

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- Automatically decrease the number of required past steps if data is missing [!29](https://github.com/dmidk/sunflow/pull/29), @KristianHMoller
+
 ## [v1.3.0]
 
 ### Added

README.md

@@ -103,6 +103,7 @@ podman run -it --rm --entrypoint="" sunflow bash
 | `SATELLITE_DATA_DIRECTORY` | `.` | Directory for input satellite data archive |
 | `PAST_STEPS` | `4` | Number of past time steps for motion field |
 | `FUTURE_STEPS` | `24` | Number of forecast time steps |
+| `SCALE_FUTURE_STEPS` | `true` | If fewer than `PAST_STEPS` input images are available, scale the number of forecast steps by the same ratio. Set to `false` to keep `FUTURE_STEPS` |
 | `INPUT_DATA_AVAILABILITY_DELAY_MINUTES` | `24` | Data availability delay (minutes) |
 | `INPUT_DATA_FREQUENCY_MINUTES` | `15` | Data frequency (minutes) |
 | `MAX_WAITING_TIME_MINUTES` | `27` | Maximum wait time for data (minutes) |

sunflow/config.py

@@ -23,6 +23,12 @@
     "75th_percentile": "p75",
     "90th_percentile": "p90",
 }
+_TRUTHY_VALUES = {"1", "true", "yes", "on"}
+
+
+def _is_truthy(value: str) -> bool:
+    """Return True if the string is a truthy flag (1/true/yes/on, any case)."""
+    return value.strip().lower() in _TRUTHY_VALUES
 
 
 # Predefined domain options
@@ -65,8 +71,7 @@ def from_env(cls) -> Self:
             bucket=os.getenv("S3_BUCKET", None),
             input_prefix=os.getenv("S3_INPUT_PREFIX", "satellite_data"),
             output_prefix=os.getenv("S3_OUTPUT_PREFIX", "nowcasts"),
-            verify_ssl=os.getenv("AWS_VERIFY_SSL", "true").strip().lower()
-            in {"1", "true", "yes", "on"},
+            verify_ssl=_is_truthy(os.getenv("AWS_VERIFY_SSL", "true")),
         )
 
 
@@ -80,6 +85,7 @@ class NowcastConfig:
     beta: float
     past_steps: int
     future_steps: int
+    scale_future_steps: bool
     input_data_availability_delay_minutes: int
     input_data_frequency_minutes: int
     max_waiting_time_minutes: int
@@ -106,6 +112,8 @@ def from_env(
         - BETA (default: 0.0 for ENS_MEMBERS=1, 0.15 for ENS_MEMBERS>1)
         - PAST_STEPS (default: 4)
         - FUTURE_STEPS (default: 24)
+        - SCALE_FUTURE_STEPS (default: true; if false, FUTURE_STEPS is kept
+          even when fewer past steps than PAST_STEPS are available)
         - INPUT_DATA_AVAILABILITY_DELAY_MINUTES (default: 24)
         - INPUT_DATA_FREQUENCY_MINUTES (default: 15)
         - MAX_WAITING_TIME_MINUTES (default: 27)
@@ -153,6 +161,7 @@ def _get_value(name: str, default: str) -> str:
             beta=float(_get_value("BETA", str(default_beta))),
             past_steps=int(_get_value("PAST_STEPS", "4")),
             future_steps=int(_get_value("FUTURE_STEPS", "24")),
+            scale_future_steps=_is_truthy(_get_value("SCALE_FUTURE_STEPS", "true")),
             input_data_availability_delay_minutes=int(
                 _get_value("INPUT_DATA_AVAILABILITY_DELAY_MINUTES", "24")
             ),

sunflow/data_io.py

@@ -376,6 +376,115 @@ def load_data_from_s3(
     return subset_to_bbox(ds, bbox)
 
 
+def scale_future_steps(
+    future_steps: int, requested_past_steps: int, effective_past_steps: int
+) -> int:
+    """Scale future steps by the relative reduction in past steps (minimum 1)."""
+    if effective_past_steps >= requested_past_steps:
+        return future_steps
+    return max(1, round(future_steps * effective_past_steps / requested_past_steps))
+
+
+def fetch_past_data_with_reduction(
+    time_steps: list[datetime],
+    run_mode: str,
+    config: dict[str, Any],
+    bbox: str,
+    dataset_name: str,
+    domain_satellite_name: str,
+    nowcast_config: NowcastConfig,
+    s3_config: S3Config,
+    min_steps: int = 2,
+) -> tuple[xr.Dataset, list[datetime]]:
+    """Fetch past data, using fewer steps if older data is unavailable.
+
+    Time steps are loaded one at a time from the most recent backwards, and
+    loading stops at the first unavailable step. Each file is therefore read
+    at most once, and the result is always a consecutive series ending at the
+    most recent time step (as required for motion estimation). This is
+    equivalent to reducing the number of past steps one at a time.
+
+    Args:
+        time_steps: Requested past time steps in chronological order, the last
+            one being the most recent.
+        run_mode: One of 'download', 'files', or 's3'.
+        config: Dataset configuration dict.
+        bbox: Bounding box string.
+        dataset_name: Name of dataset (options: KNMI, DWD).
+        domain_satellite_name: Satellite domain identifier.
+        nowcast_config: NowcastConfig object.
+        s3_config: S3Config object.
+        min_steps: Minimum number of consecutive steps required.
+
+    Returns:
+        Tuple of the dataset concatenated along 'time' and the list of time
+        steps actually loaded (chronological order).
+
+    Raises:
+        RuntimeError: If fewer than min_steps consecutive steps are available.
+    """
+    from .downloaders import download_past_data
+
+    collected: list[xr.Dataset] = []
+    loaded_steps: list[datetime] = []
+
+    for ts in reversed(time_steps):
+        ts_str = ts.strftime("%Y-%m-%dT%H:%M:%SZ")
+        try:
+            match run_mode:
+                case "download":
+                    ds = download_past_data([ts], config, bbox, dataset_name)
+                case "files":
+                    ds = load_data_from_files(
+                        [ts],
+                        dataset_name,
+                        domain_satellite_name,
+                        nowcast_config.satellite_data_directory,
+                        "past data",
+                        config["filename_format"],
+                        bbox=bbox,
+                    )
+                case "s3":
+                    ds = load_data_from_s3(
+                        [ts],
+                        dataset_name,
+                        domain_satellite_name,
+                        s3_config,
+                        "past data",
+                        config["filename_format"],
+                        bbox=bbox,
+                    )
+                case _:
+                    raise ValueError(f"Unknown run_mode: {run_mode}")
+        except FileNotFoundError as e:
+            logger.warning(f"Past data not available for {ts_str}: {e}")
+            break
+
+        if ds.sizes.get("time", 0) == 0:
+            logger.warning(f"Past data not available for {ts_str}")
+            break
+
+        collected.append(ds)
+        loaded_steps.append(ts)
+
+    if len(loaded_steps) < min_steps:
+        raise RuntimeError(
+            f"Only {len(loaded_steps)} consecutive past time step(s) available "
+            f"ending at {time_steps[-1].strftime('%Y-%m-%dT%H:%M:%SZ')}; "
+            f"at least {min_steps} required. Cannot proceed."
+        )
+
+    if len(loaded_steps) < len(time_steps):
+        logger.warning(
+            f"Only {len(loaded_steps)} of {len(time_steps)} requested past time "
+            "steps are available. Proceeding with reduced past data."
+        )
+
+    collected.reverse()
+    loaded_steps.reverse()
+    return xr.concat(collected, dim="time", data_vars=None), loaded_steps
+
+
 def fetch_clearsky_with_fallback(
     time_steps: list[datetime],
     run_mode: str,
@@ -491,6 +600,7 @@ def save_forecast(
     output_mode: str,
     run_mode: str = "files",
     s3_config: S3Config | None = None,
+    warning: str | None = None,
 ) -> str:
     """Save forecast data to a CF-compliant NetCDF4 file.
 
@@ -514,6 +624,8 @@ def save_forecast(
             or 'full_ensemble').
         run_mode: One of 'files' (local) or 's3'. Defaults to 'files'.
         s3_config: S3Config object; required when run_mode is 's3'.
+        warning: Optional text written as a global 'warning' attribute
+            (e.g. when fewer input images than requested were used).
 
     Returns:
         Filename (basename only) of the written NetCDF file.
@@ -622,6 +734,7 @@ def save_forecast(
             ),
             "model_version": model_version,
             **({"statistics": statistics_attr} if statistics_attr else {}),
+            **({"warning": warning} if warning else {}),
         },
     )
 

sunflow/main.py

@@ -2,7 +2,7 @@
 import argparse
 import os
 import sys
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 
@@ -17,11 +17,10 @@
 from .data_io import (
     fetch_clearsky_with_fallback,
     fetch_current_data_with_retry,
-    load_data_from_files,
-    load_data_from_s3,
+    fetch_past_data_with_reduction,
     save_forecast,
+    scale_future_steps,
 )
-from .downloaders import download_past_data
 from .forecast import (
     compute_ensemble_statistics,
     make_pvlib_clearsky_dataset,
@@ -297,40 +296,42 @@ def run_nowcast(
         "past_observations",
     )
 
-    logger.info(f"Loading past data for {len(past_time_steps)} time steps...")
-    match run_mode:
-        case "download":
-            data = download_past_data(
-                past_time_steps,
-                config,
-                domain_satellite,
-                dataset_name,
-            )
-        case "files":
-            data = load_data_from_files(
-                past_time_steps,
-                dataset_name,
-                domain_satellite_name,
-                nowcast_config.satellite_data_directory,
-                "past data",
-                config["filename_format"],
-                bbox=domain_satellite,
+    logger.info(f"Loading past data for up to {len(past_time_steps)} time steps...")
+    data, past_time_steps = fetch_past_data_with_reduction(
+        past_time_steps,
+        run_mode,
+        config,
+        domain_satellite,
+        dataset_name,
+        domain_satellite_name,
+        nowcast_config,
+        s3_config,
+    )
+    n_past = len(past_time_steps)
+    logger.info(f"Loaded {n_past} past data timesteps")
+    output_warning = None
+    if n_past < nowcast_config.past_steps:
+        if nowcast_config.scale_future_steps:
+            future_steps = scale_future_steps(
+                nowcast_config.future_steps, nowcast_config.past_steps, n_past
             )
-        case "s3":
-            data = load_data_from_s3(
-                past_time_steps,
-                dataset_name,
-                domain_satellite_name,
-                s3_config,
-                "past data",
-                config["filename_format"],
-                bbox=domain_satellite,
+        else:
+            future_steps = nowcast_config.future_steps
+        output_warning = (
+            f"Only {n_past} satellite images were used instead of the requested "
+            f"{nowcast_config.past_steps} because input data was unavailable."
+        )
+        if future_steps != nowcast_config.future_steps:
+            output_warning += (
+                f" The number of forecast steps was reduced from "
+                f"{nowcast_config.future_steps} to {future_steps}."
             )
-
-    n_loaded = len(data.time) if "time" in data.coords else 0
-    logger.info(f"Loaded {n_loaded} past data timesteps")
-    if n_loaded == 0:
-        raise RuntimeError("No past data loaded. Cannot proceed.")
+        logger.warning(
+            f"Reducing past_steps from {nowcast_config.past_steps} to {n_past}. "
+            f"future_steps: {nowcast_config.future_steps} -> {future_steps}."
+        )
+        # Use a copy so the shared config is not modified between runs
+        nowcast_config = replace(nowcast_config, future_steps=future_steps)
     if run_mode in {"files", "s3"}:
         validate_dataset_covers_domain(data, domain_satellite, "Input dataset")
 
@@ -514,6 +515,7 @@ def run_nowcast(
         output_mode,
         run_mode,
         s3_config,
+        warning=output_warning,
     )
 
     logger.info(f"Nowcast completed: {filename}")

sunflow/validation.py

@@ -90,6 +90,14 @@ def validate_nowcast_config(nowcast_config: NowcastConfig) -> None:
         )
         sys.exit(1)
 
+    if nowcast_config.past_steps < 2:
+        logger.error(
+            f"Invalid nowcast configuration: past_steps must be at least 2 "
+            f"(the most recent time step and one more are required). "
+            f"Current value: {nowcast_config.past_steps}. Exiting.\n"
+        )
+        sys.exit(1)
+
 
 def validate_run_mode(run_mode: str, dataset_name: str) -> None:
     """Validate that the run mode is compatible with the dataset.