diff --git a/climate_tookit/calculate_hazards/ensemble_hazards.py b/climate_tookit/calculate_hazards/ensemble_hazards.py index ef2bed2..b4a6aea 100644 --- a/climate_tookit/calculate_hazards/ensemble_hazards.py +++ b/climate_tookit/calculate_hazards/ensemble_hazards.py @@ -367,8 +367,9 @@ def calculate_ensemble(crop: str, lat: float, lon: float, if fixed_season else None) jobs = [(m, sc) for sc in scenarios for m in models] - # max_workers <= 0 -> auto: one worker per job, capped at 16. - workers = max(1, min(len(jobs), 16) if max_workers <= 0 else min(max_workers, len(jobs))) + # max_workers <= 0 -> auto: one worker per job, capped at 10 to match the + # GEE HTTP connection pool (avoids "connection pool is full" churn). + workers = max(1, min(len(jobs), 10) if max_workers <= 0 else min(max_workers, len(jobs))) print(f"\nNEX-GDDP ensemble: {crop} at ({lat:.4f}, {lon:.4f}) " f"{start_year}-{end_year}") diff --git a/climate_tookit/climate_statistics/ensemble_statistics.py b/climate_tookit/climate_statistics/ensemble_statistics.py index d90637f..9261770 100644 --- a/climate_tookit/climate_statistics/ensemble_statistics.py +++ b/climate_tookit/climate_statistics/ensemble_statistics.py @@ -307,8 +307,9 @@ def _run_model(model: str): raw_by_model: Dict[str, Dict[str, Any]] = {} failed: List[Dict[str, str]] = [] - # max_workers <= 0 -> auto: one worker per model, capped at 16. - workers = max(1, min(len(active), 16) if max_workers <= 0 else min(max_workers, len(active))) + # max_workers <= 0 -> auto: one worker per model, capped at 10 to match the + # GEE HTTP connection pool (avoids "connection pool is full" churn). + workers = max(1, min(len(active), 10) if max_workers <= 0 else min(max_workers, len(active))) def _record(model, r, err, idx): if err is not None or r is None: diff --git a/climate_tookit/climatology/long_term_climatology.py b/climate_tookit/climatology/long_term_climatology.py index 1e8893d..1e5b405 100644 --- a/climate_tookit/climatology/long_term_climatology.py +++ b/climate_tookit/climatology/long_term_climatology.py @@ -945,8 +945,9 @@ def _run_model(model: str) -> Tuple[str, Optional[Dict[str, Any]], Optional[str] per_model_results: Dict[str, Dict[str, Any]] = {} failed: List[Dict[str, str]] = [] - # max_workers <= 0 -> auto: one worker per model, capped at 16. - workers = max(1, min(len(active), 16) if max_workers <= 0 else min(max_workers, len(active))) + # max_workers <= 0 -> auto: one worker per model, capped at 10 to match the + # GEE HTTP connection pool (avoids "connection pool is full" churn). + workers = max(1, min(len(active), 10) if max_workers <= 0 else min(max_workers, len(active))) def _record(model: str, r: Optional[Dict[str, Any]], err: Optional[str], idx: int) -> None: if err is not None or r is None: diff --git a/climate_tookit/compare_periods/ensemble_periods.py b/climate_tookit/compare_periods/ensemble_periods.py index b3404a4..443ba56 100644 --- a/climate_tookit/compare_periods/ensemble_periods.py +++ b/climate_tookit/compare_periods/ensemble_periods.py @@ -661,8 +661,9 @@ def _run_model(model: str): by_model: Dict[str, Dict[str, Any]] = {} failed: List[Dict[str, str]] = [] - # max_workers <= 0 -> auto: one worker per model, capped at 16. - workers = max(1, min(len(active), 16) if max_workers <= 0 else min(max_workers, len(active))) + # max_workers <= 0 -> auto: one worker per model, capped at 10 to match the + # GEE HTTP connection pool (avoids "connection pool is full" churn). + workers = max(1, min(len(active), 10) if max_workers <= 0 else min(max_workers, len(active))) def _record(model, r, err, idx): if err is not None or r is None: diff --git a/climate_tookit/season_analysis/ensemble.py b/climate_tookit/season_analysis/ensemble.py index b3be697..8fa213e 100644 --- a/climate_tookit/season_analysis/ensemble.py +++ b/climate_tookit/season_analysis/ensemble.py @@ -359,7 +359,7 @@ def aggregate_overall(model_results: List[Dict]): return ensemble, model_averages # Top-level orchestrator -def run_ensemble(lat, lon, start_year, end_year, scenarios, models, fixed_arg=None, verbose=True, max_workers=0): +def run_ensemble(lat, lon, start_year, end_year, scenarios, models, fixed_arg=None, verbose=True): results = {} mode = 'fixed' if fixed_arg else 'auto' @@ -370,61 +370,42 @@ def run_ensemble(lat, lon, start_year, end_year, scenarios, models, fixed_arg=No f"{start_year}–{end_year} | mode={mode}") print('=' * 70) - # Each model is an independent, I/O-bound GEE run; fetch them - # concurrently and restore the input order afterwards. - def _run_model(model): + # NOTE: models are processed SERIALLY here on purpose. analyze_one_model + # relies on process-global state — use_nex_gddp() monkeypatches + # seasons.get_climate_data and redirect_stdout() swaps sys.stdout — which + # is not thread-safe: concurrent models would clobber each other's + # model/scenario binding (e.g. a fetch running under the wrong scenario). + # The big speed win for this path comes from the single span-fetch in + # seasons.fetch_and_analyze_years_fixed (one GEE call per model instead + # of one per year), which applies regardless of concurrency. + per_model = [] + for i, model in enumerate(models, 1): + if verbose: + print(f" [{i:02d}/{len(models):02d}] {model:<22}", end=' ', flush=True) try: s_dict, a_dict, skip_info = analyze_one_model( lat, lon, start_year, end_year, model, scenario, fixed_arg) - return model, { + per_model.append({ 'model': model, 'seasons_dict': s_dict, 'annual_dict': a_dict, 'skip_info': skip_info, - }, None + }) + if verbose: + n_seasons = sum(len(v) for v in s_dict.values()) + n_years = sum(1 for v in s_dict.values() if v) + extra = '' + if mode == 'auto' and skip_info['perhumid_years']: + extra = f" [perhumid: {len(skip_info['perhumid_years'])}y]" + print(f"✓ {n_seasons} season(s) over {n_years} year(s){extra}") except Exception as exc: - return model, { + per_model.append({ 'model': model, 'seasons_dict': {}, 'annual_dict': {}, 'skip_info': {'perhumid_years': [], 'no_season_years': [], 'analyzed_years': []}, 'error': f"{type(exc).__name__}: {exc}", - }, f"{type(exc).__name__}: {exc}" - - def _log(idx, entry, err): - if not verbose: - return - if err: - print(f" [{idx:02d}/{len(models):02d}] {model_name(entry):<22} ✗ {err}", flush=True) - else: - s_dict = entry['seasons_dict'] - n_seasons = sum(len(v) for v in s_dict.values()) - n_years = sum(1 for v in s_dict.values() if v) - extra = '' - if mode == 'auto' and entry['skip_info'].get('perhumid_years'): - extra = f" [perhumid: {len(entry['skip_info']['perhumid_years'])}y]" - print(f" [{idx:02d}/{len(models):02d}] {model_name(entry):<22} " - f"✓ {n_seasons} season(s) over {n_years} year(s){extra}", flush=True) - - def model_name(entry): - return entry.get('model', '?') - - by_model = {} - # max_workers <= 0 -> auto: one worker per model, capped at 16. - workers = max(1, min(len(models), 16) if max_workers <= 0 else min(max_workers, len(models))) - if workers == 1: - for i, model in enumerate(models, 1): - _, entry, err = _run_model(model) - by_model[model] = entry - _log(i, entry, err) - else: - from concurrent.futures import ThreadPoolExecutor, as_completed - order = {m: i for i, m in enumerate(models, 1)} - with ThreadPoolExecutor(max_workers=workers) as ex: - futs = {ex.submit(_run_model, m): m for m in models} - for fut in as_completed(futs): - m, entry, err = fut.result() - by_model[m] = entry - _log(order[m], entry, err) - per_model = [by_model[m] for m in models if m in by_model] + }) + if verbose: + print(f"✗ {type(exc).__name__}: {exc}") ok = sum(1 for r in per_model if not r.get('error')) diagnostics = _aggregate_skip_info(per_model) @@ -667,8 +648,6 @@ def main(): p.add_argument('--list-models', action='store_true', help='Print models and exit') p.add_argument('--output', help='Save JSON result here') p.add_argument('--quiet', action='store_true') - p.add_argument('--workers', type=int, default=0, - help='Parallel GEE fetch workers across models (default: auto = one per model, capped at 16; use 1 to disable)') args = p.parse_args() if args.source_key: @@ -705,7 +684,6 @@ def main(): scenarios = scenarios, models = models, fixed_arg = args.fixed_season, verbose = not args.quiet, - max_workers = args.workers, ) if not args.quiet: diff --git a/climate_tookit/season_analysis/seasons.py b/climate_tookit/season_analysis/seasons.py index ef4ec16..0154bfc 100644 --- a/climate_tookit/season_analysis/seasons.py +++ b/climate_tookit/season_analysis/seasons.py @@ -615,35 +615,48 @@ def fetch_and_analyze_years_fixed( annual_dict : Dict[int, Dict] = {} force = None if source == "auto" else source + # Resolve all season windows up front so we know the full span to fetch. + resolved_by_year: Dict[int, List] = {} + max_cess = date(end_year, 12, 31) for year in range(start_year, end_year + 1): - print(f"\nFixed-season year {year} | source={source}") - - # Resolve all season dates for this year resolved = [] for sd in fixed_seasons: (o_m, o_d) = sd["onset_md"] (c_m, c_d) = sd["cessation_md"] cess_year = year + 1 if (c_m, c_d) < (o_m, o_d) else year try: - resolved.append((date(year, o_m, o_d), date(cess_year, c_m, c_d))) + cd = date(cess_year, c_m, c_d) + resolved.append((date(year, o_m, o_d), cd)) + if cd > max_cess: + max_cess = cd except ValueError as exc: print(f" [WARNING] Invalid date: {exc}") + resolved_by_year[year] = resolved + + # Fetch the entire period (plus any year-crossing tail) in ONE call, then + # slice per year in memory. The per-year helpers filter internally by year + # / window, so passing the full master frame yields identical results + # while replacing N_years fetches with one. + overall_start = f"{start_year}-01-01" + overall_end = max_cess.strftime("%Y-%m-%d") + print(f" Fetching {overall_start} to {overall_end} (full span, one call) ...") + try: + master = get_climate_data(lat, lon, overall_start, overall_end, force_source=force) + master = add_et0(master, lat) + print(f" Retrieved {len(master)} days") + except Exception as exc: + print(f" ✗ Data fetch failed: {exc} — stats will be n/a") + master = None + + for year in range(start_year, end_year + 1): + print(f"\nFixed-season year {year} | source={source}") + + resolved = resolved_by_year.get(year, []) if not resolved: annual_dict[year] = {} continue - # Fetch data: full calendar year + any year-crossing tail, in one call - fetch_start = f"{year}-01-01" - fetch_end = max(date(year, 12, 31), max(c for _, c in resolved)).strftime("%Y-%m-%d") - print(f" Fetching {fetch_start} to {fetch_end} ...") - - try: - df = get_climate_data(lat, lon, fetch_start, fetch_end, force_source=force) - df = add_et0(df, lat) - print(f" Retrieved {len(df)} days") - except Exception as exc: - print(f" ✗ Data fetch failed: {exc} — stats will be n/a") - df = None + df = master # helpers slice to the year / window internally # Annual stats (calendar year only) if df is not None and not df.empty: