Weather tools

cycles.weather

download_forcing(data_path, forcing, date_start, date_end)

Download forcing netCDF files for a date range.

Parameters:
  • data_path (Path | str) –

    Directory where forcing files are stored.

  • forcing (str) –

    Name of reanalysis product.

  • date_start (datetime | int) –

    Start datetime or start year.

  • date_end (datetime | int) –

    End datetime or end year.

Source code in cycles/weather/weather.py
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def download_forcing(data_path: Path | str, forcing: str, date_start: datetime | int, date_end: datetime | int) -> None:
    """Download forcing netCDF files for a date range.

    Args:
        data_path: Directory where forcing files are stored.
        forcing: Name of reanalysis product.
        date_start: Start datetime or start year.
        date_end: End datetime or end year.
    """
    # Create data directory if it doesn't exist
    data_path = Path(data_path)
    data_path.mkdir(parents=True, exist_ok=True)

    if isinstance(date_start, int) and isinstance(date_end, int):
        date_start = datetime(date_start, 1, 1)
        date_end = datetime(date_end, 12, 31)

    assert isinstance(date_start, datetime) and isinstance(date_end, datetime)

    reanalysis = REANALYSIS[forcing]
    if reanalysis is REANALYSIS.gridMET:
        _download_gridmet(data_path, reanalysis, date_start.year, date_end.year)
    else:
        _download_xldas(data_path, reanalysis, date_start, date_end)

find_grids(forcing, *, locations=None, screen_output=True, remove_duplicates=True)

Find nearest valid weather grids for provided locations.

Parameters:
  • forcing (str) –

    Name of reanalysis product.

  • locations (LocationInput, default: None ) –

    Optional coordinates or named coordinates.

  • screen_output (bool, default: True ) –

    If True, print selected grid summary.

  • remove_duplicates (bool, default: True ) –

    If True, collapse duplicate grid selections.

Returns:
  • str | list[str]

    Weather filename for one location or list of filenames.

Source code in cycles/weather/weather.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def find_grids(forcing: str, *, locations: LocationInput=None, screen_output: bool=True, remove_duplicates: bool=True) -> str | list[str]:
    """Find nearest valid weather grids for provided locations.

    Args:
        forcing: Name of reanalysis product.
        locations: Optional coordinates or named coordinates.
        screen_output: If True, print selected grid summary.
        remove_duplicates: If True, collapse duplicate grid selections.

    Returns:
        Weather filename for one location or list of filenames.
    """
    df = _find_grids(REANALYSIS[forcing], locations, screen_output=screen_output, remove_duplicates=remove_duplicates)

    return df['weather_file'].iloc[0] if len(df) == 1 else df['weather_file'].tolist()

generate_weather_files(data_path, weather_path, forcing, date_start, date_end, *, hourly=False, locations=None, header=True)

Generate Cycles weather files for selected locations and dates.

Parameters:
  • data_path (Path | str) –

    Directory containing downloaded forcing files.

  • weather_path (Path | str) –

    Output directory for generated weather files.

  • forcing (str) –

    Name of reanalysis product.

  • date_start (datetime) –

    Start date of generated records.

  • date_end (datetime) –

    End date of generated records.

  • hourly (bool, default: False ) –

    If True, write hourly weather files.

  • locations (LocationInput, default: None ) –

    Optional coordinates or named coordinates.

  • header (bool, default: True ) –

    If True, write weather file headers.

Source code in cycles/weather/weather.py
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def generate_weather_files(data_path: Path | str, weather_path: Path | str, forcing: str, date_start: datetime, date_end: datetime, *,
    hourly: bool=False, locations: LocationInput=None, header: bool=True) -> None:
    """Generate Cycles weather files for selected locations and dates.

    Args:
        data_path: Directory containing downloaded forcing files.
        weather_path: Output directory for generated weather files.
        forcing: Name of reanalysis product.
        date_start: Start date of generated records.
        date_end: End date of generated records.
        hourly: If True, write hourly weather files.
        locations: Optional coordinates or named coordinates.
        header: If True, write weather file headers.
    """
    reanalysis = REANALYSIS[forcing]
    resolution = Resolution.HOURLY if hourly else Resolution.DAILY

    Path(weather_path).mkdir(parents=True, exist_ok=True)

    grid_df = _find_grids(reanalysis, locations, screen_output=False, remove_duplicates=True)

    if reanalysis is REANALYSIS.gridMET:
        weather_data = _process_gridmet(Path(data_path), date_start, date_end, grid_df)
    elif reanalysis in [REANALYSIS.GLDAS, REANALYSIS.NLDAS]:
        weather_data = _process_xldas(Path(data_path), reanalysis, date_start, date_end, grid_df, resolution)

    if resolution is Resolution.HOURLY:
        start = max(date_start, reanalysis.start_time)
        freq = '1h'
    elif resolution is Resolution.DAILY:
        start = date_start
        freq = '1d'

    time_index = pd.date_range(start=start, end=date_end + timedelta(days=1), freq=freq, inclusive='left')

    weather_data = {key: _interpolate_to_hourly(reanalysis, np.array(value)) if hourly and reanalysis.data_interval > 1 else np.array(value) for key, value in weather_data.items()}

    _write_weather_files(Path(weather_path), time_index, weather_data, grid_df, header, resolution)