diff --git a/CHANGELOG.md b/CHANGELOG.md index 5521bab..483c9c9 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 81116dc..2cc5323 100644 --- a/README.md +++ b/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) | diff --git a/sunflow/config.py b/sunflow/config.py index 7173cbc..fff0e2d 100644 --- a/sunflow/config.py +++ b/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") ), diff --git a/sunflow/data_io.py b/sunflow/data_io.py index d0a810d..6758a3e 100644 --- a/sunflow/data_io.py +++ b/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 {}), }, ) diff --git a/sunflow/main.py b/sunflow/main.py index 1be43f0..89108a0 100644 --- a/sunflow/main.py +++ b/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}") diff --git a/sunflow/validation.py b/sunflow/validation.py index b32e475..afc5df2 100644 --- a/sunflow/validation.py +++ b/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.