from __future__ import annotations import calendar,gzip,io,os,random,re,sys,tarfile,tempfile,threading,time,requests,json,rasterio,rioxarray,math,uuid,hashlib, shutil, logging from collections import deque,OrderedDict,defaultdict from datetime import datetime,timedelta,timezone from functools import lru_cache from pathlib import Path from typing import Optional,Callable,Any import numpy as np import xarray as xr from fastapi import FastAPI,HTTPException,Query,Response,Body from fastapi.staticfiles import StaticFiles from rasterio.crs import CRS from rasterio.enums import Resampling from rasterio.transform import Affine,from_bounds from rasterio.warp import reproject from rio_cogeo.cogeo import cog_translate from rio_cogeo.profiles import cog_profiles from rio_tiler.errors import TileOutsideBounds from rio_tiler.io import COGReader from rio_tiler.utils import render from starlette.responses import PlainTextResponse,Response from huggingface_hub import HfApi,snapshot_download,hf_hub_download from huggingface_hub._commit_api import CommitOperationAdd,CommitOperationDelete from zoneinfo import ZoneInfo import sqlite3 from rio_tiler.colormap import cmap from fastapi.responses import Response as FastResponse from pydantic import BaseModel from starlette.middleware.gzip import GZipMiddleware import huggingface_hub.utils._http as _hf_http from requests.adapters import HTTPAdapter import concurrent.futures import requests.adapters as _req_adapters os.environ["TZ"] = "America/Chicago" time.tzset() logging.Formatter.converter = time.localtime os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "600" os.environ["HF_HUB_ETAG_TIMEOUT"] = "600" _orig_adapter_send = _req_adapters.HTTPAdapter.send def _timeout_600_send(self, request, *args, **kwargs): if "timeout" not in kwargs or kwargs["timeout"] is None or (isinstance(kwargs.get("timeout"), (int, float)) and kwargs["timeout"] <= 30): kwargs["timeout"] = 600 return _orig_adapter_send(self, request, *args, **kwargs) _req_adapters.HTTPAdapter.send = _timeout_600_send LOCAL_TZ=ZoneInfo(os.environ.get("LOCAL_TZ","America/Chicago")) LOWRES_Z=int(os.environ.get("LOWRES_Z","6")) LOWRES_MAX_REQUEST_Z=int(os.environ.get("LOWRES_MAX_REQUEST_Z","14")) LOWRES_BOX_ZMIN=int(os.environ.get("LOWRES_BOX_ZMIN","4")) LOWRES_BOX_ZMAX=int(os.environ.get("LOWRES_BOX_ZMAX","6")) HF_MAX_OPS_PER_COMMIT=int(os.environ.get("HF_MAX_OPS_PER_COMMIT","300")) HF_PUSH_SLEEP_SEC=float(os.environ.get("HF_PUSH_SLEEP_SEC","0.5")) ROLL_WINDOW_DAYS=int(os.environ.get("ROLL_WINDOW_DAYS","6")) SCHEDULE_HOUR_LOCAL=int(os.environ.get("SCHEDULE_HOUR_LOCAL","10")) #hour SCHEDULE_MIN_LOCAL=int(os.environ.get("SCHEDULE_MIN_LOCAL","00")) #minute WARM_ZMIN=int(os.environ.get("WARM_ZMIN","6")) WARM_ZMAX=int(os.environ.get("WARM_ZMAX","11")) WARM_RADIUS=int(os.environ.get("WARM_RADIUS","1")) WARM_CAP_PER_ZOOM=int(os.environ.get("WARM_CAP_PER_ZOOM","7000")) WARM_MAX_TILES_TOTAL=int(os.environ.get("WARM_MAX_TILES_TOTAL","50000")) WARM_DOM=os.environ.get("WARM_DOM","zz").lower() WARM_MILES=float(os.environ.get("WARM_MILES","8")) BOOT_ROLL_DEFAULT=os.environ.get("RUN_BOOT_ROLL","0") #set to 1 to run the download sequence on boot, 0 to bypass WARM_HOURS_LIST=[24,72] CN_MBTILES_LOCAL=Path("cache")/"tracks"/"cn_tracks.mbtiles" CN_MBTILES_REPO_PATH="tracks/cn_tracks.mbtiles" CN_BOUNDS_REPO_PATH="tracks/cn_tracks.bounds.txt" _VIEWGRID_LOCK=threading.Lock() _VIEWGRID_CACHE:dict[tuple,tuple[float,Any]]={} _VIEWGRID_TTL_S=45.0 _MELTCOGS_LOCK=threading.Lock() _MELTCOGS_CACHE:dict[tuple,tuple[float,Any]]={} _MELTCOGS_TTL_S=300.0 KEEP_TILE_DAYS=int(os.environ.get("KEEP_TILE_DAYS","6")) KEEP_FORECAST_COG_DAYS=int(os.environ.get("KEEP_FORECAST_COG_DAYS","1")) _COGREADER_CACHE_MAX=int(os.environ.get("COGREADER_CACHE_MAX","12")) _COGREADER_CACHE:"OrderedDict[str,COGReader]"=OrderedDict() _COGREADER_CACHE_LK=threading.Lock() ROLL_SINGLETON_KEY="roll_job" _ACTIVE_CLIENT: dict[str, dict[str, Any]] = {} _ACTIVE_CLIENT_LK = threading.Lock() _ACTIVE_CLIENT_TTL_S = 180.0 CACHE_CONTROL_BY_DATE=os.environ.get("CACHE_CONTROL_BY_DATE","public, max-age=31536000, immutable") CACHE_CONTROL_LATEST=os.environ.get("CACHE_CONTROL_LATEST","public, max-age=300, stale-while-revalidate=600") PNG_MAGIC=b"\x89PNG\r\n\x1a\n" ZIP_MAGIC=b"PK\x03\x04" RAW_TILE_EXT=".npz" _COG_LRU_MAX=int(os.environ.get("COGREADER_LRU_MAX","36")) _COG_LRU:"OrderedDict[str,COGReader]"=OrderedDict() _COG_LRU_LK=threading.Lock() _RATE_LIMIT_WINDOW=10.0 _RATE_LIMIT_MAX_REQUESTS=50 _PUSH_RESUME_TIMER = None _PUSH_TIMER_LK = threading.Lock() def _pause_push_for_user(): """Pause background HF push for 30s to prioritize user requests.""" global _PUSH_RESUME_TIMER _PUSH_PAUSED.set() with _PUSH_TIMER_LK: if _PUSH_RESUME_TIMER: _PUSH_RESUME_TIMER.cancel() _PUSH_RESUME_TIMER = threading.Timer(30.0, lambda: _PUSH_PAUSED.clear()) _PUSH_RESUME_TIMER.daemon = True _PUSH_RESUME_TIMER.start() def _pick_cache_root()->Path: env=(os.environ.get("SNODAS_CACHE")or"").strip() candidates=([env]if env else[])+["/data/snodas-cache","/home/user/snodas-cache","/tmp/snodas-cache","./cache"] for c in candidates: p=Path(c).expanduser() try: p.mkdir(parents=True,exist_ok=True) t=p/".wtest";t.write_text("ok");t.unlink(missing_ok=True) return p.resolve() except:pass return Path(tempfile.mkdtemp(prefix="snodas-cache-")).resolve() CACHE=_pick_cache_root() _HF_NEGATIVE_CACHE: dict[str, float] = {} _HF_NEGATIVE_CACHE_LK = threading.Lock() _HF_NEGATIVE_TTL = 300.0 # 5 minutes _HF_NEGATIVE_MAX = 10000 # cap size def _hf_negative_check(rel: str) -> bool: with _HF_NEGATIVE_CACHE_LK: ts = _HF_NEGATIVE_CACHE.get(rel) if ts is None: return False if (time.time() - ts) > _HF_NEGATIVE_TTL: del _HF_NEGATIVE_CACHE[rel] return False return True def _hf_negative_mark(rel: str) -> None: with _HF_NEGATIVE_CACHE_LK: _HF_NEGATIVE_CACHE[rel] = time.time() # Prevent unbounded growth if len(_HF_NEGATIVE_CACHE) > _HF_NEGATIVE_MAX: # Remove oldest 20% sorted_keys = sorted(_HF_NEGATIVE_CACHE, key=_HF_NEGATIVE_CACHE.get) for k in sorted_keys[:_HF_NEGATIVE_MAX // 5]: del _HF_NEGATIVE_CACHE[k] _NOAA_FILE_SIGS: dict[str, dict] = {} _NOAA_FILE_SIGS_LK = threading.Lock() _NOAA_SIGS_PATH = CACHE / "_noaa_file_sigs.json" def _load_noaa_sigs() -> None: global _NOAA_FILE_SIGS try: if _NOAA_SIGS_PATH.exists(): with _NOAA_FILE_SIGS_LK: _NOAA_FILE_SIGS = json.loads(_NOAA_SIGS_PATH.read_text("utf-8")) except: pass def _save_noaa_sigs() -> None: try: with _NOAA_FILE_SIGS_LK: _NOAA_SIGS_PATH.write_text(json.dumps(_NOAA_FILE_SIGS, sort_keys=True), "utf-8") except: pass def _get_noaa_file_sig(url: str) -> dict | None: """HEAD request to get Last-Modified and Content-Length from NOAA.""" try: r = requests.head(url, timeout=15, allow_redirects=True, headers={"User-Agent": "snodas/1.3"}) if r.status_code != 200: return None return { "last_modified": r.headers.get("Last-Modified", ""), "content_length": r.headers.get("Content-Length", ""), "etag": r.headers.get("ETag", ""), } except: return None def _record_baked_sig(valid:str,dom:str,url:str,source_type:str|None=None)->None: """Record NOAA file sig. Key is now date|dom (not valid|dom) for consistency.""" sig=_get_noaa_file_sig(url) if not sig:return date_ymd=valid[:8]if len(valid)>=8 else valid key=f"{date_ymd}|{dom}" sig["url"]=url sig["baked_at"]=time.time() sig["valid"]=valid if source_type is None: if"ttnats"in url.lower(): source_type="ttna_11z"if url.lower().endswith("11dp000.nc")or"11dp000"in url.lower()else"ttna_05z" else: source_type="forecast" sig["source_type"]=source_type with _NOAA_FILE_SIGS_LK: _NOAA_FILE_SIGS[key]=sig _save_noaa_sigs() _log("INFO",f"[staleness] recorded sig for {key}: type={source_type} LM={sig['last_modified']} CL={sig['content_length']}") def _record_baked_sig_v2(date_ymd:str,dom:str,url:str,source_type:str)->None: _record_baked_sig(date_ymd,dom,url,source_type=source_type) # 1. NEW: Find 11Z TTNA melt files def _find_melt_11044_ttna_11z(valid_date_ymd:str,dom_prefer:str="zz")->list[str]: """Find TTNA 11Z melt files for a given date (YYYYMMDD).""" dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid_date_ymd}11";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low:continue if"__t0024ttnats"not in low:continue if want_ts not in low:continue if not low.endswith(".nc"):continue out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) # 2. NEW: Find 05Z TTNA melt files def _find_melt_11044_ttna_05z(valid_date_ymd:str,dom_prefer:str="zz")->list[str]: """Find TTNA 05Z melt files for a given date (YYYYMMDD).""" dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid_date_ymd}05";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low:continue if"__t0024ttnats"not in low:continue if want_ts not in low:continue if not low.endswith(".nc"):continue out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) def _pick_best_run_init_for_valid_11z(valid:str,dom_prefer:str="zz")->str|None: """Find the latest TT run init that has a TS...11 file.""" valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";rows=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low or"__t0024tt"not in low or want_ts not in low or not low.endswith(".nc"):continue if"ttnats"in low:continue # Skip TTNA files, only want forecast TT...06 tt=_tt_from_name(base) if not tt:continue rows.append((tt,_score_dom(_as_url(base),dom_prefer))) if not rows:return None rows.sort(key=lambda t:(t[0],-t[1]));return rows[-1][0] def _find_melt_11044_t0024_11z(run_init:str,valid:str,dom_prefer:str="zz")->list[str]: """Find forecast melt files matching a specific TT run init and 11Z valid time.""" run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low or"__t0024tt"not in low or want_tt not in low or want_ts not in low or not low.endswith(".nc"):continue if"ttnats"in low:continue # Skip TTNA out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) def _find_best_melt_for_date(date_ymd:str,dom_prefer:str="zz")->dict: """Returns {ok, melt_urls, valid, run_init, source_type} preferring 11Z > 05Z.""" dom_prefer=(dom_prefer or"zz").lower() # 1. TTNA 11Z urls_11z=_find_melt_11044_ttna_11z(date_ymd,dom_prefer=dom_prefer) if urls_11z: return{"ok":True,"melt_urls":urls_11z,"valid":f"{date_ymd}11","run_init":"TTNA","source_type":"ttna_11z"} # 2. Forecast 11Z (TT...06TS...11) valid_11=f"{date_ymd}11" run_init_11=_pick_best_run_init_for_valid_11z(valid_11,dom_prefer=dom_prefer) if run_init_11: melt_urls_11=_find_melt_11044_t0024_11z(run_init_11,valid_11,dom_prefer=dom_prefer) if melt_urls_11: return{"ok":True,"melt_urls":melt_urls_11,"valid":valid_11,"run_init":run_init_11,"source_type":"forecast_11z"} # 3. TTNA 05Z urls_05z=_find_melt_11044_ttna_05z(date_ymd,dom_prefer=dom_prefer) if urls_05z: return{"ok":True,"melt_urls":urls_05z,"valid":f"{date_ymd}05","run_init":"TTNA","source_type":"ttna_05z"} # 4. Forecast 05Z (TT...06TS...05) valid_05=_norm_ts(f"{date_ymd}05") run_init=_pick_best_run_init_for_valid_t0024(valid_05,dom_prefer=dom_prefer) if not run_init: return{"ok":False,"error":"no melt files found (11z, 05z, or forecast)","date":date_ymd} melt_urls=_find_melt_11044_t0024(run_init,valid_05,dom_prefer=dom_prefer) if not melt_urls: return{"ok":False,"error":"forecast run_init found but no melt files","run_init":run_init,"date":date_ymd} return{"ok":True,"melt_urls":melt_urls,"valid":valid_05,"run_init":run_init,"source_type":"forecast_05z"} # 4. NEW: Find snowpack files matching the valid time we chose # def _find_snowpack_for_best_melt(best_melt:dict,dom_prefer:str="zz")->list[str]: # """Find snowpack files that match the valid time from _find_best_melt_for_date.""" # dom_prefer=(dom_prefer or"zz").lower() # valid=best_melt.get("valid","") # run_init=best_melt.get("run_init","") # source_type=best_melt.get("source_type","") # if source_type in("ttna_11z","ttna_05z"): # # For TTNA, search by valid time # html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[] # for n in names: # base=n.split("?",1)[0].strip();low=base.lower() # if"ssmv11034"not in low and"ssmv11036"not in low:continue # if"hp001"not in low:continue # if want_ts not in low:continue # if not(low.endswith(".grib2")or low.endswith(".nc")):continue # out.append(_as_url(base)) # if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) # # Fallback: try 05Z snowpack if we're on 11Z melt # if source_type=="ttna_11z": # want_ts_05=f"ts{valid[:8]}05" # for n in names: # base=n.split("?",1)[0].strip();low=base.lower() # if"ssmv11034"not in low and"ssmv11036"not in low:continue # if"hp001"not in low:continue # if want_ts_05 not in low:continue # if not(low.endswith(".grib2")or low.endswith(".nc")):continue # out.append(_as_url(base)) # if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) # return[] # # Forecast source — use existing logic # if run_init and run_init!="TTNA": # return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer) # return[] def _find_snowpack_for_best_melt(best_melt:dict,dom_prefer:str="zz")->list[str]: dom_prefer=(dom_prefer or"zz").lower() valid=best_melt.get("valid","") run_init=best_melt.get("run_init","") source_type=best_melt.get("source_type","") html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}" if source_type in("ttna_11z","ttna_05z"): # Prefer TTNA snowpack matching the valid time ttna_out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if("ssmv11034"not in low and"ssmv11036"not in low):continue if"hp001"not in low:continue if want_ts not in low:continue if"ttnats"not in low:continue # Only TTNA snowpack if not(low.endswith(".grib2")or low.endswith(".nc")):continue ttna_out.append(_as_url(base)) if ttna_out: return sorted(set(ttna_out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) # Fallback: TTNA snowpack at 05Z if we're on 11Z melt if source_type=="ttna_11z": want_ts_05=f"ts{valid[:8]}05" for n in names: base=n.split("?",1)[0].strip();low=base.lower() if("ssmv11034"not in low and"ssmv11036"not in low):continue if"hp001"not in low:continue if want_ts_05 not in low:continue if"ttnats"not in low:continue if not(low.endswith(".grib2")or low.endswith(".nc")):continue ttna_out.append(_as_url(base)) if ttna_out: return sorted(set(ttna_out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) if source_type in("forecast_11z","forecast_05z","forecast"): # Use snowpack from the same run init only if run_init and run_init!="TTNA": return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer) # Last resort: any snowpack matching valid time but limit to preferred domain out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if("ssmv11034"not in low and"ssmv11036"not in low):continue if"hp001"not in low:continue if want_ts not in low:continue if not(low.endswith(".grib2")or low.endswith(".nc")):continue if _dom_from_url(_as_url(base))==dom_prefer: out.append(_as_url(base)) if out:return sorted(set(out),key=lambda u:_dom_pri(_dom_from_url(u))) return[] def _check_and_rebuild_stale_dates()->None: dom=(WARM_DOM or"zz").lower() window=_rolling_dates_local_window(lookback=2,lookahead=3) with _NOAA_FILE_SIGS_LK: known_sigs={k:dict(v)for k,v in _NOAA_FILE_SIGS.items()} stale_dates=[] for ymd in window: key=f"{ymd}|{dom}" old_sig=known_sigs.get(key) urls_11z=_find_melt_11044_ttna_11z(ymd,dom_prefer=dom) if urls_11z: url_11z=urls_11z[0] if old_sig and old_sig.get("source_type")=="ttna_11z": new_sig=_get_noaa_file_sig(url_11z) if new_sig: old_lm=old_sig.get("last_modified","");new_lm=new_sig.get("last_modified","") old_cl=old_sig.get("content_length","");new_cl=new_sig.get("content_length","") if(old_lm and new_lm and old_lm!=new_lm)or(old_cl and new_cl and old_cl!=new_cl): _log("INFO",f"[staleness] 11Z updated for {ymd}: LM {old_lm} -> {new_lm}") stale_dates.append((ymd,url_11z)) continue _log("INFO",f"[staleness] 11Z now available for {ymd} — was {old_sig.get('source_type','unknown') if old_sig else 'not baked'}, rebuilding") stale_dates.append((ymd,url_11z)) continue if not old_sig: best=_find_best_melt_for_date(ymd,dom_prefer=dom) if best.get("ok")and best.get("melt_urls"): _record_baked_sig_v2(ymd,dom,best["melt_urls"][0],best.get("source_type","unknown")) continue if old_sig and old_sig.get("source_type")in("forecast","forecast_05z","ttna_05z"): # Check if forecast 11Z is now available valid_11=f"{ymd}11" ri_11=_pick_best_run_init_for_valid_11z(valid_11,dom_prefer=dom) if ri_11: urls_11=_find_melt_11044_t0024_11z(ri_11,valid_11,dom_prefer=dom) if urls_11: _log("INFO",f"[staleness] forecast 11Z now available for {ymd}") stale_dates.append((ymd,urls_11[0])) continue old_url=old_sig.get("url","") best=_find_best_melt_for_date(ymd,dom_prefer=dom) if best.get("ok")and best.get("melt_urls"): new_url=best["melt_urls"][0] if new_url!=old_url: _log("INFO",f"[staleness] newer forecast run available for {ymd}: {old_url.split('/')[-1]} -> {new_url.split('/')[-1]}") stale_dates.append((ymd,new_url)) continue new_sig=_get_noaa_file_sig(new_url) if new_sig: old_lm=old_sig.get("last_modified","");new_lm=new_sig.get("last_modified","") old_cl=old_sig.get("content_length","");new_cl=new_sig.get("content_length","") if(old_lm and new_lm and old_lm!=new_lm)or(old_cl and new_cl and old_cl!=new_cl): _log("INFO",f"[staleness] forecast file modified for {ymd}: LM {old_lm} -> {new_lm}") stale_dates.append((ymd,new_url)) continue if not stale_dates:return # Priority: day 2 first, then day 1, then day 3, then others today=datetime.now(LOCAL_TZ).date() def _rebuild_priority(item): ymd=item[0] try: d=datetime.strptime(ymd,"%Y%m%d").date() offset=(d-today).days if offset==2:return 0 # day 2 first if offset==1:return 1 # day 1 second if offset==3:return 2 # day 3 third return 3+abs(offset) # everything else after except:return 99 stale_dates.sort(key=_rebuild_priority) _log("INFO",f"[staleness] rebuild order: {[d[0] for d in stale_dates]}") for ymd,url in stale_dates: _log("INFO",f"[staleness] hot-swapping {ymd}") try:_hot_swap_rebuild_date(ymd,dom) except Exception as e:_log("WARN",f"[staleness] rebuild failed for {ymd}: {e!r}") def _staleness_monitor_loop()->None: _load_noaa_sigs() _log("INFO","[staleness] monitor started") time.sleep(300) while True: try:_check_and_rebuild_stale_dates() except Exception as e:_log("WARN",f"[staleness] monitor error: {e!r}") time.sleep(20*60) def _nuke_and_rebuild_date(date:str,dom:str)->None: dom=(dom or"zz").lower();valid_05=f"{date}05";valid_11=f"{date}11" # 1. Delete local COGs for both 05Z and 11Z for valid in[valid_05,valid_11]: for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]: for p in _forecast_dir().glob(pattern): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True) except:pass # 2. Delete local tile cache for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: d=CACHE/kind/dom/str(h)/date if d.exists():shutil.rmtree(d,ignore_errors=True) # 3. Delete tar markers marker_dir=CACHE/"_tar_markers" if marker_dir.exists(): for mk in marker_dir.glob(f"*_{date}.tar.gz.done"): try:mk.unlink(missing_ok=True) except:pass # 4. Delete tar archives from HF tok,repo=_hf_cfg();api=_hf_api() if tok and repo and api is not None: for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: archive_path=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz" try: api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=archive_path)],commit_message=f"nuke: delete {archive_path}") _log("INFO",f"[nuke] deleted HF archive: {archive_path}") except Exception as e: err=repr(e) if"404"not in err and"EntryNotFound"not in err and"does not exist"not in err: _log("WARN",f"[nuke] HF delete failed: {archive_path}: {e!r}") # 5. Also delete 72h COGs and tiles for next 2 dates for offset in[1,2]: try: future_date=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") for fv in[f"{future_date}05",f"{future_date}11"]: for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True);_log("INFO",f"[nuke] deleted 72h COG: {p.name}") except:pass for kind in["tilecache","rawtilecache"]: d=CACHE/kind/dom/"72"/future_date if d.exists():shutil.rmtree(d,ignore_errors=True);_log("INFO",f"[nuke] deleted {kind}/72h tiles for {future_date}") if tok and repo and api is not None: for kind in["tilecache","rawtilecache"]: ap=f"archives/{kind}_{dom}_h72_{future_date}.tar.gz" try: api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"nuke: delete stale 72h {ap}") _log("INFO",f"[nuke] deleted HF 72h archive: {ap}") except Exception as e: err=repr(e) if"404"not in err and"EntryNotFound"not in err and"does not exist"not in err: _log("WARN",f"[nuke] HF 72h delete failed: {ap}: {e!r}") except:pass # 6. Clear negative cache with _HF_NEGATIVE_CACHE_LK: to_del=[k for k in _HF_NEGATIVE_CACHE if date in k] for k in to_del:del _HF_NEGATIVE_CACHE[k] # 7. Rebuild — _pick_forecast_for_valid now auto-selects 11Z > 05Z > forecast _log("INFO",f"[nuke] rebuilding COGs for {date} (will auto-select best available)") best=_find_best_melt_for_date(date,dom_prefer=dom) _log("INFO",f"[nuke] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}") # Use the valid time from best melt selection actual_valid=best.get("valid",f"{date}05") r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom) if not r.get("ok"):_log("WARN",f"[nuke] COG rebuild failed for {date}: {r}");return # 8. Re-bake tiles — sequential for CPU efficiency _log("INFO",f"[nuke] re-baking tiles for {date}") for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: try: r=fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) _log("INFO",f"[nuke] {date}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[nuke] {date}: {label} failed: {e!r}") for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) except:pass # 9. Rebake 72h for next 2 dates — sequential, with tile invalidation for offset in[1,2]: try: future_date=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") future_best=_find_best_melt_for_date(future_date,dom_prefer=dom) future_valid=future_best.get("valid",f"{future_date}05")if future_best.get("ok")else f"{future_date}05" try:_forecast_melt72h_end_cog(future_valid,dom_prefer=dom) except:pass _log("INFO",f"[nuke] rebaking 72h tiles for {future_date}") # Delete stale 72h tiles (nuke already deleted them in step 5, # but belt-and-suspenders in case the path changed) for kind in["tilecache","rawtilecache"]: d=CACHE/kind/dom/"72"/future_date if d.exists():shutil.rmtree(d,ignore_errors=True) for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: try:fn(date_yyyymmdd=future_date,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) except:pass for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try:fn(date_yyyymmdd=future_date,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) except:pass except Exception as e:_log("WARN",f"[nuke] 72h rebake for +{offset} failed: {e!r}") # 10. Push corrected tiles push_dates=[date] for offset in[1,2]: try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")) except:pass _log("INFO",f"[nuke] pushing corrected tiles for {push_dates}") _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST) # 11. Record signature if best.get("ok")and best.get("melt_urls"): _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type")) _log("INFO",f"[nuke] {date}: rebuild complete (source={best.get('source_type','?')})") def _client_mark_active(client_id: str, active_date: str, gen: int | None = None) -> None: with _ACTIVE_CLIENT_LK: _ACTIVE_CLIENT[client_id] = { "date": active_date, "gen": int(gen) if gen is not None else 0, "ts": time.time(), } def _client_is_active_date(client_id: str, date_yyyymmdd: str) -> bool: now = time.time() with _ACTIVE_CLIENT_LK: st = _ACTIVE_CLIENT.get(client_id) if not st: return True if (now - float(st.get("ts", 0.0))) > _ACTIVE_CLIENT_TTL_S: return True return str(st.get("date", "")) == str(date_yyyymmdd) def _client_is_active_gen(client_id: str, gen: int | None) -> bool: if not client_id or gen is None: return True # no generation tracking → allow now = time.time() with _ACTIVE_CLIENT_LK: st = _ACTIVE_CLIENT.get(client_id) if not st: return True if (now - float(st.get("ts", 0.0))) > _ACTIVE_CLIENT_TTL_S: return True server_gen = int(st.get("gen", 0)) return int(gen) >= server_gen def _make_cancel_check(client_id: str | None, gen: int | None): if not client_id or gen is None: return None def _is_cancelled() -> bool: return not _client_is_active_gen(client_id, gen) return _is_cancelled class CogCache: def __init__(self): self._datasets: dict[str, COGReader] = {} self._lock = threading.Lock() def get(self, cog: Path) -> COGReader: key = cog.as_posix() with self._lock: r = self._datasets.get(key) if r is None: r = COGReader(key) self._datasets[key] = r return r def evict(self, path: Path) -> None: with self._lock: ds = self._datasets.pop(str(path), None) if ds: try: ds.close() except: pass def close_all(self): with self._lock: for r in self._datasets.values(): try: r.close() except: pass self._datasets.clear() def __enter__(self): return self def __exit__(self, *args): self.close_all() COG_CACHE=CogCache() def _cogreader_get(path:Path)->COGReader: key=path.as_posix() with _COGREADER_CACHE_LK: r=_COGREADER_CACHE.get(key) if r is not None:_COGREADER_CACHE.move_to_end(key,last=True);return r r=COGReader(key);_COGREADER_CACHE[key]=r;_COGREADER_CACHE.move_to_end(key,last=True) while len(_COGREADER_CACHE)>_COGREADER_CACHE_MAX: _,old_r=_COGREADER_CACHE.popitem(last=False) try:old_r.close() except:pass return r class _RateLimitBuckets: def __init__(self): self._buckets:dict[str,list[float]]=defaultdict(list) self._locks:dict[str,threading.Lock]=defaultdict(threading.Lock) self._cleanup_lock=threading.Lock();self._last_cleanup=0.0 def check(self,client_id:str)->bool: if not client_id:return True now=time.time();cutoff=now-_RATE_LIMIT_WINDOW with self._locks[client_id]: bucket=self._buckets[client_id];bucket[:]=[t for t in bucket if t>cutoff] if len(bucket)>=_RATE_LIMIT_MAX_REQUESTS:return False bucket.append(now) if now-self._last_cleanup>60.0:self._cleanup_old_clients(cutoff) return True def _cleanup_old_clients(self,cutoff:float)->None: if not self._cleanup_lock.acquire(blocking=False):return try: self._last_cleanup=time.time() empty=[cid for cid,bucket in self._buckets.items()if not bucket or all(t<=cutoff for t in bucket)] for cid in empty:self._buckets.pop(cid,None);self._locks.pop(cid,None) finally:self._cleanup_lock.release() _rate_limiter=_RateLimitBuckets() def _check_rate_limit(client_id:str)->bool:return _rate_limiter.check(client_id) def _is_valid_png_file(p:Path)->bool: try: if not p.exists():return False if p.stat().st_size<64:return False with p.open("rb")as f:head=f.read(8) return head==PNG_MAGIC except:return False def _ymd_from_cache_relpath(rel:str)->str|None: try: p=Path(rel).as_posix();parts=p.split("/") if len(parts)>=4 and parts[0]in("tilecache","rawtilecache"): ymd=parts[3];return ymd if re.fullmatch(r"[0-9]{8}",ymd)else None if parts and parts[0]=="forecast": m=re.search(r"([0-9]{8})",p);return m.group(1)if m and re.fullmatch(r"[0-9]{8}",m.group(1))else None base=parts[-1]if parts else p m=re.match(r"bydate_[a-z0-9]+_h\d+_([0-9]{8})_z\d+_\d+_\d+\.(png|npz)$",base) return m.group(1)if m else None except:return None def _validate_npz_bytes(data:bytes)->bool: if not data or len(data)<16:return False if data[0:4]!=b'PK\x03\x04':return False return True def _mem_cache_get(cache:dict,key:Any,ttl_s:int)->Any|None: try: v=cache.get(key) if not v:return None ts,payload=v if(time.time()-ts)>ttl_s: try:del cache[key] except:pass return None return payload except:return None def _mem_cache_put(cache:dict,key:Any,payload:Any)->None: try:cache[key]=(time.time(),payload) except:pass def _safe_unlink(p:Path)->None: try:p.unlink(missing_ok=True) except:pass def _tile_cache_dir()->Path: d=CACHE/"tilecache";d.mkdir(parents=True,exist_ok=True);return d def _tile_cache_path(dom:str,hours:int,date_yyyymmdd:str,z:int,x:int,y:int)->Path: dom=(dom or"zz").lower() return CACHE/"tilecache"/dom/str(int(hours))/date_yyyymmdd/str(int(z))/str(int(x))/f"{int(y)}.png" def _raw_tile_cache_path(dom:str,hours:int,date_yyyymmdd:str,z:int,x:int,y:int)->Path: dom=(dom or"zz").lower() return CACHE/"rawtilecache"/dom/str(int(hours))/date_yyyymmdd/str(int(z))/str(int(x))/f"{int(y)}.npz" def _raw_tile_cache_dir()->Path: d=CACHE/"rawtilecache";d.mkdir(parents=True,exist_ok=True);return d def _is_valid_png_bytes(b:bytes)->bool:return bool(b)and len(b)>=64 and b[:8]==PNG_MAGIC def _is_valid_npz_bytes(b:bytes)->bool:return bool(b)and len(b)>=16 and b[:4]==ZIP_MAGIC def _atomic_write_bytes(path:Path,data:bytes)->None: path.parent.mkdir(parents=True,exist_ok=True) tmp=path.with_name(path.name+f".tmp.{uuid.uuid4().hex}") with tmp.open("wb")as f:f.write(data);f.flush();os.fsync(f.fileno()) os.replace(tmp,path) def _is_valid_npz_file(p:Path)->bool: try: if not p.exists():return False if p.stat().st_size<60:return False with p.open("rb")as f:head=f.read(2) return head==b"PK" except:return False # def _raw_tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int, # cancel_check=None) -> bytes | None: # p = _raw_tile_cache_path(dom, hours, ymd, z, x, y) # try: # if _is_valid_npz_file(p): # return p.read_bytes() # elif p.exists(): # _safe_unlink(p) # except: # pass # if not _hf_pull_on_miss_enabled(): # return None # if cancel_check and cancel_check(): # return None # rel = p.relative_to(CACHE).as_posix() # if _hf_negative_check(rel): # return None # try: # if _hf_try_pull_file(rel, cancel_check=cancel_check): # if cancel_check and cancel_check(): # return None # if _is_valid_npz_file(p): # return p.read_bytes() # elif p.exists(): # _safe_unlink(p) # _hf_negative_mark(rel) # except Exception as e: # if "404" not in str(e).lower(): # _log("WARN", f"[hf_pull_raw] unexpected error for {rel}: {e!r}") # else: # _hf_negative_mark(rel) # return None def _raw_tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int, cancel_check=None) -> bytes | None: """Check local raw tile cache. No HF pull — same reasoning as _tile_cache_get.""" p = _raw_tile_cache_path(dom, hours, ymd, z, x, y) try: if _is_valid_npz_file(p): return p.read_bytes() elif p.exists(): _safe_unlink(p) except: pass # Per-tile HF pull removed: see _tile_cache_get comment. return None def _tile_cache_put(dom,hours,ymd,z,x,y,png:bytes)->Path: p=_tile_cache_path(dom,hours,ymd,z,x,y) if not _is_valid_png_bytes(png): if p.exists():_safe_unlink(p) return p _atomic_write_bytes(p,png);return p def _raw_tile_cache_put(dom,hours,ymd,z,x,y,npz_bytes:bytes)->Path: p=_raw_tile_cache_path(dom,hours,ymd,z,x,y) if not _is_valid_npz_bytes(npz_bytes): if p.exists():_safe_unlink(p) return p _atomic_write_bytes(p,npz_bytes);return p _UPLOAD_STATE_DIR=CACHE/"_upload_state" _UPLOAD_STATE_DIR.mkdir(parents=True,exist_ok=True) def _upload_manifest_path(date_yyyymmdd:str)->Path:return _UPLOAD_STATE_DIR/f"uploaded_{date_yyyymmdd}.json" def _load_upload_manifest(date_yyyymmdd:str)->dict: p=_upload_manifest_path(date_yyyymmdd) if not p.exists():return{"tilecache":{},"rawtilecache":{}} try:return json.loads(p.read_text()) except:return{"tilecache":{},"rawtilecache":{}} def _save_upload_manifest(date_yyyymmdd:str,manifest:dict)->None: try:_upload_manifest_path(date_yyyymmdd).write_text(json.dumps(manifest,sort_keys=True)) except:pass def _load_upload_manifest_global()->dict: try: p=CACHE/"_hf_upload_manifest.json" if p.exists():return json.loads(p.read_text("utf-8")) except:pass return{} def _save_upload_manifest_global(m:dict)->None: try: p=CACHE/"_hf_upload_manifest.json";tmp=p.with_suffix(".tmp") tmp.write_text(json.dumps(m,separators=(",",":"),sort_keys=True),"utf-8");tmp.replace(p) except:pass def _folder_signature(local_folder:Path,stable_age_sec:float=8.0)->tuple[int,int,float]: now=time.time();file_count=0;total_bytes=0;newest_mtime=0.0 for p in local_folder.rglob("*"): try: if not p.is_file():continue st=p.stat() if st.st_size<=0:continue if(now-st.st_mtime)dict: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"} if not local_folder.exists():return{"ok":True,"skipped":True,"reason":"folder_not_exists"} file_count,total_bytes,newest_mtime=_folder_signature(local_folder,stable_age_sec=8.0) if file_count<=0:return{"ok":True,"skipped":True,"reason":"empty_or_unstable_folder"} max_files=int(os.environ.get("HF_UPLOAD_MAX_FILES","8000")) if file_count>max_files: _log("WARN",f"[hf_upload_folder] skip too many files kind={kind} {repo_path} files={file_count} max={max_files}") return{"ok":True,"skipped":True,"reason":"too_many_files","files":file_count,"max":max_files} manifest=_load_upload_manifest_global();key=f"{kind}|{dom}|{int(hours)}|{repo_path}";prev=manifest.get(key) if prev: if prev.get("file_count")==file_count and prev.get("total_bytes")==total_bytes and prev.get("newest_mtime",0.0)>=newest_mtime: return{"ok":True,"skipped":True,"reason":"no_change"} try: api.upload_folder(folder_path=str(local_folder),repo_id=repo,repo_type="dataset",path_in_repo=repo_path,token=tok, commit_message=f"{kind}: upload {date_yyyymmdd} dom={dom} h={hours} ({file_count} files)",allow_patterns=["**/*"]) manifest[key]={"file_count":file_count,"total_bytes":total_bytes,"newest_mtime":newest_mtime,"ts":time.time()} _save_upload_manifest_global(manifest) return{"ok":True,"skipped":False,"files":file_count,"bytes":total_bytes} except Exception as e: _log("WARN",f"[hf_upload_folder] failed kind={kind} date={date_yyyymmdd} err={e!r}") return{"ok":False,"error":repr(e)} def _iter_zoom_dirs(date_folder: Path): if not date_folder.exists(): return [] out = [] for p in sorted(date_folder.iterdir()): if p.is_dir() and p.name.isdigit(): out.append(p) return out def _hf_tiles_uploader_loop() -> None: _log("INFO", "[hf_uploader] disabled — roll job handles HF push directly") return def _pack_raw_tile_npz(*,melt_mm:np.ndarray,valid_mask_u8:np.ndarray)->bytes: try: mm=melt_mm.astype("float32",copy=False);m=valid_mask_u8.astype("uint8",copy=False);valid=(m>0) inches=(mm/25.4).astype("float32",copy=False);q=np.zeros(inches.shape,dtype="uint16") qv=np.rint(np.clip(inches*100.0,0.0,65535.0)).astype("uint16");q[valid]=qv[valid] bio=io.BytesIO();np.savez_compressed(bio,q=q,m=m,s=np.uint8(100));return bio.getvalue() except: bio=io.BytesIO();np.savez_compressed(bio,q=np.zeros((256,256),dtype="uint16"),m=np.zeros((256,256),dtype="uint8"),s=np.uint8(100)) return bio.getvalue() _LOCK_ROOT=Path(os.environ.get("SNODAS_LOCK_DIR","/tmp/snodas-locks")).expanduser() _LOCK_ROOT.mkdir(parents=True,exist_ok=True) _LOCK_STALE_SECONDS=int(os.environ.get("LOCK_STALE_SECONDS",str(48*3600))) def _acquire_singleton_lock(name:str)->bool: p=_LOCK_ROOT/f"{name}.lock";fd=None try: fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR) os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()}\n".encode("utf-8"));return True except FileExistsError: try: age=time.time()-p.stat().st_mtime if age>float(_LOCK_STALE_SECONDS): try:p.unlink(missing_ok=True) except:pass fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR) os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()} stale_recovered=1\n".encode("utf-8"));return True except:pass return False except:return False finally: if fd is not None: try:os.close(fd) except:pass def _hf_cfg_norm(): cfg=_hf_cfg() if isinstance(cfg,dict): ok=bool(cfg.get("ok"));repo=cfg.get("repo")or cfg.get("dataset_repo")or cfg.get("repo_id");token=cfg.get("token")or cfg.get("hf_token") return ok,repo,token,cfg if isinstance(cfg,tuple): if len(cfg)==0:return False,None,None,{"raw":cfg} if len(cfg)==1:return bool(cfg[0]),None,None,{"raw":cfg} if len(cfg)==2:return bool(cfg[0]),cfg[1],None,{"raw":cfg} return bool(cfg[0]),cfg[1],(cfg[2]if len(cfg)>=3 else None),{"raw":cfg} return False,None,None,{"raw":cfg} _ROLL_LOCK_NAME="roll_job" _ROLL_LOCK_STALE_SECONDS=int(os.environ.get("ROLL_LOCK_STALE_SECONDS",str(16*3600))) def _run_lock_path(name:str)->Path:return _LOCK_ROOT/f"{name}.runlock" def _try_acquire_run_lock(name:str,*,stale_seconds:int)->bool: p=_run_lock_path(name);fd=None try: fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR) os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()}\n".encode("utf-8"));return True except FileExistsError: try: age=time.time()-p.stat().st_mtime if age>float(stale_seconds): try:p.unlink(missing_ok=True) except:pass fd=os.open(p.as_posix(),os.O_CREAT|os.O_EXCL|os.O_RDWR) os.write(fd,f"pid={os.getpid()} utc={datetime.utcnow().isoformat()} stale_recovered=1\n".encode("utf-8"));return True except:pass return False except:return False finally: if fd is not None: try:os.close(fd) except:pass def _release_run_lock(name:str)->None: try:_run_lock_path(name).unlink(missing_ok=True) except:pass CFGRIB_DIR=CACHE/"cfgrib" CFGRIB_DIR.mkdir(parents=True,exist_ok=True) os.environ.setdefault("CFGRIB_INDEXPATH",str(CFGRIB_DIR)) COLLAB_BASE="https://www.nohrsc.noaa.gov/pub/products/collaborators" NSIDC_BASE="https://noaadata.apps.nsidc.org/NOAA/G02158" _DIRLIST={"ts":0.0,"txt":""} _COLLAB_LAST={"ts":0.0,"status":None,"error":None,"len":0} _BUILD_LOCKS:dict[str,threading.Lock]={} _BUILD_LOCKS_LOCK=threading.Lock() _LOG=deque(maxlen=2000) TILE_CACHE_DIR=(CACHE/"tile_png_cache"/"forecast_by_date") TILE_CACHE_DIR.mkdir(parents=True,exist_ok=True) def _q(v:float,step:float)->float:return round(v/step)*step def _lock(key:str)->threading.Lock: with _BUILD_LOCKS_LOCK: l=_BUILD_LOCKS.get(key) if l is None:l=_BUILD_LOCKS[key]=threading.Lock() return l # def _tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int, # cancel_check=None) -> bytes | None: # p = _tile_cache_path(dom, hours, ymd, z, x, y) # try: # if _is_valid_png_file(p): # return p.read_bytes() # elif p.exists(): # _safe_unlink(p) # except: # pass # if not _hf_pull_on_miss_enabled(): # return None # if cancel_check and cancel_check(): # return None # rel = p.relative_to(CACHE).as_posix() # if _hf_negative_check(rel): # return None # try: # if _hf_try_pull_file(rel, cancel_check=cancel_check): # if cancel_check and cancel_check(): # return None # if _is_valid_png_file(p): # return p.read_bytes() # elif p.exists(): # _safe_unlink(p) # _hf_negative_mark(rel) # except Exception as e: # if "404" not in str(e).lower(): # _log("WARN", f"[hf_pull] unexpected error for {rel}: {e!r}") # else: # _hf_negative_mark(rel) # return None def _tile_cache_get(dom: str, hours: int, ymd: str, z: int, x: int, y: int, cancel_check=None) -> bytes | None: """Check local tile cache. No HF pull — tiles only exist as tar archives in HF, not as individual files, so per-tile pulls always 404.""" p = _tile_cache_path(dom, hours, ymd, z, x, y) try: if _is_valid_png_file(p): return p.read_bytes() elif p.exists(): _safe_unlink(p) except: pass # Per-tile HF pull removed: individual tiles don't exist in the HF repo. # Tiles get populated via boot-time tar extraction or local baking. # Cache misses fall through to on-demand generation (fast w/ preloaded COGs). return None def _hf_date_exists_remote(dom: str, hours: int, ymd: str) -> bool: """Check if HF repo has any tiles for this date/hours combo. Fast — only lists one dir.""" tok, repo = _hf_cfg() if not tok or not repo: return False api = _hf_api() if api is None: return False prefix = f"tilecache/{dom}/{hours}/{ymd}" try: items = api.list_repo_tree(repo_id=repo, repo_type="dataset", path_in_repo=prefix, recursive=False) for _ in items: return True except Exception: pass return False def _date_is_baked(dom: str, ymd: str, min_tiles: int = 50) -> bool: if _date_has_local_tiles_both(dom, ymd, min_tiles=min_tiles): return True # Temporarily check HF remote for testing for h in WARM_HOURS_LIST: if not _hf_date_exists_remote(dom, h, ymd): return False return True def _incremental_hf_push(tile_paths:list[Path],repo_prefix:str,batch_name:str)->dict: if not tile_paths:return{"ok":True,"pushed":0,"skipped":0} if repo_prefix in("tilecache","rawtilecache"):return{"ok":True,"pushed":0,"skipped":len(tile_paths),"disabled":True} tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured","pushed":0,"skipped":0} ops:list[CommitOperationAdd]=[];skipped=0 for p in tile_paths: try: if(not p.exists())or(p.stat().st_size<=0):skipped+=1;continue rel=p.relative_to(CACHE).as_posix() if not(rel.startswith("forecast/")or rel.startswith("tracks/")):skipped+=1;continue ops.append(CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix())) except:skipped+=1 if not ops:return{"ok":True,"pushed":0,"skipped":len(tile_paths)} dedup:dict[str,CommitOperationAdd]={} for op in ops:dedup[op.path_in_repo]=op ops=list(dedup.values());attempts=3;last_err:str|None=None for i in range(attempts): try: api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"{repo_prefix}: {batch_name} ({len(ops)} files)") return{"ok":True,"pushed":len(ops),"skipped":(len(tile_paths)-len(ops))+skipped} except Exception as e: msg=repr(e);last_err=msg is_timeout=("ReadTimeout"in msg)or("Timeout"in msg)or("timed out"in msg.lower()) _log("WARN",f"[incremental_push] failed attempt={i+1}/{attempts}: {msg}") if not is_timeout:return{"ok":False,"error":msg,"pushed":0,"skipped":0} time.sleep(2.0*(2**i)+random.random()) return{"ok":False,"error":last_err or"unknown_error","pushed":0,"skipped":0} def _ensure_cn_mbtiles()->Path: CN_MBTILES_LOCAL.parent.mkdir(parents=True,exist_ok=True) if CN_MBTILES_LOCAL.exists()and CN_MBTILES_LOCAL.stat().st_size>1024*1024:return CN_MBTILES_LOCAL tok,repo=_hf_cfg() if not tok or not repo:raise RuntimeError("HF not configured; cannot download cn_tracks.mbtiles") p=hf_hub_download(repo_id=repo,repo_type="dataset",filename=CN_MBTILES_REPO_PATH,token=tok,local_dir=str(CN_MBTILES_LOCAL.parent))#,local_dir_use_symlinks=False) downloaded=Path(p) if downloaded!=CN_MBTILES_LOCAL: try:downloaded.replace(CN_MBTILES_LOCAL) except:pass return CN_MBTILES_LOCAL def _mbtiles_overzoom(z:int,x:int,y:int,zmax:int)->tuple[int,int,int]: z=int(z);x=int(x);y=int(y);zmax=int(zmax) if z<=zmax:return z,x,y dz=z-zmax;return zmax,(x>>dz),(y>>dz) def _cogreader_lru_invalidate(cog:Path)->None: key=cog.as_posix() with _COG_LRU_LK:r=_COG_LRU.pop(key,None) if r is not None: try:r.close() except:pass def _atomic_replace(src:Path,dst:Path)->None: dst.parent.mkdir(parents=True,exist_ok=True) try:src.replace(dst) except: if dst.exists(): try:dst.unlink(missing_ok=True) except:pass src.replace(dst) def _atomic_cog_translate(src_tif:Path,out_cog:Path)->None: out_cog.parent.mkdir(parents=True,exist_ok=True) tmp_cog=out_cog.with_suffix(out_cog.suffix+".tmp") try:tmp_cog.unlink(missing_ok=True) except:pass prof=cog_profiles.get("deflate") if prof is None:raise HTTPException(status_code=500,detail="rio_cogeo deflate profile missing") cog_translate(src_tif.as_posix(),tmp_cog.as_posix(),prof,quiet=True,overview_resampling="nearest") # Verify the COG is readable before replacing try: with rasterio.open(tmp_cog.as_posix())as verify: verify.read(1,window=rasterio.windows.Window(0,0,min(256,verify.width),min(256,verify.height))) except Exception as e: _log("WARN",f"[cog_translate] verification failed for {out_cog.name}: {e!r}") try:tmp_cog.unlink(missing_ok=True) except:pass raise HTTPException(status_code=500,detail=f"COG verification failed: {e!r}") _atomic_replace(tmp_cog,out_cog) try:src_tif.unlink(missing_ok=True) except:pass def _mbtiles_xyz_to_tms(z:int,y:int)->int:return(1<sqlite3.Connection: # mb=_ensure_cn_mbtiles();return sqlite3.connect(f"file:{mb}?mode=ro",uri=True,check_same_thread=False) _CN_CONN: sqlite3.Connection | None = None _CN_CONN_LK = threading.Lock() def _cn_mbtiles_conn() -> sqlite3.Connection: global _CN_CONN if _CN_CONN is not None: return _CN_CONN with _CN_CONN_LK: if _CN_CONN is not None: return _CN_CONN mb = _ensure_cn_mbtiles() _CN_CONN = sqlite3.connect(f"file:{mb}?mode=ro", uri=True, check_same_thread=False) return _CN_CONN def _mbtiles_get_tile(conn:sqlite3.Connection,z:int,x:int,y:int)->bytes|None: ty=_mbtiles_xyz_to_tms(z,y) try: cur=conn.execute("SELECT tile_data FROM tiles WHERE zoom_level=? AND tile_column=? AND tile_row=?",(z,x,ty)) row=cur.fetchone();return row[0]if row else None except:return None @lru_cache(maxsize=1) def _cn_mbtiles_maxzoom() -> int: try: conn=_cn_mbtiles_conn() cur=conn.execute("SELECT value FROM metadata WHERE name='maxzoom'") row=cur.fetchone() if row:return int(row[0]) cur2=conn.execute("SELECT MAX(zoom_level) FROM tiles") row2=cur2.fetchone();return int(row2[0])if row2 and row2[0]else 10 except:return 10 def _cn_overzoom_parent(z:int,x:int,y:int,maxz:int)->tuple[int,int,int]: if z<=maxz:return z,x,y dz=z-maxz;return maxz,x>>dz,y>>dz def _tiles_for_boxes_zoom_range(boxes:list[tuple[float,float,float,float]],zmin:int,zmax:int,cap_per_zoom:int=5000)->list[tuple[int,int,int]]: out=[] for z in range(zmin,zmax+1): n=1<x1:x0,x1=x1,x0 if y0>y1:y0,y1=y1,y0 for xx in range(x0,x1+1): for yy in range(y0,y1+1):zset.add((z,xx,yy)) for t in list(zset)[:cap_per_zoom]:out.append(t) return out def _generate_forecast_by_date_png_impl(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,max_in:float|None)->tuple[bytes,dict]: dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") if int(hours)==24: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail) try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom) except Exception as e:raise HTTPException(status_code=503,detail={"error":"melt24_union_cog_unavailable","date":date_yyyymmdd,"hours":24,"upstream_error":repr(e)}) try:melt_mm, melt_mask, oob = _tile_arrays_from_cog(melt24_union, z, x, y, cog_cache=None) except Exception as e: _log("WARN",f"[tile_read] melt24 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e!r}") return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Read-Error":"melt24"} if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24"} melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255) snow=snow_valid=None;snow_info="snowpack_unavailable" if sel.get("snowpack_urls"): try: snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom) if snow_union: try: snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_union,z,x,y,cog_cache=COG_CACHE) if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr;snow_info=f"snowpack_union:{snow_union.name}" except Exception as e2:_log("WARN",f"[tile_read] snowpack24 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e2!r}");snow_info=f"snowpack_unavailable:{e2!r}" except Exception as e:snow_info=f"snowpack_unavailable:{e!r}" png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_IN) return png,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24"} try: melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom) except HTTPException as e: raise HTTPException(status_code=503,detail={"error":"melt72_cog_unavailable","date":date_yyyymmdd,"hours":72,"upstream_error":str(e.detail)if hasattr(e,"detail")else str(e)}) except Exception as e: raise HTTPException(status_code=503,detail={"error":"melt72_cog_unavailable","date":date_yyyymmdd,"hours":72,"upstream_error":repr(e)}) try: melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=None) except Exception as e: _log("WARN",f"[tile_read] melt72 failed date={date_yyyymmdd} dom={dom} z={z} x={x} y={y} err={e!r}") return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-72h-COG":melt72_cog.name,"X-Read-Error":"melt72"} if oob or melt_mm is None or melt_mask is None: return _transparent_png_256(),{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-72h-COG":melt72_cog.name} # Snowpack overlay is optional — try run_init but don't block on it run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom) snow=snow_valid=None;snow_info="snowpack_unavailable" if run_init: snow_urls=_find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom) if snow_urls: try: snow_cogs=_build_forecast_snowpack_cogs(valid,snow_urls) snow,snow_valid,snow_info=_union_tile(snow_cogs,z,x,y,label="snowpack") except Exception as e:snow_info=f"snowpack_unavailable:{e!r}" png=_melt_to_png(melt_mm,melt_mask,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_72H_IN) return png,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init or"none","X-Forecast-72h-COG":melt72_cog.name,"X-SnowPack-INFO":snow_info} def _generate_forecast_by_date_png(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,max_in:float|None)->tuple[bytes,dict]: return _generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=max_in) def _union_tile_cached(cogs:list[Path],z:int,x:int,y:int,*,label:str,cog_cache:CogCache,valid_pred:Optional[Callable[[np.ndarray,np.ndarray],np.ndarray]]=None)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]: if not cogs:return None,None,f"{label}_no_cogs" out_arr:Optional[np.ndarray]=None;out_valid:Optional[np.ndarray]=None;used:list[str]=[] for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))): try: r=cog_cache.get(cog);data,mask=r.tile(x,y,z,tilesize=256,resampling_method="nearest") arr=data[0].astype("float32");m=mask[0];valid=(m>0)&np.isfinite(arr)&(arr>-9990.0) if valid_pred is not None: try:valid=valid&valid_pred(arr,m) except:pass if out_arr is None:out_arr=arr.copy();out_valid=valid.copy() else: fill=valid&(~out_valid) if fill.any():out_arr[fill]=arr[fill];out_valid[fill]=True used.append(cog.name) except TileOutsideBounds:continue except:continue if out_arr is None or out_valid is None:return None,None,f"{label}_unavailable_all_domains" info=f"{label}_union:"+",".join(used[:4])+("..."if len(used)>4 else"") return out_arr,out_valid,info def _log(level:str,msg:str)->None: _LOG.append({"ts":int(datetime.utcnow().timestamp()*1000),"level":level,"msg":msg}) print(f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} | {level} | {msg}",flush=True) _MODE=threading.local() def _mode_get():return getattr(_MODE,"m","tiles") class _Mode: def __init__(self,m):self.m=m def __enter__(self):_MODE.m=self.m def __exit__(self,*a):_MODE.m="tiles" _SESSION=requests.Session() POOL=int(os.environ.get("HTTP_POOL_SIZE","64")) _ADAPTER=requests.adapters.HTTPAdapter(pool_connections=POOL,pool_maxsize=POOL,max_retries=0,pool_block=True) _SESSION.mount("http://",_ADAPTER);_SESSION.mount("https://",_ADAPTER) HF_PULL_ON_MISS_DEFAULT=(os.environ.get("HF_PULL_ON_MISS","1").strip()!="0") _HF_PULL_ON_MISS_LOCAL=threading.local() def _hf_pull_on_miss_enabled()->bool: v=getattr(_HF_PULL_ON_MISS_LOCAL,"enabled",None) return bool(HF_PULL_ON_MISS_DEFAULT)if v is None else bool(v) class _HfPullOnMiss: def __init__(self,enabled:bool):self.enabled=bool(enabled);self.prev=None def __enter__(self): self.prev=getattr(_HF_PULL_ON_MISS_LOCAL,"enabled",None);_HF_PULL_ON_MISS_LOCAL.enabled=self.enabled;return self def __exit__(self,*a): if self.prev is None and hasattr(_HF_PULL_ON_MISS_LOCAL,"enabled"): try:delattr(_HF_PULL_ON_MISS_LOCAL,"enabled") except:_HF_PULL_ON_MISS_LOCAL.enabled=None else:_HF_PULL_ON_MISS_LOCAL.enabled=self.prev _CPU_SEM_LOCK=threading.Lock() def _cpu_sem_limits()->tuple[int,int]: t=int(os.environ.get("CPU_TILES_MAX_INFLIGHT","12"));v=int(os.environ.get("CPU_VALUE_MAX_INFLIGHT","16"));return t,v def _cpu_set_limits(*,tiles:int|None=None,value:int|None=None)->None: global CPU_SEM_TILES,CPU_SEM_VALUE tiles0,value0=_cpu_sem_limits();t=int(tiles if tiles is not None else tiles0);v=int(value if value is not None else value0) t=max(1,t);v=max(1,v) with _CPU_SEM_LOCK:CPU_SEM_TILES=threading.BoundedSemaphore(t);CPU_SEM_VALUE=threading.BoundedSemaphore(v) class _CpuLimits: def __init__(self,*,tiles:int|None=None,value:int|None=None):pass def __enter__(self):return self def __exit__(self,*a):pass def _retry_get(url:str,*,timeout=(10,400),tries:int=6,backoff:float=1.6,headers=None,stream:bool=False,allow_redirects:bool=True)->requests.Response: hdr={"User-Agent":"snodas/1.3"} if headers:hdr.update(headers) last_exc=None for i in range(int(tries)): sem=HTTP_SEM_VALUE if _mode_get()=="value"else HTTP_SEM_TILES with sem: try:r=_SESSION.get(url,timeout=timeout,allow_redirects=allow_redirects,headers=hdr,stream=stream) except(requests.exceptions.Timeout,requests.exceptions.ConnectionError)as e:last_exc=e;r=None if r is None: if iPath: dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid) out=_forecast_dir()/f"{valid}_{dom_prefer}_melt24_union_cog.tif" if out.exists()and out.stat().st_size>0:return out sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer) if not sel.get("ok"):raise HTTPException(status_code=503,detail={"error":"pick_failed","valid":valid,"dom":dom_prefer,"detail":sel}) melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]) if not melt_cogs:raise HTTPException(status_code=503,detail={"error":"no_melt_cogs","valid":valid,"dom":dom_prefer}) cogs=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p))) with rasterio.open(cogs[0])as base: acc=base.read(1).astype("float32");base_transform=base.transform;base_crs=base.crs;base_w,base_h=base.width,base.height acc_valid=np.isfinite(acc)&(acc>-9990.0) for p in cogs[1:]: with rasterio.open(p)as src: arr=src.read(1).astype("float32") if(src.transform!=base_transform)or(src.crs!=base_crs)or(src.width!=base_w)or(src.height!=base_h): tmp=np.full((base_h,base_w),np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base_transform,dst_crs=base_crs,resampling=Resampling.nearest) arr=tmp v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid) if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out) try:_hf_enqueue_files([out]) except:pass tmp_tif.unlink(missing_ok=True) _log("INFO",f"[build_forecast] melt24_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}") return out def _forecast_snowpack_union_cog(valid:str,dom_prefer:str="zz")->Path|None: dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid) out=_forecast_dir()/f"{valid}_{dom_prefer}_snow_union_cog.tif" if out.exists()and out.stat().st_size>0:return out sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer) if not sel.get("ok")or not sel.get("snowpack_urls"):return None try:snow_cogs=_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"]) except:return None if not snow_cogs:return None cogs=sorted(snow_cogs,key=lambda p:_dom_pri(_cog_dom(p))) with rasterio.open(cogs[0])as base: acc=base.read(1).astype("float32");base_transform=base.transform;base_crs=base.crs;base_w,base_h=base.width,base.height acc_valid=np.isfinite(acc)&(acc>-9990.0) for p in cogs[1:]: with rasterio.open(p)as src: arr=src.read(1).astype("float32") if(src.transform!=base_transform)or(src.crs!=base_crs)or(src.width!=base_w)or(src.height!=base_h): tmp=np.full((base_h,base_w),np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base_transform,dst_crs=base_crs,resampling=Resampling.nearest) arr=tmp v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid) if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out) try:_hf_enqueue_files([out]) except:pass tmp_tif.unlink(missing_ok=True) _log("INFO",f"[build_forecast] snow_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}") return out def _forecast_melt24h_union_end_cog(valid:str,dom_prefer:str="zz")->Path: valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() out=_forecast_dir()/f"fcst_melt24h_union_end_{valid}_{dom_prefer}_cog.tif" if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out with _lock(f"fcst_melt24h_union_{valid}_{dom_prefer}"): if out.exists()and out.stat().st_size>0:return out sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer) if not sel.get("ok"):raise HTTPException(status_code=503,detail={"error":"forecast_selection_unavailable_for_24h_union","valid":valid,"dom":dom_prefer,"sel":sel}) _log("INFO",f"[debug_union] valid={valid} run_init={sel['run_init']} melt_urls={sel['melt_urls']}") melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]) _log("INFO",f"[debug_union] melt_cogs built: {[c.name for c in melt_cogs]}") if not melt_cogs:raise HTTPException(status_code=503,detail={"error":"no_melt_cogs_for_24h_union","valid":valid,"dom":dom_prefer}) cogs=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p))) _log("INFO",f"[debug_union] sorted cogs for union: {[c.name for c in cogs]}") with rasterio.open(cogs[0])as base: acc=base.read(1).astype("float32");acc_valid=np.isfinite(acc)&(acc>-9990.0) _log("INFO",f"[debug_union] base cog={cogs[0].name} valid_pixels={acc_valid.sum()} mean_where_valid={float(np.nanmean(acc[acc_valid])):.2f}") for c in cogs[1:]: with rasterio.open(c)as src: arr=src.read(1).astype("float32") if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height): tmp=np.full_like(acc,np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest) arr=tmp v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid) _log("INFO",f"[debug_union] overlay cog={c.name} valid_pixels={v.sum()} fill_pixels={fill.sum()} mean_where_valid={float(np.nanmean(arr[v])):.2f}") if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True _log("INFO",f"[debug_union] final: total_valid={acc_valid.sum()} mean={float(np.nanmean(acc[acc_valid])):.2f} max={float(np.nanmax(acc[acc_valid])):.2f}") tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out) try:_hf_enqueue_files([out]) except:pass tmp_tif.unlink(missing_ok=True) _log("INFO",f"[build_forecast] melt24h_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}") return out def _forecast_snowpack_union_end_cog(valid:str,dom_prefer:str="zz")->Path|None: valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() out=_forecast_dir()/f"fcst_snowpack_union_end_{valid}_{dom_prefer}_cog.tif" if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out with _lock(f"fcst_snowpack_union_{valid}_{dom_prefer}"): if out.exists()and out.stat().st_size>0:return out sel=_pick_forecast_for_valid(valid,dom_prefer=dom_prefer) if not sel.get("ok")or not sel.get("snowpack_urls"):return None try:snow_cogs=_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"]) except:return None if not snow_cogs:return None cogs=sorted(snow_cogs,key=lambda p:_dom_pri(_cog_dom(p))) with rasterio.open(cogs[0])as base: acc=base.read(1).astype("float32");acc_valid=np.isfinite(acc)&(acc>-9990.0) for c in cogs[1:]: with rasterio.open(c)as src: arr=src.read(1).astype("float32") if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height): tmp=np.full_like(acc,np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest) arr=tmp v=np.isfinite(arr)&(arr>-9990.0);fill=v&(~acc_valid) if fill.any():acc[fill]=arr[fill];acc_valid[fill]=True tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out) try:_hf_enqueue_files([out]) except:pass tmp_tif.unlink(missing_ok=True) _log("INFO",f"[build_forecast] snowpack_union OK valid={valid} dom_prefer={dom_prefer} -> {out.name}") return out app=FastAPI(title="SNODAS Snowmelt Tiles",version="1.3.0") app.add_middleware(GZipMiddleware, minimum_size=500) app.mount("/web",StaticFiles(directory="web",html=True),name="web") @app.get("/favicon.ico") def favicon():return Response(status_code=204) _TS10=re.compile(r"^\d{10}$") _TS8U2=re.compile(r"^(\d{8})[_-](\d{2})$") _HREF_RE=re.compile(r'href\s*=\s*(?P[\'"])(?P.*?)(?P=q)',re.I) _TS_IN_NAME=re.compile(r"(?:TS|NATS)(\d{10})",re.I) _TT_IN_NAME=re.compile(r"TT(\d{10})",re.I) def _norm_ts(ts:str)->str: ts=(ts or"").strip() if _TS10.fullmatch(ts):return ts m=_TS8U2.fullmatch(ts) if m:return f"{m.group(1)}{m.group(2)}" raise HTTPException(status_code=404,detail="unknown forecast key") US_BORDER_LAT=49.0 def _tile2lon(x:int,z:int)->float:return(float(x)/(1<float: n=np.pi-(2.0*np.pi*float(y)/(1<tuple[float,float,float,float]: z=int(z);x=int(x);y=int(y);return _tile2lon(x,z),_tile2lat(y+1,z),_tile2lon(x+1,z),_tile2lat(y,z) def _box_intersects(a,b)->bool: aw,as_,ae,an=a;bw,bs,be,bn=b;return not(aebool: z=int(z);x=int(x);y=int(y);n=1<=n or x<0 or x>=n:return True tb=_tile_bounds(z,x,y);box_a=(-170.0,49.0,-40.0,90.0);box_b=(-97.0,37.0,-63.0,49.0) return _box_intersects(tb,box_a)or _box_intersects(tb,box_b) HTTP_SEM_TILES=threading.BoundedSemaphore(int(os.environ.get("HTTP_TILES_MAX_INFLIGHT","24"))) HTTP_SEM_VALUE=threading.BoundedSemaphore(int(os.environ.get("HTTP_VALUE_MAX_INFLIGHT","12"))) _cpu_set_limits() @lru_cache(maxsize=1) def _transparent_png_256()->bytes: rgb=np.zeros((3,256,256),dtype="uint8");a=np.zeros((256,256),dtype="uint8");return render(rgb,mask=a,img_format="PNG") @lru_cache(maxsize=1) def _zero_raw_npz_256()->bytes: return _pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8")) _REMOTE_KEEP_DAYS=7 def _hf_cfg()->tuple[Optional[str],Optional[str]]: tok=(os.environ.get("HF_TOKEN")or"").strip();repo=(os.environ.get("HF_DATASET_REPO")or"").strip() return(tok,repo)if tok and repo else(None,None) _HF_API_SINGLETON:Optional[HfApi]=None def _hf_api()->Optional[HfApi]: global _HF_API_SINGLETON tok,repo=_hf_cfg() if not tok or not repo:return None if _HF_API_SINGLETON is None:_HF_API_SINGLETON=HfApi() return _HF_API_SINGLETON def _date_ymd_utc(d:datetime.date|None=None)->str: if d is None:d=datetime.utcnow().date() return d.strftime("%Y%m%d") # def _keep_cutoff_ymd(keep_days:int=_REMOTE_KEEP_DAYS)->str: # return(datetime.utcnow().date()-timedelta(days=int(keep_days)-1)).strftime("%Y%m%d") def _file_ymd_from_name(name:str)->Optional[str]: m=re.search(r"(?:^|_)(\d{8})(?:_|\.|$)",name);return m.group(1)if m else None def _is_cache_file_we_manage(rel:str)->bool: n=rel.replace("\\","/") if n.startswith("forecast/")and n.endswith("_cog.tif"):return True if n.startswith("tilecache/")and n.endswith(".png"):return True if n.startswith("rawtilecache/")and n.endswith(".npz"):return True base=n.rsplit("/",1)[-1] if re.fullmatch(r"bydate_[a-z0-9]+_h\d+_\d{8}_z\d+_\d+_\d+\.png",base):return True if re.fullmatch(r"bydate_[a-z0-9]+_h\d+_\d{8}_z\d+_\d+_\d+\.npz",base):return True return False def _local_cache_paths_to_consider()->list[Path]: out:list[Path]=[] for p in CACHE.glob("melt24h_*_cog.tif"):out.append(p) for p in CACHE.glob("melt72h_end_*_cog.tif"):out.append(p) fc=_forecast_dir() if fc.exists(): for p in fc.glob("*_cog.tif"):out.append(p) return out def _is_lfs_pointer_bytes(b:bytes)->bool: if not b:return False if b.startswith(b"version https://git-lfs.github.com/spec/v1"):return True if b"git-lfs.github.com/spec/v1"in b[:200]:return True return False def _hf_push_baked_as_tar(baked_ymds:list[str],dom:str,hours_list:list[int])->dict: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"} dom=(dom or"zz").lower();pushed=0;errors=0;archives_pushed=0 for ymd in baked_ymds: for h in hours_list: for kind in["tilecache","rawtilecache"]: base=CACHE/kind/dom/str(int(h))/ymd if not base.exists():continue while _PUSH_PAUSED.is_set():time.sleep(2.0) ext=".png"if kind=="tilecache"else".npz";min_sz=64 if kind=="tilecache"else 60 files=[] for p in base.rglob(f"*{ext}"): try: if p.stat().st_size=9.0: _log("INFO",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd}: timeout after {elapsed:.1f}s — likely succeeded");pushed+=len(files);archives_pushed+=1;ok=True;break _log("WARN",f"[hf_push_tar] {kind}/{dom}/h{h}/{ymd} attempt {attempt+1}: {type(e).__name__} after {elapsed:.1f}s") if attempt<2:time.sleep(10.0*(attempt+1)) if not ok:errors+=len(files) del tar_bytes,tar_buf _log("INFO",f"[hf_push_tar] done: pushed={pushed} errors={errors} archives={archives_pushed}") return{"ok":errors==0,"pushed":pushed,"errors":errors,"archives":archives_pushed} def _hf_pull_tar_archives()->int: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return 0 window=set(_rolling_dates_local_window(lookback=2,lookahead=3)) _log("INFO",f"[hf_pull_tar] window={sorted(window)}") try: items=list(api.list_repo_tree(repo_id=repo,repo_type="dataset",path_in_repo="archives/",recursive=False)) archive_paths=[getattr(it,"path","")for it in items if getattr(it,"path","").endswith(".tar.gz")] except Exception as e:_log("WARN",f"[hf_pull_tar] list failed: {e!r}");return 0 if not archive_paths:_log("INFO","[hf_pull_tar] no archives found");return 0 _log("INFO",f"[hf_pull_tar] found {len(archive_paths)} archives") pulled=0;staging=CACHE/"_tar_staging";staging.mkdir(parents=True,exist_ok=True) marker_dir=CACHE/"_tar_markers";marker_dir.mkdir(parents=True,exist_ok=True) for ap in archive_paths: name=ap.rsplit("/",1)[-1];m=re.search(r"_(\d{8})\.tar\.gz$",name) if not m:continue ymd=m.group(1) if ymd not in window:continue marker=marker_dir/f"{name}.done" if marker.exists():continue parts=name.replace(".tar.gz","").split("_") if len(parts)>=4: edir=CACHE/parts[0]/parts[1]/parts[2].replace("h","")/parts[3] if edir.exists(): ext="*.png"if parts[0]=="tilecache"else"*.npz" try: cnt=sum(1 for _ in edir.rglob(ext)) if cnt>100:_log("INFO",f"[hf_pull_tar] skip {name}: {cnt} local files");marker.write_text(f"skipped:{cnt}");continue except:pass _log("INFO",f"[hf_pull_tar] downloading {ap}");t0=time.time() try: local=hf_hub_download(repo_id=repo,repo_type="dataset",filename=ap,token=tok,local_dir=str(staging)) lp=Path(local) if not lp.exists():continue with lp.open("rb")as f:magic=f.read(2) if magic!=b"\x1f\x8b":_log("WARN",f"[hf_pull_tar] {name}: not gzip");lp.unlink(missing_ok=True);continue with tarfile.open(str(lp),mode="r:gz")as tar: safe=[m for m in tar.getmembers()if not m.name.startswith("/")and".."not in m.name] tar.extractall(path=str(CACHE),members=safe);pulled+=len(safe) _log("INFO",f"[hf_pull_tar] unpacked {name}: {len(safe)} files in {time.time()-t0:.1f}s") marker.write_text(f"unpacked:{len(safe)}:ts:{time.time()}") try:lp.unlink(missing_ok=True) except:pass except Exception as e:_log("WARN",f"[hf_pull_tar] failed {ap}: {e!r}") try:shutil.rmtree(staging,ignore_errors=True) except:pass _log("INFO",f"[hf_pull_tar] complete: {pulled} files");return pulled # def _hf_pull_cache()->None: # tok,repo=_hf_cfg() # if not tok or not repo:_log("INFO","[hf] pull cache disabled");return # window=_rolling_dates_local_window(lookback=1,lookahead=3);dom=(WARM_DOM or"zz").lower() # _log("INFO",f"[hf] pull COGs for window={window}");pulled=0 # for ymd in window: # for name in[f"forecast/fcst_melt24h_union_end_{ymd}05_{dom}_cog.tif",f"forecast/fcst_melt72h_end_{ymd}05_{dom}_cog.tif",f"forecast/fcst_snowpack_union_end_{ymd}05_{dom}_cog.tif"]: # dest=CACHE/name # if dest.exists()and dest.stat().st_size>1000:continue # try:hf_hub_download(repo_id=repo,repo_type="dataset",token=tok,filename=name,local_dir=str(CACHE));pulled+=1 # except:pass # _log("INFO",f"[hf] pull COGs done: {pulled}") # _log("INFO","[hf] pull tar archives") # try:pt=_hf_pull_tar_archives();_log("INFO",f"[hf] pull tar archives done: {pt} files") # except Exception as e:_log("WARN",f"[hf] pull tar archives failed: {e!r}") def _hf_pull_cache()->None: tok,repo=_hf_cfg() if not tok or not repo:_log("INFO","[hf] pull cache disabled");return window=_rolling_dates_local_window(lookback=1,lookahead=3);dom=(WARM_DOM or"zz").lower() _log("INFO",f"[hf] pull COGs for window={window}");pulled=0 for ymd in window: # Try both 11z and 05z COGs — staleness monitor may have upgraded to 11z for valid_suffix in ["11", "05"]: for prefix in ["fcst_melt24h_union_end", "fcst_melt72h_end", "fcst_snowpack_union_end"]: name = f"forecast/{prefix}_{ymd}{valid_suffix}_{dom}_cog.tif" dest = CACHE / name if dest.exists() and dest.stat().st_size > 1000: continue try: hf_hub_download(repo_id=repo, repo_type="dataset", token=tok, filename=name, local_dir=str(CACHE)) pulled += 1 except: pass _log("INFO",f"[hf] pull COGs done: {pulled}") _log("INFO","[hf] pull tar archives") try:pt=_hf_pull_tar_archives();_log("INFO",f"[hf] pull tar archives done: {pt} files") except Exception as e:_log("WARN",f"[hf] pull tar archives failed: {e!r}") def _hf_commit(ops:list,message:str)->None: api=_hf_api();tok,repo=_hf_cfg() if api is None or not repo or not ops:return api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message) def _hf_commit_with_backoff(ops:list,message:str)->None: api=_hf_api();tok,repo=_hf_cfg() if api is None or not repo or not ops:return for i in range(6): try: api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message) if HF_PUSH_SLEEP_SEC>0:time.sleep(HF_PUSH_SLEEP_SEC) return except Exception as e: s=repr(e).lower();is_rate=("429"in s)or("rate"in s and"limit"in s)or("too many requests"in s) if(i<5)and is_rate:time.sleep(2.0*(2**i)+random.uniform(0,0.25));continue raise _TILE_Z_RE=re.compile(r"_z(\d+)_") def _hf_push_tilecache_for_window(window:list[str],dom:str,hours_list:list[int],zmin:int|None=None,zmax:int|None=None)->dict: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"} dom=(dom or"zz").lower();root=_tile_cache_dir() if not root.exists():return{"ok":True,"pushed":0,"skipped":0} want_dates=set(window);want_hours=set(int(h)for h in hours_list) ops:list[CommitOperationAdd]=[];pushed=0;skipped=0 for p in root.rglob("bydate_*.png"): name=p.name;m=re.match(r"^bydate_([a-z0-9]+)_h(\d+)_(\d{8})_z\d+_\d+_\d+\.png$",name) if not m:continue f_dom,f_hours,f_ymd=m.group(1),int(m.group(2)),m.group(3) if f_dom!=dom or f_hours not in want_hours or f_ymd not in want_dates:continue mz=_TILE_Z_RE.search(name);z=int(mz.group(1))if mz else None if z is not None and zmin is not None and zint(zmax):continue if not _is_valid_png_file(p):skipped+=1;continue ops.append(CommitOperationAdd(path_in_repo=f"tilecache/{name}",path_or_fileobj=p.as_posix())) if len(ops)>=int(HF_MAX_OPS_PER_COMMIT): api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"tilecache: update dom={dom} files={len(ops)}") pushed+=len(ops);ops=[] if HF_PUSH_SLEEP_SEC>0:time.sleep(float(HF_PUSH_SLEEP_SEC)) if ops: api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"tilecache: update dom={dom} files={len(ops)}") pushed+=len(ops) return{"ok":True,"pushed":pushed,"skipped":skipped,"dom":dom} def _hf_push_files(paths:list[Path])->None: tok,repo=_hf_cfg() if not tok or not repo:return ops=[] for p in paths: try: if not p.exists()or p.stat().st_size<=0:continue rel=p.relative_to(CACHE).as_posix() if not _is_cache_file_we_manage(rel):continue ops.append(CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix())) except:continue try: _hf_commit(ops,message=f"cache: add/update {len(ops)} files") if ops:_log("INFO",f"[hf_cache] push OK files={len(ops)}") except Exception as e:_log("INFO",f"[hf_cache] push FAIL err={e!r}") def _hf_prune_remote_and_local(keep_days:int=_REMOTE_KEEP_DAYS,*,keep_ymds:set[str]|None=None)->None: if keep_ymds is None:cutoff=_keep_cutoff_ymd(keep_days) else:cutoff=min(keep_ymds)if keep_ymds else datetime.now(LOCAL_TZ).date().strftime("%Y%m%d") _log("INFO",f"[hf_prune] keep_ymds={sorted(keep_ymds)if keep_ymds else'None'} cutoff={cutoff}") local_dirs=0 for kind in["tilecache","rawtilecache"]: root=CACHE/kind if not root.exists():continue for dom_dir in root.iterdir(): if not dom_dir.is_dir():continue for hours_dir in dom_dir.iterdir(): if not hours_dir.is_dir():continue for ymd_dir in hours_dir.iterdir(): if not ymd_dir.is_dir():continue ymd=ymd_dir.name if not re.fullmatch(r"[0-9]{8}",ymd):continue if(keep_ymds and ymd not in keep_ymds)or(not keep_ymds and ymddatetime:return datetime.strptime(ts10,"%Y%m%d%H").replace(tzinfo=timezone.utc) def _yyyymmdd(d)->str:return d.strftime("%Y%m%d") def _track_points_path()->Path: for p in[Path("web")/"cn_track_points.npz",Path("cn_track_points.npz"),Path("cache")/"cn_track_points.npz"]: if p.exists():return p return Path("web")/"cn_track_points.npz" @lru_cache(maxsize=4) def _load_track_points(max_points:int=35000)->list[tuple[float,float]]: p=_track_points_path() if not p.exists():_log("INFO",f"[prewarm] cn_track_points.npz not found at {p}");return[] try: data=np.load(str(p));arr=data["lonlat"] if arr.ndim!=2 or arr.shape[1]!=2:_log("INFO",f"[prewarm] cn_track_points.npz has unexpected shape {arr.shape}");return[] arr=arr.astype(np.float64,copy=False) except Exception as e:_log("INFO",f"[prewarm] failed to read cn_track_points.npz err={e!r}");return[] if arr.shape[0]==0:return[] if arr.shape[0]>int(max_points):step=max(1,int(arr.shape[0])//int(max_points));arr=arr[::step] return[(float(lon),float(lat))for lon,lat in arr] def _lonlat_to_tile(lon:float,lat:float,z:int)->tuple[int,int]: lat=max(-85.05112878,min(85.05112878,float(lat)));lon=float(lon);n=2**int(z) x=int((lon+180.0)/360.0*n);lat_rad=np.deg2rad(lat) y=int((1.0-np.log(np.tan(lat_rad)+1.0/np.cos(lat_rad))/np.pi)/2.0*n) return max(0,min(n-1,x)),max(0,min(n-1,y)) def _collab_dirlist(ttl_sec:float=240.0)->str: now=datetime.utcnow().timestamp() if _DIRLIST["txt"]and(now-float(_DIRLIST["ts"]))list[str]:return[m.group("u").strip()for m in _HREF_RE.finditer(html or"")if(m.group("u")or"").strip()] def _ts_from_name(name:str)->Optional[str]:m=_TS_IN_NAME.search(Path(name).name);return m.group(1)if m else None def _tt_from_name(name:str)->Optional[str]:m=_TT_IN_NAME.search(Path(name).name);return m.group(1)if m else None def _dom_from_url(url:str)->str:n=Path(url).name.lower();return(n.split("_",1)[0]if"_"in n else n[:2]).strip()or"xx" def _dom_pri(dom:str)->int: d=(dom or"").lower() if d=="zz":return 0 if d=="dp":return 1 if d=="qc":return 2 if d=="us":return 3 return 9 def _as_url(n:str)->str:return n if n.startswith("http")else f"{COLLAB_BASE}/{n}" def _score_dom(url:str,dom_prefer:str)->int:d=_dom_from_url(url);return 0 if d==dom_prefer else(_dom_pri(d)+10) def _valid_ts_for_days(days:int,now_utc:Optional[datetime]=None)->str: now_utc=now_utc or datetime.utcnow().replace(tzinfo=timezone.utc) return f"{(now_utc.date()+timedelta(days=int(days))).strftime('%Y%m%d')}05" def _is_data_name(low:str)->bool:low=low.lower();return low.endswith(".nc")or low.endswith(".grz")or low.endswith(".grib2") def _list_available_valid_ts_t0024(dom_prefer:str="zz")->list[str]: dom_prefer=(dom_prefer or"zz").lower();html=_collab_dirlist();names=_hrefs(html);ts_set:set[str]=set() for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low or"__t0024tt"not in low or not low.endswith(".nc"):continue ts=_ts_from_name(base) if ts and _TS10.fullmatch(ts):ts_set.add(ts) return sorted(ts_set) def _lon0360_to_180(x:np.ndarray)->np.ndarray:return((x+180.0)%360.0)-180.0 def _best_state_nc_near_valid(code:str,valid:str,dom_prefer:str="zz",max_after_h:float=18.0,max_before_h:float=240.0)->dict: code=(code or"").lower();dom_prefer=(dom_prefer or"zz").lower();valid=_norm_ts(valid) html=_collab_dirlist();names=_hrefs(html);cands=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if code not in low or"__t0001ttnats"not in low or"hp001"not in low or not low.endswith(".nc"):continue ts=_ts_from_name(base) if not ts or not _TS10.fullmatch(ts):continue url=_as_url(base) try:dh=(_ts_to_dt(ts)-_ts_to_dt(valid)).total_seconds()/3600.0 except:continue cands.append((dh,ts,_dom_from_url(url),url)) if not cands:return{"picked_ts":None,"url":None,"delta_h":None} preferred=[r for r in cands if(r[2]or"").lower()==dom_prefer];pool=preferred if preferred else cands before=[r for r in pool if r[0]<=0];after=[r for r in pool if r[0]>0];pick=None if before:before.sort(key=lambda t:abs(t[0]));pick=before[0]if abs(before[0][0])<=float(max_before_h)else None if pick is None and after:after.sort(key=lambda t:t[0]);pick=after[0]if after[0][0]<=float(max_after_h)else None if pick is None:pool.sort(key=lambda t:abs(t[0]));pick=pool[0] dh,ts,dom,url=pick;return{"picked_ts":ts,"url":url,"delta_h":dh,"dom":dom} def _find_fcst_11034_grib2_tt_ts(run_init:str,valid:str,dom_prefer:str="zz")->list[str]: run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11034"not in low or"__t0001tt"not in low or"hp001"not in low:continue if want_tt not in low or want_ts not in low or not low.endswith(".grib2"):continue out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) def _pick_best_run_init_for_valid_t0024(valid:str,dom_prefer:str="zz")->Optional[str]: valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower();html=_collab_dirlist();names=_hrefs(html) want_ts=f"ts{valid}";rows=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low or"__t0024tt"not in low or want_ts not in low or not low.endswith(".nc"):continue tt=_tt_from_name(base) if not tt:continue rows.append((tt,_score_dom(_as_url(base),dom_prefer))) if not rows:return None rows.sort(key=lambda t:(t[0],-t[1]));return rows[-1][0] def _find_melt_11044_t0024(run_init:str,valid:str,dom_prefer:str="zz")->list[str]: run_init=_norm_ts(run_init);valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_tt=f"tt{run_init}";want_ts=f"ts{valid}";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low or"__t0024tt"not in low or want_tt not in low or want_ts not in low or not low.endswith(".nc"):continue out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) def _pick_forecast_for_valid(valid:str,dom_prefer:str="zz")->dict: dom_prefer=(dom_prefer or"zz").lower() # Extract date from valid (could be YYYYMMDD05 or YYYYMMDD11 or just YYYYMMDD) if len(valid)>=8: date_ymd=valid[:8] else: date_ymd=valid best=_find_best_melt_for_date(date_ymd,dom_prefer=dom_prefer) if not best.get("ok"): _log("DEBUG",f"[pick_forecast] no melt found for {date_ymd} dom={dom_prefer}: {best.get('error','unknown')}") return{"ok":False,"valid":valid,"error":best.get("error","no melt files found")} snowpack_urls=_find_snowpack_for_best_melt(best,dom_prefer=dom_prefer) actual_valid=best["valid"] return{ "ok":True, "run_init":best["run_init"], "valid":actual_valid, "lead":"t0024", "melt_urls":best["melt_urls"], "snowpack_ts":actual_valid, "snowpack_urls":snowpack_urls or[], "snowpack_missing":not bool(snowpack_urls), "source_type":best["source_type"], } def _pick_forecast_for_days(days:int,dom_prefer:str="zz")->dict: if days not in(2,3):raise HTTPException(status_code=400,detail="days must be 2 or 3") return{"ok":True,"run_init":run_init,"valid":valid,"lead":"t0024","melt_urls":melt_urls,"snowpack_ts":valid,"snowpack_urls":snowpack_urls or[],"snowpack_missing":not bool(snowpack_urls)} now=datetime.utcnow().replace(tzinfo=timezone.utc);valid=_valid_ts_for_days(days,now_utc=now) sel=_pick_forecast_for_valid(valid,dom_prefer=(dom_prefer or"zz").lower()) if sel.get("ok"):sel["days"]=days;sel["now_utc"]=now.isoformat() return sel def _find_melt_11044_ttna(valid:str,dom_prefer:str="zz")->list[str]: """Find TTNA (analysis) melt files for a given valid time.""" valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low:continue if"__t0024ttnats"not in low:continue if want_ts not in low:continue if not low.endswith(".nc"):continue out.append(_as_url(base)) return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) def _find_snowpack_for_valid(valid:str,dom_prefer:str="zz")->list[str]: """Find snowpack files for a given valid time, checking both TTNA and forecast patterns.""" valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() html=_collab_dirlist();names=_hrefs(html);want_ts=f"ts{valid}";out=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11034"not in low and"ssmv11036"not in low:continue if"hp001"not in low:continue if want_ts not in low:continue if not(low.endswith(".grib2")or low.endswith(".nc")):continue out.append(_as_url(base)) if out:return sorted(set(out),key=lambda u:(0 if _dom_from_url(u)==dom_prefer else _dom_pri(_dom_from_url(u))+10)) # Fallback: try finding via the forecast TT approach run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom_prefer) if run_init:return _find_fcst_11034_grib2_tt_ts(run_init,valid,dom_prefer=dom_prefer) return[] def _nsidc_month_folder(d)->str:return f"{d.month:02d}_{calendar.month_abbr[d.month].title()}" def _nsidc_url(date,masked:bool)->str:return f"{NSIDC_BASE}/{date.year}/{_nsidc_month_folder(date)}/{_yyyymmdd(date)}/SNODAS_{'masked'if masked else'unmasked'}_{_yyyymmdd(date)}.tar" def _nsidc_exists(date)->tuple[bool,str]: hdr={"User-Agent":"snodas/1.3"} for masked in(False,True): url=_nsidc_url(date,masked) try: r=requests.head(url,timeout=20,allow_redirects=True,headers=hdr) if r.status_code==200:return True,url if r.status_code in(403,405): rg=_retry_get(url,stream=True,timeout=(10,60),tries=2,headers=hdr);rg.close() if rg.status_code==200:return True,url except:pass return False,_nsidc_url(date,masked=False) @lru_cache(maxsize=512) def _nsidc_exists_cached(dstr:str)->tuple[bool,str]:return _nsidc_exists(datetime.strptime(dstr,"%Y%m%d").date()) def _http_get(url:str,timeout:int=180)->bytes: r=_retry_get(url,timeout=(15,timeout),tries=3,headers={"User-Agent":"snodas/1.3"});r.raise_for_status();return r.content @lru_cache(maxsize=1) def _resolve_latest_nsidc_day(days_back:int=120)->Optional[str]: utc_today=datetime.utcnow().date() for i in range(days_back+1): d=utc_today-timedelta(days=i);ok,_=_nsidc_exists(d) if ok:return _yyyymmdd(d) return None def _download_day_tar(date)->Path: dstr=_yyyymmdd(date);out=CACHE/f"SNODAS_{dstr}.tar" if out.exists()and out.stat().st_size>0:return out ok,url=_nsidc_exists_cached(dstr) if not ok:raise HTTPException(status_code=404,detail=f"NSIDC not available for {dstr}") with _lock(f"nsidc_tar_{dstr}"): if out.exists()and out.stat().st_size>0:return out out.write_bytes(_http_get(url,timeout=360)) return out _KV_RE=re.compile(r"^\s*(?P[^:=#;]+?)\s*(?:[:=]\s*|\s+)(?P.+?)\s*$") def _parse_hdr(path:Path)->tuple[dict,str]: txt=path.read_text(encoding="utf-8",errors="ignore");meta:dict[str,str]={} for raw in txt.splitlines(): line=raw.strip() if not line or line.startswith(("#",";")):continue m=_KV_RE.match(line) if not m:continue k=m.group("k").strip().lower();v=m.group("v").strip().split("#",1)[0].split(";",1)[0].strip() meta[k]=v;meta[re.sub(r"[^a-z0-9]+","",k)]=v return meta,txt def _dtype_from_hdr(nbits:int,pixeltype:str,byteorder:str)->np.dtype: pixeltype=(pixeltype or"").upper();byteorder=(byteorder or"I").upper() endian="<"if byteorder in("I","LSBFIRST","LITTLEENDIAN")else">" if pixeltype=="FLOAT":return np.dtype(endian+("f4"if nbits<=32 else"f8")) if pixeltype in("UNSIGNEDINT","UNSIGNED","UINTEGER"):return np.dtype(endian+("u1"if nbits<=8 else"u2"if nbits<=16 else"u4")) if pixeltype in("SIGNEDINT","SIGNED","INTEGER"):return np.dtype(endian+("i1"if nbits<=8 else"i2"if nbits<=16 else"i4")) return np.dtype(endian+"f4") def _bil_to_geotiff(hdr:Path,bil:Path,out_tif:Path)->Path: meta,raw=_parse_hdr(hdr);raw_l=raw.lower() def gi(*keys,default=None): for k in keys: kk=re.sub(r"[^a-z0-9]+","",k.lower()) for cand in(k.lower(),kk): v=meta.get(cand) if v is None:continue try:return int(float(str(v).strip())) except:pass return default def gf(*keys,default=None): for k in keys: kk=re.sub(r"[^a-z0-9]+","",k.lower()) for cand in(k.lower(),kk): v=meta.get(cand) if v is None:continue try:return float(str(v).strip()) except:pass return default def re_float(pat:str)->Optional[float]: m=re.search(pat,raw,re.I);return float(m.group(1))if m else None ncols=gi("ncols","samples","nsamples","columns") nrows=gi("nrows","lines","nlines","rows") if ncols is None:m=re.search(r"Number of columns:\s*([0-9]+)",raw,re.I);ncols=int(m.group(1))if m else None if nrows is None:m=re.search(r"Number of rows:\s*([0-9]+)",raw,re.I);nrows=int(m.group(1))if m else None if ncols is None or nrows is None:raise HTTPException(status_code=502,detail="BIL header missing ncols/nrows") bands=gi("bands","nbands","number of bands",default=1)or 1 nbits=gi("nbits","bitspersample","bits per sample",default=32)or 32 dbpp=gi("data bytes per pixel",default=None) if dbpp:nbits=int(dbpp)*8 pixeltype=meta.get("pixeltype")or meta.get("datatype")or"FLOAT" byteorder=meta.get("byteorder")or"I" slope=gf("data slope","slope",default=1.0)or 1.0 intercept=gf("data intercept","intercept",default=0.0)or 0.0 nodata_raw=gf("no data value","nodata","nodata_value",default=None) ulx=gf("ulxmap","upper left x","ulx",default=None) uly=gf("ulymap","upper left y","uly",default=None) xll=gf("xllcorner","xllcenter","lower left x","xll",default=None) yll=gf("yllcorner","yllcenter","lower left y","yll",default=None) xres=gf("x-axis resolution","x axis resolution","xdim","x dim","cellsize",default=None) yres=gf("y-axis resolution","y axis resolution","ydim","y dim","cellsize",default=None) if xres is None:xres=re_float(r"X[- ]axis resolution:\s*([0-9.eE+-]+)") if yres is None:yres=re_float(r"Y[- ]axis resolution:\s*([0-9.eE+-]+)") xmin=gf("minimum x-axis coordinate","min x","minx",default=None) xmax=gf("maximum x-axis coordinate","max x","maxx",default=None) ymin=gf("minimum y-axis coordinate","min y","miny",default=None) ymax=gf("maximum y-axis coordinate","max y","maxy","maximumy",default=None) if xmin is None:xmin=re_float(r"Minimum x[- ]axis coordinate:\s*([0-9.eE+-]+)") if xmax is None:xmax=re_float(r"Maximum x[- ]axis coordinate:\s*([0-9.eE+-]+)") if ymin is None:ymin=re_float(r"Minimum y[- ]axis coordinate:\s*([0-9.eE+-]+)") if ymax is None:ymax=re_float(r"Maximum y[- ]axis coordinate:\s*([0-9.eE+-]+)") if xres is None and xmin is not None and xmax is not None:xres=(xmax-xmin)/float(ncols) if yres is None and ymin is not None and ymax is not None:yres=(ymax-ymin)/float(nrows) if ulx is None and xll is not None:ulx=float(xll) if uly is None and yll is not None and yres is not None:uly=float(yll)+abs(float(yres))*float(nrows) if ulx is None and xmin is not None:ulx=float(xmin) if uly is None and ymax is not None:uly=float(ymax) if ulx is None or uly is None or xres is None or yres is None:raise HTTPException(status_code=502,detail="BIL header missing geo params") units_line=(meta.get("data units")or meta.get("dataunits")or"").strip();denom=None m=re.search(r"Data units:\s*Meters\s*/\s*([0-9.]+)",raw,re.I) if m: try:denom=float(m.group(1)) except:denom=None if denom is None and units_line and"/"in units_line: try:denom=float(units_line.split("/",1)[1].strip()) except:denom=None is_nohrsc_gisrs="nohrsc gis/rs raster file"in raw_l file_bytes=bil.stat().st_size;bytes_per_sample=max(1,int(nbits)//8) expected_samples=int(nrows)*int(ncols)*max(1,int(bands));expected_bytes=expected_samples*bytes_per_sample if is_nohrsc_gisrs:dtp=np.dtype(">i2");expected_bytes=expected_samples*2 else:dtp=_dtype_from_hdr(int(nbits),str(pixeltype),str(byteorder)) if file_bytes!=expected_bytes: inferred=None for bps in(1,2,4): if file_bytes==expected_samples*bps:inferred=bps;break if inferred is None:raise HTTPException(status_code=502,detail=f"BIL byte mismatch file={file_bytes} expected={expected_bytes}") if is_nohrsc_gisrs and inferred!=2:raise HTTPException(status_code=502,detail="NOHRSC GIS/RS expected 2 bytes/sample") if not is_nohrsc_gisrs: if inferred==2:dtp=np.dtype("xr.DataArray: da=da.astype("float32");da=da.where(np.isfinite(da)).where(da>-9990.0) da=xr.where(da<0.0,0.0,da);da=xr.where(da>100000.0,np.nan,da);return da def _open_grib2(path:Path)->xr.DataArray: idx=(CFGRIB_DIR/(path.name+".idx")).as_posix();ds=xr.open_dataset(path.as_posix(),engine="cfgrib",backend_kwargs={"indexpath":idx}) try: da=_pick_grib_var(ds) if"step"in da.dims:da=da.isel(step=0) attrs=dict(da.attrs or{}) for k in("_FillValue","missingValue","missing_value"): if k in attrs: try:da=da.where(da!=float(attrs[k])) except:pass da=da.where(np.isfinite(da)).where(da<1.0e20).astype("float32") units=str(attrs.get("units","")).strip().lower() if units in("m","meter","metre","m of water equivalent","meters"):da=da*1000.0 dims_l={d.lower():d for d in da.dims};rename={} if"longitude"in dims_l:rename[dims_l["longitude"]]="x" if"latitude"in dims_l:rename[dims_l["latitude"]]="y" if rename:da=da.rename(rename) if"x"in da.coords: try: x=np.asarray(da["x"].values) if x.ndim==1 and x.size>2: if float(np.nanmin(x))>=0.0 and float(np.nanmax(x))>180.0:da=da.assign_coords(x=_lon0360_to_180(x));da=da.sortby("x") except:pass if"x"in da.dims and"y"in da.dims: try:da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False) except:pass try:da=da.rio.write_crs("EPSG:4326",inplace=False) except:pass try:da=_ensure_xy_transform(da) except:pass return _sanitize_mm(da) finally: try:ds.close() except:pass def _open_bil(hdr:Path,dat:Path)->xr.DataArray: tmp_tif=dat.with_suffix(".tif") if not tmp_tif.exists()or tmp_tif.stat().st_size==0:_bil_to_geotiff(hdr,dat,tmp_tif) da=xr.open_dataarray(tmp_tif.as_posix(),engine="rasterio").astype("float32") try: if da.rio.crs is None:da=da.rio.write_crs("EPSG:4326",inplace=False) da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False) except:pass return _sanitize_mm(da) def _pick_netcdf_raster_var(ds:xr.Dataset)->xr.DataArray: bad=("bounds","bnds","crs","grid_mapping","lat_bounds","lon_bounds","time_bounds") bad2=("mask","qc","quality","flag","prob","probability","index","idx") good=("melt","snowmelt","smelt","we","water","equiv","mm","kg","swe") best=None;best_score=-10**9 for name,da in ds.data_vars.items(): n=name.lower() if any(t in n for t in bad)or any(t in n for t in bad2)or da.ndim<2:continue score=300 if any(t in n for t in good)else 0 score+=200 if da.ndim==2 else 50 dims=[d.lower()for d in da.dims] if("y"in dims and"x"in dims)or("south_north"in dims and"west_east"in dims):score+=200 if("lat"in dims and"lon"in dims)or("latitude"in dims and"longitude"in dims):score+=150 try:score+=int((da.shape[-1]*da.shape[-2])//50000) except:pass if score>best_score:best_score=score;best=da if best is None:raise HTTPException(status_code=502,detail=f"No suitable raster variable in NetCDF. vars={list(ds.data_vars)[:50]}") return best def _open_netcdf(path:Path)->xr.DataArray: last=None;ds=None for eng in("netcdf4","h5netcdf","scipy"): try:ds=xr.open_dataset(path.as_posix(),engine=eng);break except Exception as e:last=e;ds=None if ds is None:raise HTTPException(status_code=502,detail=f"NetCDF backend unavailable/failed. last_err={last!r}") try: da=_pick_netcdf_raster_var(ds);dims_l={d.lower():d for d in da.dims};rename={} if"south_north"in dims_l and"west_east"in dims_l:rename[dims_l["south_north"]]="y";rename[dims_l["west_east"]]="x" elif"lat"in dims_l and"lon"in dims_l:rename[dims_l["lat"]]="y";rename[dims_l["lon"]]="x" elif"latitude"in dims_l and"longitude"in dims_l:rename[dims_l["latitude"]]="y";rename[dims_l["longitude"]]="x" elif"rlat"in dims_l and"rlon"in dims_l:rename[dims_l["rlat"]]="y";rename[dims_l["rlon"]]="x" if rename:da=da.rename(rename) spatial={"x","y"} for d in list(da.dims): if d not in spatial:da=da.isel({d:int(da.sizes.get(d,1)or 1)-1}) da=da.astype("float32");units=str(da.attrs.get("units","")).strip().lower() if units in("m","meter","metre","m of water equivalent","meters"):da=da*1000.0 try: lon=da.coords.get("longitude")or da.coords.get("lon");lat=da.coords.get("latitude")or da.coords.get("lat") if lon is not None and lat is not None: if lon.ndim==2 and lat.ndim==2: h,w=da.shape[-2],da.shape[-1] tfm=from_bounds(float(np.nanmin(lon.values)),float(np.nanmin(lat.values)),float(np.nanmax(lon.values)),float(np.nanmax(lat.values)),w,h) da=da.rio.write_crs("EPSG:4326",inplace=False).rio.write_transform(tfm,inplace=False) except:pass # Set spatial dims FIRST try: if"x"in da.dims and"y"in da.dims:da=da.rio.set_spatial_dims(x_dim="x",y_dim="y",inplace=False) except:pass try:da=_ensure_xy_transform(da) except:pass # Force CRS — drop any conflicting spatial_ref first try: if da.rio.crs is None: if"spatial_ref"in da.coords:da=da.drop_vars("spatial_ref") da=da.rio.write_crs("EPSG:4326",inplace=False) except:pass try: if da.rio.crs is None: da=da.rio.write_crs("EPSG:4326",inplace=False) except Exception as e: _log("WARN",f"[open_netcdf] CRS write failed: {e!r}") return _sanitize_mm(da) finally: try:ds.close() except:pass def _open_grid(desc:dict)->xr.DataArray: t=desc.get("type") if t=="grib2":return _open_grib2(desc["path"]) if t=="bil":return _open_bil(desc["hdr"],desc["dat"]) if t=="netcdf":return _open_netcdf(desc["path"]) raise HTTPException(status_code=500,detail="unknown grid type") def _is_probably_html(blob:bytes)->bool:head=blob[:512].lstrip();return head.startswith(b" bool: if cancel_check and cancel_check(): return False tok, repo = _hf_cfg() if not tok or not repo: return False rel = (rel or "").lstrip("/") if not rel: return False dest = (CACHE / rel).resolve() dest.parent.mkdir(parents=True, exist_ok=True) result = {"tmp": None, "error": None} def do_download(): try: result["tmp"] = hf_hub_download( repo_id=repo, repo_type="dataset", filename=rel, token=tok, local_dir=str(CACHE) ) except Exception as e: result["error"] = e t = threading.Thread(target=do_download, daemon=True) t.start() poll_interval = 0.5 elapsed = 0.0 while t.is_alive() and elapsed < float(timeout_sec): if cancel_check and cancel_check(): _log("DEBUG", f"[hf_pull] cancelled (gen changed) for {rel}") return False t.join(timeout=poll_interval) elapsed += poll_interval if t.is_alive(): _log("DEBUG", f"[hf_pull] timeout after {timeout_sec}s for {rel}") return False if cancel_check and cancel_check(): return False if result["error"] is not None: e = result["error"] err_str = str(e).lower() if "404" in err_str or "not found" in err_str: return False if "timeout" not in err_str and "connection" not in err_str: _log("WARN", f"[hf_pull] error for {rel}: {e!r}") return False try: tmp = result["tmp"] if tmp is None: return False p = Path(tmp) if not p.exists(): return False b = p.read_bytes() if _is_lfs_pointer_bytes(b) or len(b) < 200: try: p.unlink(missing_ok=True) dest.unlink(missing_ok=True) except: pass return False if p.resolve() != dest: try: dest.write_bytes(b) except: pass return True except Exception as e: _log("WARN", f"[hf_pull] validation failed for {rel}: {e!r}") return False def _is_netcdf_bytes(blob:bytes)->bool: if len(blob)>=4 and blob[:3]==b"CDF"and blob[3]in(1,2):return True return len(blob)>=8 and blob[:8]==b"\x89HDF\r\n\x1a\n" _HF_OPS_LOCK=threading.Lock() _HF_PENDING:dict[str,CommitOperationAdd]={} def _hf_enqueue_files(paths:list[Path])->int: tok,repo=_hf_cfg() if not tok or not repo:return 0 added=0 with _HF_OPS_LOCK: for p in paths: try: if(not p.exists())or p.stat().st_size<=0:continue rel=p.relative_to(CACHE).as_posix() if not _is_cache_file_we_manage(rel):continue _HF_PENDING[rel]=CommitOperationAdd(path_in_repo=rel,path_or_fileobj=p.as_posix());added+=1 except:continue return added def _hf_flush_pending(message:str)->None: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return with _HF_OPS_LOCK:ops=list(_HF_PENDING.values());_HF_PENDING.clear() if not ops:return for attempt in range(6): try: api.create_commit(repo_id=repo,repo_type="dataset",operations=ops,commit_message=message) _log("INFO",f"[hf_cache] flush OK files={len(ops)}");return except Exception as e: wait=(1.8**attempt)+random.uniform(0,1.0);_log("INFO",f"[hf_cache] flush retry attempt={attempt+1} wait={wait:.1f} err={e!r}");time.sleep(wait) _log("INFO",f"[hf_cache] flush FAILED files={len(ops)}") def _download_url_to_grid(url:str,workdir:Path)->tuple[dict,str]: hdrs={"User-Agent":"snodas/1.3"};name=Path(url).name.lower();last_err=None for attempt in range(6): r=_retry_get(url,timeout=(10,360),tries=1,headers=hdrs);sc=int(r.status_code) if sc in(429,502,503,504): last_err=f"HTTP {sc}" if attempt<5: ra=r.headers.get("Retry-After") try:wait=float(ra)if ra else None except:wait=None time.sleep((wait if wait is not None else(1.6**attempt))+random.uniform(0,0.8));continue raise HTTPException(status_code=502,detail=f"{url} -> {last_err} after retries") if sc not in(200,206):raise HTTPException(status_code=404,detail=f"{url} -> HTTP {sc}") blob=r.content;ctype=(r.headers.get("Content-Type")or"").lower() if _is_probably_html(blob)or"text/html"in ctype: raise HTTPException(status_code=502,detail=f"Upstream returned HTML. url={url}") if len(blob)>=2 and blob[:2]==b"\x1f\x8b": try:blob=gzip.decompress(blob) except Exception as e:raise HTTPException(status_code=502,detail=f"gzip decompress failed: {e!r}") if blob.startswith(b"GRIB"):p=workdir/"data.grib2";p.write_bytes(blob);return{"type":"grib2","path":p},name if _is_netcdf_bytes(blob):p=workdir/"data.nc";p.write_bytes(blob);return{"type":"netcdf","path":p},name looks_tar=(len(blob)%512==0)or(b"ustar"in blob[:4096]) if looks_tar: try: with tarfile.open(fileobj=io.BytesIO(blob),mode="r:*")as tf: grib_member=hdr_member=dat_member=nc_member=None for m in tf.getmembers(): n=m.name.lower() if n.endswith(".grib2")and grib_member is None:grib_member=m elif n.endswith(".nc")and nc_member is None:nc_member=m elif n.endswith(".hdr")and hdr_member is None:hdr_member=m elif(n.endswith(".dat")or n.endswith(".bil"))and dat_member is None:dat_member=m if grib_member is not None: p=workdir/"data.grib2";f=tf.extractfile(grib_member) if f is None:raise HTTPException(status_code=502,detail="TAR member unreadable") p.write_bytes(f.read(grib_member.size));return{"type":"grib2","path":p},name if nc_member is not None: p=workdir/"data.nc";f=tf.extractfile(nc_member) if f is None:raise HTTPException(status_code=502,detail="TAR member unreadable") p.write_bytes(f.read(nc_member.size));return{"type":"netcdf","path":p},name if hdr_member is not None and dat_member is not None: hdr=workdir/"data.hdr";dat=workdir/"data.bil" fh=tf.extractfile(hdr_member);fd=tf.extractfile(dat_member) if fh is None or fd is None:raise HTTPException(status_code=502,detail="TAR member unreadable") hdr.write_bytes(fh.read(hdr_member.size));dat.write_bytes(fd.read(dat_member.size)) return{"type":"bil","hdr":hdr,"dat":dat},name except HTTPException:raise except Exception as e:raise HTTPException(status_code=502,detail=f"TAR parse failed: {e!r}") raise HTTPException(status_code=502,detail=f"payload unrecognized. url={url}") raise HTTPException(status_code=504,detail=f"upstream retries exhausted. url={url} last_err={last_err}") def _da_to_cog(da:xr.DataArray,out:Path)->Path: out.parent.mkdir(parents=True,exist_ok=True) try:_=da.rio.transform();ok_transform=True except:ok_transform=False if not ok_transform and"x"in da.coords and"y"in da.coords: try:da=_ensure_xy_transform(da) except:pass nodata=np.float32(-9999.0) try:crs=da.rio.crs except:crs=None if crs is None: try: if"spatial_ref"in da.coords:da=da.drop_vars("spatial_ref") da=da.rio.write_crs("EPSG:4326",inplace=False) except:pass try:crs=da.rio.crs except:crs=None if crs is None:da=da.rio.write_crs("EPSG:4326",inplace=False) da=da.astype("float32").where(np.isfinite(da)).where(da>-9990.0);da=da.where(~da.isnull(),nodata) with tempfile.NamedTemporaryFile(suffix=".tif",dir=str(out.parent),delete=False)as tf:tmp=Path(tf.name) try: da.rio.to_raster(tmp.as_posix(),dtype="float32",compress="DEFLATE",nodata=float(nodata),tiled=True) # Force CRS into the written file if rio.to_raster didn't set it with rasterio.open(tmp.as_posix(),"r")as check: if check.crs is None: _log("INFO",f"[da_to_cog] patching missing CRS for {out.name}") data=check.read();prof=check.profile.copy() prof["crs"]=CRS.from_epsg(4326) tmp2=tmp.with_suffix(".fix.tif") with rasterio.open(tmp2.as_posix(),"w",**prof)as ds2: ds2.write(data) tmp2.replace(tmp) _atomic_cog_translate(tmp,out) finally:tmp.unlink(missing_ok=True) return out def _extract_var(tar_path:Path,code:str)->tuple[Path,Path]: outdir=CACHE/f"{tar_path.stem}_x";outdir.mkdir(exist_ok=True) hdr_path:Optional[Path]=None;dat_path:Optional[Path]=None with tarfile.open(tar_path)as tf: for m in tf.getmembers(): name=Path(m.name).name;low=name.lower() if code not in low:continue if low.endswith(".hdr"): outp=outdir/name if not outp.exists(): f=tf.extractfile(m) if f is not None:outp.write_bytes(f.read()) hdr_path=outp elif low.endswith(".dat")or low.endswith(".bil"): outp=outdir/name if not outp.exists(): f=tf.extractfile(m) if f is not None:outp.write_bytes(f.read()) dat_path=outp if not hdr_path or not dat_path:raise HTTPException(status_code=500,detail=f"Missing .hdr or .dat for var {code}") return hdr_path,dat_path def build_24h_cog(date)->Path: dstr=_yyyymmdd(date);out=CACHE/f"melt24h_{dstr}_cog.tif" if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out with _lock(f"melt24h_{dstr}"): if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out tar=_download_day_tar(date);hdr,dat=_extract_var(tar,"1044");da=_open_bil(hdr,dat);p=_da_to_cog(da,out) try:_hf_enqueue_files([p]) except:pass return p def build_72h_cog(end_date)->Path: dstr=_yyyymmdd(end_date);out=CACHE/f"melt72h_end_{dstr}_cog.tif" if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out with _lock(f"melt72h_{dstr}"): if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out cogs=[build_24h_cog(end_date-timedelta(days=2)),build_24h_cog(end_date-timedelta(days=1)),build_24h_cog(end_date)] with rasterio.open(cogs[0])as base: acc=base.read(1).astype("float32") for c in cogs[1:]: with rasterio.open(c)as src: arr=src.read(1).astype("float32") if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height): tmp=np.full_like(acc,np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest) arr=tmp m=np.isfinite(arr);acc[m]=np.nan_to_num(acc[m],nan=0.0)+arr[m] tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out);tmp_tif.unlink(missing_ok=True) try:_hf_enqueue_files([out]) except:pass return out def _forecast_dir()->Path:d=CACHE/"forecast";d.mkdir(parents=True,exist_ok=True);return d def _forecast_melt_cog(run_init:str,valid:str,lead:str,dom:str)->Path: ri=run_init if run_init!="TTNA"else"TTNA" return _forecast_dir()/f"fcst_melt_{lead}_{ri}_{valid}_{dom}_cog.tif" def _forecast_snow_cog(ts:str,dom:str,code:str)->Path:return _forecast_dir()/f"fcst_snow_{code}_{ts}_{dom}_cog.tif" def _forecast_melt72h_end_cog(valid:str,dom_prefer:str="zz")->Path: valid=_norm_ts(valid);dom_prefer=(dom_prefer or"zz").lower() out=_forecast_dir()/f"fcst_melt72h_end_{valid}_{dom_prefer}_cog.tif" if out.exists()and out.stat().st_size>0:return out _hf_try_pull_file(out.relative_to(CACHE).as_posix()) if out.exists()and out.stat().st_size>0:return out with _lock(f"fcst_melt72h_{valid}_{dom_prefer}"): if out.exists()and out.stat().st_size>0:return out v0=valid;v1=_shift_ts(v0,-24);v2=_shift_ts(v0,-48) sels=[_pick_forecast_for_valid(v,dom_prefer=dom_prefer)for v in(v2,v1,v0)] if not all(s.get("ok")for s in sels):raise HTTPException(status_code=503,detail={"error":"forecast_selection_unavailable_for_72h","need":[v2,v1,v0],"sels":sels}) day_cogs:list[Path]=[] for s,v in zip(sels,(v2,v1,v0)): melt_cogs=_build_forecast_melt_cogs(s["run_init"],v,"t0024",s["melt_urls"]) pick=None for p in melt_cogs: if _cog_dom(p)==dom_prefer:pick=p;break if pick is None:pick=sorted(melt_cogs,key=lambda p:_dom_pri(_cog_dom(p)))[0] day_cogs.append(pick) with rasterio.open(day_cogs[0])as base: acc=base.read(1).astype("float32") for c in day_cogs[1:]: with rasterio.open(c)as src: arr=src.read(1).astype("float32") if(src.transform!=base.transform)or(src.crs!=base.crs)or(src.width!=base.width)or(src.height!=base.height): tmp=np.full_like(acc,np.nan,dtype="float32") reproject(arr,tmp,src_transform=src.transform,src_crs=src.crs,dst_transform=base.transform,dst_crs=base.crs,resampling=Resampling.nearest) arr=tmp m=np.isfinite(arr)&(arr>-9990.0);acc[m]=np.nan_to_num(acc[m],nan=0.0)+arr[m] tmp_tif=out.with_suffix(".tmp.tif");prof=base.profile.copy();prof.update({"compress":"DEFLATE","tiled":True}) with rasterio.open(tmp_tif,"w",**prof)as ds:ds.write(acc,1) _atomic_cog_translate(tmp_tif,out) try:_hf_enqueue_files([out]) except:pass tmp_tif.unlink(missing_ok=True);_log("INFO",f"[build_forecast] melt72h OK valid={valid} dom_prefer={dom_prefer} -> {out.name}") return out def _pick_grib_var(ds:xr.Dataset)->xr.DataArray: best=None;best_score=-10**9 for name,da in ds.data_vars.items(): a=dict(da.attrs or{});units=str(a.get("units","")).lower() txt=" ".join([name.lower(),str(a.get("GRIB_name","")).lower(),str(a.get("standard_name","")).lower(),str(a.get("long_name","")).lower()]) score=0 if"snow"in txt or"swe"in txt or"water equivalent"in txt:score+=500 if"depth"in txt:score+=250 if units in("m","meter","metre","mm"):score+=120 if units in("k","kelvin","degc","c","degree_celsius"):score-=400 if da.ndim>=2:score+=50 if score>best_score:best_score=score;best=da if best is None:return ds[next(iter(ds.data_vars))] return best def _ensure_xy_transform(da:xr.DataArray)->xr.DataArray: if not("x"in da.coords and"y"in da.coords):return da try: x=np.asarray(da["x"].values);y=np.asarray(da["y"].values) if x.ndim!=1 or y.ndim!=1 or x.size<2 or y.size<2:return da dx=float(np.nanmedian(np.diff(x)));dy=float(np.nanmedian(np.diff(y))) if not np.isfinite(dx)or not np.isfinite(dy)or dx==0.0 or dy==0.0:return da left=float(x[0])-dx/2.0;top=float(y[0])+(-dy)/2.0 if dy<0 else float(y[0])-dy/2.0 return da.rio.write_transform(Affine(dx,0.0,left,0.0,dy,top),inplace=False) except:return da def _build_forecast_melt_cogs(run_init:str,valid:str,lead:str,melt_urls:list[str])->list[Path]: run_init_safe=run_init if run_init!="TTNA"else"TTNA" run_init_norm=_norm_ts(run_init)if run_init!="TTNA"else"TTNA" valid=_norm_ts(valid);lead=(lead or"").lower() if not melt_urls:raise HTTPException(status_code=502,detail=f"No melt URLs provided for run_init={run_init} valid={valid} lead={lead}") dom2urls:dict[str,list[str]]={} for u in melt_urls:dom2urls.setdefault(_dom_from_url(u),[]).append(u) built:list[Path]=[];errors:list[str]=[] with _lock(f"melt_{lead}_{run_init_safe}_{valid}"): for dom in sorted(dom2urls.keys(),key=_dom_pri): out=_forecast_melt_cog(run_init_safe,valid,lead,dom) if out.exists()and out.stat().st_size>0:built.append(out);continue last_err=None for url in dom2urls[dom]: low=Path(url).name.lower() if lead and lead not in low:continue with tempfile.TemporaryDirectory()as td: try: desc,_=_download_url_to_grid(url,Path(td));da=_open_grid(desc);_da_to_cog(da,out) try:_hf_enqueue_files([out]) except:pass if out.exists()and out.stat().st_size>0: _log("INFO",f"[build_forecast] melt OK run_init={run_init_safe} valid={valid} lead={lead} dom={dom} -> {out.name}") built.append(out);break last_err="empty COG after build" except HTTPException as e:last_err=f"{e.status_code} {e.detail}" except Exception as e:last_err=repr(e) if out not in built:errors.append(f"dom={dom} last_err={last_err}");_log("WARN",f"[build_forecast] melt FAIL run_init={run_init_safe} valid={valid} lead={lead} dom={dom}") if not built:raise HTTPException(status_code=502,detail={"error":"No usable MELT COGs","run_init":run_init,"valid":valid,"lead":lead,"errors":errors[:5]}) return built def _build_forecast_snowpack_cogs(ts:str,urls:list[str])->list[Path]: ts=_norm_ts(ts)if ts and _TS10.fullmatch(ts)else ts dom2urls:dict[str,list[str]]={} for u in urls:dom2urls.setdefault(_dom_from_url(u),[]).append(u) built:list[Path]=[] with _lock(f"snow_{ts}"): for dom in sorted(dom2urls.keys(),key=_dom_pri): last_err=None for url in dom2urls[dom]: low=Path(url).name.lower() code="ssmv11036"if"ssmv11036"in low else"ssmv11034"if"ssmv11034"in low else"ssmv11005"if"ssmv11005"in low else"ssmv_unknown" out=_forecast_snow_cog(ts or"na",dom,code) if out.exists()and out.stat().st_size>0:built.append(out);break with tempfile.TemporaryDirectory()as td: try: desc,_=_download_url_to_grid(url,Path(td));da=_open_grid(desc);_da_to_cog(da,out) try:_hf_enqueue_files([out]) except:pass if out.exists()and out.stat().st_size>0: _log("INFO",f"[build_forecast] snow OK ts={ts} dom={dom} code={code} -> {out.name}") built.append(out);break last_err="empty COG" except HTTPException as e:last_err=f"{e.status_code} {e.detail}" except Exception as e:last_err=repr(e) if not any(f"_{dom}_"in p.name for p in built):_log("INFO",f"[build_forecast] snow FAIL ts={ts} dom={dom} last_err={last_err}") if not built:raise HTTPException(status_code=502,detail=f"No usable snow field for {ts}") return built def _cog_dom(p:Path)->str:parts=p.name.split("_");return parts[-2]if len(parts)>=2 else"xx" def _union_tile(cogs:list[Path],z:int,x:int,y:int,*,label:str,valid_pred:Optional[Callable[[np.ndarray,np.ndarray],np.ndarray]]=None)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]: if not cogs:return None,None,f"{label}_no_cogs" out_arr:Optional[np.ndarray]=None;out_valid:Optional[np.ndarray]=None;used:list[str]=[] for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))): try: with COGReader(cog.as_posix())as r:data,mask=r.tile(x,y,z,tilesize=256,resampling_method="nearest") arr=data[0].astype("float32");m=mask[0];valid=(m>0)&np.isfinite(arr)&(arr>-9990.0) if valid_pred is not None: try:valid=valid&valid_pred(arr,m) except:pass if out_arr is None:out_arr=arr.copy();out_valid=valid.copy() else: fill=valid&(~out_valid) if fill.any():out_arr[fill]=arr[fill];out_valid[fill]=True used.append(cog.name) except TileOutsideBounds:continue except:continue if out_arr is None or out_valid is None:return None,None,f"{label}_unavailable_all_domains" return out_arr,out_valid,f"{label}_union:"+",".join(used[:4])+("..."if len(used)>4 else"") def _read_mask_tile(ts:str,urls:list[str],z:int,x:int,y:int)->tuple[Optional[np.ndarray],Optional[np.ndarray],str]: ts=_norm_ts(ts)if ts else ts if not urls:return None,None,"mask_no_urls" try:cogs=_build_forecast_snowpack_cogs(ts or"0000000000",urls) except HTTPException as e:return None,None,f"mask_build_fail:{e.status_code}:{e.detail}" except Exception as e:return None,None,f"mask_build_fail:{e!r}" return _union_tile(cogs,z,x,y,label="mask") def _cog_point_data_mask(r:COGReader,lon:float,lat:float)->tuple[np.ndarray,np.ndarray]: out=r.point(lon,lat) if isinstance(out,(tuple,list))and len(out)==2:return np.asarray(out[0]),np.asarray(out[1]) data=getattr(out,"data",None);mask=getattr(out,"mask",None) if data is None or mask is None:raise ValueError(f"Unexpected COGReader.point() return type: {type(out)!r}") return np.asarray(data),np.asarray(mask) INCH_TO_MM=25.4 BIN_EDGES_IN=np.array([0.01,0.02,0.04,0.08,0.12,0.16,0.20,0.39,0.59,0.79,0.98,1.4,2.0,3.0,4.0,5.0,9999.0],dtype="float32") BIN_EDGES_72H_IN=np.array([0.01,0.04,0.08,0.16,0.24,0.31,0.39,0.79,1.4,2.0,3.0,3.9,5.0,6.5,8.0,9.8,9999.0],dtype="float32") MELT_COLORS=np.array([[0xDC,0xDC,0xDC],[0x99,0x00,0x00],[0xCC,0x33,0x33],[0xFF,0x66,0x66],[0xCC,0xCC,0x66],[0xFF,0x8C,0x00],[0xFF,0xFF,0x66],[0x99,0xFF,0x99],[0x00,0xFF,0x00],[0x66,0xCC,0x66],[0x66,0x66,0xFF],[0x33,0x33,0xCC],[0x00,0x00,0x99],[0x99,0x00,0x99],[0x4B,0x00,0x82],[0x2F,0x00,0x4F],[0x00,0x00,0x00]],dtype="uint8") @lru_cache(maxsize=32) def _corridor_tiles_set_for_z(z:int)->set[tuple[int,int]]: pts=_load_track_points() if not pts:return set() tiles=_tiles_within_miles(pts,int(z),WARM_MILES,cap=WARM_CAP_PER_ZOOM) return set((x,y)for(_,x,y)in tiles) def _is_corridor_tile(z:int,x:int,y:int)->bool:return(int(x),int(y))in _corridor_tiles_set_for_z(int(z)) def _effective_request_tile(z:int,x:int,y:int)->tuple[int,int,int,str]: z=int(z);x=int(x);y=int(y) if not _tile_allowed(z,x,y):return z,x,y,"none" if _is_corridor_tile(z,x,y):return z,x,y,"hi" z_lr=min(int(LOWRES_Z),z) if z<=z_lr:return z,x,y,"lo" shift=z-z_lr;return z_lr,(x>>shift),(y>>shift),"lo" def _miles_to_meters(mi:float)->float:return float(mi)*1609.344 def _meters_per_pixel_webmercator(z:int,lat:float)->float: lat=max(-85.05112878,min(85.05112878,float(lat))) return 156543.03392*max(0.0,float(np.cos(np.deg2rad(lat))))/(2**int(z)) def _tile_radius_for_miles(z:int,lat:float,miles:float)->int: mpp=_meters_per_pixel_webmercator(int(z),float(lat)) if not np.isfinite(mpp)or mpp<=0:return 0 return max(0,int(np.ceil(_miles_to_meters(float(miles))/(mpp*256.0)))) def _tiles_within_miles(points:list[tuple[float,float]],z:int,miles:float,cap:int=8000)->list[tuple[int,int,int]]: if not points:return[] z=int(z);n=2**z;seen:set[tuple[int,int]]=set();out:list[tuple[int,int,int]]=[];miles=float(miles) for lon,lat in points: tx,ty=_lonlat_to_tile(lon,lat,z);r=_tile_radius_for_miles(z,lat,miles) for dx in range(-r,r+1): x=tx+dx if x<0 or x>=n:continue for dy in range(-r,r+1): y=ty+dy if y<0 or y>=n:continue k=(x,y) if k in seen:continue seen.add(k);out.append((z,x,y)) if len(out)>=int(cap):return out return out def _generate_forecast_by_date_raw(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str)->tuple[bytes,dict]: sem=CPU_SEM_VALUE if _mode_get()=="value"else CPU_SEM_TILES with sem:return _generate_forecast_by_date_raw_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom) def _generate_forecast_by_date_raw_impl(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,cog_cache:CogCache|None=None)->tuple[bytes,dict]: dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") if int(hours)==24: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail) melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom) melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt24_union,z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask is None: raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8")) return raw,{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":"snowpack_unavailable","X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Raw-Scale":"10"} melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255) snow_info="snowpack_unavailable" if sel.get("snowpack_urls"): try: snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom) if snow_union:snow_info=f"snowpack_union:{snow_union.name}" except Exception as e:snow_info=f"snowpack_unavailable:{e!r}" raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask_u8) return raw,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":f"melt24_union:{melt24_union.name}","X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel["valid"],"X-Forecast-RunInit-TT":sel["run_init"],"X-Forecast-Hours":"24","X-Raw-Scale":"10"} run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom) if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom,"suggestion":"forecast data may not be available yet for this date"}) melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom) melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask is None: raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8")) return raw,{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init,"X-Forecast-72h-COG":melt72_cog.name,"X-Raw-Scale":"10"} raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask.astype("uint8",copy=False)) return raw,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":run_init,"X-Forecast-72h-COG":melt72_cog.name,"X-Raw-Scale":"10"} _BAKING_MODE=threading.local() def _is_baking_mode()->bool:return getattr(_BAKING_MODE,"active",False) class _BakingMode: def __init__(self):self.prev=None def __enter__(self):self.prev=getattr(_BAKING_MODE,"active",False);_BAKING_MODE.active=True;return self def __exit__(self,*args):_BAKING_MODE.active=self.prev if self.prev is not None else False def _bake_lowres_box_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,conc:int=2,push_every:int=300)->dict: dom=(dom or"zz").lower();_log("DEBUG",f"[bake_lowres] Starting for {date_yyyymmdd}") boxes=[(-170.0,49.0,-40.0,90.0),(-97.0,37.0,-63.0,49.0)] tiles=_tiles_for_boxes_zoom_range(boxes,zmin=int(zmin),zmax=int(zmax)) if not tiles:return{"ok":False,"error":"no_lowres_tiles"} q=deque() for hours in hours_list: for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y))) total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_paths_count=0 def worker(): nonlocal ok,skip,fail,pending_paths_count with _HfPullOnMiss(False): while True: with lk: if not q:return hours,z,x,y=q.popleft();processed=total_jobs-len(q) try: p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y) if _is_valid_png_file(p): with lk:skip+=1 continue png,_=_generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=None) _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png) with lk:ok+=1;pending_paths_count+=1 if processed%200==0:_log("DEBUG",f"[bake_lowres] {date_yyyymmdd}: {processed}/{total_jobs}") except Exception as e: with lk: if fail<5:_log("WARN",f"[bake_lowres] tile failed z={z} x={x} y={y} err={e!r}") fail+=1 threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))] for t in threads:t.start() for t in threads:t.join() _log("INFO",f"[bake_lowres] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={pending_paths_count}") return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"} def _bake_lowres_box_raw_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,conc:int=2,push_every:int=300)->dict: dom=(dom or"zz").lower();_log("DEBUG",f"[bake_lowres_raw] Starting for {date_yyyymmdd}") boxes=[(-170.0,49.0,-40.0,90.0),(-97.0,37.0,-63.0,49.0)] tiles=_tiles_for_boxes_zoom_range(boxes,zmin=int(zmin),zmax=int(zmax)) if not tiles:return{"ok":False,"error":"no_lowres_tiles"} q=deque() for hours in hours_list: for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y))) total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_paths_count=0 def worker(): nonlocal ok,skip,fail,pending_paths_count with _HfPullOnMiss(False): while True: with lk: if not q:return hours,z,x,y=q.popleft();processed=total_jobs-len(q) try: p=_raw_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y) if _is_valid_npz_file(p): with lk:skip+=1 continue raw,_=_generate_forecast_by_date_raw_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom) _raw_tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,raw) with lk:ok+=1;pending_paths_count+=1 if processed%200==0:_log("DEBUG",f"[bake_lowres_raw] {date_yyyymmdd}: {processed}/{total_jobs}") except Exception as e: with lk: if fail<5:_log("WARN",f"[bake_lowres_raw] tile failed z={z} x={x} y={y} err={e!r}") fail+=1 threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))] for t in threads:t.start() for t in threads:t.join() _log("INFO",f"[bake_lowres_raw] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={pending_paths_count}") return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"} def _hf_push_rawtilecache_for_window(window:list[str],dom:str,hours_list:list[int],zmin:int|None=None,zmax:int|None=None)->dict: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return{"ok":False,"error":"hf_not_configured"} dom=(dom or"zz").lower();root=_raw_tile_cache_dir() if not root.exists():return{"ok":True,"pushed":0,"skipped":0} want_dates=set(window);want_hours=set(int(h)for h in hours_list) ops:list[CommitOperationAdd]=[];pushed=0;skipped=0 for p in root.rglob(f"bydate_*{RAW_TILE_EXT}"): name=p.name;m=re.match(rf"^bydate_([a-z0-9]+)_h(\d+)_(\d{{8}})_z\d+_\d+_\d+\{re.escape(RAW_TILE_EXT)}$",name) if not m:continue f_dom,f_hours,f_ymd=m.group(1),int(m.group(2)),m.group(3) if f_dom!=dom or f_hours not in want_hours or f_ymd not in want_dates:continue mz=_TILE_Z_RE.search(name);z=int(mz.group(1))if mz else None if z is not None and zmin is not None and zint(zmax):continue if not _is_valid_npz_file(p):skipped+=1;continue ops.append(CommitOperationAdd(path_in_repo=f"rawtilecache/{name}",path_or_fileobj=p.as_posix())) if len(ops)>=int(HF_MAX_OPS_PER_COMMIT): api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"rawtilecache: update dom={dom} files={len(ops)}") pushed+=len(ops);ops=[] if HF_PUSH_SLEEP_SEC>0:time.sleep(float(HF_PUSH_SLEEP_SEC)) if ops: api.create_commit(repo_id=repo,repo_type="dataset",token=tok,operations=ops,commit_message=f"rawtilecache: update dom={dom} files={len(ops)}") pushed+=len(ops) return{"ok":True,"pushed":pushed,"skipped":skipped,"dom":dom} def _png(rgb:np.ndarray,a:np.ndarray)->bytes:return render(rgb,mask=a.astype("uint8"),img_format="PNG") def _dilate_bool_1px(m:np.ndarray)->np.ndarray: m=m.astype(bool,copy=False) if m.size==0:return m p=np.pad(m,1,mode="constant",constant_values=False);out=p[1:-1,1:-1].copy() out|=p[:-2,1:-1];out|=p[2:,1:-1];out|=p[1:-1,:-2];out|=p[1:-1,2:] out|=p[:-2,:-2];out|=p[:-2,2:];out|=p[2:,:-2];out|=p[2:,2:] return out def _melt_to_png(melt_mm:np.ndarray,melt_mask:np.ndarray,max_in:Optional[float],snow_mm:Optional[np.ndarray],snow_valid:Optional[np.ndarray],*,snow_allow_min_mm:float=0.0,snow_underlay_min_mm:float=0.01,dilate_px:int=1,bin_edges_in:Optional[np.ndarray]=None)->bytes: mm=melt_mm.astype("float32");melt_valid=(melt_mask>0)&np.isfinite(mm)&(mm>-9990.0) mm=np.where(melt_valid,np.maximum(mm,0.0),np.nan).astype("float32") snow_underlay=np.zeros_like(melt_valid,dtype=bool);snow_allow=None if snow_mm is not None and snow_valid is not None: snow=snow_mm.astype("float32");snow_ok=snow_valid&np.isfinite(snow)&(snow>-9990.0) snow_underlay=snow_ok&(snow>=float(snow_underlay_min_mm));snow_allow=snow_ok&(snow>=float(snow_allow_min_mm)) if dilate_px and dilate_px>0: for _ in range(int(dilate_px)):snow_allow=_dilate_bool_1px(snow_allow) melt_valid=melt_valid&snow_allow inches=mm/INCH_TO_MM if max_in is not None:inches=np.minimum(inches,float(max_in)) melt_vis=np.isfinite(inches)&(inches>0.0)&melt_valid;snow_only=snow_underlay&(~melt_vis) h,w=mm.shape;rgb=np.zeros((3,h,w),dtype="uint8");alpha=np.zeros((h,w),dtype="uint8") if snow_only.any():rgb[:,snow_only]=np.array(MELT_COLORS[0],dtype="uint8").reshape(3,1);alpha[snow_only]=255 if melt_vis.any(): filled=np.where(melt_vis,inches,0.0).astype("float32") edges=bin_edges_in if bin_edges_in is not None else BIN_EDGES_IN idx=np.clip(np.digitize(filled,edges,right=True),0,len(MELT_COLORS)-1) rgb[:,melt_vis]=np.moveaxis(MELT_COLORS[idx],-1,0)[:,melt_vis];alpha[melt_vis]=255 return _png(rgb,alpha) CACHE_CONTROL_TILES=os.environ.get("TILES_CACHE_CONTROL","public, max-age=604800, immutable") def _resp_png(data:bytes,*,cache_control:str="public, max-age=31536000, immutable",**extra_headers)->Response: etag=hashlib.md5(data).hexdigest()[:16] headers={"Cache-Control":cache_control,"ETag":f'"{etag}"',"Content-Type":"image/png","X-Content-Type-Options":"nosniff"} for k,v in extra_headers.items(): if v is not None:headers[k]=str(v) return Response(content=data,media_type="image/png",headers=headers) _COG_READ_LOCKS: dict[str, threading.Lock] = {} _COG_READ_LOCKS_LK = threading.Lock() def _cog_read_lock(cog: Path) -> threading.Lock: key = cog.as_posix() with _COG_READ_LOCKS_LK: lk = _COG_READ_LOCKS.get(key) if lk is None: lk = threading.Lock(); _COG_READ_LOCKS[key] = lk return lk def _tile_arrays_from_cog(cog: Path, z: int, x: int, y: int, *, cog_cache: CogCache = None) -> tuple[Optional[np.ndarray], Optional[np.ndarray], bool]: use_global = (cog_cache is None) if use_global: cog_cache = COG_CACHE if (not cog.exists()) or (cog.stat().st_size <= 0): raise HTTPException(status_code=500, detail=f"COG missing: {cog.name}") def _read_once(): r = cog_cache.get(cog) data, mask = r.tile(x, y, z, tilesize=256, resampling_method="nearest") return data[0].astype("float32"), mask[0].astype("uint8"), False if use_global: lk = _cog_read_lock(cog) def _read_locked(): with lk: return _read_once() read_fn = _read_locked else: read_fn = _read_once try: return read_fn() except TileOutsideBounds: return None, None, True except Exception: try: cog_cache.evict(cog) except: pass try: return read_fn() except TileOutsideBounds: return None, None, True except Exception as e2: msg = str(e2); _log("WARN", f"[cog_read] failed cog={cog.name} z={z} x={x} y={y} err={e2!r}") if ("TIFFReadEncodedTile" in msg) or ("ZIPDecode" in msg) or ("Decoding error" in msg): try: cog_cache.evict(cog) except: pass if use_global: try: _safe_unlink(cog); _log("WARN", f"[cog_read] deleted corrupted COG: {cog.name}") except: pass return None, None, True def _tile_png_from_cog(cog:Path,z:int,x:int,y:int,max_in:Optional[float])->bytes: arr,mask,oob=_tile_arrays_from_cog(cog,z,x,y,cog_cache=COG_CACHE) if oob or arr is None or mask is None:return _transparent_png_256() return _melt_to_png(arr,mask,max_in,None,None) def _serve_forecast_tile(run_init:str,valid:str,lead:str,melt_urls:list[str],snow_ts:str,snow_urls:list[str],z:int,x:int,y:int,max_in:float|None)->Response: run_init=_norm_ts(run_init);valid=_norm_ts(valid);lead=(lead or"").lower() melt_cogs=_build_forecast_melt_cogs(run_init,valid,lead,melt_urls) snow=snow_valid=None;snow_info="mask_unavailable" if snow_urls:snow,snow_valid,snow_info=_read_mask_tile(snow_ts or valid,snow_urls,z,x,y) melt_mm,melt_valid,melt_info=_union_tile(melt_cogs,z,x,y,label="melt") if melt_mm is None or melt_valid is None:return _resp_png(_transparent_png_256(),**{"X-Melt-Info":melt_info,"X-NoData":"1"}) melt_mask_u8=(melt_valid.astype("uint8")*255) png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2) return _resp_png(png,**{"X-Forecast-RunInit-TT":run_init,"X-Forecast-Valid-TS":valid,"X-Forecast-Lead":lead,"X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-SnowMask-TS":(snow_ts or valid or"")}) def _corridor_tiles_for_zoom_range(zmin:int,zmax:int,*,miles:float,cap_per_zoom:int)->list[tuple[int,int,int]]: pts=_load_track_points() if not pts:return[] out:list[tuple[int,int,int]]=[];seen:set[tuple[int,int,int]]=set() for z in range(int(zmin),int(zmax)+1): tiles=_tiles_within_miles(pts,z,float(miles),cap=int(cap_per_zoom)) for(zz,x,y)in tiles: if not _tile_allowed(zz,x,y):continue t=(zz,x,y) if t in seen:continue seen.add(t);out.append(t) return out def _point_mm_from_cog(cog:Path,lon:float,lat:float)->tuple[Optional[float],str]: if not cog.exists()or cog.stat().st_size<=0:return None,f"missing:{cog.name}" try: with COGReader(cog.as_posix())as r:data,mask=_cog_point_data_mask(r,lon,lat) v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0]) if m<=0 or(not np.isfinite(v))or(v<=-9990.0):return None,f"nodata:{cog.name}" return v,f"ok:{cog.name}" except TileOutsideBounds:return None,f"oob:{cog.name}" except Exception as e:return None,f"error:{cog.name}:{e!r}" def _union_point_mm(cogs:list[Path],lon:float,lat:float)->tuple[Optional[float],str]: for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))): v,info=_point_mm_from_cog(cog,lon,lat) if v is not None:return v,info return None,"all_domains_failed" def _bake_corridor_tiles_for_date(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,conc:int=2,push_every:int=500)->dict: dom=(dom or"zz").lower();_log("DEBUG",f"[bake_corridor] Starting for {date_yyyymmdd}") tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom)) if not tiles:return{"ok":False,"error":"no_corridor_tiles"} tiles=tiles[:int(max_tiles_total)];_log("DEBUG",f"[bake_corridor] {len(tiles)} tiles to process") q=deque() for hours in hours_list: for(z,x,y)in tiles:q.append((int(hours),int(z),int(x),int(y))) total_jobs=len(q);ok=0;skip=0;fail=0;lk=threading.Lock();pending_push:list[Path]=[];total_pushed=0;push_lk=threading.Lock() def maybe_push(): nonlocal pending_push,total_pushed with push_lk: if len(pending_push)>=push_every: batch=pending_push[:push_every];pending_push=pending_push[push_every:] result=_incremental_hf_push(batch,"tilecache",f"{date_yyyymmdd}_corridor_png_{total_pushed}") if result.get("ok"):total_pushed+=result.get("pushed",0);_log("DEBUG",f"[bake_corridor] Pushed {result.get('pushed',0)} tiles to HF (total: {total_pushed})") else:_log("WARN",f"[bake_corridor] Push failed: {result}") def worker(): nonlocal ok,skip,fail,pending_push with _HfPullOnMiss(False): while True: with lk: if not q:return hours,z,x,y=q.popleft();processed=total_jobs-len(q) try: p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y) if _is_valid_png_file(p): with lk:skip+=1 continue png,_=_generate_forecast_by_date_png_impl(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,max_in=None) _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png) with lk:ok+=1;pending_push.append(p) maybe_push() if processed%500==0:_log("DEBUG",f"[bake_corridor] {date_yyyymmdd}: {processed}/{total_jobs} (ok={ok}, skip={skip}, fail={fail})") except Exception as e: with lk: if fail<5:_log("WARN",f"[bake_corridor] tile failed z={z} x={x} y={y} err={e!r}") fail+=1 threads=[threading.Thread(target=worker,daemon=True)for _ in range(int(conc))] for t in threads:t.start() for t in threads:t.join() with push_lk: if pending_push: result=_incremental_hf_push(pending_push,"tilecache",f"{date_yyyymmdd}_corridor_png_final") if result.get("ok"):total_pushed+=result.get("pushed",0) _log("DEBUG",f"[bake_corridor] Final push: {result}");pending_push=[] _log("INFO",f"[bake_corridor] {date_yyyymmdd} complete: ok={ok}, skipped={skip}, fail={fail}, pushed={total_pushed}") return{"ok":True,"date":date_yyyymmdd,"dom":dom,"hours_list":hours_list,"tiles":len(tiles),"jobs":total_jobs,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":total_pushed,"zmin":zmin,"zmax":zmax} def _rolling_dates_local_window(today_local=None, *, lookback: int = 1, lookahead: int = 3) -> list[str]: if today_local is None: today_local = datetime.now(LOCAL_TZ).date() return sorted([(today_local + timedelta(days=d)).strftime("%Y%m%d") for d in range(-int(lookback), int(lookahead) + 1)]) def _keep_cutoff_ymd(keep_days: int = _REMOTE_KEEP_DAYS) -> str: k = int(keep_days) today_local = datetime.now(LOCAL_TZ).date() if k <= 0: cutoff_date = today_local else: cutoff_date = today_local - timedelta(days=(k - 1)) return cutoff_date.strftime("%Y%m%d") def _prune_tile_cache(keep_ymds: set[str], keep_days_fallback: int = 0) -> int: deleted = 0 root = _tile_cache_dir() if not root.exists(): return 0 for dom_dir in root.iterdir(): if not dom_dir.is_dir(): continue for hours_dir in dom_dir.iterdir(): if not hours_dir.is_dir(): continue for ymd_dir in hours_dir.iterdir(): if not ymd_dir.is_dir(): continue ymd = ymd_dir.name if not re.fullmatch(r"[0-9]{8}", ymd): continue if ymd in keep_ymds: continue try: shutil.rmtree(ymd_dir, ignore_errors=True) deleted += 1 _log("INFO", f"[prune_tile] deleted local tilecache dir: {ymd_dir.relative_to(CACHE)}") except: pass return deleted def _prune_raw_tile_cache(keep_ymds: set[str], keep_days_fallback: int = 0) -> int: deleted = 0 root = _raw_tile_cache_dir() if not root.exists(): return 0 for dom_dir in root.iterdir(): if not dom_dir.is_dir(): continue for hours_dir in dom_dir.iterdir(): if not hours_dir.is_dir(): continue for ymd_dir in hours_dir.iterdir(): if not ymd_dir.is_dir(): continue ymd = ymd_dir.name if not re.fullmatch(r"[0-9]{8}", ymd): continue # ONLY keep dates explicitly in the keep set if ymd in keep_ymds: continue try: shutil.rmtree(ymd_dir, ignore_errors=True) deleted += 1 _log("INFO", f"[prune_raw] deleted local rawtilecache dir: {ymd_dir.relative_to(CACHE)}") except: pass return deleted def _prune_forecast_cogs(keep_days: int = 0, *, keep_ymds: set[str] | None = None) -> int: deleted = 0 fc = _forecast_dir() if not fc.exists(): return 0 for p in fc.glob("*_cog.tif"): ymd = _file_ymd_from_name(p.name) if not ymd: continue if keep_ymds is not None: if ymd in keep_ymds: continue else: cutoff = _keep_cutoff_ymd(keep_days) if ymd >= cutoff: continue try: p.unlink(missing_ok=True) deleted += 1 _log("INFO", f"[prune_cog] deleted forecast COG: {p.name}") except: pass return deleted def _ensure_forecast_cogs_for_date(date_yyyymmdd:str,dom:str)->dict: dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):return{"ok":False,"date":date_yyyymmdd,"error":"pick_failed","detail":sel} try:_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]) except Exception as e:return{"ok":False,"date":date_yyyymmdd,"valid":sel["valid"],"error":f"melt_build_failed:{e!r}"} if sel.get("snowpack_urls"): try:_build_forecast_snowpack_cogs(sel.get("snowpack_ts")or sel["valid"],sel["snowpack_urls"]) except:pass melt24_union=snow_union=melt72=None try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom) except Exception as e:_log("WARN",f"[ensure_forecast] 24h UNION melt failed for {sel['valid']}: {e!r}") try:snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom) except Exception as e:_log("WARN",f"[ensure_forecast] UNION snowpack failed for {sel['valid']}: {e!r}") try:melt72=_forecast_melt72h_end_cog(sel["valid"],dom_prefer=dom) except Exception as e:_log("WARN",f"[ensure_forecast] 72h COG failed for {sel['valid']}: {e!r}") return{"ok":True,"date":date_yyyymmdd,"valid":sel["valid"],"tt":sel["run_init"],"dom":dom,"melt24_union_cog":(melt24_union.name if melt24_union else None),"snow_union_cog":(snow_union.name if snow_union else None),"melt72_cog":(melt72.name if melt72 else None)} def _ensure_forecast_for_valid(valid: str, dom_prefer: str = "zz") -> dict: dom_prefer = (dom_prefer or "zz").lower() valid = _norm_ts(valid) date_yyyymmdd = valid[:8] sel = _pick_forecast_for_valid(valid, dom_prefer=dom_prefer) if not sel.get("ok"): return { "ok": False, "date": date_yyyymmdd, "valid": valid, "error": "pick_failed", "detail": sel, } try: _build_forecast_melt_cogs( sel["run_init"], sel["valid"], "t0024", sel["melt_urls"], ) except Exception as e: return { "ok": False, "date": date_yyyymmdd, "valid": valid, "error": f"melt_build_failed:{e!r}", } if sel.get("snowpack_urls"): try: _build_forecast_snowpack_cogs( sel.get("snowpack_ts") or sel["valid"], sel["snowpack_urls"], ) except Exception: pass melt24_union = None snow_union = None melt72 = None try: melt24_union = _forecast_melt24h_union_end_cog( sel["valid"], dom_prefer=dom_prefer, ) except Exception as e: _log("WARN", f"[ensure_forecast] 24h UNION melt failed for {valid}: {e!r}") try: snow_union = _forecast_snowpack_union_end_cog( sel.get("snowpack_ts") or sel["valid"], dom_prefer=dom_prefer, ) except Exception as e: _log("WARN", f"[ensure_forecast] UNION snowpack failed for {valid}: {e!r}") try: melt72 = _forecast_melt72h_end_cog( sel["valid"], dom_prefer=dom_prefer, ) except Exception as e: _log("WARN", f"[ensure_forecast] 72h COG failed for {valid}: {e!r}") return { "ok": True, "date": date_yyyymmdd, "valid": sel["valid"], "tt": sel["run_init"], "dom": dom_prefer, "melt24_union_cog": (melt24_union.name if melt24_union else None), "snow_union_cog": (snow_union.name if snow_union else None), "melt72_cog": (melt72.name if melt72 else None), } def _next_run_local(now:datetime|None=None)->datetime: now=now or datetime.now(LOCAL_TZ);target=now.replace(hour=SCHEDULE_HOUR_LOCAL,minute=SCHEDULE_MIN_LOCAL,second=0,microsecond=0) return target+timedelta(days=1)if target<=now else target def _bake_corridor_tiles_for_date_optimized(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,push_every:int=300)->dict: dom=(dom or"zz").lower();_log("INFO",f"[bake_corridor_opt] Starting for {date_yyyymmdd}") tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom)) if not tiles:return{"ok":False,"error":"no_tiles"} tiles=tiles[:int(max_tiles_total)];jobs=[(int(h),int(z),int(x),int(y))for h in hours_list for(z,x,y)in tiles] total_jobs=len(jobs);_log("INFO",f"[bake_corridor_opt] {total_jobs} tiles to process");ok=0;skip=0;fail=0;local_written=0 with CogCache()as cog_cache: _log("INFO",f"[bake_corridor_opt] Pre-loading COGs for {date_yyyymmdd}");preloaded=_preload_forecast_for_date(date_yyyymmdd,dom,cog_cache) _log("INFO",f"[bake_corridor_opt] Preload complete: h24={preloaded.get('h24',{}).get('ok',False)}, h72={preloaded.get('h72',{}).get('ok',False)}") for i,(hours,z,x,y)in enumerate(jobs): try: p=_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y) if _is_valid_png_file(p):skip+=1;continue png,is_valid=_generate_tile_from_preloaded(z,x,y,hours,preloaded,cog_cache=cog_cache) _tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,png);ok+=1;local_written+=1 except Exception as e: if fail<5:_log("WARN",f"[bake_corridor_opt] tile failed z={z} x={x} y={y}: {e!r}") fail+=1 if(i+1)%500==0:_log("DEBUG",f"[bake_corridor_opt] {date_yyyymmdd}: {i+1}/{total_jobs} (ok={ok}, skip={skip}, fail={fail})") _log("INFO",f"[bake_corridor_opt] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, local_written={local_written}") return{"ok":True,"date":date_yyyymmdd,"jobs":total_jobs,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":0,"remote_push":"deferred"} def _bake_corridor_raw_tiles_for_date_optimized(*,date_yyyymmdd:str,dom:str,hours_list:list[int],zmin:int,zmax:int,miles:float,cap_per_zoom:int,max_tiles_total:int,push_every:int=300)->dict: dom=(dom or"zz").lower();_log("INFO",f"[bake_corridor_raw_opt] Starting for {date_yyyymmdd}") tiles=_corridor_tiles_for_zoom_range(zmin,zmax,miles=float(miles),cap_per_zoom=int(cap_per_zoom)) if not tiles:return{"ok":False,"error":"no_tiles"} tiles=tiles[:int(max_tiles_total)];jobs=[(int(h),int(z),int(x),int(y))for h in hours_list for(z,x,y)in tiles] total_jobs=len(jobs);_log("INFO",f"[bake_corridor_raw_opt] {total_jobs} tiles to process");ok=0;skip=0;fail=0;pending_push:list[Path]=[];total_pushed=0 with CogCache()as cog_cache: _log("INFO",f"[bake_corridor_raw_opt] Pre-loading COGs for {date_yyyymmdd}") preloaded=_preload_forecast_for_date(date_yyyymmdd,dom,cog_cache) _log("INFO",f"[bake_corridor_raw_opt] Preload complete: h24={preloaded.get('h24',{}).get('ok',False)}, h72={preloaded.get('h72',{}).get('ok',False)}") for i,(hours,z,x,y)in enumerate(jobs): try: p=_raw_tile_cache_path(dom,hours,date_yyyymmdd,z,x,y) if _is_valid_npz_file(p):skip+=1;continue raw,_=_generate_forecast_by_date_raw_preloaded(z=z,x=x,y=y,date_yyyymmdd=date_yyyymmdd,hours=hours,dom=dom,preloaded=preloaded,cog_cache=cog_cache) _raw_tile_cache_put(dom,hours,date_yyyymmdd,z,x,y,raw);ok+=1;pending_push.append(p) except Exception as e: if fail<5:_log("WARN",f"[bake_corridor_raw_opt] tile failed z={z} x={x} y={y}: {e!r}") fail+=1 if len(pending_push)>=push_every: result=_incremental_hf_push(pending_push,"rawtilecache",f"{date_yyyymmdd}_corridor_raw_{total_pushed}") if result.get("ok"):total_pushed+=result.get("pushed",0) pending_push=[] if(i+1)%500==0:_log("DEBUG",f"[bake_corridor_raw_opt] {date_yyyymmdd}: {i+1}/{total_jobs} (ok={ok}, skip={skip}, fail={fail}, pushed={total_pushed})") if pending_push: result=_incremental_hf_push(pending_push,"rawtilecache",f"{date_yyyymmdd}_corridor_raw_final") if result.get("ok"):total_pushed+=result.get("pushed",0) _log("INFO",f"[bake_corridor_raw_opt] {date_yyyymmdd} complete: ok={ok}, skip={skip}, fail={fail}, pushed={total_pushed}") return{"ok":True,"date":date_yyyymmdd,"ok_count":ok,"skip_count":skip,"fail_count":fail,"pushed_count":total_pushed} _PUSH_PAUSED = threading.Event() _PUSH_PAUSED.clear() def _hf_push_baked_dates(baked_ymds: list[str], dom: str, hours_list: list[int]) -> dict: tok, repo = _hf_cfg(); api = _hf_api() if not tok or not repo or api is None: return {"ok": False, "error": "hf_not_configured"} dom = (dom or "zz").lower(); pushed = 0; errors = 0 for ymd in baked_ymds: for h in hours_list: for kind in ["tilecache", "rawtilecache"]: base = CACHE / kind / dom / str(int(h)) / ymd if not base.exists(): continue while _PUSH_PAUSED.is_set(): time.sleep(2.0) ext = ".png" if kind == "tilecache" else ".npz" ops = []; total_bytes = 0 for p in base.rglob(f"*{ext}"): sz = p.stat().st_size if sz < 60: continue try: rel = p.relative_to(CACHE).as_posix() ops.append(CommitOperationAdd(path_in_repo=rel, path_or_fileobj=str(p))) total_bytes += sz except: pass if not ops: continue _log("INFO", f"[hf_push] {kind}/{ymd}/h{h}: {len(ops)} files, {total_bytes/1024/1024:.1f} MB total") for i in range(0, len(ops),750): while _PUSH_PAUSED.is_set(): time.sleep(2.0) batch = ops[i:i+750] batch_bytes = sum(Path(op.path_or_fileobj).stat().st_size for op in batch if hasattr(op, 'path_or_fileobj')) batch_num = i // 750+ 1 ok = False for attempt in range(3): t0 = time.time() try: api.create_commit(repo_id=repo, repo_type="dataset", operations=batch, commit_message=f"{kind}: {ymd} h{h} batch {batch_num} ({len(batch)} files)") elapsed = time.time() - t0 pushed += len(batch) ok = True _log("INFO", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num}: OK {len(batch)} files, {batch_bytes/1024:.0f} KB in {elapsed:.1f}s") break except Exception as e: elapsed = time.time() - t0 err_str = repr(e) if "ReadTimeout" in err_str and elapsed >= 9.0: _log("INFO", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num}: timeout after {elapsed:.1f}s — likely succeeded (attempt {attempt+1})") pushed += len(batch); ok = True; time.sleep(2.0); break _log("WARN", f"[hf_push] {kind}/{ymd}/h{h} batch {batch_num} attempt {attempt+1}: {type(e).__name__} after {elapsed:.1f}s — {len(batch)} files, {batch_bytes/1024:.0f} KB") time.sleep(10.0 * (attempt + 1)) if not ok: errors += len(batch) else: time.sleep(2.0) _log("INFO", f"[hf_push] {kind}/{ymd}/h{h}: done (pushed={pushed} errors={errors})") _log("INFO", f"[hf_push_baked] done: pushed={pushed} errors={errors}") return {"ok": True, "pushed": pushed, "errors": errors} def _run_daily_roll_job_impl()->dict: if not _try_acquire_run_lock(_ROLL_LOCK_NAME,stale_seconds=_ROLL_LOCK_STALE_SECONDS): _log("INFO","[roll] skipped: already running");return{"ok":True,"skipped":True,"reason":"already_running"} try: import gc;started=datetime.now(LOCAL_TZ).isoformat();today_local=datetime.now(LOCAL_TZ).date() window=_rolling_dates_local_window(today_local=today_local,lookback=2,lookahead=3);keep_set=set(window) _log("INFO",f"[roll] local_today={today_local} window={window}") need=[ymd for ymd in window if not _date_is_baked(WARM_DOM,ymd)] already=[y for y in window if y not in need] _log("INFO",f"[roll] need={need} already_baked={already}") if not need: _log("INFO","[roll] all dates baked — prune only") try:_hf_prune_remote_and_local(keep_ymds=keep_set) except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}") return{"ok":True,"skipped":True,"reason":"nothing_needed","window":window,"finished":datetime.now(LOCAL_TZ).isoformat()} _log("INFO","[roll] Step 2: Pulling forecast COGs + tar archives") try:_hf_pull_cache() except Exception as e:_log("WARN",f"[roll] HF pull failed: {e!r}") need=[ymd for ymd in need if not _date_is_baked(WARM_DOM,ymd)] _log("INFO",f"[roll] after pull, still need={need}") if not need: try:_hf_prune_remote_and_local(keep_ymds=keep_set) except:pass return{"ok":True,"skipped":True,"reason":"hf_pull_satisfied","window":window,"finished":datetime.now(LOCAL_TZ).isoformat()} _log("INFO","[roll] Step 3: Pruning old tiles") try: dt=_prune_tile_cache(keep_set);dr=_prune_raw_tile_cache(keep_set);dc=_prune_forecast_cogs(keep_ymds=keep_set) _log("INFO",f"[roll] pruned tiles={dt} raw={dr} cogs={dc}") except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}") try:_tile_png_cached.cache_clear() except:pass today_str=today_local.strftime("%Y%m%d");tmrw_str=(today_local+timedelta(days=1)).strftime("%Y%m%d") need=sorted(need,key=lambda y:(0 if y==today_str else 1 if y==tmrw_str else 2,y)) _log("INFO",f"[roll] Step 4: Baking {len(need)} date(s) (priority order: {need})") results={"ok":True,"started":started,"window":window,"need":need,"dates":{}};baked_ymds=[];failed_ymds=[] for i,ymd in enumerate(need): _log("INFO",f"[roll] === {ymd} ({i+1}/{len(need)}) ===");dr={"date":ymd} cog_ok=False for cog_attempt in range(5): try: r=_ensure_forecast_for_valid(_norm_ts(f"{ymd}05"),dom_prefer=WARM_DOM) if r.get("ok"):cog_ok=True;dr["cog"]={"ok":True};break else:_log("WARN",f"[roll] {ymd}: COG attempt {cog_attempt+1}/5 failed") except Exception as e:_log("WARN",f"[roll] {ymd}: COG attempt {cog_attempt+1}/5 exception: {e!r}") if cog_attempt<4:sleep_sec=10*(2**cog_attempt);_log("INFO",f"[roll] {ymd}: retrying COG in {sleep_sec}s");time.sleep(sleep_sec) if not cog_ok:_log("WARN",f"[roll] {ymd}: COG failed after 5 attempts");dr["cog"]={"ok":False};results["dates"][ymd]=dr;failed_ymds.append(ymd);continue with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool: futs={} # Bake sequentially: PNG first, then raw for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: try: r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) dr[label]={"ok":r.get("ok_count",0),"skip":r.get("skip_count",0),"fail":r.get("fail_count",0)} _log("INFO",f"[roll] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[roll] {ymd}: {label} exception: {e!r}");dr[label]={"error":repr(e)} for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try: r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) dr[label]={"ok":r.get("ok_count",0),"skip":r.get("skip_count",0),"fail":r.get("fail_count",0)} _log("INFO",f"[roll] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[roll] {ymd}: {label} exception: {e!r}");dr[label]={"error":repr(e)} try: best=_find_best_melt_for_date(ymd,dom_prefer=WARM_DOM) if best.get("ok")and best.get("melt_urls"):_record_baked_sig(ymd,WARM_DOM,best["melt_urls"][0],source_type=best.get("source_type")) except:pass results["dates"][ymd]=dr;baked_ymds.append(ymd);_log("INFO",f"[roll] === {ymd} bake complete ===");gc.collect() if baked_ymds: _log("INFO",f"[roll] Step 5: Pushing {len(baked_ymds)} dates to HF via tar archives") push_result=_hf_push_baked_as_tar(baked_ymds,WARM_DOM,WARM_HOURS_LIST);results["push"]=push_result _log("INFO",f"[roll] Push complete: {push_result}") else:_log("INFO","[roll] Step 5: No dates to push") _log("INFO","[roll] Step 6: Pruning old data") try:_hf_prune_remote_and_local(keep_ymds=keep_set) except Exception as e:_log("WARN",f"[roll] prune failed: {e!r}") results["finished"]=datetime.now(LOCAL_TZ).isoformat() _log("INFO",f"[roll] ========== COMPLETE at {results['finished']} ==========") if failed_ymds: _log("INFO",f"[roll] Step 7: Starting retry loop for failed dates: {failed_ymds}") def _retry_failed(): remaining=list(failed_ymds);max_retries=48 for retry_round in range(max_retries): if not remaining:break _log("INFO",f"[roll_retry] round {retry_round+1}: waiting 30 min before retrying {remaining}");time.sleep(30*60) still_failed=[] for ymd in remaining: if _date_is_baked(WARM_DOM,ymd):_log("INFO",f"[roll_retry] {ymd}: already baked");continue cog_ok=False for cog_attempt in range(5): try: r=_ensure_forecast_for_valid(_norm_ts(f"{ymd}05"),dom_prefer=WARM_DOM) if r.get("ok"):cog_ok=True;break except:pass if cog_attempt<4:time.sleep(10*(2**cog_attempt)) if not cog_ok:_log("WARN",f"[roll_retry] {ymd}: COG still unavailable");still_failed.append(ymd);continue _log("INFO",f"[roll_retry] {ymd}: COG OK, baking") try: with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool: futs={} for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: futs[label]=pool.submit(fn,date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) for label,fut in futs.items(): try:r=fut.result();_log("INFO",f"[roll_retry] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[roll_retry] {ymd}: {label} exception: {e!r}") for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try:r=fn(date_yyyymmdd=ymd,dom=WARM_DOM,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra);_log("INFO",f"[roll_retry] {ymd}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[roll_retry] {ymd}: {label} exception: {e!r}") _hf_push_baked_as_tar([ymd],WARM_DOM,WARM_HOURS_LIST);_log("INFO",f"[roll_retry] {ymd}: bake + push complete") try: best=_find_best_melt_for_date(ymd,dom_prefer=WARM_DOM) if best.get("ok")and best.get("melt_urls"):_record_baked_sig(ymd,WARM_DOM,best["melt_urls"][0],source_type=best.get("source_type")) except:pass except Exception as e:_log("WARN",f"[roll_retry] {ymd}: bake failed: {e!r}");still_failed.append(ymd) gc.collect() remaining=still_failed if remaining:_log("WARN",f"[roll_retry] giving up on: {remaining} after {max_retries} rounds") else:_log("INFO","[roll_retry] all failed dates recovered") threading.Thread(target=_retry_failed,daemon=True,name="roll_retry").start() return results finally:_release_run_lock(_ROLL_LOCK_NAME) def _daily_scheduler_loop()->None: time.sleep(0.25);_log("INFO","[scheduler] start") while True: try: now=datetime.now(LOCAL_TZ);nxt=_next_run_local(now);sleep_s=max(1.0,(nxt-now).total_seconds()) _log("INFO",f"[scheduler] next_run_local={nxt.isoformat()} sleep_s={int(sleep_s)}");time.sleep(sleep_s) r=_run_daily_roll_job_impl() if isinstance(r,dict)and r.get("skipped"):_log("INFO",f"[scheduler] roll skipped reason={r.get('reason')}") except Exception as e:_log("INFO",f"[scheduler] crashed err={e!r}");time.sleep(30.0) def _preload_forecast_for_date(date_yyyymmdd:str,dom:str,cog_cache:CogCache)->dict: dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") result={"date":date_yyyymmdd,"valid":valid,"dom":dom,"h24":{"ok":False},"h72":{"ok":False}} try: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if sel.get("ok"): melt24_union=snow_union=None try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom) except Exception as e:_log("WARN",f"[preload] melt24 union failed for {date_yyyymmdd}: {e!r}");melt24_union=None if sel.get("snowpack_urls"): try:snow_union=_forecast_snowpack_union_end_cog(sel.get("snowpack_ts")or sel["valid"],dom_prefer=dom) except:snow_union=None if melt24_union is not None: try:cog_cache.get(melt24_union) except:pass if snow_union is not None: try:cog_cache.get(snow_union) except:pass if melt24_union is not None and melt24_union.exists()and melt24_union.stat().st_size>0: result["h24"]={"ok":True,"run_init":sel["run_init"],"valid":sel["valid"],"melt24_union_cog":melt24_union,"snow_union_cog":snow_union} except Exception as e:_log("WARN",f"[preload] h24 failed for {date_yyyymmdd}: {e!r}") try: run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom) if run_init: melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom) if melt72_cog.exists()and melt72_cog.stat().st_size>0: try:cog_cache.get(melt72_cog) except:pass result["h72"]={"ok":True,"run_init":run_init,"valid":valid,"melt72_cog":melt72_cog,"snow_union_cog":result["h24"].get("snow_union_cog")if result["h24"].get("ok")else None} except Exception as e:_log("WARN",f"[preload] h72 failed for {date_yyyymmdd}: {e!r}") return result def _generate_tile_from_preloaded(z:int,x:int,y:int,hours:int,preloaded:dict,max_in:float|None=None,*,cog_cache:CogCache|None=None)->tuple[bytes,bool]: if hours==24: h24=preloaded.get("h24",{}) if not h24.get("ok"):return _transparent_png_256(),False melt_cog=h24.get("melt24_union_cog") if not melt_cog:return _transparent_png_256(),False melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt_cog,z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),False melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255) snow=snow_valid=None;snow_cog=h24.get("snow_union_cog") if snow_cog: snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_cog,z,x,y,cog_cache=cog_cache) if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_IN) return png,True if hours==72: h72=preloaded.get("h72",{}) if not h72.get("ok"):return _transparent_png_256(),False melt72_cog=h72.get("melt72_cog") if not melt72_cog:return _transparent_png_256(),False melt_mm,melt_mask,oob=_tile_arrays_from_cog(melt72_cog,z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask is None:return _transparent_png_256(),False melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0);melt_mask_u8=(melt_valid.astype("uint8")*255) snow=snow_valid=None;snow_cog=h72.get("snow_union_cog") if snow_cog: snow_arr,snow_mask,oob2=_tile_arrays_from_cog(snow_cog,z,x,y,cog_cache=cog_cache) if(not oob2)and(snow_arr is not None)and(snow_mask is not None):snow_valid=(snow_mask>0)&np.isfinite(snow_arr)&(snow_arr>-9990.0);snow=snow_arr png=_melt_to_png(melt_mm,melt_mask_u8,max_in,snow,snow_valid,snow_allow_min_mm=0.0,snow_underlay_min_mm=0.0001,dilate_px=2,bin_edges_in=BIN_EDGES_72H_IN) return png,True return _transparent_png_256(),False def _generate_forecast_by_date_raw_preloaded(*,z:int,x:int,y:int,date_yyyymmdd:str,hours:int,dom:str,preloaded:dict,cog_cache:CogCache)->tuple[bytes,dict]: dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") if int(hours)==24: h24=preloaded.get("h24",{}) if not h24.get("ok"):raise HTTPException(status_code=503,detail={"error":"forecast_unavailable","valid":valid,"dom":dom}) sel=h24.get("sel")or{};melt_info="melt_unavailable" if h24.get("melt24_union_cog"): melt_mm,melt_mask,oob=_tile_arrays_from_cog(h24["melt24_union_cog"],z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask is None:melt_mm=None;melt_valid=None else:melt_valid=(melt_mask>0)&np.isfinite(melt_mm)&(melt_mm>-9990.0) melt_info=f"melt24_union:{h24['melt24_union_cog'].name}" else:melt_mm,melt_valid,melt_info=_union_tile_cached(h24.get("melt_cogs",[]),z,x,y,label="melt",cog_cache=cog_cache) snow_info="snowpack_unavailable" try: if h24.get("snow_union_cog"):snow_info=f"snowpack_union:{h24['snow_union_cog'].name}" elif h24.get("snow_cogs"):_,_,snow_info=_union_tile_cached(h24["snow_cogs"],z,x,y,label="snowpack",cog_cache=cog_cache) except Exception as e:snow_info=f"snowpack_unavailable:{e!r}" if melt_mm is None or melt_valid is None: raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8")) return raw,{"X-Allowed":"1","X-OOB":"1","X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel.get("valid",valid),"X-Forecast-RunInit-TT":sel.get("run_init",""),"X-Forecast-Hours":"24","X-Raw-Scale":"10"} raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=(melt_valid.astype("uint8")*255)) return raw,{"X-Allowed":"1","X-OOB":"0","X-Melt-Info":melt_info,"X-SnowMask-Info":snow_info,"X-Forecast-Valid":sel.get("valid",valid),"X-Forecast-RunInit-TT":sel.get("run_init",""),"X-Forecast-Hours":"24","X-Raw-Scale":"10"} h72=preloaded.get("h72",{}) if not h72.get("ok"):raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom,"suggestion":"forecast data may not be available yet for this date"}) melt_mm,melt_mask_u8,oob=_tile_arrays_from_cog(h72["melt72_cog"],z,x,y,cog_cache=cog_cache) if oob or melt_mm is None or melt_mask_u8 is None: raw=_pack_raw_tile_npz(melt_mm=np.zeros((256,256),dtype="float32"),valid_mask_u8=np.zeros((256,256),dtype="uint8")) return raw,{"X-Allowed":"1","X-OOB":"1","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":h72.get("run_init",""),"X-Forecast-72h-COG":h72["melt72_cog"].name,"X-Raw-Scale":"10"} raw=_pack_raw_tile_npz(melt_mm=melt_mm,valid_mask_u8=melt_mask_u8.astype("uint8",copy=False)) return raw,{"X-Allowed":"1","X-OOB":"0","X-Forecast-Hours":"72","X-Forecast-Valid":valid,"X-Forecast-RunInit-TT":h72.get("run_init",""),"X-Forecast-72h-COG":h72["melt72_cog"].name,"X-Raw-Scale":"10"} def _shift_ts(ts10:str,hours:int)->str:return(_ts_to_dt(_norm_ts(ts10))+timedelta(hours=hours)).strftime("%Y%m%d%H") def _union_point(cogs:list[Path],lon:float,lat:float)->tuple[Optional[float],str]: if not cogs:return None,"no_cogs" used=[] for cog in sorted(cogs,key=lambda p:_dom_pri(_cog_dom(p))): try: with COGReader(cog.as_posix())as r:data,mask=_cog_point_data_mask(r,lon,lat) v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0]);used.append(cog.name) if m>0 and np.isfinite(v)and v>-9990.0:return v,"ok:"+cog.name except:continue return None,"nodata:"+",".join(used[:3]) def _rolling_window_dates(lookback_days: int = 1, lookahead_days: int = 3) -> list[str]: today = datetime.now(LOCAL_TZ).date() return sorted([(today + timedelta(days=d)).strftime("%Y%m%d") for d in range(-int(lookback_days), int(lookahead_days) + 1)]) def _check_date_has_local_tiles(dom: str, hours: int, ymd: str) -> bool: dom = (dom or "zz").lower() for kind, ext in [("tilecache", "*.png"), ("rawtilecache", "*.npz")]: d = CACHE / kind / dom / str(hours) / ymd if d.exists(): try: next(d.rglob(ext)) return True except StopIteration: pass return False def _date_has_local_tiles_both(dom: str, ymd: str, min_tiles: int = 50) -> bool: dom = (dom or "zz").lower() for h in WARM_HOURS_LIST: d = CACHE / "tilecache" / dom / str(h) / ymd if not d.exists(): return False try: count = sum(1 for _ in d.rglob("*.png")) if count < min_tiles: return False except: return False return True _AVAILABLE_DATES_CACHE:dict={"ts":0.0,"data":None,"key":None} _AVAILABLE_DATES_CACHE_TTL=120.0 @app.get("/forecast/sample_melt") def forecast_sample_melt(date_yyyymmdd:str=Query(...,regex=r"^\d{8}$"),hours:int=Query(24,ge=24,le=72),lon:float=Query(...,ge=-180.0,le=180.0),lat:float=Query(...,ge=-90.0,le=90.0),dom:str=Query("zz")): dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05");mm=None;info="" if int(hours)==72: cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom) try: r=_cogreader_get(cog);data,mask=_cog_point_data_mask(r,lon,lat);v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0]) if m>0 and np.isfinite(v)and v>-9990.0:mm=v;info="ok:"+cog.name else:info="nodata:"+cog.name except Exception as e:info=f"err:{e!r}" else: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="forecast data may not be available yet for this date";raise HTTPException(status_code=503,detail=detail) try:melt24_union=_forecast_melt24h_union_end_cog(sel["valid"],dom_prefer=dom) except Exception as e:raise HTTPException(status_code=503,detail={"error":"melt24_union_cog_unavailable","date_yyyymmdd":date_yyyymmdd,"valid":sel.get("valid",valid),"dom":dom,"upstream_error":repr(e)}) try: r=_cogreader_get(melt24_union);data,mask=_cog_point_data_mask(r,lon,lat);v=float(np.ravel(data)[0]);m=int(np.ravel(mask)[0]) if m>0 and np.isfinite(v)and v>-9990.0:mm=v;info="ok:"+melt24_union.name else:info="nodata:"+melt24_union.name except Exception as e:info=f"err:{e!r}" if mm is None:return{"ok":True,"value_in":None,"info":info,"hours":int(hours),"date_yyyymmdd":date_yyyymmdd} inches=max(0.0,float(mm)/INCH_TO_MM);return{"ok":True,"value_in":round(inches,1),"info":info,"hours":int(hours),"date_yyyymmdd":date_yyyymmdd} @app.on_event("startup") def _startup()->None: _transparent_png_256();_zero_raw_npz_256() if _acquire_singleton_lock("hf_pull"): def _background_hf_pull_and_cn(): try:_log("INFO","[startup] background hf_pull_cache starting");_hf_pull_cache();_log("INFO","[startup] background hf_pull_cache finished") except Exception as e:_log("WARN",f"[startup] background hf_pull_cache failed: {e!r}") try:_log("INFO","[startup] background ensure_cn_mbtiles starting");_ensure_cn_mbtiles();_log("INFO","[startup] background ensure_cn_mbtiles finished") except Exception as e:_log("WARN",f"[tracks] failed to ensure cn mbtiles (background): {e!r}") t_hf=threading.Thread(target=_background_hf_pull_and_cn,daemon=True,name="hf_pull_bg");t_hf.start();_log("INFO","[startup] hf pull scheduled in background (singleton)") else:_log("INFO","[startup] hf pull skipped (another worker holds lock)") if _acquire_singleton_lock("daily_scheduler"):t=threading.Thread(target=_daily_scheduler_loop,daemon=True);t.start();_log("INFO","[startup] daily scheduler enabled (singleton)") else:_log("INFO","[startup] daily scheduler disabled (another worker holds lock)") if(str(BOOT_ROLL_DEFAULT).strip()!="0")and _acquire_singleton_lock("boot_roll"):t2=threading.Thread(target=_run_daily_roll_job_impl,daemon=True);t2.start();_log("INFO","[startup] boot roll job kicked off") else:_log("INFO","[startup] boot roll job skipped") if _acquire_singleton_lock("hf_uploader"):t3=threading.Thread(target=_hf_tiles_uploader_loop,daemon=True);t3.start();_log("INFO","[startup] hf uploader enabled (singleton)") else:_log("INFO","[startup] hf uploader disabled (another worker holds lock)") if _acquire_singleton_lock("staleness_monitor"): t_stale = threading.Thread(target=_staleness_monitor_loop, daemon=True, name="staleness_monitor") t_stale.start() _log("INFO", "[startup] staleness monitor enabled (singleton)") else: _log("INFO", "[startup] staleness monitor disabled (another worker holds lock)") @app.get("/",response_class=PlainTextResponse) def root():return"SNODAS Snowmelt Tiles. See /web for demo and /docs for API." @app.get("/healthz",response_class=PlainTextResponse) def health():return"ok" @app.get("/config.js",response_class=PlainTextResponse) def config_js():return f"window.__SNODAS_CONFIG__ = {{ MAPBOX_TOKEN: '{os.environ.get('MAPBOX_TOKEN','')}' }};" @app.get("/tiles/24h/latest/{z}/{x}/{y}.png") def tiles_24h_latest(z:int,x:int,y:int,max:float|None=Query(None)): d=_resolve_latest_nsidc_day() if d is None:raise HTTPException(status_code=503,detail="No recent NSIDC day available") if max is None: try: hit=_tile_cache_get("nsidc",24,d,z,x,y) if hit is not None:return _resp_png(hit,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"24h-latest","X-Cache":"tilecache-hit","X-Latest-Day":d}) except:pass cog=build_24h_cog(datetime.strptime(d,"%Y%m%d").date());png=_tile_png_from_cog(cog,z,x,y,max) if max is None: try:_tile_cache_put("nsidc",24,d,z,x,y,png) except:pass return _resp_png(png,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"24h-latest","X-Cache":("tilecache-store"if max is None else"bypass-max"),"X-Latest-Day":d}) @app.get("/tiles/24h/{date_yyyymmdd:regex(\\d{8})}/{z}/{x}/{y}.png") def tiles_24h(date_yyyymmdd:str,z:int,x:int,y:int,max:float|None=Query(None)): try:d=datetime.strptime(date_yyyymmdd,"%Y%m%d").date() except:raise HTTPException(status_code=400,detail="Date must be YYYYMMDD") return _resp_png(_tile_png_from_cog(build_24h_cog(d),z,x,y,max),cache_control=CACHE_CONTROL_BY_DATE,**{"X-Route":"24h-date"}) @app.get("/tiles/72h/{end_yyyymmdd:regex(\\d{8})}/{z}/{x}/{y}.png") def tiles_72h(end_yyyymmdd:str,z:int,x:int,y:int,max:float|None=Query(None)): try:d=datetime.strptime(end_yyyymmdd,"%Y%m%d").date() except:raise HTTPException(status_code=400,detail="Date must be YYYYMMDD") return _resp_png(_tile_png_from_cog(build_72h_cog(d),z,x,y,max),cache_control=CACHE_CONTROL_BY_DATE,**{"X-Route":"72h"}) @app.get("/forecast/ending") def forecast_ending(days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72)): now=datetime.utcnow().replace(tzinfo=timezone.utc);end_dt=_ts_to_dt(_valid_ts_for_days(days,now_utc=now)) return{"days":int(days),"hours":int(hours),"valid":end_dt.strftime("%Y%m%d%H"),"end_iso":end_dt.isoformat(),"label":f"{int(hours)} Hour snowmelt ending on {end_dt.strftime('%Y-%m-%d %HZ')}","now_utc":now.isoformat(),"anchor_date_05z":now.date().isoformat()} @app.get("/tiles/forecast/latest/{z}/{x}/{y}.png") def tiles_forecast_latest(z:int,x:int,y:int,max:float|None=Query(None),days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72)): z_eff,x_eff,y_eff,tier=_effective_request_tile(z,x,y) if tier=="none":return _resp_png(_transparent_png_256(),cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"forecast-latest","X-Allowed":"0"}) sel=_pick_forecast_for_days(days,dom_prefer="zz") if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);detail["suggestion"]="latest forecast may not be available yet";raise HTTPException(status_code=503,detail=detail) valid_ymd=(sel["valid"]or"")[:8] if max is None and valid_ymd: try: hit=_tile_cache_get("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff) if hit is not None:return _resp_png(hit,cache_control=CACHE_CONTROL_LATEST,**{"X-Route":"forecast-latest","X-Allowed":"1","X-Cache":"hit","X-Forecast-Valid":sel["valid"],"X-Forecast-Hours":str(int(hours)),"X-Forecast-RunInit-TT":sel["run_init"],"X-Tier":tier}) except:pass try: png,headers=_generate_forecast_by_date_png(z=z_eff,x=x_eff,y=y_eff,date_yyyymmdd=valid_ymd,hours=hours,dom="zz",max_in=max) cache_status="generated-no-cache" if max is None and valid_ymd: try:_tile_cache_put("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff,png);_hf_enqueue_files([_tile_cache_path("zz",int(hours),valid_ymd,z_eff,x_eff,y_eff)]);cache_status="generated-and-cached" except:cache_status="generated-only" headers["X-Cache"]=cache_status;headers["X-Tier"]=tier;headers["X-Route"]="forecast-latest" return _resp_png(png,cache_control=CACHE_CONTROL_LATEST,**headers) except HTTPException as e:_log("WARN",f"[tiles] forecast/latest failed: {e.detail}");raise def _is_gzip(data:bytes)->bool:return len(data)>=2 and data[0]==0x1F and data[1]==0x8B @app.get("/cn_tiles/{z}/{x}/{y}.pbf") def cn_tiles(z:int,x:int,y:int): conn=_cn_mbtiles_conn();zmax=_cn_mbtiles_maxzoom();z2,x2,y2=_mbtiles_overzoom(z,x,y,zmax) try:data=_mbtiles_get_tile(conn,z2,x2,y2) except Exception as e:_log("WARN",f"[cn_tiles] read fail z={z} x={x} y={y} -> z2={z2} x2={x2} y2={y2} err={e!r}");data=None if not data:return Response(status_code=404) headers={"Content-Type":"application/x-protobuf","Cache-Control":"public, max-age=86400"} if _is_gzip(data):headers["Content-Encoding"]="gzip" return FastResponse(content=data,headers=headers) @app.get("/value/forecast/by_date") def value_forecast_by_date(date_yyyymmdd:str=Query(...,regex=r"^\d{8}$"),hours:int=Query(24,ge=24,le=72),lon:float=Query(...),lat:float=Query(...),dom:str=Query("zz")): with _Mode("value"): dom=(dom or"zz").lower();valid=_norm_ts(f"{date_yyyymmdd}05") if int(hours)==24: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);raise HTTPException(status_code=503,detail=detail) melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]);mm,info=_union_point_mm(melt_cogs,float(lon),float(lat)) inches=None if mm is None else(float(mm)/INCH_TO_MM);return{"ok":True,"hours":24,"valid":sel["valid"],"run_init":sel["run_init"],"mm":mm,"inches":inches,"info":info} run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom) if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom}) melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom);mm,info=_point_mm_from_cog(melt72_cog,float(lon),float(lat)) inches=None if mm is None else(float(mm)/INCH_TO_MM);inches_1dp=None if inches is None else round(inches,1) return{"ok":True,"hours":72,"valid":valid,"run_init":run_init,"mm":None if mm is None else float(mm),"inches":inches,"inches_1dp":inches_1dp,"info":info} @app.post("/value/forecast/viewport_grid") def value_forecast_viewport_grid(payload:dict): with _Mode("value"): try: date_yyyymmdd=payload.get("date_yyyymmdd");hours=int(payload.get("hours",24));dom=(payload.get("dom")or"zz").lower();bounds=payload.get("bounds");resolution=int(payload.get("resolution",20)) if not date_yyyymmdd or not bounds or len(bounds)!=4:raise HTTPException(status_code=400,detail="Missing date_yyyymmdd or bounds") west,south,east,north=[float(b)for b in bounds];resolution=max(5,min(30,resolution));south=max(-85.0,min(85.0,south));north=max(-85.0,min(85.0,north)) if east Response: _pause_push_for_user() dom = (dom or "zz").lower() hours_i = int(hours) if not _tile_allowed(z, x, y): return _resp_png( _transparent_png_256(), cache_control="public, max-age=31536000, immutable", **{"X-Route": "forecast-by-date", "X-Allowed": "0"}, ) if client_id and gen is not None and not _client_is_active_gen(client_id, gen): return _resp_png( _transparent_png_256(), cache_control="no-store", **{"X-Route": "forecast-by-date", "X-Stale": "gen", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)}, ) if client_id and not _client_is_active_date(client_id, date_yyyymmdd): return _resp_png( _transparent_png_256(), cache_control="no-store", **{"X-Route": "forecast-by-date", "X-Stale": "1", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)}, ) cancel_check = _make_cancel_check(client_id, gen) try: cached = _tile_cache_get(dom, hours_i, date_yyyymmdd, z, x, y, cancel_check=cancel_check) if cached is not None: return _resp_png( cached, cache_control="public, max-age=31536000, immutable", **{"X-Route": "forecast-by-date", "X-Cache": "hit"}, ) except: pass if cancel_check and cancel_check(): return _resp_png( _transparent_png_256(), cache_control="no-store", **{"X-Route": "forecast-by-date", "X-Stale": "gen-pre-gen", "X-Date": date_yyyymmdd}, ) #_log("INFO", f"[tile] generating on-demand: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}") png, headers = _generate_forecast_by_date_png( z=z, x=x, y=y, date_yyyymmdd=date_yyyymmdd, hours=hours_i, dom=dom, max_in=max_in, ) if not _is_valid_png_bytes(png): _log("WARN", f"[tile] generated invalid PNG: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}") return _resp_png( _transparent_png_256(), cache_control="no-store", **{"X-Route": "forecast-by-date", "X-Bad-PNG": "1", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i)}, ) if cancel_check and cancel_check(): return _resp_png( _transparent_png_256(), cache_control="no-store", **{"X-Route": "forecast-by-date", "X-Stale": "gen-post-gen", "X-Date": date_yyyymmdd}, ) try: _tile_cache_put(dom, hours_i, date_yyyymmdd, z, x, y, png) except: pass extra = dict(headers) if isinstance(headers, dict) else {} extra.setdefault("X-Route", "forecast-by-date") extra.setdefault("X-Cache", "miss") return _resp_png(png, cache_control="public, max-age=31536000, immutable", **extra) @app.get("/tiles/forecast_raw/by_date/{z}/{x}/{y}.npz") def tiles_forecast_by_date_raw( z: int, x: int, y: int, date_yyyymmdd: str = Query(..., regex=r"^\d{8}$"), hours: int = Query(24), dom: str = Query("zz"), client_id: str | None = Query(None), gen: int | None = Query(None), ) -> Response: _pause_push_for_user() if client_id and not _check_rate_limit(client_id): return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Rate-Limited": "1", "Retry-After": "5", }, ) dom = (dom or "zz").lower() hours_i = int(hours) if client_id and gen is not None and not _client_is_active_gen(client_id, gen): return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Stale": "gen", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i), }, ) if client_id and not _client_is_active_date(client_id, date_yyyymmdd): return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Stale": "1", "X-Date": date_yyyymmdd, "X-Hours": str(hours_i), }, ) cancel_check = _make_cancel_check(client_id, gen) try: cached = _raw_tile_cache_get(dom, hours_i, date_yyyymmdd, z, x, y, cancel_check=cancel_check) if cached is not None: if not _validate_npz_bytes(cached): _log("WARN", f"[tile_raw] Cached NPZ failed header check: {date_yyyymmdd} z{z} x{x} y{y}") try: _safe_unlink(_raw_tile_cache_path(dom, hours_i, date_yyyymmdd, z, x, y)) except: pass else: return Response( content=cached, media_type="application/octet-stream", headers={ "Cache-Control": "public, max-age=31536000, immutable", "X-Route": "forecast-raw-by-date", "X-Cache": "hit", }, ) except Exception as e: _log("WARN", f"[tile_raw] cache-read error: {e!r}") if cancel_check and cancel_check(): return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Stale": "gen-pre-gen", "X-Date": date_yyyymmdd, }, ) #_log("INFO", f"[tile_raw] generating on-demand: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}") raw, headers = _generate_forecast_by_date_raw( z=z, x=x, y=y, date_yyyymmdd=date_yyyymmdd, hours=hours_i, dom=dom, ) if not _validate_npz_bytes(raw): _log("WARN", f"[tile_raw] generated invalid NPZ header: {date_yyyymmdd} h{hours_i} z{z} x{x} y{y}") return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Bad-NPZ": "1", }, ) if cancel_check and cancel_check(): return Response( content=_zero_raw_npz_256(), media_type="application/octet-stream", headers={ "Cache-Control": "no-store", "X-Route": "forecast-raw-by-date", "X-Stale": "gen-post-gen", "X-Date": date_yyyymmdd, }, ) try: _raw_tile_cache_put(dom, hours_i, date_yyyymmdd, z, x, y, raw) except Exception as e: _log("WARN", f"[tile_raw] failed to write raw cache for {date_yyyymmdd} z{z} x{x} y{y}: {e!r}") extra = dict(headers) if isinstance(headers, dict) else {} h = { "Cache-Control": "public, max-age=31536000, immutable", "X-Route": "forecast-raw-by-date", "X-Cache": "miss", } for k, v in extra.items(): try: h[str(k)] = str(v) except: pass return Response(content=raw, media_type="application/octet-stream", headers=h) @app.get("/__debug/record_all_sigs") def __debug_record_all_sigs(): dom=(WARM_DOM or"zz").lower() window=_rolling_dates_local_window(lookback=2,lookahead=3) recorded=[] for ymd in window: best=_find_best_melt_for_date(ymd,dom_prefer=dom) if best.get("ok")and best.get("melt_urls"): _record_baked_sig(ymd,dom,best["melt_urls"][0],source_type=best.get("source_type")) recorded.append({"date":ymd,"source_type":best.get("source_type"),"url":best["melt_urls"][0]}) else: recorded.append({"date":ymd,"error":best.get("error","no melt found")}) return{"recorded":recorded} @app.get("/tiles/forecast/{ts_key:regex(\\d{8}(?:[_-]?\\d{2}))}/{z}/{x}/{y}.png") def tiles_forecast(ts_key:str,z:int,x:int,y:int,max:float|None=Query(None),lead:str=Query("t0024"),valid:str|None=Query(None),dom:str=Query("zz")): if not _tile_allowed(z,x,y):return _resp_png(_transparent_png_256(),**{"X-Route":"forecast-direct","X-Allowed":"0"}) ts_key=_norm_ts(ts_key);lead=(lead or"").lower() if lead!="t0024":raise HTTPException(status_code=400,detail="This service now serves direct 24h melt files (t0024). Use lead=t0024.") if not valid:raise HTTPException(status_code=400,detail="valid is required (TSYYYYMMDDHH) for direct 24h melt") valid=_norm_ts(valid);melt_urls=_find_melt_11044_t0024(ts_key,valid,dom_prefer=dom) if not melt_urls:raise HTTPException(status_code=404,detail=f"No melt file for TT={ts_key} TS={valid} lead=t0024") sd=_best_state_nc_near_valid("ssmv11036",valid,dom_prefer=dom) if not sd.get("url"):raise HTTPException(status_code=404,detail="No snowdepth 11036 state .nc found") resp=_serve_forecast_tile(ts_key,valid,"t0024",melt_urls,sd.get("picked_ts")or"",[sd["url"]],z,x,y,max);resp.headers["X-Allowed"]="1";return resp @app.get("/forecast/latest_ts") def forecast_latest_ts(dom:str="zz"): now=datetime.utcnow().replace(tzinfo=timezone.utc);sel2=_pick_forecast_for_days(2,dom_prefer=dom);sel3=_pick_forecast_for_days(3,dom_prefer=dom) return{"now_utc":now.isoformat(),"d2":sel2,"d3":sel3,"collab_last":dict(_COLLAB_LAST)} @app.get("/forecast/selection") def forecast_selection(dom:str="zz"):return forecast_latest_ts(dom=dom) @app.post("/value/forecast/by_date_batch") def value_forecast_by_date_batch(payload:dict): with _Mode("value"): try: date_yyyymmdd=payload.get("date_yyyymmdd");hours=int(payload.get("hours",24));dom=(payload.get("dom")or"zz").lower();pts=payload.get("pts",[]) if not date_yyyymmdd or not pts:raise HTTPException(status_code=400,detail="Missing date_yyyymmdd or pts") valid=_norm_ts(f"{date_yyyymmdd}05") if int(hours)==24: sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):detail=dict(sel);detail["collab_last"]=dict(_COLLAB_LAST);raise HTTPException(status_code=503,detail=detail) melt_cogs=_build_forecast_melt_cogs(sel["run_init"],sel["valid"],"t0024",sel["melt_urls"]);values=[] for pt in pts: if not isinstance(pt,(list,tuple))or len(pt)<2:values.append(None);continue lon,lat=float(pt[0]),float(pt[1]);mm,info=_union_point_mm(melt_cogs,lon,lat) if mm is None:values.append(None) else:inches=float(mm)/INCH_TO_MM;values.append({"mm":float(mm),"inches":inches,"inches_1dp":round(inches,1)}) return{"ok":True,"hours":24,"valid":sel["valid"],"run_init":sel["run_init"],"count":len(pts),"values":values} run_init=_pick_best_run_init_for_valid_t0024(valid,dom_prefer=dom) if not run_init:raise HTTPException(status_code=503,detail={"error":"no_run_init_for_valid","valid":valid,"dom":dom}) melt72_cog=_forecast_melt72h_end_cog(valid,dom_prefer=dom);values=[] for pt in pts: if not isinstance(pt,(list,tuple))or len(pt)<2:values.append(None);continue lon,lat=float(pt[0]),float(pt[1]);mm,info=_point_mm_from_cog(melt72_cog,lon,lat) if mm is None:values.append(None) else:inches=float(mm)/INCH_TO_MM;values.append({"mm":float(mm),"inches":inches,"inches_1dp":round(inches,1)}) return{"ok":True,"hours":72,"valid":valid,"run_init":run_init,"count":len(pts),"values":values} except HTTPException:raise except Exception as e:_log("ERROR",f"[batch] unexpected error: {e!r}");raise HTTPException(status_code=500,detail=f"batch_error: {e!r}") @app.get("/forecast/avail") def forecast_avail(lead:str="t0024"): html=_collab_dirlist();names=_hrefs(html);lead=(lead or"").lower();tts:list[str]=[] for n in names: base=n.split("?",1)[0].strip();low=base.lower() if"ssmv11044"not in low:continue if lead and lead not in low:continue if not _is_data_name(low):continue tt=_tt_from_name(base) if tt:tts.append(tt) tts=sorted(set(tts));return{"lead":lead,"count":len(tts),"min_tt":tts[0]if tts else None,"max_tt":tts[-1]if tts else None,"sample_tail":tts[-10:]if len(tts)>10 else tts} _AVAIL_CACHE={"ts":0.0,"key":None,"resp":None} def _available_dates_from_hf_repo(dom:str,hours:int,lookback_days:int)->list[str]: tok,repo=_hf_cfg();api=_hf_api() if not tok or not repo or api is None:return[] try:files=api.list_repo_files(repo_id=repo,repo_type="dataset",token=tok) except Exception as e:_log("WARN",f"[available_days] hf list_repo_files failed: {e!r}");return[] now=datetime.utcnow().replace(tzinfo=timezone.utc);min_ymd=(now.date()-timedelta(days=int(lookback_days))).strftime("%Y%m%d") prefix=f"tilecache/{dom}/{int(hours)}/";out=set() for f in files: if not f.startswith(prefix):continue parts=f.split("/") if len(parts)<4:continue ymd=parts[3] if len(ymd)==8 and ymd.isdigit()and ymd>=min_ymd:out.add(ymd) return sorted(out) _HF_FILESET_CACHE={"ts":0.0,"files":None} def _hf_repo_files_cached(ttl_s:float=60.0)->list[str]: now=time.time() if _HF_FILESET_CACHE["files"]is not None and(now-_HF_FILESET_CACHE["ts"])bool: dom=(dom or"zz").lower();hours=int(hours);pref_png=f"tilecache/{dom}/{hours}/{ymd}/";pref_raw=f"rawtilecache/{dom}/{hours}/{ymd}/" files=_hf_repo_files_cached(ttl_s=120.0)or[] for f in files: if not f:continue f=f.strip() if f.startswith(pref_png)or f.startswith(pref_raw):return True parts=f.split("/",1) if len(parts)==2: tail=parts[1] if tail.startswith(pref_png)or tail.startswith(pref_raw):return True try: tok,repo=_hf_cfg() if not repo:return False api=HfApi() for c in[f"tilecache/{dom}/{hours}/{ymd}/0/0/0.png",f"tilecache/{dom}/{hours}/{ymd}/1/0/0.png",f"tilecache/{dom}/{hours}/{ymd}/2/0/0.png"]: try:hf_hub_download(repo_id=repo,filename=c,repo_type="dataset",token=tok,repo_type_kwargs=None);return True except:continue except Exception as e:_log("WARN",f"[hf_probe] error probing HF for day {ymd}: {e!r}") return False @app.get("/forecast/available_days") def forecast_available_days(dom: str = Query("zz"), lookback_days: int = Query(1, ge=0, le=7), lookahead_days: int = Query(3, ge=1, le=5)): cache_key = (dom, int(lookback_days), int(lookahead_days)) if _AVAILABLE_DATES_CACHE.get("data") is not None and _AVAILABLE_DATES_CACHE.get("key") == cache_key and (time.time() - float(_AVAILABLE_DATES_CACHE.get("ts", 0.0))) < _AVAILABLE_DATES_CACHE_TTL: return _AVAILABLE_DATES_CACHE["data"] window_dates = _rolling_window_dates(lookback_days=int(lookback_days), lookahead_days=int(lookahead_days)) local_status = {} for ymd in window_dates: has_24 = _check_date_has_local_tiles(dom, 24, ymd) has_72 = _check_date_has_local_tiles(dom, 72, ymd) local_status[ymd] = {"has_24h": has_24, "has_72h": has_72, "has_any": (has_24 or has_72)} now_local = datetime.now(LOCAL_TZ) resp = {"ok": True, "now_local": now_local.isoformat(), "today": now_local.strftime("%Y%m%d"), "lookback_days": int(lookback_days), "lookahead_days": int(lookahead_days), "available_dates": window_dates, "local_tile_status": local_status, "source": "rolling_window", "dom": dom} _AVAILABLE_DATES_CACHE["ts"] = time.time(); _AVAILABLE_DATES_CACHE["key"] = cache_key; _AVAILABLE_DATES_CACHE["data"] = resp return resp @app.get("/__debug/inspect_snow") def __debug_inspect_snow(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")): def _safe(v): try: f=float(v) if not math.isfinite(f):return None return f except:return None dom=(dom or"zz").lower() sel=_pick_forecast_for_valid(f"{date}05",dom_prefer=dom) if not sel.get("ok"):return{"ok":False,"error":"pick_failed","detail":str(sel)} snow_info={"snowpack_urls":sel.get("snowpack_urls",[]),"snowpack_ts":sel.get("snowpack_ts",""),"run_init":sel.get("run_init",""),"source_type":sel.get("source_type","")} # Check snowpack COG if it exists valid=sel.get("valid",f"{date}05") snow_cog_name=f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif" snow_cog=_forecast_dir()/snow_cog_name cog_info={"exists":snow_cog.exists(),"name":snow_cog_name} if snow_cog.exists(): try: with rasterio.open(snow_cog.as_posix())as src: cog_info["crs"]=str(src.crs) cog_info["transform"]=str(src.transform) cog_info["width"]=src.width cog_info["height"]=src.height cog_info["bounds"]=list(src.bounds) except Exception as e:cog_info["error"]=repr(e) # Also check the melt COG melt_cog_name=f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif" melt_cog=_forecast_dir()/melt_cog_name melt_info={"exists":melt_cog.exists(),"name":melt_cog_name} if melt_cog.exists(): try: with rasterio.open(melt_cog.as_posix())as src: melt_info["crs"]=str(src.crs) melt_info["transform"]=str(src.transform) melt_info["width"]=src.width melt_info["height"]=src.height melt_info["bounds"]=list(src.bounds) except Exception as e:melt_info["error"]=repr(e) return{"ok":True,"date":date,"selection":snow_info,"snow_cog":cog_info,"melt_cog":melt_info} @app.get("/__debug/prewarm_cn") def __debug_prewarm_cn(days:int=Query(2,ge=2,le=3),hours:int=Query(24,ge=24,le=72),dom:str=Query("zz"),zmin:int=Query(6,ge=0,le=14),zmax:int=Query(12,ge=0,le=14),miles:float=Query(8.0,ge=0.0,le=50.0),cap_per_zoom:int=Query(6000,ge=100,le=20000),max_tiles_total:int=Query(20000,ge=100,le=100000)): dom=(dom or"zz").lower();now=datetime.utcnow().replace(tzinfo=timezone.utc);valid=_valid_ts_for_days(int(days),now_utc=now) sel=_pick_forecast_for_valid(valid,dom_prefer=dom) if not sel.get("ok"):return{"ok":False,"error":"forecast_selection_failed","detail":sel} try:return _bake_corridor_tiles_for_date_optimized(date_yyyymmdd=sel["valid"][:8],dom=dom,hours_list=[int(hours)],zmin=int(zmin),zmax=int(zmax),miles=float(miles),cap_per_zoom=int(cap_per_zoom),max_tiles_total=int(max_tiles_total)) except Exception as e:return{"ok":False,"error":f"{e!r}"} @app.get("/__debug/roll_now") def __debug_roll_now():return _run_daily_roll_job_impl() @app.get("/__logs") def __logs(since_ms:int|None=None,limit:int=200): try:limit=max(1,min(int(limit),1000)) except:limit=200 rows=list(_LOG) if since_ms is not None: try:rows=[r for r in rows if r["ts"]>int(since_ms)] except:pass return{"count":len(rows[-limit:]),"items":rows[-limit:]} @lru_cache(maxsize=8000) def _tile_png_cached(sig: str, z: int, x: int, y: int, max_in: float | None) -> bytes: name = sig.split(":", 1)[0] p = _forecast_dir().parent / "forecast" / name if name.endswith(".tif") else None if p is None or not p.exists(): raise HTTPException(status_code=500, detail="cached tile missing COG") return _tile_png_from_cog(p, z, x, y, max_in) class ActiveDatePayload(BaseModel): client_id: str | None = None date_yyyymmdd: str | None = None gen: int | None = None @app.post("/ui/active_date") def ui_active_date(payload: ActiveDatePayload = Body(...)): client_id = (payload.client_id or "").strip() date_yyyymmdd = (payload.date_yyyymmdd or "").strip() gen = payload.gen if not client_id: raise HTTPException(status_code=422, detail="client_id required") if not re.match(r"^\d{8}$", date_yyyymmdd): raise HTTPException(status_code=422, detail="date_yyyymmdd must be YYYYMMDD") _pause_push_for_user() _client_mark_active(client_id, date_yyyymmdd, gen=gen) return {"ok": True, "client_id": client_id, "active_date": date_yyyymmdd, "gen": gen} @app.get("/__debug/purge_bad_npz") def __debug_purge_bad_npz(dry_run:bool=Query(True),check_hf_repo:bool=Query(False),min_valid_size:int=Query(100)): results={"ok":True,"dry_run":dry_run,"local_cache":{"path":None,"exists":False,"found":0,"files":[]},"hf_repo":{"enabled":check_hf_repo,"found":0,"files":[]}} raw_cache=CACHE/"rawtilecache";results["local_cache"]["path"]=str(raw_cache) if raw_cache.exists(): results["local_cache"]["exists"]=True;bad_files=[] for p in raw_cache.rglob("*.npz"): try: size=p.stat().st_size if size-9990.0)&(vals<100000.0) valid_vals=vals[valid_mask] raw_info={"type":desc.get("type"),"note":"not netcdf"} if desc.get("type")=="netcdf": try: ds=xr.open_dataset(desc["path"].as_posix(),engine="netcdf4") raw_var=_pick_netcdf_raster_var(ds) raw_attrs=dict(raw_var.attrs or{}) raw_vals=raw_var.values.astype("float32").ravel() raw_valid=raw_vals[np.isfinite(raw_vals)&(raw_vals>-9990.0)&(raw_vals<1e20)] raw_info={"var_name":raw_var.name,"units":str(raw_attrs.get("units","")),"raw_mean":_safe(np.mean(raw_valid)),"raw_max":_safe(np.max(raw_valid)),"raw_min":_safe(np.min(raw_valid)),"raw_median":_safe(np.median(raw_valid)),"shape":list(raw_var.shape),"all_attrs":{k:str(v)for k,v in raw_attrs.items()}} ds.close() except Exception as e: raw_info={"error":repr(e)} samples={} for label,lon,lat in[("Rouses_Point",-73.365,44.991),("Ogdensburg",-75.486,44.694),("Belleville_ON",-77.385,44.163),("Montreal",-73.567,45.502),("Ottawa",-75.698,45.421)]: try: px=da.sel(x=lon,y=lat,method="nearest").values v=float(px) samples[label]={"mm":_safe(v),"inches":_safe(v/25.4)if np.isfinite(v)else None} except Exception as e: samples[label]={"error":str(e)} return{ "ok":True,"url":url,"filename":name,"valid":valid,"run_init":sel.get("run_init"),"source_type":"ttna"if"ttnats"in url.lower()else"forecast", "after_open_grid":{"units":units,"mean_mm":_safe(np.mean(valid_vals)),"max_mm":_safe(np.max(valid_vals)),"min_mm":_safe(np.min(valid_vals)),"median_mm":_safe(np.median(valid_vals)),"valid_pixels":int(valid_mask.sum()),"total_pixels":int(vals.size),"attrs":{k:str(v)for k,v in attrs.items()}}, "raw_file":raw_info, "point_samples_after_processing":samples, } # def _hot_swap_rebuild_date(date:str,dom:str)->None: # """Rebuild tiles from latest data, then swap out old tiles atomically. # Users always see baked tiles during the transition.""" # dom=(dom or"zz").lower() # valid_05=f"{date}05";valid_11=f"{date}11" # # 1. Build new COGs from latest NOAA data (old tiles still being served) # _log("INFO",f"[hot_swap] {date}: building fresh COGs") # # Delete only COGs, not tiles # for valid in[valid_05,valid_11]: # for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]: # for p in _forecast_dir().glob(pattern): # try:COG_CACHE.evict(p) # except:pass # try:p.unlink(missing_ok=True) # except:pass # # Also delete 72h COGs for next 2 dates # for offset in[1,2]: # try: # fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") # for fv in[f"{fd}05",f"{fd}11"]: # for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"): # try:COG_CACHE.evict(p) # except:pass # try:p.unlink(missing_ok=True) # except:pass # except:pass # # Clear negative cache # with _HF_NEGATIVE_CACHE_LK: # to_del=[k for k in _HF_NEGATIVE_CACHE if date in k] # for k in to_del:del _HF_NEGATIVE_CACHE[k] # best=_find_best_melt_for_date(date,dom_prefer=dom) # _log("INFO",f"[hot_swap] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}") # actual_valid=best.get("valid",f"{date}05") # r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom) # if not r.get("ok"):_log("WARN",f"[hot_swap] {date}: COG build failed: {r}");return # # 2. Delete old tiles so bake doesn't skip them # _log("INFO",f"[hot_swap] {date}: deleting old tiles") # for kind in["tilecache","rawtilecache"]: # for h in WARM_HOURS_LIST: # d=CACHE/kind/dom/str(h)/date # if d.exists():shutil.rmtree(d,ignore_errors=True) # # 3. Bake NEW tiles # _log("INFO",f"[hot_swap] {date}: baking new tiles") # with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool: # futs={} # for label,fn,extra in[("corridor_png",_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),("corridor_raw",_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: # futs[label]=pool.submit(fn,date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) # for label,fut in futs.items(): # try: # r=fut.result();_log("INFO",f"[hot_swap] {date}: {label} ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") # except Exception as e:_log("WARN",f"[hot_swap] {date}: {label} failed: {e!r}") # for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: # try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) # except:pass # for offset in[1,2]: # try: # fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") # fb=_find_best_melt_for_date(fd,dom_prefer=dom) # fv=fb.get("valid",f"{fd}05")if fb.get("ok")else f"{fd}05" # try:_forecast_melt72h_end_cog(fv,dom_prefer=dom) # except:pass # _log("INFO",f"[hot_swap] rebaking 72h for {fd}") # with concurrent.futures.ThreadPoolExecutor(max_workers=2)as pool: # for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: # pool.submit(fn,date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) # for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: # try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) # except:pass # except Exception as e:_log("WARN",f"[hot_swap] 72h rebake +{offset} failed: {e!r}") # # 4. NOW delete old HF archives and push new ones # _log("INFO",f"[hot_swap] {date}: deleting old HF archives") # tok,repo=_hf_cfg();api=_hf_api() # if tok and repo and api is not None: # # Delete old archives # for kind in["tilecache","rawtilecache"]: # for h in WARM_HOURS_LIST: # ap=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz" # try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old {ap}") # except:pass # # Delete 72h archives for next 2 dates # for offset in[1,2]: # try: # fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") # for kind in["tilecache","rawtilecache"]: # ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz" # try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old 72h {ap}") # except:pass # except:pass # # Delete tar markers # marker_dir=CACHE/"_tar_markers" # if marker_dir.exists(): # for mk in marker_dir.glob(f"*_{date}.tar.gz.done"): # try:mk.unlink(missing_ok=True) # except:pass # # 5. Push new tiles # push_dates=[date] # for offset in[1,2]: # try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")) # except:pass # _log("INFO",f"[hot_swap] pushing new tiles for {push_dates}") # _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST) # # 6. Record signature # if best.get("ok")and best.get("melt_urls"): # _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type")) # _log("INFO",f"[hot_swap] {date}: complete (source={best.get('source_type','?')})") def _hot_swap_rebuild_date(date:str,dom:str)->None: dom=(dom or"zz").lower() valid_05=f"{date}05";valid_11=f"{date}11" # 1. Build new COGs from latest NOAA data # Old COGs are deleted so they get rebuilt from fresh NOAA data. # Old TILES are NOT deleted yet — they keep serving users. _log("INFO",f"[hot_swap] {date}: building fresh COGs") for valid in[valid_05,valid_11]: for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]: for p in _forecast_dir().glob(pattern): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True) except:pass for offset in[1,2]: try: fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") for fv in[f"{fd}05",f"{fd}11"]: for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True) except:pass except:pass with _HF_NEGATIVE_CACHE_LK: to_del=[k for k in _HF_NEGATIVE_CACHE if date in k] for k in to_del:del _HF_NEGATIVE_CACHE[k] best=_find_best_melt_for_date(date,dom_prefer=dom) _log("INFO",f"[hot_swap] {date}: selected source_type={best.get('source_type','?')} valid={best.get('valid','?')}") actual_valid=best.get("valid",f"{date}05") r=_ensure_forecast_for_valid(actual_valid,dom_prefer=dom) if not r.get("ok"):_log("WARN",f"[hot_swap] {date}: COG build failed: {r}");return # 2. Bake NEW tiles into staging directories # Old tiles remain on disk at the real paths, still serving users. _log("INFO",f"[hot_swap] {date}: baking new tiles to staging (old tiles still serving)") staging_root=CACHE/"_staging_swap"/date try: if staging_root.exists():shutil.rmtree(staging_root,ignore_errors=True) except:pass old_dirs=[] for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: d=CACHE/kind/dom/str(h)/date if d.exists(): backup=CACHE/"_staging_swap"/f"{kind}_{dom}_{h}_{date}_old" backup.parent.mkdir(parents=True,exist_ok=True) try: # Rename is atomic on same filesystem — instant, no downtime d.rename(backup) old_dirs.append((d,backup)) _log("DEBUG",f"[hot_swap] moved {d} -> {backup}") except Exception as e: # If rename fails (cross-device?), fall back to just deleting _log("WARN",f"[hot_swap] rename failed, deleting old tiles: {e!r}") shutil.rmtree(d,ignore_errors=True) # 3. Bake tiles — they go to the real paths (which are now empty) # PNG first so users see visual tiles sooner, then raw for mouse readouts _log("INFO",f"[hot_swap] {date}: baking corridor PNG tiles") try: r=_bake_corridor_tiles_for_date_optimized(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL) _log("INFO",f"[hot_swap] {date}: corridor_png ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[hot_swap] {date}: corridor_png failed: {e!r}") _log("INFO",f"[hot_swap] {date}: baking corridor raw tiles") try: r=_bake_corridor_raw_tiles_for_date_optimized(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=WARM_ZMIN,zmax=WARM_ZMAX,miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL) _log("INFO",f"[hot_swap] {date}: corridor_raw ok={r.get('ok_count',0)} skip={r.get('skip_count',0)} fail={r.get('fail_count',0)}") except Exception as e:_log("WARN",f"[hot_swap] {date}: corridor_raw failed: {e!r}") for label,fn,extra in[("lowres_png",_bake_lowres_box_tiles_for_date,dict(conc=1)),("lowres_raw",_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try:fn(date_yyyymmdd=date,dom=dom,hours_list=WARM_HOURS_LIST,zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) except:pass # 4. Clean up old backup dirs (new tiles are now in place) for orig,backup in old_dirs: try:shutil.rmtree(backup,ignore_errors=True) except:pass try: if staging_root.exists():shutil.rmtree(staging_root,ignore_errors=True) except:pass # 5. Rebake 72h for future dates (+1, +2) for offset in[1,2]: try: fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") fb=_find_best_melt_for_date(fd,dom_prefer=dom) fv=fb.get("valid",f"{fd}05")if fb.get("ok")else f"{fd}05" try:_forecast_melt72h_end_cog(fv,dom_prefer=dom) except:pass _log("INFO",f"[hot_swap] rebaking 72h for {fd}") # Delete old 72h tiles so bake doesn't skip them for kind in["tilecache","rawtilecache"]: d=CACHE/kind/dom/"72"/fd if d.exists(): shutil.rmtree(d,ignore_errors=True) _log("DEBUG",f"[hot_swap] deleted stale 72h tiles: {d}") # Also delete old 72h lowres tiles for kind in["tilecache","rawtilecache"]: d=CACHE/kind/dom/"72"/fd # Lowres tiles are at the same path under different zoom levels, # so the rmtree above already handles them # Bake fresh 72h tiles sequentially for fn,extra in[(_bake_corridor_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL)),(_bake_corridor_raw_tiles_for_date_optimized,dict(miles=WARM_MILES,cap_per_zoom=WARM_CAP_PER_ZOOM,max_tiles_total=WARM_MAX_TILES_TOTAL))]: try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=WARM_ZMIN,zmax=WARM_ZMAX,**extra) except:pass for fn,extra in[(_bake_lowres_box_tiles_for_date,dict(conc=1)),(_bake_lowres_box_raw_tiles_for_date,dict(conc=1))]: try:fn(date_yyyymmdd=fd,dom=dom,hours_list=[72],zmin=LOWRES_BOX_ZMIN,zmax=LOWRES_BOX_ZMAX,**extra) except:pass except Exception as e:_log("WARN",f"[hot_swap] 72h rebake +{offset} failed: {e!r}") # 6. Delete old HF archives and push new ones _log("INFO",f"[hot_swap] {date}: deleting old HF archives") tok,repo=_hf_cfg();api=_hf_api() if tok and repo and api is not None: for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: ap=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz" try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old {ap}") except:pass for offset in[1,2]: try: fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") for kind in["tilecache","rawtilecache"]: ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz" try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"hot_swap: delete old 72h {ap}") except:pass except:pass marker_dir=CACHE/"_tar_markers" if marker_dir.exists(): for mk in marker_dir.glob(f"*_{date}.tar.gz.done"): try:mk.unlink(missing_ok=True) except:pass # 7. Push new tiles push_dates=[date] for offset in[1,2]: try:push_dates.append((datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d")) except:pass _log("INFO",f"[hot_swap] pushing new tiles for {push_dates}") _hf_push_baked_as_tar(push_dates,dom,WARM_HOURS_LIST) # 8. Record signature if best.get("ok")and best.get("melt_urls"): _record_baked_sig(date,dom,best["melt_urls"][0],source_type=best.get("source_type")) _log("INFO",f"[hot_swap] {date}: complete (source={best.get('source_type','?')})") @app.get("/__debug/nuke_date") def __debug_nuke_date(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")): dom=(dom or"zz").lower();deleted_cogs=[];deleted_archives=[] # 1. Delete local COGs (forces rebuild from NOAA) for valid in[f"{date}05",f"{date}11"]: for pattern in[f"fcst_melt24h_union_end_{valid}_{dom}_cog.tif",f"fcst_melt72h_end_{valid}_{dom}_cog.tif",f"fcst_snowpack_union_end_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_{dom}_cog.tif",f"fcst_melt_t0024_*_{valid}_*_cog.tif"]: for p in _forecast_dir().glob(pattern): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True);deleted_cogs.append(p.name) except:pass # 2. Do NOT delete tiles — they keep serving users until rebake replaces them tile_dirs_present=0 for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: d=CACHE/kind/dom/str(h)/date if d.exists():tile_dirs_present+=1 # 3. Delete local tar markers so they don't block re-pull marker_dir=CACHE/"_tar_markers" markers_cleared=0 if marker_dir.exists(): for mk in marker_dir.glob(f"*_{date}.tar.gz.done"): try:mk.unlink(missing_ok=True);markers_cleared+=1 except:pass # 4. Delete tar archives from HF repo tok,repo=_hf_cfg();api=_hf_api() if tok and repo and api is not None: for kind in["tilecache","rawtilecache"]: for h in WARM_HOURS_LIST: archive_path=f"archives/{kind}_{dom}_h{h}_{date}.tar.gz" try: api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=archive_path)],commit_message=f"nuke: delete {archive_path}") deleted_archives.append(archive_path);_log("INFO",f"[nuke] deleted HF archive: {archive_path}") except Exception as e: err=repr(e) if"404"in err or"EntryNotFound"in err or"does not exist"in err: deleted_archives.append(f"{archive_path} (already gone)") else: deleted_archives.append(f"{archive_path}: FAIL {err}") # Also handle 72h for future dates for offset in[1,2]: try: fd=(datetime.strptime(date,"%Y%m%d")+timedelta(days=offset)).strftime("%Y%m%d") for fv in[f"{fd}05",f"{fd}11"]: for p in _forecast_dir().glob(f"fcst_melt72h_end_{fv}_*_cog.tif"): try:COG_CACHE.evict(p) except:pass try:p.unlink(missing_ok=True);deleted_cogs.append(p.name) except:pass if tok and repo and api is not None: for kind in["tilecache","rawtilecache"]: ap=f"archives/{kind}_{dom}_h72_{fd}.tar.gz" try:api.create_commit(repo_id=repo,repo_type="dataset",operations=[CommitOperationDelete(path_in_repo=ap)],commit_message=f"nuke: delete stale 72h {ap}") except:pass except:pass # 5. Clear negative cache with _HF_NEGATIVE_CACHE_LK: to_del=[k for k in _HF_NEGATIVE_CACHE if date in k] for k in to_del:del _HF_NEGATIVE_CACHE[k] return{ "deleted_cogs":deleted_cogs, "tile_dirs_kept":tile_dirs_present, "tiles_note":"Old tiles kept for serving — rebake will replace them", "hf_archives_deleted":deleted_archives, "tar_markers_cleared":markers_cleared, "negative_cache_cleared":len(to_del), "date":date, "message":f"Nuked COGs for {date} (tiles kept). Hit /__debug/rebake_date?date={date} to rebuild with fresh data." } @app.get("/__debug/rebake_date") def __debug_rebake_date(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")): dom=(dom or"zz").lower() def _do_rebake(): try:_hot_swap_rebuild_date(date,dom) except Exception as e:_log("WARN",f"[rebake] {date}: fatal error: {e!r}") threading.Thread(target=_do_rebake,daemon=True,name=f"rebake_{date}").start() return{"ok":True,"date":date,"message":f"Hot-swap rebake started for {date}. Old tiles served until new ones are ready."} @app.get("/__debug/noaa_sigs") def __debug_noaa_sigs(): with _NOAA_FILE_SIGS_LK: return{"sigs":{k:v for k,v in _NOAA_FILE_SIGS.items()}} @app.get("/__debug/inspect_full_tile") def __debug_inspect_full_tile(date:str=Query(...,regex=r"^\d{8}$"),dom:str=Query("zz")): dom=(dom or"zz").lower() sel=_pick_forecast_for_valid(f"{date}05",dom_prefer=dom) best=_find_best_melt_for_date(date,dom_prefer=dom) melt_valid=best.get("valid","?") snow_urls=sel.get("snowpack_urls",[]) snow_ts=sel.get("snowpack_ts","?") # Check what snowpack COG exists snow_cogs=[] for v in[f"{date}11",f"{date}05"]: p=_forecast_dir()/f"fcst_snowpack_union_end_{v}_{dom}_cog.tif" if p.exists():snow_cogs.append({"name":p.name,"valid":v,"size":p.stat().st_size}) return{ "date":date, "melt_source":best.get("source_type","?"), "melt_valid":melt_valid, "melt_url":best.get("melt_urls",["?"])[0].split("/")[-1]if best.get("melt_urls")else"?", "snowpack_ts_from_sel":snow_ts, "snowpack_urls_count":len(snow_urls), "snowpack_urls":[u.split("/")[-1]for u in snow_urls[:3]], "snowpack_cogs_on_disk":snow_cogs, "run_init":best.get("run_init","?"), } if __name__=="__main__": import uvicorn host=os.environ.get("HOST","0.0.0.0");port=int(os.environ.get("PORT","8000"));workers=int(os.environ.get("UVICORN_WORKERS","3")or"3") module=(os.environ.get("APP_MODULE","")or"").strip() if workers>1 and not module:raise SystemExit("UVICORN_WORKERS>1 requires APP_MODULE like 'main:app'. Example: APP_MODULE=main:app UVICORN_WORKERS=3 python main.py") if module:uvicorn.run(module,host=host,port=port,workers=workers,log_level="info",access_log=False) else:uvicorn.run(app,host=host,port=port,workers=1,log_level="info",access_log=False)