import os import sys import tempfile import logging import subprocess import shutil from datetime import datetime, timedelta import numpy as np import xarray as xr from ecmwf.opendata import Client import requests # Setup logging first logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Add current directory to path for Arctic extractor import sys.path.append(os.path.dirname(os.path.abspath(__file__))) # Import ONLY the working Arctic GRIB handler for Docker production try: from arctic_grib_handler import ArcticGRIBHandler ARCTIC_HANDLER_AVAILABLE = True logger.info("✅ Arctic GRIB Handler loaded (Docker production version)") except ImportError as e: logger.error(f"❌ Arctic GRIB handler not available: {e}") ARCTIC_HANDLER_AVAILABLE = False class GRIBWavePuller: def __init__(self): self.client = Client("ecmwf") self.output_dir = os.getenv('OUTPUT_DIR', '/tmp/wave_data') os.makedirs(self.output_dir, exist_ok=True) # Set ECCODES environment variables to handle polar stereographic issues self._setup_eccodes_environment() def _setup_eccodes_environment(self): """Setup ECCODES environment variables to handle projection issues""" try: # Set environment variables that might help with polar stereographic processing os.environ['ECCODES_GRIB_STRICT_PARSING'] = '0' # Relaxed parsing os.environ['ECCODES_GRIB_IGNORE_GRID_DEFINITION'] = '1' # Ignore grid definition errors logger.info("Set ECCODES environment variables for relaxed parsing") except Exception as e: logger.warning(f"Could not set ECCODES environment variables: {e}") def _check_cdo_available(self): """Check if CDO (Climate Data Operators) is available""" try: result = subprocess.run(['cdo', '--version'], capture_output=True, text=True, timeout=10) return result.returncode == 0 except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError): return False def _reproject_arctic_with_cdo(self, grib_file_path): """Reproject Arctic GRIB file using CDO as alternative to wgrib2""" try: if not self._check_cdo_available(): logger.warning("CDO not available for Arctic reprojection") return None logger.info("Attempting to reproject Arctic GRIB file using CDO") # Create temporary file for reprojected data temp_reprojected = tempfile.NamedTemporaryFile(delete=False, suffix='_cdo_reprojected.grib2') temp_reprojected.close() # Use CDO to reproject to regular lat-lon grid # remapbil = bilinear interpolation to regular lat-lon grid cmd = [ 'cdo', 'remapbil,r720x360', # 0.5° resolution global grid grib_file_path, temp_reprojected.name ] logger.info(f"Running CDO command: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode == 0: logger.info("Successfully reprojected Arctic GRIB file with CDO") return temp_reprojected.name else: logger.error(f"CDO failed: {result.stderr}") if os.path.exists(temp_reprojected.name): os.unlink(temp_reprojected.name) return None except Exception as e: logger.error(f"Error reprojecting with CDO: {e}") return None def fetch_ecmwf_wave_grib(self, forecast_time=0): """Fetch global wave data from ECMWF open data""" try: logger.info("Fetching ECMWF global wave GRIB data...") # Create temporary file for GRIB data temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.grib2') try: # Fetch wave height data from ECMWF self.client.retrieve( type="fc", # forecast param=["swh"], # significant wave height time=0, # 00 UTC step=forecast_time, # forecast hours ahead target=temp_file.name ) logger.info(f"GRIB file downloaded: {temp_file.name}") return temp_file.name except Exception as e: logger.error(f"Failed to fetch ECMWF data: {e}") # Clean up temp file on error if os.path.exists(temp_file.name): os.unlink(temp_file.name) return None except Exception as e: logger.error(f"Error in fetch_ecmwf_wave_grib: {e}") return None def fetch_noaa_wave_grib(self, forecast_hour=0): """Fetch global wave data from NOAA WW3 model""" try: logger.info(f"Fetching NOAA WW3 global wave GRIB data for forecast hour {forecast_hour}...") # NOAA GFS/WW3 wave data URL pattern base_url = "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod" # Try current date and previous days (in case of delayed updates) now = datetime.utcnow() dates_to_try = [ now.strftime("%Y%m%d"), (now - timedelta(days=1)).strftime("%Y%m%d"), (now - timedelta(days=2)).strftime("%Y%m%d") ] # Try different model runs (00, 06, 12, 18 UTC) to find available data model_runs = ["00", "06", "12", "18"] current_hour = now.hour # Start with the most recent available run if current_hour >= 18: preferred_runs = ["18", "12", "06", "00"] elif current_hour >= 12: preferred_runs = ["12", "06", "00", "18"] elif current_hour >= 6: preferred_runs = ["06", "00", "18", "12"] else: preferred_runs = ["00", "18", "12", "06"] # Try different dates and model runs for date_str in dates_to_try: logger.info(f"Trying date: {date_str}") for hour in preferred_runs: try: # Format forecast hour with leading zeros (f000, f001, f002, etc.) forecast_str = f"f{forecast_hour:03d}" # Download multiple regional files for global coverage successful_downloads = [] # Try different regional GRIB files available on NOAA regional_files = [ (f"gfswave.t{hour}z.atlocn.0p16.{forecast_str}.grib2", "Atlantic"), (f"gfswave.t{hour}z.epacif.0p16.{forecast_str}.grib2", "East_Pacific"), (f"gfswave.t{hour}z.arctic.9km.{forecast_str}.grib2", "Arctic"), # Add more regional files if available (f"gfswave.t{hour}z.global.0p16.{forecast_str}.grib2", "Global"), # Try global if it exists ] # Try to download each regional file for filename, region_name in regional_files: try: url = f"{base_url}/gfs.{date_str}/{hour}/wave/gridded/{filename}" logger.info(f"Attempting to download {region_name} region: {filename}") temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.grib2') response = requests.get(url, timeout=300) if response.status_code == 200: temp_file.write(response.content) temp_file.close() successful_downloads.append((temp_file.name, region_name, hour, forecast_hour)) logger.info(f"{region_name} GRIB file downloaded: {temp_file.name}") else: logger.debug(f"HTTP {response.status_code} for {region_name}") os.unlink(temp_file.name) continue except Exception as file_error: logger.debug(f"Error downloading {region_name}: {file_error}") continue # If we got at least one regional file, return the list if successful_downloads: logger.info(f"Successfully downloaded {len(successful_downloads)} regional files") return successful_downloads except Exception as run_error: logger.warning(f"Error trying {hour}Z run on {date_str}: {run_error}") continue logger.error("Failed to download NOAA data from any model run") return None, None, None except Exception as e: logger.error(f"Error in fetch_noaa_wave_grib: {e}") return None, None, None def _check_wgrib2_available(self): """Check if wgrib2 command-line tool is available""" try: result = subprocess.run(['wgrib2', '-version'], capture_output=True, text=True, timeout=10) return result.returncode == 0 except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError): return False def _reproject_arctic_with_wgrib2(self, grib_file_path): """Reproject Arctic GRIB file to lat-lon grid using wgrib2""" try: if not self._check_wgrib2_available(): logger.warning("wgrib2 not available for Arctic reprojection") return None logger.info("Attempting to reproject Arctic GRIB file using wgrib2") # Create temporary file for reprojected data temp_reprojected = tempfile.NamedTemporaryFile(delete=False, suffix='_reprojected.grib2') temp_reprojected.close() # Use wgrib2 to reproject to lat-lon grid # This covers Arctic regions with reasonable resolution cmd = [ 'wgrib2', grib_file_path, '-new_grid', 'latlon', '0:720:0.5', '50:71:0.5', # 0.5° res, 50-85°N, 0-360°E temp_reprojected.name ] logger.info(f"Running wgrib2 command: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode == 0: logger.info("Successfully reprojected Arctic GRIB file") return temp_reprojected.name else: logger.error(f"wgrib2 failed: {result.stderr}") if os.path.exists(temp_reprojected.name): os.unlink(temp_reprojected.name) return None except Exception as e: logger.error(f"Error reprojecting with wgrib2: {e}") return None def _process_arctic_without_coordinates(self, grib_file_path): """Process Arctic GRIB file bypassing coordinate processing entirely""" try: logger.info("Attempting to process Arctic file without coordinate processing") # Try to open with minimal processing - just get the data values try: # Use cfgrib with very restrictive read_keys to bypass coordinate issues ds = xr.open_dataset( grib_file_path, engine='cfgrib', decode_timedelta=True, backend_kwargs={ 'read_keys': ['paramId', 'shortName', 'name', 'units'], # Only read essential keys 'errors': 'ignore', 'indexpath': '' } ) logger.info(f"Successfully opened Arctic file with restricted processing") logger.info(f"Available variables: {list(ds.variables.keys())}") # Look for wave height data wave_var = None for var_name in ['swh', 'HTSGW', 'htsgw']: if var_name in ds.variables: wave_var = var_name break if wave_var is None: # Try broader search for var_name in ds.variables: if any(keyword in var_name.lower() for keyword in ['wave', 'height', 'swh']): wave_var = var_name break if wave_var: wave_data = ds[wave_var].values logger.info(f"Found wave data: {wave_var}, shape: {wave_data.shape}") # Create fake coordinate grid for Arctic region (approximate) # This is a fallback when we can't get real coordinates if len(wave_data.shape) == 2: rows, cols = wave_data.shape # Create approximate Arctic coordinate grid fake_lats = np.linspace(85, 60, rows) # 85°N to 60°N fake_lons = np.linspace(-180, 180, cols) # Full longitude range lon_grid, lat_grid = np.meshgrid(fake_lons, fake_lats) # Flatten and filter valid data flat_lats = lat_grid.flatten() flat_lons = lon_grid.flatten() flat_waves = wave_data.flatten() # Remove invalid data valid_mask = (~np.isnan(flat_waves)) & (flat_waves >= 0) & (flat_waves < 50) if np.any(valid_mask): filtered_lats = flat_lats[valid_mask] filtered_lons = flat_lons[valid_mask] filtered_waves = flat_waves[valid_mask] logger.info(f"Arctic fallback: {len(filtered_lats)} approximate points") return filtered_lats, filtered_lons, filtered_waves, None, None ds.close() return None, None except Exception as restricted_error: logger.warning(f"Restricted processing failed: {restricted_error}") return None, None except Exception as e: logger.error(f"Error in coordinate-bypass processing: {e}") return None, None def process_grib_file(self, grib_file_path, region_name=None): """Process GRIB file and extract wave data including direction and period""" try: logger.info(f"Processing GRIB file: {grib_file_path}") # Check if this is an Arctic file that needs special handling is_arctic = (region_name and 'arctic' in region_name.lower()) or 'arctic' in grib_file_path.lower() # Try to open GRIB file and extract all available wave parameters try: datasets = [] if is_arctic: # Multi-layered approach for Arctic polar stereographic projection logger.info("Processing Arctic GRIB file with enhanced multi-layered fallback approach") # Approach 1: Try wgrib2 reprojection first (most reliable) reprojected_file = self._reproject_arctic_with_wgrib2(grib_file_path) if reprojected_file: try: logger.info("Processing wgrib2-reprojected Arctic file") ds_height = xr.open_dataset(reprojected_file, engine='cfgrib', decode_timedelta=True) datasets.append(ds_height) # Clean up reprojected file after processing os.unlink(reprojected_file) except Exception as reprojected_error: logger.warning(f"Failed to process reprojected Arctic file: {reprojected_error}") if os.path.exists(reprojected_file): os.unlink(reprojected_file) datasets = [] # Approach 2: If wgrib2 failed, try CDO reprojection if not datasets: cdo_reprojected_file = self._reproject_arctic_with_cdo(grib_file_path) if cdo_reprojected_file: try: logger.info("Processing CDO-reprojected Arctic file") ds_height = xr.open_dataset(cdo_reprojected_file, engine='cfgrib', decode_timedelta=True) datasets.append(ds_height) # Clean up reprojected file after processing os.unlink(cdo_reprojected_file) except Exception as cdo_error: logger.warning(f"Failed to process CDO-reprojected Arctic file: {cdo_error}") if os.path.exists(cdo_reprojected_file): os.unlink(cdo_reprojected_file) datasets = [] # Approach 3: Try coordinate-bypass method if not datasets: try: logger.info("Trying coordinate-bypass method for Arctic processing") result = self._process_arctic_without_coordinates(grib_file_path) if result and result[0] is not None: return result except Exception as bypass_error: logger.warning(f"Coordinate-bypass method failed: {bypass_error}") # Approach 4: If all reprojection methods failed, try cfgrib with relaxed settings if not datasets: try: logger.info("Trying cfgrib with relaxed error handling for Arctic") ds_height = xr.open_dataset(grib_file_path, engine='cfgrib', backend_kwargs={'errors': 'ignore'}, decode_timedelta=True) datasets.append(ds_height) logger.info("Successfully opened Arctic file with relaxed cfgrib settings") except Exception as cfgrib_error: logger.warning(f"Failed to process Arctic file with cfgrib: {cfgrib_error}") # Use ONLY the proven working Arctic GRIB handler if not datasets: try: logger.info("🧊 Using proven Arctic GRIB handler (Docker production)") if not ARCTIC_HANDLER_AVAILABLE: logger.error("❌ Arctic GRIB handler not available") return None, None arctic_handler = ArcticGRIBHandler() result = arctic_handler.get_compatible_format(grib_file_path, sample_points=200) if result['status'] == 'success' and result['sampled_points'] > 0: logger.info(f"✅ Arctic extraction successful: {result['sampled_points']} points") data = result['data'] # Convert to expected format (latitude, longitude, value) return (data['latitude'], data['longitude'], data['value'], None, None) else: logger.error(f"❌ Arctic extraction failed: {result['message']}") return None, None except Exception as extraction_error: logger.error(f"❌ Arctic GRIB handler failed: {extraction_error}") return None, None else: # Normal processing for non-Arctic files with timedelta fix ds_height = xr.open_dataset(grib_file_path, engine='cfgrib', decode_timedelta=True) datasets.append(ds_height) # Try to get wave direction and period by opening with different filters try: ds_ocean = xr.open_dataset(grib_file_path, engine='cfgrib', filter_by_keys={'discipline': 10}, decode_timedelta=True) if ds_ocean.variables.keys() != ds_height.variables.keys(): datasets.append(ds_ocean) except: logger.info("Could not open oceanographic discipline data separately") # Combine all available variables all_vars = {} for ds in datasets: all_vars.update(ds.variables) logger.info(f"Available variables: {list(all_vars.keys())}") except Exception as e: error_msg = str(e) logger.error(f"Error opening GRIB file: {error_msg}") # Check if this is an Arctic polar stereographic error if is_arctic and any(keyword in error_msg.lower() for keyword in [ 'polar stereographic', 'spherical earth', 'geoiterator', 'geographic attributes', 'unable to create iterator' ]): logger.info("Detected Arctic polar stereographic error - using eccodes-based Arctic extraction") try: logger.info("🧊 Arctic error detected - using proven handler") if not ARCTIC_HANDLER_AVAILABLE: logger.error("❌ Arctic GRIB handler not available") return None, None arctic_handler = ArcticGRIBHandler() result = arctic_handler.get_compatible_format(grib_file_path, sample_points=200) if result['status'] == 'success' and result['sampled_points'] > 0: logger.info(f"✅ Arctic error handling successful: {result['sampled_points']} points") data = result['data'] # Convert to expected format (latitude, longitude, value) return (data['latitude'], data['longitude'], data['value'], None, None) else: logger.error(f"❌ Arctic error handling failed: {result['message']}") return None, None except Exception as arctic_error: logger.error(f"❌ Arctic handler error processing failed: {arctic_error}") return None, None else: # Non-Arctic error or different error type return None, None # Extract wave height data wave_height_var = None wave_heights = None for var_name in ['swh', 'HTSGW', 'htsgw']: if var_name in all_vars: wave_height_var = var_name wave_heights = all_vars[var_name].values logger.info(f"Using wave height variable: {wave_height_var}") break if wave_heights is None: # Try broader search for var_name in all_vars: if any(keyword in var_name.lower() for keyword in ['wave', 'height', 'swh']): wave_height_var = var_name wave_heights = all_vars[var_name].values logger.info(f"Found wave height variable: {wave_height_var}") break if wave_heights is None: logger.error("No wave height variables found in GRIB file") for ds in datasets: ds.close() return None, None # Extract wave direction data wave_directions = None wave_dir_var = None for var_name in ['dirpw', 'DIRPW', 'dp', 'wvdir', 'WVDIR', 'dir']: if var_name in all_vars: wave_dir_var = var_name wave_directions = all_vars[var_name].values logger.info(f"Found wave direction variable: {wave_dir_var}") break # Extract wave period data wave_periods = None wave_period_var = None for var_name in ['perpw', 'PERPW', 'tp', 'wvper', 'WVPER', 'per']: if var_name in all_vars: wave_period_var = var_name wave_periods = all_vars[var_name].values logger.info(f"Found wave period variable: {wave_period_var}") break # Get coordinates from the first dataset ds_main = datasets[0] lats = ds_main.latitude.values if 'latitude' in ds_main else ds_main.lat.values lons = ds_main.longitude.values if 'longitude' in ds_main else ds_main.lon.values # Log what we found if wave_directions is not None: logger.info(f"Wave directions shape: {wave_directions.shape}, range: {np.nanmin(wave_directions):.1f}-{np.nanmax(wave_directions):.1f} degrees") if wave_periods is not None: logger.info(f"Wave periods shape: {wave_periods.shape}, range: {np.nanmin(wave_periods):.1f}-{np.nanmax(wave_periods):.1f} seconds") # Create structured data processed_data = { 'timestamp': datetime.utcnow().isoformat(), 'data_source': 'ECMWF_GRIB' if 'ecmwf' in grib_file_path.lower() else 'NOAA_GRIB', 'parameters_found': { 'wave_height': wave_height_var, 'wave_direction': wave_dir_var, 'wave_period': wave_period_var, 'has_velocity_components': wave_dir_var is not None }, 'grid_info': { 'lat_min': float(np.min(lats)), 'lat_max': float(np.max(lats)), 'lon_min': float(np.min(lons)), 'lon_max': float(np.max(lons)), 'lat_resolution': float(lats[1] - lats[0]) if len(lats) > 1 else None, 'lon_resolution': float(lons[1] - lons[0]) if len(lons) > 1 else None, 'grid_shape': wave_heights.shape }, 'wave_statistics': { 'max_wave_height': float(np.nanmax(wave_heights)), 'min_wave_height': float(np.nanmin(wave_heights)), 'mean_wave_height': float(np.nanmean(wave_heights)), 'std_wave_height': float(np.nanstd(wave_heights)) }, 'sample_points': self._extract_sample_points_with_vectors(lats, lons, wave_heights, wave_directions, wave_periods) } # Add direction/period statistics if available if wave_directions is not None: processed_data['direction_statistics'] = { 'mean_direction': float(np.nanmean(wave_directions)), 'direction_std': float(np.nanstd(wave_directions)) } if wave_periods is not None: processed_data['period_statistics'] = { 'max_period': float(np.nanmax(wave_periods)), 'min_period': float(np.nanmin(wave_periods)), 'mean_period': float(np.nanmean(wave_periods)) } # Close all datasets for ds in datasets: ds.close() return processed_data, grib_file_path except Exception as e: logger.error(f"Error processing GRIB file: {e}") return None, None def process_multiple_regional_files(self, regional_files): """Process multiple regional GRIB files and combine data for global coverage""" try: logger.info(f"Processing {len(regional_files)} regional GRIB files for global coverage...") combined_sample_points = [] all_wave_heights = [] all_wave_directions = [] all_wave_periods = [] global_lat_min = float('inf') global_lat_max = float('-inf') global_lon_min = float('inf') global_lon_max = float('-inf') parameters_found = { 'wave_height': None, 'wave_direction': None, 'wave_period': None, 'has_velocity_components': False } regions_processed = [] for grib_file_path, region_name, model_run, forecast_hour in regional_files: try: logger.info(f"Processing {region_name} region: {grib_file_path}") # Process this regional file result = self.process_grib_file(grib_file_path, region_name=region_name) # Handle different return formats (normal processing vs Arctic pygrib) if result is None or (isinstance(result, tuple) and len(result) == 2 and result[0] is None): logger.warning(f"Failed to process {region_name} region") continue # Check if this is Arctic data with raw coordinates (from pygrib) if (isinstance(result, tuple) and len(result) == 5 and not isinstance(result[0], dict)): # Arctic data: (lats, lons, heights, directions, periods) lats, lons, heights, directions, periods = result logger.info(f"Processing Arctic raw coordinate data: {len(lats)} points") # Convert to sample points format regional_points = [] for i in range(len(lats)): point = { 'lat': float(lats[i]), 'lon': float(lons[i]), 'wave_height': float(heights[i]) if heights[i] is not None else None, 'wave_direction': float(directions[i]) if directions is not None and i < len(directions) else None, 'wave_period': float(periods[i]) if periods is not None and i < len(periods) else None, 'u_velocity': None, 'v_velocity': None } regional_points.append(point) combined_sample_points.extend(regional_points) # Update bounds for Arctic if lats is not None and len(lats) > 0: global_lat_min = min(global_lat_min, float(np.min(lats))) global_lat_max = max(global_lat_max, float(np.max(lats))) global_lon_min = min(global_lon_min, float(np.min(lons))) global_lon_max = max(global_lon_max, float(np.max(lons))) regions_processed.append(region_name) logger.info(f"Successfully processed Arctic {region_name}: {len(regional_points)} points") else: # Normal processed data format regional_data, _ = result if regional_data and 'sample_points' in regional_data: # Add regional sample points to global collection regional_points = regional_data['sample_points'] combined_sample_points.extend(regional_points) # Update global bounds grid_info = regional_data.get('grid_info', {}) if grid_info.get('lat_min') is not None: global_lat_min = min(global_lat_min, grid_info['lat_min']) global_lat_max = max(global_lat_max, grid_info['lat_max']) global_lon_min = min(global_lon_min, grid_info['lon_min']) global_lon_max = max(global_lon_max, grid_info['lon_max']) # Collect wave data for statistics for point in regional_points: if point.get('wave_height') is not None: all_wave_heights.append(point['wave_height']) if point.get('wave_direction') is not None: all_wave_directions.append(point['wave_direction']) if point.get('wave_period') is not None: all_wave_periods.append(point['wave_period']) # Update parameters found regional_params = regional_data.get('parameters_found', {}) if not parameters_found['wave_height']: parameters_found['wave_height'] = regional_params.get('wave_height') if not parameters_found['wave_direction']: parameters_found['wave_direction'] = regional_params.get('wave_direction') if not parameters_found['wave_period']: parameters_found['wave_period'] = regional_params.get('wave_period') if regional_params.get('has_velocity_components'): parameters_found['has_velocity_components'] = True regions_processed.append(region_name) logger.info(f"Successfully processed {region_name}: {len(regional_points)} points") # Clean up temp file if os.path.exists(grib_file_path): os.unlink(grib_file_path) except Exception as e: logger.error(f"Error processing {region_name} region: {e}") # Clean up temp file on error if os.path.exists(grib_file_path): os.unlink(grib_file_path) continue if not combined_sample_points: logger.error("No valid data found in any regional file") return None logger.info(f"Combined data from {len(regions_processed)} regions: {regions_processed}") logger.info(f"Total sample points: {len(combined_sample_points)}") # Create combined global dataset combined_data = { 'timestamp': datetime.utcnow().isoformat(), 'data_source': f'NOAA_MULTI_REGIONAL_GRIB ({",".join(regions_processed)})', 'parameters_found': parameters_found, 'grid_info': { 'lat_min': float(global_lat_min) if global_lat_min != float('inf') else None, 'lat_max': float(global_lat_max) if global_lat_max != float('-inf') else None, 'lon_min': float(global_lon_min) if global_lon_min != float('inf') else None, 'lon_max': float(global_lon_max) if global_lon_max != float('-inf') else None, 'regions_included': regions_processed, 'total_points': len(combined_sample_points) }, 'wave_statistics': { 'max_wave_height': float(max(all_wave_heights)) if all_wave_heights else None, 'min_wave_height': float(min(all_wave_heights)) if all_wave_heights else None, 'mean_wave_height': float(np.mean(all_wave_heights)) if all_wave_heights else None, 'std_wave_height': float(np.std(all_wave_heights)) if all_wave_heights else None }, 'sample_points': combined_sample_points } # Add direction/period statistics if available if all_wave_directions: combined_data['direction_statistics'] = { 'mean_direction': float(np.mean(all_wave_directions)), 'direction_std': float(np.std(all_wave_directions)) } if all_wave_periods: combined_data['period_statistics'] = { 'max_period': float(max(all_wave_periods)), 'min_period': float(min(all_wave_periods)), 'mean_period': float(np.mean(all_wave_periods)) } return combined_data except Exception as e: logger.error(f"Error processing multiple regional files: {e}") # Clean up any remaining temp files for grib_file_path, region_name, _, _ in regional_files: if os.path.exists(grib_file_path): os.unlink(grib_file_path) return None def _extract_sample_points_with_vectors(self, lats, lons, wave_heights, wave_directions=None, wave_periods=None, num_samples=100): """Extract sample points with velocity vector components for visualization""" try: # Create meshgrid for coordinates lon_grid, lat_grid = np.meshgrid(lons, lats) # Flatten arrays flat_lats = lat_grid.flatten() flat_lons = lon_grid.flatten() flat_waves = wave_heights.flatten() flat_dirs = None flat_periods = None if wave_directions is not None: flat_dirs = wave_directions.flatten() if wave_periods is not None: flat_periods = wave_periods.flatten() # Remove NaN values valid_mask = ~np.isnan(flat_waves) if flat_dirs is not None: valid_mask = valid_mask & ~np.isnan(flat_dirs) valid_lats = flat_lats[valid_mask] valid_lons = flat_lons[valid_mask] valid_waves = flat_waves[valid_mask] if flat_dirs is not None: valid_dirs = flat_dirs[valid_mask] else: valid_dirs = None if flat_periods is not None: valid_periods = flat_periods[valid_mask] else: valid_periods = None if len(valid_waves) == 0: return [] # Sample points for visualization (to avoid too many points) sample_size = min(num_samples, len(valid_waves)) sample_indices = np.random.choice(len(valid_waves), size=sample_size, replace=False) sample_points = [] for idx in sample_indices: point = { 'lat': float(valid_lats[idx]), 'lon': float(valid_lons[idx]), 'wave_height': float(valid_waves[idx]) } if valid_dirs is not None: direction = float(valid_dirs[idx]) point['wave_direction'] = direction # Calculate velocity components (u, v) from direction # Wave direction is "coming from" in meteorological convention # Convert to radians and calculate u,v components dir_rad = np.radians(direction) # Use wave height as a proxy for wave energy/velocity magnitude magnitude = point['wave_height'] * 0.1 # Scale factor for visualization # Components: u = eastward, v = northward # Direction 0° = from North, 90° = from East, etc. point['u_component'] = magnitude * np.sin(dir_rad) # eastward component point['v_component'] = -magnitude * np.cos(dir_rad) # northward component (negative because "from") if valid_periods is not None: point['wave_period'] = float(valid_periods[idx]) sample_points.append(point) return sample_points except Exception as e: logger.error(f"Error extracting sample points with vectors: {e}") return [] def _extract_sample_points(self, lats, lons, wave_heights, num_samples=100): """Legacy method - extract sample points for visualization without vectors""" return self._extract_sample_points_with_vectors(lats, lons, wave_heights, None, None, num_samples) def fetch_global_wave_data(self, forecast_hour=0): """Main method to fetch global wave data""" try: # ECMWF open data doesn't include wave parameters, so use NOAA primarily logger.info("Fetching wave data from NOAA WW3 model (ECMWF doesn't provide wave data)...") grib_file = None model_run = None # Try NOAA first for wave data result = self.fetch_noaa_wave_grib(forecast_hour) if result and isinstance(result, list): # Multiple regional files downloaded regional_files = result model_run = regional_files[0][2] if regional_files else None grib_file = None # Will process multiple files elif result and len(result) == 3: # Single file (legacy format) grib_file, model_run, actual_forecast_hour = result regional_files = None else: # NOAA failed, try ECMWF as last resort (though it likely won't have wave data) logger.info("NOAA failed, trying ECMWF as fallback (unlikely to have wave data)...") grib_file = self.fetch_ecmwf_wave_grib(forecast_hour) regional_files = None model_run = None if not grib_file and not regional_files: logger.error("Both ECMWF and NOAA failed - no real wave data available") return None # Process GRIB file(s) if regional_files: # Process multiple regional files and combine processed_data = self.process_multiple_regional_files(regional_files) grib_path = "multiple_regional_files" else: # Process single GRIB file processed_data, grib_path = self.process_grib_file(grib_file) if processed_data: # Add forecast metadata processed_data['forecast_info'] = { 'forecast_hour': forecast_hour, 'model_run': model_run, 'forecast_valid_time': (datetime.utcnow() + timedelta(hours=forecast_hour)).isoformat(), 'is_current': forecast_hour == 0 } # Clean up temporary files if grib_file and os.path.exists(grib_file): os.unlink(grib_file) elif regional_files: # Files already cleaned up in process_multiple_regional_files pass return processed_data except Exception as e: logger.error(f"Error in fetch_global_wave_data: {e}") return None def fetch_multiple_forecasts(self, forecast_hours=[0, 6, 12, 24, 48]): """Fetch multiple forecast time steps""" forecasts = {} for hour in forecast_hours: try: logger.info(f"Fetching forecast for +{hour} hours...") data = self.fetch_global_wave_data(hour) if data: forecasts[f"f{hour:03d}"] = data logger.info(f"Successfully fetched +{hour}h forecast") else: logger.warning(f"Failed to fetch +{hour}h forecast") except Exception as e: logger.error(f"Error fetching +{hour}h forecast: {e}") continue return forecasts def _generate_mock_global_data(self, forecast_hour=0): """Generate mock global wave data for testing""" logger.info(f"Generating mock global wave data for +{forecast_hour}h forecast...") # Create a grid of sample points around the world with mock vectors sample_points = [] for lat in range(-60, 61, 20): # Every 20 degrees latitude for lon in range(-180, 181, 30): # Every 30 degrees longitude # Simulate higher waves in storm-prone areas base_height = np.random.uniform(0.5, 2.0) if abs(lat) > 40: # Higher latitudes tend to have bigger waves base_height += np.random.uniform(0.5, 1.5) # Generate mock wave direction (random but realistic patterns) wave_dir = np.random.uniform(0, 360) # Calculate velocity components magnitude = base_height * 0.1 dir_rad = np.radians(wave_dir) u_comp = magnitude * np.sin(dir_rad) v_comp = -magnitude * np.cos(dir_rad) sample_points.append({ 'lat': float(lat), 'lon': float(lon), 'wave_height': round(float(base_height), 2), 'wave_direction': round(float(wave_dir), 1), 'wave_period': round(np.random.uniform(4.0, 12.0), 1), 'u_component': round(float(u_comp), 3), 'v_component': round(float(v_comp), 3) }) return { 'timestamp': datetime.utcnow().isoformat(), 'data_source': 'MOCK_GLOBAL_DATA', 'grid_info': { 'lat_min': -60.0, 'lat_max': 60.0, 'lon_min': -180.0, 'lon_max': 180.0, 'lat_resolution': 20.0, 'lon_resolution': 30.0, 'grid_shape': [7, 13] # 7 lats x 13 lons }, 'wave_statistics': { 'max_wave_height': max(p['wave_height'] for p in sample_points), 'min_wave_height': min(p['wave_height'] for p in sample_points), 'mean_wave_height': np.mean([p['wave_height'] for p in sample_points]), 'std_wave_height': np.std([p['wave_height'] for p in sample_points]) }, 'forecast_info': { 'forecast_hour': forecast_hour, 'model_run': 'MOCK', 'forecast_valid_time': (datetime.utcnow() + timedelta(hours=forecast_hour)).isoformat(), 'is_current': forecast_hour == 0 }, 'sample_points': sample_points }