### Environment Setup for DLAMP.data

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.md

Instructions for creating a Python environment using Micromamba. Includes a simple setup and an experimental setup for integrating with DLAMP.tw and physicsnemo.

```bash
micromamba env create -n [envname] -c conda-forge python=3.11 conda python-cdo python-eccodes
pip install -r requirements.txt
```

```bash
micromamba env create -n [envname] -c conda-forge python=3.11 conda

git clone https://github.com/NVIDIA/physicsnemo && cd physicsnemo
make install && cd ..

git clone https://github.com/Chia-Tung/DLAMP DLAMP.tw && cd DLAMP.tw
pip install -r requirements.txt && \
pip install hydra-core --upgrade && \
pip install onnxruntime-gpu==1.20.0 && cd ..

git clone https://github.com/YaoChuDoSomething/DLAMP.data DLAMP.data && cd DLAMP.data
pip install -r requirement.txt
```

--------------------------------

### Configure SFNO Workflow Settings

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.SFNO.md

Example YAML configuration for defining time control, I/O paths, and initial condition sources for the SFNO forecast integration.

```yaml
share:
  time_control:
    start: "2024-10-01_00:00"
    end: "2024-10-03_00:00"
    format: "%Y-%m-%d_%H:%M"
    base_step_hours: 1
  
  io_control:
    base_dir: "/path/to/DLAMP.data/"

sfno:
  initial_condition_source: "GFS"
```

--------------------------------

### Install Dependencies for SFNO Integration

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.SFNO.md

Installs the necessary Python packages required to run the SFNO preprocessing pipeline, including earth2studio, torch, and xarray.

```bash
pip install earth2studio torch xarray netcdf4 scipy pyyaml loguru tqdm
```

--------------------------------

### Custom Data Source for SFNO Forecast

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.SFNO.md

This example shows how to integrate a custom data source with the SFNODataProcessor for generating forecasts. It involves creating a custom data source class and assigning it to the processor's data_source attribute before running the forecast.

```python
from src.preproc.sfno_processor import SFNODataProcessor

# Assuming MyCustomDataSource is defined elsewhere
# processor = SFNODataProcessor("config/sfno.yaml")
# processor.data_source = MyCustomDataSource()
# timeline = processor.run_forecast()
```

--------------------------------

### Workflow Execution and Control

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.md

The main entry point for the data processing workflow is dlamp_prep.py. Users can toggle specific modules like the CDS downloader and regridder by modifying the script variables.

```python
from src.preproc.cds_downloader import CDSDataDownloader
from src.preproc.dlamp_regridder import DataRegridder

do_cds_downloader = 1
do_dlamp_regridder = 1

DLAMP_DATA_DIR = "./"
yaml_config = f"{DLAMP_DATA_DIR}/config/era5.yaml"
```

```bash
python dlamp_prep.py
```

--------------------------------

### Execute ERA5 Preprocessing Workflow

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Orchestrates the ERA5 data pipeline by initializing the CDS downloader and the data regridder based on workflow control flags.

```python
#!/usr/bin/env python
from src.preproc.cds_downloader import CDSDataDownloader
from src.preproc.dlamp_regridder import DataRegridder

do_cds_downloader = 1
do_dlamp_regridder = 1
yaml_config = "./config/era5.yaml"

if do_cds_downloader:
    downloader = CDSDataDownloader(yaml_config)
    for curr in downloader.create_timeline():
        downloader.process_download(curr)

if do_dlamp_regridder:
    regridder = DataRegridder(yaml_config)
    regridder.main_process()
```

--------------------------------

### Execute SFNO Preprocessing Workflow

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Handles the SFNO workflow including forecast generation, format conversion, and regridding, supporting both CLI arguments and programmatic execution.

```python
from src.preproc.sfno_processor import SFNODataProcessor
from src.preproc.dlamp_regridder import DataRegridder

yaml_config = "./config/sfno.yaml"
processor = SFNODataProcessor(yaml_config)
timeline = processor.run_forecast()

for curr_time in timeline:
    processor.convert_to_era5_format(curr_time)

regridder = DataRegridder(yaml_config)
regridder.main_process()
```

--------------------------------

### Download ERA5 Data with CDSDataDownloader (Python)

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Handles automated downloading of ERA5 reanalysis data from the Copernicus Climate Data Store (CDS). It supports pressure-level and surface-level datasets, converts GRIB to NetCDF with proper latitude orientation, and manages file naming and directory structures based on YAML configuration. It automatically skips existing files and handles area subsetting and pressure levels.

```python
from src.preproc.cds_downloader import CDSDataDownloader

# Initialize downloader with YAML configurationyaml_config = "./config/era5.yaml"
downloader = CDSDataDownloader(yaml_config)

# Create timeline of all timestamps to process
timeline = downloader.create_timeline()
# Returns: [datetime(2024, 9, 30, 0, 0), datetime(2024, 9, 30, 1, 0), ...]

# Download data for each timestamp
for curr_time in timeline:
    # Downloads pressure-level and surface-level data
    # Converts GRIB to NetCDF with inverted latitude
    # Saves to: ./grib/era5pl_20240930_0000.grib, ./ncdb/Pool/era5pl_20240930_0000.nc
    downloader.process_download(curr_time)

# The downloader automatically:
# - Skips existing files to avoid re-downloading
# - Handles area subsetting (north, south, east, west bounds)
# - Manages pressure levels for upper-air data
# - Uses CDO for coordinate inversion

```

--------------------------------

### Configure ERA5 Workflow via YAML

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Defines the configuration structure for ERA5 data processing, including time control, I/O paths, download parameters, and diagnostic variable registry.

```yaml
share:
  exp_code: "E5_FANAPI"
  time_control:
    start: "2024-09-30_00:00"
    end: "2024-10-03_00:00"
  io_control:
    base_dir: "/path/to/DLAMP.data/"
registry:
  varname:
    z_p:
      requires: ['z']
      function: diag_z_p
```

--------------------------------

### Integrate SFNO Global Forecasts with SFNODataProcessor (Python)

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Integrates NVIDIA's SFNO global weather forecast model into the DLAMP preprocessing pipeline. It runs SFNO forecasts using earth2studio, converts output to ERA5-like format for compatibility with existing diagnostic infrastructure, and supports both GFS and CDS data sources for initial conditions. It saves raw output and provides functions to convert forecast data to a usable format.

```python
from src.preproc.sfno_processor import SFNODataProcessor

# Initialize SFNO processor with configurationyaml_config = "./config/sfno.yaml"
processor = SFNODataProcessor(yaml_config)

# Run complete SFNO forecast
timeline = processor.run_forecast()
# Loads SFNO model via earth2studio
# Fetches initial conditions from GFS or CDS
# Runs forecast for configured duration (e.g., 48 hours)
# Saves raw output to ./sfno_raw/

# Create timeline of forecast valid times
timeline = processor.create_timeline()
# Returns: [datetime(2024, 10, 1, 0), datetime(2024, 10, 1, 1), ...]

# Convert SFNO output to ERA5-like format for a specific time
curr_time = timeline[6]  # 6 hours into forecast
pl_nc, sl_nc = processor.convert_to_era5_format(curr_time)
# Output: ('sfnopl_20241001_0600.nc', 'sfnosl_20241001_0600.nc')

# Complete workflow: forecast + conversion for all timesteps
converted_files = processor.process_forecast_for_regridding()
# Returns: [(datetime, pl_path, sl_path), ...]

```

--------------------------------

### Manage SFNO Feedback Interface

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

The SFNOFeedbackInterface class facilitates two-way data exchange between global SFNO forecasts and regional models. It supports downscaling, boundary blending, and provides hooks for updating global forecasts with regional feedback.

```python
from src.preproc.sfno_processor import SFNOFeedbackInterface

# Initialize feedback interface
yaml_config = "./config/sfno.yaml"
feedback = SFNOFeedbackInterface(yaml_config)

# Check if two-way feedback is enabled
if feedback.enabled:
    print(f"Feedback active: update every {feedback.update_frequency}h")
    print(f"Blending method: {feedback.blending_method}")

# Downscale SFNO global forecast to regional domain
regional_domain = {
    'north': 31, 'south': 17,
    'west': 114, 'east': 128
}
regional_data = feedback.downscale_to_regional(sfno_data, regional_domain)

# Apply boundary blending (when enabled)
blended_data = feedback.apply_boundary_blending(
    regional_data, global_data, domain_spec
)

# Future: Update global forecast with regional model feedback
updated_forecast = feedback.update_global_forecast(
    sfno_forecast, regional_feedback, update_time
)
```

--------------------------------

### Force CPU Usage in PyTorch

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.SFNO.md

This snippet demonstrates how to explicitly set the PyTorch device to CPU, which can be a solution for 'CUDA out of memory' errors during model loading or inference. It's a fallback mechanism when GPU resources are insufficient.

```python
import torch

device = torch.device("cpu")  # Force CPU
```

--------------------------------

### Configuration for Regridding and Diagnostics

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.md

YAML-based configuration for defining interpolation parameters and diagnostic variable registries. These files control how data is processed and which diagnostic functions are applied.

```yaml
regrid:
  target_nc: "./assets/target.nc"
  target_lat: "XLAT"
  target_lon: "XLONG"
  target_pres: "pres_levels"
  source_lat: "lat"
  source_lon: "lon"
  source_pres: "plev"
  levels: [1000, 975, 20]
  adopted_varlist: ["XLONG", "XLAT", "pres_levels", "HGT", "LANDMASK"]
  write_regrid: False
```

```yaml
registry:
  source_dataset: "ERA5"
  varname:
    z_p:
      requires: ['z']
      function: diag_z_p
    tk_p:
      requires: ['t']
      function: diag_tk_p
```

--------------------------------

### Orchestrate Diagnostic Registry

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Manages the loading and execution order of diagnostic functions. It uses topological sorting to ensure that variables are calculated only after their required dependencies are satisfied.

```python
from src.registry.diagnostic_registry import (
    load_diagnostics, sort_diagnostics_by_dependencies
)

# Load diagnostics from YAML configuration
yaml_config = "./config/era5.yaml"
diagnostics = load_diagnostics(yaml_config)

# Sort diagnostics by dependencies (topological sort)
ordered_vars = sort_diagnostics_by_dependencies(diagnostics)

# Execute diagnostics in correct order
for var in ordered_vars:
    if var not in diagnostics:
        continue

    info = diagnostics[var]
    requires = info["requires"]
    diag_func = info["function"]

    # Check all requirements are available
    if all(req in ds.data_vars or req in ds.coords for req in requires):
        result = diag_func(source_dataset, ds)
        output_ds[var] = result
```

--------------------------------

### Regrid Data and Calculate Diagnostics with DataRegridder (Python)

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Performs horizontal interpolation from source grids (ERA5 or SFNO) to target regional domains, with automatic NaN handling using nearest-neighbor fallback. It also orchestrates diagnostic variable calculation in a dependency-aware order determined by topological sorting. Outputs data in RWRF-compatible NetCDF format.

```python
from src.preproc.dlamp_regridder import DataRegridder

# Initialize regridder with YAML configurationyaml_config = "./config/era5.yaml"
regridder = DataRegridder(yaml_config)

# Build timeline from configuration
timeline = regridder.build_timeline()

# Process single timestep (regrid + diagnostics)
curr_time = timeline[0]
regridder.process_single_time(curr_time)
# Output: ./ncdb/Pool/e5dlamp_20240930_0000.nc

# Or run the entire workflow
regridder.main_process()

# The regridder performs:
# 1. Horizontal interpolation from global to regional grid
# 2. Linear interpolation with nearest-neighbor fallback for NaN values
# 3. Diagnostic variable calculation in dependency order
# 4. NetCDF output in RWRF-compatible format

# Generate I/O filenames for a timestamp
pl_nc, sl_nc, output_nc = regridder.gen_io_filename(curr_time)
# Returns: ('era5pl_20240930_0000.nc', 'era5sl_20240930_0000.nc', 'e5dlamp_20240930_0000.nc')

```

--------------------------------

### Calculate Wind Speed from Pressure Level Components

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Calculates the wind speed magnitude from u and v wind components using the Pythagorean theorem. It supports multiple source datasets by mapping specific variable names to standard calculations and returns an xarray DataArray with appropriate metadata.

```python
import numpy as np
import xarray as xr

def diag_wind_speed_p(source_dataset: str, ds: xr.Dataset) -> xr.DataArray:
    """
    Calculate wind speed from u and v components at pressure levels.
    wind_speed = sqrt(u^2 + v^2)
    """
    from src.registry.diagnostic_functions import _create_dataarray

    match source_dataset:
        case "ERA5":
            u = np.squeeze(ds["u"].values)
            v = np.squeeze(ds["v"].values)
            data = np.sqrt(u**2 + v**2)

        case "SFNO":
            u = np.squeeze(ds["u"].values)
            v = np.squeeze(ds["v"].values)
            data = np.sqrt(u**2 + v**2)

        case "RWRF":
            u = np.squeeze(ds["umet_p"].values)
            v = np.squeeze(ds["vmet_p"].values)
            data = np.sqrt(u**2 + v**2)

        case _:
            data = np.nan

    return _create_dataarray(
        data, ds, "wind_speed_p", "Wind Speed", "m s-1"
    )
```

--------------------------------

### Adding Custom Diagnostic Functions in Python

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.SFNO.md

This code illustrates how to define and register new diagnostic functions for use with the SFNO model. It involves creating a Python function that calculates a custom variable and then updating the configuration file to include this new diagnostic.

```python
import xarray as xr
import numpy as np

def diag_MY_VARIABLE(source_dataset: str, ds: xr.Dataset) -> xr.DataArray:
    """Custom diagnostic variable"""
    match source_dataset:
        case "SFNO":
            # Calculate from SFNO variables
            data = calculate_my_variable(ds) # Placeholder for actual calculation
        case _:
            data = np.nan
    
    return _create_dataarray(data, ds, "MY_VAR", "Description", "units") # Placeholder for _create_dataarray
```

--------------------------------

### Calculate Meteorological Diagnostics

Source: https://context7.com/yaochudosomething/dlamp.data/llms.txt

Provides functions to derive meteorological variables from raw model outputs like ERA5. These functions return xarray DataArrays and support custom implementation patterns for specific data sources.

```python
from src.registry.diagnostic_functions import (
    diag_z_p, diag_tk_p, diag_QVAPOR_p, diag_wa_p,
    diag_T2, diag_Q2, diag_rh2, diag_slp, diag_REFL
)
import xarray as xr

# Load regridded dataset
ds = xr.open_dataset("./ncdb/Pool/e5regrid_20240930_0000.nc")

# Calculate geopotential height from geopotential
z_p = diag_z_p("ERA5", ds)

# Calculate water vapor mixing ratio from specific humidity
qvapor = diag_QVAPOR_p("ERA5", ds)

# Calculate vertical velocity from omega
wa = diag_wa_p("ERA5", ds)

# Surface diagnostics
t2 = diag_T2("ERA5", ds)
rh2 = diag_rh2("ERA5", ds)
slp = diag_slp("ERA5", ds)
refl = diag_REFL("ERA5", ds)

# Custom diagnostic function example
def diag_my_variable(source_dataset: str, ds: xr.Dataset) -> xr.DataArray:
    from src.registry.diagnostic_functions import _create_dataarray
    import numpy as np

    match source_dataset:
        case "ERA5":
            temp = np.squeeze(ds["t"].values)
            data = temp - 273.15
        case _:
            data = np.nan

    return _create_dataarray(data, ds, "temp_c", "Temperature", "degC")
```

--------------------------------

### Implement Diagnostic Calculation Function

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.md

Implement the calculation logic in src/registry/diagnostic_functions.py. The function must accept a source dataset string and an xarray Dataset, returning an xarray DataArray.

```python
def diag_my_new_var(source_dataset: str, ds: xr.Dataset) -> xr.DataArray:
    """
    Calculates my new variable.
    """
    temp = np.squeeze(ds["t"].values)
    sfc_pressure = np.squeeze(ds["sp"].values)

    data = temp * np.log(sfc_pressure)

    return _create_dataarray(
        data, ds, "my_new_var", "My New Diagnostic Variable", "units"
    )
```

--------------------------------

### Define Diagnostic Variable in YAML Configuration

Source: https://github.com/yaochudosomething/dlamp.data/blob/main/README.md

Register a new diagnostic variable in the era5.yaml configuration file. The entry specifies the required input variables and the name of the corresponding Python function to execute.

```yaml
registry:
  varname:
    my_new_var:
      requires: ['t', 'sp']
      function: diag_my_new_var
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.